hanabi-league/src/stats.py

import enum
from typing import List, Tuple

from hanabi import hanab_game
import utils
from database import conn_manager


class GameOutcome(enum.Enum):
    win = 0
    discard_crit = 1
    bomb_crit = 2
    strikeout = 3
    bottom_deck = 4
    vote_to_kill = 5
    out_of_pace = 6


class GameAnalysisResult:
    def __init__(self, outcomes: List[GameOutcome], bdrs: List[Tuple[hanab_game.DeckCard, int]]):
        self.outcome = GameOutcome
        self.bdrs = bdrs


def analyze_game(instance: hanab_game.HanabiInstance, actions: List[hanab_game.Action]) -> GameAnalysisResult:
    # List of bdrs
    bdrs = []
    # This is the default value if we find no other reason why the game was lost (or won)
    outcomes = []
    game = hanab_game.GameState(instance)

    def handle_lost_card(card, game, play: bool):
        if not game.is_trash(card):
            if game.is_critical(card):
                outcomes.append(GameOutcome.bomb_crit if play else GameOutcome.discard_crit)
            elif card.rank != 1:
                if card in game.deck[game.progress:]:
                    bdrs.append((card, game.draw_pile_size))
            else:
                if game.deck[game.progress:].count(card) == 2:
                    bdrs.append((card, game.draw_pile_size))

    for action in actions:
        if action.type == hanab_game.ActionType.Discard:
            discarded_card = instance.deck[action.target]
            handle_lost_card(discarded_card, game, False)
        if action.type == hanab_game.ActionType.Play:
            played_card = instance.deck[action.target]
            if not game.is_playable(played_card) and not game.is_trash(played_card):
                bombed_card = instance.deck[action.target]
                handle_lost_card(bombed_card, game, True)
        game.make_action(action)
        if game.pace < 0 and GameOutcome.out_of_pace not in outcomes:
            outcomes.append(GameOutcome.out_of_pace)

    if game.strikes == 3:
        outcomes.append(GameOutcome.strikeout)
    elif actions[-1].type in [hanab_game.ActionType.EndGame, hanab_game.ActionType.VoteTerminate]:
        outcomes.append(GameOutcome.vote_to_kill)
    if game.score == 5 * instance.num_suits:
        outcomes.append(GameOutcome.win)

    return GameAnalysisResult(outcomes, bdrs)


def update_user_statistics():
    """
    Update the cumulative user statistics for this user, assuming that the corresponding game statistics have
    been computed already.
    @param user_ids:
    @return:
    """
    # Note that some of these statistics could be computed by updating them on each new game insertion.
    # However, it would be tedious to ensure that *every* new game triggers an update of these statistics.
    # Also, this would be error-prone, since doing a mistake once means that values will be off forever
    # (unless the DB is reset).
    # Since it is cheap to accumulate some values over the whole DB, we therefore recreate the statistics as a whole,
    # reusing only the individual results (that never change and therefore can only be missing, but never wrong)
    cur = conn_manager.get_new_cursor()

    # Update total number of moves
    for clue_starved in [True, False]:
        rating_type = utils.get_rating_type(clue_starved)
        # We insert 0 here to ensure that we have an entry for each player
        # Note that this will immediately be changed by the next query in case it is nonzero,
        # so the zero value never shows up in the database if it was nonzero before.
        cur.execute(
            "INSERT INTO user_statistics (user_id, variant_type, total_game_moves, games_played, games_won)"
            "  ("
            "  SELECT id, %s, 0, 0, 0 FROM users"
            "  )"
            "ON CONFLICT (user_id, variant_type) DO UPDATE "
            "SET"
            "  (total_game_moves, games_played, games_won)"
            "  ="
            "  (EXCLUDED.total_game_moves, EXCLUDED.games_played, EXCLUDED.games_won)",
            (rating_type,)
        )
        cur.execute(
            "INSERT INTO user_statistics (user_id, variant_type, total_game_moves)"
            "  ("
            "  SELECT users.id, %s, SUM(games.num_turns) FROM users "
            "  INNER JOIN game_participants "
            "    ON game_participants.user_id = users.id "
            "  INNER JOIN games "
            "    ON game_participants.game_id = games.id "
            "  INNER JOIN variants"
            "    ON variants.id = games.variant_id "
            "  WHERE variants.clue_starved = %s OR variants.clue_starved IS NULL"
            "  GROUP BY users.id "
            "  ) "
            "ON CONFLICT (user_id, variant_type) DO UPDATE "
            "SET total_game_moves = EXCLUDED.total_game_moves",
            (rating_type, clue_starved)
        )
        cur.execute(
            "INSERT INTO user_statistics (user_id, variant_type, games_played, games_won)"
            "  ("
            "  SELECT"
            "    users.id,"
            "    %s,"
            "    COUNT(games.id),"
            "    COUNT(games.id) FILTER ( WHERE variants.num_suits * 5 = games.score )"
            "  FROM users "
            "  INNER JOIN game_participants "
            "    ON game_participants.user_id = users.id "
            "  INNER JOIN games "
            "    ON game_participants.game_id = games.id "
            "  INNER JOIN variants "
            "    ON variants.id = games.variant_id "
            "  WHERE variants.clue_starved = %s OR variants.clue_starved IS NULL "
            "   GROUP BY users.id"
            "  )"
            "ON CONFLICT (user_id, variant_type) DO UPDATE "
            "SET (games_played, games_won) = (EXCLUDED.games_played, EXCLUDED.games_won)",
            (rating_type, clue_starved)
        )
    conn_manager.get_connection().commit()