make analyzer a tool not a service

1 files changed, 13 insertions, 74 deletions

diff --git a/worker.py b/worker.py
index 4520d6f..c5db566 100644
--- a/worker.py
+++ b/worker.py
@@ -13,23 +13,14 @@ from sqlalchemy import create_engine
 import tensorflow as tf
 import numpy as np
 
-import pika
 
-
-RABBITMQ_URI = os.environ.get("RABBITMQ_URI") or "amqp://localhost"
 DATABASE_URI = os.environ["DATABASE_URI"]
-BATCHSIZE = os.environ.get("BATCHSIZE") or 1000  # objects
-IDLE_TIMEOUT = os.environ.get("IDLE_TIMEOUT") or 1  # s
 MODEL_ROOT = os.path.join(os.getcwd(), os.path.dirname(__file__))\
         + "/models/"
 
 # ORM definitions
 Match = Roster = Participant = ParticipantStats = Player = None
-db = rabbit = channel = None
-
-# batch storage
-queue = []
-timer = None
+db = None
 
 # models
 mvpmodel = None
@@ -37,7 +28,7 @@ mvpmodel = None
 
 def connect():
     global Match, Roster, Participant, ParticipantStats, Player
-    global db, rabbit, channel
+    global db
 
     # generate schema from db
     Base = automap_base()
@@ -64,6 +55,9 @@ def connect():
     Participant.player = relationship(
         "player", foreign_keys="player.api_id",
         primaryjoin="and_(player.api_id == participant.player_api_id)")
+    Participant.hero = relationship(
+        "hero", foreign_keys="hero.id",
+        primaryjoin="and_(hero.id == participant.hero_id)")
     ParticipantStats = Base.classes.participant_stats
     Participant.participant_stats = relationship(
         "participant_stats", foreign_keys="participant_stats.participant_api_id",
@@ -72,18 +66,6 @@ def connect():
 
     db = Session(engine)
 
-    while True:
-        try:
-            rabbit = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URI))
-            break
-        except pika.exceptions.ConnectionClosed as err:
-            logging.error(err)
-            time.sleep(5)
-    channel = rabbit.channel()
-    channel.queue_declare(queue="analyze", durable=True)
-    channel.basic_qos(prefetch_count=BATCHSIZE)
-    channel.basic_consume(newjob, queue="analyze")
-
 
 class Model(object):
     # override this configuration
@@ -97,9 +79,8 @@ class Model(object):
 
     def __init__(self, db):
         self._db = db
-        for feat in self.features:
-            self._feature_cols = [tf.contrib.layers.real_valued_column(
-                feat, dimension=1)]
+        self._feature_cols = [tf.contrib.layers.real_valued_column(
+            feat, dimension=1) for feat in self.features]
         if self.type == "linear":
             self._model = tf.contrib.learn.LinearClassifier(
                 feature_columns=self._feature_cols,
@@ -183,7 +164,8 @@ class MVPScoreModel(Model):
     def __init__(self, db):
         self.features = ["participant_stats.kills",
                          "participant_stats.deaths",
-                         "participant_stats.assists"]
+                         "participant_stats.assists",
+                         "roster.hero_kills"]
         self.label = "participant_stats.impact_score"
         self.type = "linear"
         self.batches = 1
@@ -200,54 +182,11 @@ class MVPScoreModel(Model):
         return [w for l, w in self.predict(ids)]
 
 
-def newjob(_, method, properties, body):
-    global timer, queue
-    queue.append((method, properties, body))
-    if timer is None:
-        timer = rabbit.add_timeout(IDLE_TIMEOUT, process)
-    if len(queue) == BATCHSIZE:
-        process()
-
-
-def process():
-    global timer, queue, db, mvpmodel
-    if timer is not None:
-        rabbit.remove_timeout(timer)
-        timer = None
-    jobs = queue[:]
-    queue = []
-
-    logging.info("analyzing batch %s", str(len(jobs)))
-    ids = list(set([str(id, "utf-8") for _, _, id in jobs]))
-    ratings = mvpmodel.rate(ids)
-
-    with db.no_autoflush:
-        for c in range(len(ratings)):
-            rating = int(100 * float(ratings[c]))
-            pstat = db.query(ParticipantStats).filter(
-                ParticipantStats.participant_api_id == ids[c]).first()
-            # manual upsert
-            if pstat is None:
-                db.add(ParticipantStats(participant_api_id=ids[c],
-                                        impact_score=rating))
-            else:
-                pstat.impact_score = rating
-
-    db.commit()
-    # ack all until this one
-    logging.info("acking batch")
-    channel.basic_ack(jobs[-1][0].delivery_tag, multiple=True)
-    # notify web
-    for api_id in ids:
-        channel.basic_publish("amq.topic", "participant." + api_id,
-                              "stats_update")
-
-
 logging.basicConfig(level=logging.INFO)
 if __name__ == "__main__":
     connect()
     mvpmodel = MVPScoreModel(db)
-    mvpmodel.train()
-    logging.info(mvpmodel._model.evaluate(input_fn=mvpmodel._batch(), steps=1))
-
-    channel.start_consuming()
+    mvpmodel.train(force=True)
+    for name in mvpmodel._model.get_variable_names():
+        logging.info("%s: %s", name,
+                     mvpmodel._model.get_variable_value(name))