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, current_streak, total_bdr)" " (" " SELECT id, %s, 0, 0, 0, 0, 0 FROM users" " )" "ON CONFLICT (user_id, variant_type) DO UPDATE " "SET" " (total_game_moves, games_played, games_won, current_streak, total_bdr)" " =" " (EXCLUDED.total_game_moves, EXCLUDED.games_played, EXCLUDED.games_won, EXCLUDED.current_streak, EXCLUDED.total_bdr)", (rating_type,) ) cur.execute( "INSERT INTO user_statistics (user_id, variant_type, total_game_moves, games_played, games_won, total_bdr)" " (" " SELECT" " users.id," " CASE WHEN clue_starved THEN %s ELSE %s END," " SUM(games.num_turns)," " COUNT(*)," " COUNT(*) FILTER ( WHERE variants.num_suits * 5 = games.score )," " SUM (game_statistics.bottom_deck_risk)" "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 " " LEFT OUTER JOIN game_statistics" " ON games.id = game_statistics.game_id" " GROUP BY users.id, clue_starved " " ) " "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)", (utils.get_rating_type(True), utils.get_rating_type(False)) ) cur.execute( "INSERT INTO user_statistics (user_id, variant_type, current_streak)" " (" " SELECT" " user_id," " CASE WHEN clue_starved THEN %s ELSE %s END," " MAX(streak_length) AS max_streak_length FROM" " (" " SELECT" " *," " CASE" " WHEN num_suits * 5 = score" " THEN" " COUNT(*)" " OVER (PARTITION BY user_id, clue_starved, group_id ORDER BY league_id)" " END" " AS streak_length " " FROM" " (" " SELECT" " users.id AS user_id," " variants.clue_starved," " variants.num_suits," " games.score," " games.league_id," " COUNT(*) " " FILTER (WHERE variants.num_suits * 5 != games.score)" " OVER (PARTITION BY users.id, variants.clue_starved ORDER BY games.league_id)" " - CASE WHEN variants.num_suits * 5 != games.score THEN 1 ELSE 0 END" " AS group_id" " 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 " " ) AS games_grouped_by_streak " " ) AS games_with_streaks " " GROUP BY user_id, clue_starved" " )" "ON CONFLICT (user_id, variant_type) DO UPDATE " "SET current_streak = EXCLUDED.current_streak", (utils.get_rating_type(True), utils.get_rating_type(False)) ) conn_manager.get_connection().commit()