From 472e21344cbc0a18ef60ef66ed19e8b0639c7114 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 5 Mar 2017 21:50:44 +0100 Subject: rewrite as service --- .gitignore | 91 ++++++++++++++++ .gitmodules | 3 + api.py | 313 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ cli.py | 66 ++++++++++++ joblib | 1 + kdarate.py | 250 -------------------------------------------- requirements.txt | 7 +- 7 files changed, 478 insertions(+), 253 deletions(-) create mode 100644 .gitignore create mode 100644 .gitmodules create mode 100644 api.py create mode 100644 cli.py create mode 160000 joblib delete mode 100644 kdarate.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9a05e2d --- /dev/null +++ b/.gitignore @@ -0,0 +1,91 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +env/ +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*,cover +.hypothesis/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# dotenv +.env + +# virtualenv +.venv/ +venv/ +ENV/ + +# Spyder project settings +.spyderproject + +# Rope project settings +.ropeproject diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..52948d0 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "joblib"] + path = joblib + url = https://github.com/vainglorygame/joblib diff --git a/api.py b/api.py new file mode 100644 index 0000000..f1b8233 --- /dev/null +++ b/api.py @@ -0,0 +1,313 @@ +#!/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"] + self._classes = ["loss", "win"] + self._label = "win" + + # learn configuration + self._steps = 200 # per batch + self._max_batches = 20 # number of batches to train + self._testset = 1000 # size of testing sample + + # database mappings + self._paths = { + "kills": "kills", + "deaths": "deaths", + "assists": "assists", +# "teamkills": "(SELECT hero_kills FROM roster WHERE roster.api_id=participant.roster_api_id)", + "win": "winner" + } + + self._step = self._testset # saves batch learning state + + def connect(self, **args): + self._conn = psycopg2.connect(**args) + + # TODO this should be more random + def _get_sample(self, size="ALL", offset=0, filter=""): + """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 + :type identified: bool + """ + query = "SELECT " + elements = [] + # TODO use parameters + for name, path in self._paths.items(): + elements.append( + (""" + ARRAY(SELECT + {3} + FROM participant """ + filter + """ + ORDER BY id + LIMIT {0} OFFSET {1} + ) AS {2} + """).format(size, offset, name, path)) + + query += ", ".join(elements) + + cur = self._conn.cursor() + cur.execute(query) + data = cur.fetchone() + data = {"kills": data[0], "deaths": data[1], "assists": data[2], "win": data[3]} + cur.close() + return data + + 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, batchsize=500, limit=None): + """Fetches up to `limit` number of training items + from the database in batches.""" + # TODO maybe you can use an iterator? + if limit: + if self._step > limit: +# self._step = 0 + # TODO ???? + raise tf.errors.OutOfRangeError + sample = self._get_sample( + size=batchsize, offset=self._step) + self._step += batchsize + if len(sample["kills"]) < batchsize: # TODO! + logging.error("data exhausted!") + 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, offset=0), train=True), + eval_steps=1, + every_n_steps=20, + early_stopping_metric="accuracy", + early_stopping_metric_minimize=False, + early_stopping_rounds=100 + ) + + # validation monitor stops learning if the accuracy does not increase of 100 steps + for _ in range(self._max_batches): + logging.info("training batch %s", _) + 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 = self._get_sample( + filter="WHERE api_id in ('"+"','".join(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=100) + + +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() diff --git a/cli.py b/cli.py new file mode 100644 index 0000000..8765e7d --- /dev/null +++ b/cli.py @@ -0,0 +1,66 @@ +#!/usr/bin/python3 + +import os +import argparse +import asyncio +import asyncpg + +import joblib.joblib + +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" +} + + +async def main(qdb, sdb, name): + queue = joblib.joblib.JobQueue() + await queue.connect(**qdb) + await queue.setup() + pool = await asyncpg.create_pool(**sdb) + + async with pool.acquire() as con: + async with con.transaction(): + participants = await con.fetch(""" +select +unnest(array[ +roster.participant_1, +roster.participant_2, +roster.participant_3 +]) AS api_id +from roster where roster.match_api_id in ( +select +match.api_id +from player +join participant on participant.player_api_id=player.api_id +join roster on participant.roster_api_id=roster.api_id +join match on roster.match_api_id=match.api_id +where player.name=$1 +) + """, name) + payload = [{ + "id": part["api_id"], + "type": "participant" + } for part in participants] + await queue.request(jobtype="analyze", + payload=payload) + +parser = argparse.ArgumentParser(description="Request a Vainsocial analyze.") +parser.add_argument("-n", "--player", + help="Player name", + type=str) +args = parser.parse_args() + +loop = asyncio.get_event_loop() +loop.run_until_complete(main(queue_db, db_config, args.player)) diff --git a/joblib b/joblib new file mode 160000 index 0000000..4a8262f --- /dev/null +++ b/joblib @@ -0,0 +1 @@ +Subproject commit 4a8262f5cd9d319817ae7bf7ee69fc587808b0ad diff --git a/kdarate.py b/kdarate.py deleted file mode 100644 index beef8b8..0000000 --- a/kdarate.py +++ /dev/null @@ -1,250 +0,0 @@ -#!/usr/bin/python - -import os -import itertools -import json -import asyncio -import asyncpg -import numpy as np -import tensorflow as tf -tf.logging.set_verbosity(tf.logging.INFO) - - -class KDAClassifier(object): - """A DNN that classifies loss/win based on KDA.""" - def __init__(self): - self.model = None - self.modeldir = "/tmp/vgstats-tf-actor-model" - self._pool = None - - # DNN configuration - self._categories = [] - self._continuous = ["kills", "deaths", "assists"] - self._classes = ["loss", "win"] - self._label = "win" - - # learn configuration - self._steps = 200 # per batch - self._max_batches = 20 # number of batches to train - - # database mappings - self._paths = { - "id": "id", - "kills": "kills", - "deaths": "deaths", - "assists": "assists", -# "teamkills": "(SELECT hero_kills FROM roster WHERE roster.api_id=participant.roster_api_id)", -# "actor": "data->'attributes'->>'actor'", - "win": "winner" - } -# for n in range(0, 6): -# self._paths["item"+str(n)] = \ -# "COALESCE(data->'attributes'->'stats'->'items'->>" + str(n) +", '')" - - self._step = 0 # saves batch learning state - - async def _async_connect(self, dbstring): - """Connect to the database.""" - # TODO with current synchronous implementation, pooling isn't needed - self._pool = await asyncpg.create_pool(dbstring) - - def connect(self, dbstring): - """See _async_connect.""" - asyncio.get_event_loop().run_until_complete( - self._async_connect(dbstring)) - - # TODO maybe don't use asyncpg - async def _async_get_sample(self, size="ALL", offset=0): - """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 - :type identified: bool - """ - query = "SELECT " - elements = [] - for name, path in self._paths.items(): - elements.append( - """ - ARRAY(SELECT - {3} - FROM participant - ORDER BY id - LIMIT {0} OFFSET {1} - ) AS {2} - """.format(size, offset, name, path)) - - query += ", ".join(elements) - - async with self._pool.acquire() as conn: - async with conn.transaction(): - return (await conn.fetch(query))[0] - - def _get_sample(self, size="ALL", offset=0): - """See _async_get_sample.""" - return asyncio.get_event_loop().run_until_complete( - self._async_get_sample(size, offset)) - - 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 - ) - ) - - 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, batchsize=500, limit=None): - """Fetches up to `limit` number of training items - from the database in batches.""" - # TODO maybe you can use an iterator? - if limit: - if self._step > limit: -# self._step = 0 - # TODO ???? - raise tf.errors.OutOfRangeError - sample = self._get_sample( - size=batchsize, offset=self._step) - self._step += batchsize - if len(sample["kills"]) < batchsize: # TODO! - print("!!!!!!!!!! data exhausted !!!!!!!!!!!!!!!") - 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): - print("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, offset=0), train=True), - eval_steps=1, - every_n_steps=20, - early_stopping_metric="accuracy", - early_stopping_metric_minimize=False, - early_stopping_rounds=100 - ) - - # validation monitor stops learning if the accuracy does not increase of 100 steps - 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)) - - async def classify_db(self): - # TODO rewrite - """Classify all data in the data base and insert.""" - sample = self._get_sample(identified=True) - - # split sample (with participant ids) into data and ids - data = np.array( - [[val for key, val in s.items() if key != "id"] for s in sample] - ) - ids = [s["id"] for s in sample] - # let the DNN do the work - predicts = self.classify(data) - - # convert numpy->dict->json for db - predicts = [ - json.dumps({ - k: float(v) for k, v in p.items() - }) for p in predicts - ] - dbdata = list(zip(ids, predicts)) - async with self._pool.acquire() as conn: - async with conn.transaction(): - await conn.executemany( - """INSERT INTO enhanced_participant_wip(participant_id, actor) - VALUES ($1, $2) - ON CONFLICT (participant_id) DO - UPDATE SET data=$2 - """, dbdata) - - -def main(): - classifier = KDAClassifier() - classifier.connect("postgres://vainweb:vainweb@localhost/vainsocial-web") - sample = classifier._get_sample(size=10, offset=0) # DEBUG - classifier._step = 1000 # test batch size - print(sample) - # TODO train only for the latest patch. - print("training") - classifier.train() - print("done training") - sample = classifier._get_sample(size=10, offset=0) # DEBUG - print(sample) - # TODO train batches/fixed number - print("classifying debug data") - d = classifier.classify(sample) - for l in itertools.islice(d, 10): - print(l) - print(sample["win"]) - print("classified") - #classifier.classify_db() - -main() diff --git a/requirements.txt b/requirements.txt index dacd504..31091be 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,9 @@ -appdirs==1.4.0 -asyncpg==0.8.4 +appdirs==1.4.2 +asyncpg==0.9.0 numpy==1.12.0 packaging==16.8 protobuf==3.2.0 +psycopg2==2.7 pyparsing==2.1.10 six==1.10.0 -tensorflow-gpu==0.12.1 +tensorflow-gpu==1.0.0 -- cgit v1.3.1