path: root/worker.py

#!/usr/bin/python3
import os
import sys
import time
import logging

from sqlalchemy.orm import sessionmaker, relationship, load_only
from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine

import pika
import trueskill
import mpmath
mpmath.mp.dps = 50  # increase if FloatingPointError


RABBITMQ_URI = os.environ.get("RABBITMQ_URI") or "amqp://localhost"
DATABASE_URI = os.environ["DATABASE_URI"]
BATCHSIZE = int(os.environ.get("BATCHSIZE") or 500)  # matches
CHUNKSIZE = int(os.environ.get("CHUNKSIZE") or 100)  # matches
IDLE_TIMEOUT = float(os.environ.get("IDLE_TIMEOUT") or 1)  # s
QUEUE = os.environ.get("QUEUE") or "analyze"
DOCRUNCHMATCH = os.environ.get("DOCRUNCHMATCH") == "true"
CRUNCH_QUEUE = os.environ.get("CRUNCH_QUEUE") or "crunch_global"
DOTELESUCKMATCH = os.environ.get("DOTELESUCKMATCH") == "true"
TELESUCK_QUEUE = os.environ.get("TELESUCK_QUEUE") or "telesuck"
DOSEWMATCH = os.environ.get("DOSEWMATCH") == "true"
SEW_QUEUE = os.environ.get("SEW_QUEUE") or "sew"
UNKNOWN_PLAYER_SIGMA = int(os.environ.get("UNKNOWN_PLAYER_SIGMA") or 500)
TAU = float(os.environ.get("TAU") or 1000/100.0)

# mapping from Tier (-1 - 30) to average skill tier points
vst_points = {
    -1: 1,
    0: 1
}
for c in range(1, 12):
    vst_points[c] = (109 + 1/11) * (c + 0.5)
for c in range(1, 5):
    vst_points[11 + c] = vst_points[11] + 50 * (c + 0.5)
for c in range(1, 10):
    vst_points[15 + c] = vst_points[15] + (66 + 2/3) * (c + 0.5)
for c in range(1, 4):
    vst_points[24 + c] = vst_points[24] + (133 + 1/3) * (c + 0.5)
for c in range(1, 3):
    vst_points[27 + c] = vst_points[27] + 200 * (c + 0.5)
# ---


# ORM definitions
Match = Asset = Roster = Participant = ParticipantStats = ParticipantItems = Player = None
db = rabbit = channel = None

# batch storage
queue = []
timer = None


def connect():
    global Match, Asset, Roster, Participant, ParticipantStats, ParticipantItems, Player
    global Session, rabbit, channel

    # generate schema from db
    Base = automap_base()
    engine = create_engine(DATABASE_URI, pool_size=1, pool_recycle=3600)
    Session = sessionmaker(bind=engine, autoflush=False)
    Base.prepare(engine, reflect=True)

    # definitions
    # TODO check whether the primaryjoin clause is the best method to do this
    Asset = Base.classes.asset
    Match = Base.classes.match
    Match.rosters = relationship(
        "roster", foreign_keys="roster.match_api_id",
        primaryjoin="and_(match.api_id == roster.match_api_id)")
    Match.participants = relationship(
        "participant", foreign_keys="participant.match_api_id",
        primaryjoin="and_(match.api_id == participant.match_api_id)")
    Roster = Base.classes.roster
    Roster.match = relationship(
        "match", foreign_keys="match.api_id",
        primaryjoin="and_(match.api_id == roster.match_api_id)")
    Roster.participants = relationship(
        "participant", foreign_keys="participant.roster_api_id",
        primaryjoin="and_(roster.api_id == participant.roster_api_id)")
    Participant = Base.classes.participant
    Participant.roster = relationship(
        "roster", foreign_keys="roster.api_id",
        primaryjoin="and_(roster.api_id == participant.roster_api_id)")
    Participant.match = relationship(
        "match", foreign_keys="match.api_id",
        primaryjoin="and_(match.api_id == participant.match_api_id)")
    Participant.player = relationship(
        "player", foreign_keys="player.api_id",
        primaryjoin="and_(player.api_id == participant.player_api_id)")
    ParticipantStats = Base.classes.participant_stats
    Participant.participant_stats = relationship(
        "participant_stats", foreign_keys="participant_stats.participant_api_id",
        primaryjoin="and_(participant_stats.participant_api_id == participant.api_id)")
    ParticipantItems = Base.classes.participant_items
    Participant.participant_items = relationship(
        "participant_items", foreign_keys="participant_items.participant_api_id",
        primaryjoin="and_(participant_items.participant_api_id == participant.api_id)")
    Player = Base.classes.player

    rabbit = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URI))
    channel = rabbit.channel()
    channel.queue_declare(queue=QUEUE, durable=True)
    channel.queue_declare(queue=QUEUE + "_failed", durable=True)
    channel.queue_declare(queue=CRUNCH_QUEUE, durable=True)
    channel.queue_declare(queue=TELESUCK_QUEUE, durable=True)
    channel.basic_qos(prefetch_count=BATCHSIZE)
    channel.basic_consume(newjob, queue=QUEUE)


def newjob(_, method, properties, body):
    global timer, queue
    queue.append((method, properties, body))
    if timer is None:
        timer = rabbit.add_timeout(IDLE_TIMEOUT, try_process)
    if len(queue) == BATCHSIZE:
        try_process()

def try_process():
    global timer, queue
    if timer is not None:
        rabbit.remove_timeout(timer)
        timer = None
    try:
        process()
    except Exception as e:
        logger.error(e)
        for meth, prop, body in queue:
            # move to error queue and NACK
            channel.basic_publish(exchange="",
                                  routing_key=QUEUE+"_failed",
                                  body=body,
                                  properties=prop)
            channel.basic_nack(meth.delivery_tag, requeue=False)
        queue = []
        return

    logger.info("acking batch")
    db = None
    if DOTELESUCKMATCH:
        # ro session to get Telemetry URLs
        db = Session(autocommit=True, autoflush=False)

    for meth, prop, body in queue:
        channel.basic_ack(meth.delivery_tag)

        # notify web
        if prop.headers.get("notify"):
            channel.basic_publish("amq.topic", prop.headers.get("notify"),
                                  "analyze_update")

        if DOCRUNCHMATCH:
            # forward to cruncher_global
            channel.basic_publish(exchange="",
                                  routing_key=CRUNCH_QUEUE,
                                  body=body,
                                  properties=prop)
        if DOSEWMATCH:
            # forward to sewer
            channel.basic_publish(exchange="",
                                  routing_key=SEW_QUEUE,
                                  body=body,
                                  properties=prop)
        if DOTELESUCKMATCH:
            # forward asset url to telesucker
            id = str(body, "utf-8")
            for asset in db.query(Asset).options(\
                        load_only("url", "match_api_id"))\
                    .filter(Asset.match_api_id == id).all():
                channel.basic_publish(exchange="",
                                      routing_key=TELESUCK_QUEUE,
                                      body=asset.url,
                                      properties=pika.BasicProperties(
                                          headers={
                                              "match_api_id": asset.match_api_id
                                          }
                                      ))

    if db is not None:
        db.close()

    queue = []


def process():
    global timer, queue, Session
    logger.info("analyzing batch %s", str(len(queue)))
    ids = list(set([str(id, "utf-8") for _, _, id in queue]))

    db = Session()
    try:
        env = trueskill.TrueSkill(
            backend="mpmath",
            mu=1500,
            sigma=1000,
            beta=10.0/30*3000,
            tau=TAU,
            draw_probability=0
        )
        for match in db.query(Match).order_by(Match.created_at.asc()).options(
            load_only("api_id", "game_mode")
            .selectinload(Match.rosters)
                .load_only("api_id", "match_api_id", "winner")
            .selectinload(Roster.participants)
                .load_only("api_id", "match_api_id", "roster_api_id",
                           "player_api_id", "skill_tier", "went_afk")
                .selectinload(Participant.player)
                    .load_only("api_id",
                               "rank_points_ranked",
                               "trueskill_sigma", "trueskill_mu",
                               "trueskill_casual_sigma", "trueskill_casual_mu",
                               "trueskill_ranked_sigma", "trueskill_ranked_mu",
                               "trueskill_blitz_sigma", "trueskill_blitz_mu",
                               "trueskill_br_sigma", "trueskill_br_mu")
         ).filter(Match.api_id.in_(ids)).yield_per(CHUNKSIZE):
            trueskill_column = None  # on player
            if match.game_mode == "casual":  # mode names mapped by processor!!!
                trueskill_column = "trueskill_casual"
            if match.game_mode == "ranked":
                trueskill_column = "trueskill_ranked"
            if match.game_mode == "blitz":
                trueskill_column = "trueskill_blitz"
            if match.game_mode == "br":
                trueskill_column = "trueskill_br"
            if trueskill_column is None:
                logger.info("got unsupported game mode %s", match.game_mode)

            matchup_shared = []  # trueskill shared across all modes
            matchup = []

            anyAfk = False
            if len(match.rosters) != 2 \
                    or len(match.rosters[0].participants) != 3 \
                    or len(match.rosters[1].participants) != 3:
                logger.error("got an invalid matchup %s", match.api_id)
                anyAfk = True

            for participant in match.participants:
                participant.participant_items[0].any_afk = False
                if participant.went_afk == 1:
                    logger.info("got an afk matchup %s", match.api_id)
                    anyAfk = True
                    break

            if anyAfk:
                match.trueskill_quality = 0
                for participant in match.participants:
                    participant.participant_items[0].any_afk = True
                continue

            for roster in match.rosters:
                team_shared = []
                team = []
                for participant in roster.participants:
                    player = participant.player[0]

                    # calculate TrueSkill shared across all modes = starting point for all modes
                    if player.trueskill_mu is not None:
                        sigma_shared = player.trueskill_mu
                        mu_shared = player.trueskill_sigma
                    else:
                        # fallback 1 - approximate by max(ranked, blitz) points
                        rank_points = None
                        if player.rank_points_ranked is not None and player.rank_points_ranked != 0:
                            rank_points = player.rank_points_ranked
                        if player.rank_points_blitz is not None and player.rank_points_blitz != 0:
                            if rank_points is None:
                                rank_points = player.rank_points_blitz
                            else:
                                if player.rank_points_blitz > rank_points:
                                    rank_points = player.rank_points_blitz

                        if rank_points is not None:
                            sigma_shared = UNKNOWN_PLAYER_SIGMA * (2.0/3.0)  # more accurate than skill tier = more trust
                            mu_shared = float(rank_points) + sigma_shared
                        else:
                            # fallback 2 - approximate rank points by skill tier
                            sigma_shared = UNKNOWN_PLAYER_SIGMA
                            mu_shared = vst_points[participant.skill_tier] + sigma_shared

                    team_shared.append(env.create_rating(float(mu_shared), float(sigma_shared)))

                    # calculate queue specific TrueSkill
                    if getattr(player, trueskill_column + "_mu") is not None:
                        sigma = getattr(player, trueskill_column + "_sigma")
                        mu = getattr(player, trueskill_column + "_mu")
                    else:
                        # fallback 1 - approximate by shared trueskill
                        sigma = sigma_shared
                        mu = mu_shared

                    team.append(env.create_rating(float(mu), float(sigma)))


                matchup_shared.append(team_shared)
                matchup.append(team)

            logger.info("got a valid matchup %s", match.api_id)

            # store the fairness of the match using the shared TrueSkill
            match.trueskill_quality = env.quality(matchup)

            # shared TrueSkill (participant)
            for team, roster in zip(env.rate(matchup_shared, ranks=[int(not r.winner) for r in match.rosters]),
                                    match.rosters):
                # lower rank is better = winner!
                for rating, participant in zip(team, roster.participants):
                    player = participant.player[0]
                    # delta = current - pre
                    if player.trueskill_mu is not None:
                        participant.trueskill_delta = (float(rating.mu) - float(rating.sigma)) - (float(player.trueskill_mu) - float(player.trueskill_sigma))
                    else:
                        participant.trueskill_delta = 0
                    player.trueskill_mu = rating.mu
                    participant.trueskill_mu = rating.mu
                    player.trueskill_sigma = rating.sigma
                    participant.trueskill_sigma = rating.sigma

            # queue specific TrueSkill (participant_items)
            # no delta
            for team, roster in zip(env.rate(matchup, ranks=[int(not r.winner) for r in match.rosters]),
                                    match.rosters):
                for rating, participant in zip(team, roster.participants):
                    player = participant.player[0]
                    pi = participant.participant_items[0]
                    setattr(player, trueskill_column + "_mu", rating.mu)
                    setattr(pi, trueskill_column + "_mu", rating.mu)
                    setattr(player, trueskill_column + "_sigma", rating.sigma)
                    setattr(pi, trueskill_column + "_sigma", rating.sigma)

        db.commit()
    except:
        db.rollback()
        raise
    finally:
        db.close()


# log errs to stderr, debug to stdout
class InfoFilter(logging.Filter):
    def filter(self, record):
        return record.levelno in (logging.DEBUG, logging.INFO)

logger = logging.getLogger("__name__")
logger.setLevel(logging.INFO)

h1 = logging.StreamHandler(sys.stdout)
h1.setLevel(logging.INFO)
h1.addFilter(InfoFilter())
logger.addHandler(h1)

h2 = logging.StreamHandler(sys.stderr)
h2.setLevel(logging.WARNING)
logger.addHandler(h2)

if __name__ == "__main__":
    connect()
    channel.start_consuming()