path: root/api.py

#!/usr/bin/python

import os
import itertools
import json
import asyncio
import asyncpg
import logging
import numpy as np
import tensorflow as tf
import psycopg2

import joblib.worker

#tf.logging.set_verbosity(tf.logging.WARNING)

queue_db = {
    "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "localhost",
    "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5433,
    "user": os.environ.get("POSTGRESQL_SOURCE_USER") or "vainraw",
    "password": os.environ.get("POSTGRESQL_SOURCE_PASSWORD") or "vainraw",
    "database": os.environ.get("POSTGRESQL_SOURCE_DB") or "vainsocial-raw"
}

db_config = {
    "host": os.environ.get("POSTGRESQL_DEST_HOST") or "localhost",
    "port": os.environ.get("POSTGRESQL_DEST_PORT") or 5432,
    "user": os.environ.get("POSTGRESQL_DEST_USER") or "vainweb",
    "password": os.environ.get("POSTGRESQL_DEST_PASSWORD") or "vainweb",
    "database": os.environ.get("POSTGRESQL_DEST_DB") or "vainsocial-web"
}


# TODO create abstract class, move to own file
class KDAClassifier(object):
    """A DNN that classifies loss/win based on KDA."""
    def __init__(self):
        self.model = None
        self.modeldir = os.path.realpath(
            os.path.join(os.getcwd(), os.path.dirname(__file__))) + "/models/kda-win"
        self._pool = None

        # DNN configuration
        self._categories = []
        self._continuous = ["kills", "deaths", "assists", "cs", "teamkills"]
        self._classes = ["loss", "win"]
        self._label = "win"

        # learn configuration
        self._steps = 200  # per batch
        self._batchsize = 500
        self._max_batches = 5  # number of batches to train

        # database mappings
        # TODO cache, rm duplicated code
        self._trainquery = """
            SELECT
                participant.kills,
                participant.deaths,
                participant.assists,
                participant.farm,
                roster.hero_kills,
                participant.winner,
                participant.api_id
            FROM participant
            TABLESAMPLE BERNOULLI(5)
            JOIN roster on participant.roster_api_id=roster.api_id
            JOIN match on roster.match_api_id=match.api_id
            LIMIT %s
        """
        self._testquery = """
            SELECT
                participant.kills,
                participant.deaths,
                participant.assists,
                participant.farm,
                roster.hero_kills,
                participant.winner,
                participant.api_id
            FROM participant
            JOIN roster on participant.roster_api_id=roster.api_id
            JOIN match on roster.match_api_id=match.api_id
            WHERE participant.api_id=ANY(%s)
        """

    def connect(self, **args):
        self._conn = psycopg2.connect(**args)

    def _get_sample(self, size=None, filter=None):
        """Return a data set from the database.
        :param size: (optional) The number of items to get. Defaults to `All`.
        :type size: int or str
        :param offset: (optional) The SQL OFFSET parameter.
        :type offset: int or str
        """
        cur = self._conn.cursor()
        assert (filter is None) ^ (size is None)
        if filter is None:
            cur.execute(self._trainquery, (size,))
        else:
            cur.execute(self._testquery, (filter,))
        ids = []
        data = {"kills": [], "deaths": [], "assists": [], "win": [], "cs": [], "teamkills": []}
        for rec in cur:
            data["kills"].append(rec[0])
            data["deaths"].append(rec[1])
            data["assists"].append(rec[2])
            data["cs"].append(rec[3])
            data["teamkills"].append(rec[4])
            data["win"].append(rec[5])
            ids.append(rec[6])
        cur.close()
        logging.debug(data)
        return data, ids

    def _model_setup(self):
        """Sets up a model that takes the `num_features` features as input."""
        feature_columns = [
            tf.contrib.layers.sparse_column_with_hash_bucket(feat, 100)
            for feat in self._categories
        ]
        emb_columns = [
            tf.contrib.layers.embedding_column(
                sparse_id_column=col,
                dimension=5  # log_2(number of unique features)  TODO
            )
            for col in feature_columns
        ] + [
            tf.contrib.layers.real_valued_column(feat)
            for feat in self._continuous
        ]

        self.model = tf.contrib.learn.DNNClassifier(
            feature_columns=emb_columns,
            hidden_units=[2],  # http://stats.stackexchange.com/a/1097  TODO inp+outp / 2
            n_classes=len(self._classes),
            model_dir=self.modeldir,
            config=tf.contrib.learn.RunConfig(
                save_checkpoints_secs=2
            )
        )

    # TODO tf warning - dimensions are wrong
    def _toinput(self, sample, train=True):
        """Convert sample to Tensors.
        :param sample: Data dictionary.
        :type sample: dict
        :param train: (optional) Whether to return labels too.
        :type train: bool
        :return: features, label
        :rtype: tuple
        """
        continuous = {k: tf.constant(sample[k])
                      for k in self._continuous}
        categories = {k: tf.SparseTensor(
            indices=[[i, 0] for i in range(len(sample[k]))],
            values=sample[k],
            shape=[len(sample[k]), 1])
                    for k in self._categories}

        features = {**continuous, **categories}

        if train:
            label = tf.constant(sample[self._label])
            return features, label
        else:
            return features

    # TODO use asyncpg cursor https://magicstack.github.io/asyncpg/current/api/index.html#cursors
    def _more(self):
        """Fetches up to `limit` number of training items
           from the database in batches."""
        # TODO maybe you can use an iterator?
        sample = self._get_sample(size=self._batchsize)[0]
        return self._toinput(sample, train=True)

    def train(self):
        """Train a DNN on a static, algorithmic guess."""
        self._model_setup()

        if os.path.isdir(self.modeldir):
            logging.warning("already trained, not training again")
            return

        # get one batch of testing data
        # TODO either terminate with `eval_steps` or with OutOfRangeError
        validation_monitor = tf.contrib.learn.monitors.ValidationMonitor(
            input_fn=lambda: self._toinput(self._get_sample(size=1000)[0], train=True),
            eval_steps=1,
            every_n_steps=10
        )

        for _ in range(self._max_batches):
            self.model.fit(
                input_fn=lambda: self._more(),
                steps=self._steps,
                monitors=[validation_monitor]
            )

    def classify(self, sample, only_best=False):
        """Classify a data set.

        :param only_best: (optional) Return the predicted result
                          instead of a dict of propabilities.
        :type only_best: bool
        :return: Prediction results.
        :rtype: list of dict or list
        """
        if only_best:
            return self.model.predict(input_fn=lambda: self._toinput(sample, train=False))
        else:
            return self.model.predict_proba(input_fn=lambda: self._toinput(sample, train=False))

    def windup(self):
        pass

    def teardown(self, failed=False):
        if failed:
            pass
        else:
            self._conn.commit()

    def classify_db(self, objids):
        """Classify all data in the data base and insert."""
        # split sample (with participant ids) into data and ids
        # TODO use parameters
        logging.error("sample size: %s", len(objids))
        sample, objids = self._get_sample(filter=objids)
        d = self.classify(sample)
        cnt = 0
        cur = self._conn.cursor()
        for l in itertools.islice(d, len(objids)):
            cur.execute("""
                INSERT INTO participant_stats
                (patch_version, participant_api_id, score)
                VALUES(2.2, %(objid)s, %(score)s)
                ON CONFLICT(participant_api_id) DO
                UPDATE SET score=%(score)s
            """, {"objid": objids[cnt], "score": float(l[1])})
            cnt += 1
        cur.close()


class Analyzer(joblib.worker.Worker):
    def __init__(self):
        self._pool = None
        self._queries = {}
        super().__init__(jobtype="analyze")
        self.classifier = None

    async def connect(self, dbconf, queuedb):
        """Connect to database."""
        logging.warning("connecting to database")
        await super().connect(**queuedb)
        self._pool = await asyncpg.create_pool(**dbconf)
        self.classifier = KDAClassifier()
        self.classifier.connect(**dbconf)
        self.classifier._step = 1000  # TODO test batch size

    async def setup(self):
        """Setup the model."""
        self.classifier.train()

    async def _windup(self):
        self._con = await self._pool.acquire()
        self._tr = self._con.transaction()
        await self._tr.start()
        self.classifier.windup()
        self._participants = []

    async def _teardown(self, failed):
        if len(self._participants) > 0:
            # TODO if this fails, job is still marked as finished
            self.classifier.classify_db(self._participants)

        if failed:
            await self._tr.rollback()
        else:
            await self._tr.commit()
        await self._pool.release(self._con)
        self.classifier.teardown()
    
    async def _execute_job(self, jobid, payload, priority):
        object_id = payload["id"]
        object_type = payload["type"]
        if object_type != "participant":
            return
        self._participants.append(object_id)
        logging.info("%s: classifying '%s', %s", jobid,
                     object_type, object_id)

async def startup():
    for _ in range(1):
        worker = Analyzer()
        await worker.connect(db_config, queue_db)
        await worker.setup()
        await worker.start(batchlimit=1000)


logging.basicConfig(
    filename=os.path.realpath(
        os.path.join(os.getcwd(),
                 os.path.dirname(__file__))) +
        "/logs/analyzer.log",
    filemode="a",
    level=logging.DEBUG
)
console = logging.StreamHandler()
console.setLevel(logging.WARNING)
logging.getLogger("").addHandler(console)

loop = asyncio.get_event_loop()
loop.run_until_complete(startup())
loop.run_forever()