diff options
| -rw-r--r-- | .gitignore | 91 | ||||
| -rw-r--r-- | .gitmodules | 3 | ||||
| -rw-r--r-- | api.py (renamed from kdarate.py) | 219 | ||||
| -rw-r--r-- | cli.py | 66 | ||||
| m--------- | joblib | 0 | ||||
| -rw-r--r-- | requirements.txt | 7 |
6 files changed, 305 insertions, 81 deletions
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 @@ -5,16 +5,39 @@ import itertools import json import asyncio import asyncpg +import logging import numpy as np import tensorflow as tf -tf.logging.set_verbosity(tf.logging.INFO) +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 = "/tmp/vgstats-tf-actor-model" + self.modeldir = os.path.realpath( + os.path.join(os.getcwd(), os.path.dirname(__file__))) + "/models/kda-win" self._pool = None # DNN configuration @@ -26,35 +49,24 @@ class KDAClassifier(object): # 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 = { - "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) + self._step = self._testset # saves batch learning state - def connect(self, dbstring): - """See _async_connect.""" - asyncio.get_event_loop().run_until_complete( - self._async_connect(dbstring)) + def connect(self, **args): + self._conn = psycopg2.connect(**args) - # TODO maybe don't use asyncpg - async def _async_get_sample(self, size="ALL", offset=0): + # 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 @@ -64,27 +76,26 @@ class KDAClassifier(object): """ query = "SELECT " elements = [] + # TODO use parameters for name, path in self._paths.items(): elements.append( - """ + (""" ARRAY(SELECT {3} - FROM participant + FROM participant """ + filter + """ ORDER BY id LIMIT {0} OFFSET {1} ) AS {2} - """.format(size, offset, name, path)) + """).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)) + 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.""" @@ -113,6 +124,7 @@ class KDAClassifier(object): ) ) + # TODO tf warning - dimensions are wrong def _toinput(self, sample, train=True): """Convert sample to Tensors. :param sample: Data dictionary. @@ -152,7 +164,7 @@ class KDAClassifier(object): size=batchsize, offset=self._step) self._step += batchsize if len(sample["kills"]) < batchsize: # TODO! - print("!!!!!!!!!! data exhausted !!!!!!!!!!!!!!!") + logging.error("data exhausted!") return self._toinput(sample, train=True) def train(self): @@ -160,7 +172,7 @@ class KDAClassifier(object): self._model_setup() if os.path.isdir(self.modeldir): - print("already trained, not training again") + logging.warning("already trained, not training again") return # get one batch of testing data @@ -176,6 +188,7 @@ class KDAClassifier(object): # 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, @@ -196,55 +209,105 @@ class KDAClassifier(object): 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) + 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 - 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) + # 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() - # 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) + +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) -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() +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) -main() +loop = asyncio.get_event_loop() +loop.run_until_complete(startup()) +loop.run_forever() @@ -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 +Subproject 4a8262f5cd9d319817ae7bf7ee69fc587808b0a 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 |