This commit is contained in:
pavel-hudec 2024-02-06 15:43:47 +08:00
commit edcb230365
11 changed files with 516 additions and 10 deletions

View file

@ -85,3 +85,7 @@ body {
.history-bullets ul {
flex-direction: column;
}
.endgame-table td, .endgame-table th {
border: 1px solid black;
}

2
deps/py-hanabi vendored

@ -1 +1 @@
Subproject commit 3ac51d574e65aff9b3420fdebd467d7b98ea1d28
Subproject commit 51e09cd94393de64e07191d6ca544139417acb3b

View file

@ -16,10 +16,15 @@ import fetch_games
import ratings
import stats
import render_site
import endgames
from log_setup import logger
def subcommand_analyze_endgames():
endgames.work_thread()
def subcommand_init(force: bool, no_fetch_variants: bool):
tables = database.get_existing_tables()
if len(tables) > 0 and not force:
@ -101,6 +106,7 @@ def get_parser() -> argparse.ArgumentParser:
subparsers.add_parser('process-ratings', help="Process ratings of all games.")
subparsers.add_parser('process-stats', help="Process statistics for all players.")
subparsers.add_parser('generate-site', help="Generate the website from the DB.")
subparsers.add_parser('analyze-endgames', help="Run endgame analysis on games in DB. Resource intensive!")
subparsers.add_parser('run', help="Run the automatic suite: Fetch + process games and render site.")
fetch_parser = subparsers.add_parser('fetch', help='Fetch new data.')
@ -122,6 +128,7 @@ def main():
'generate-site': subcommand_generate_site,
'fetch': subcommand_fetch,
'run': subcommand_run,
'analyze-endgames': subcommand_analyze_endgames
}[args.command]
if args.verbose:

View file

@ -359,3 +359,45 @@ CREATE TABLE user_statistics (
average_game_moves REAL GENERATED ALWAYS AS (CASE WHEN games_played != 0 THEN CAST(total_game_moves AS REAL) / games_played ELSE NULL END) STORED,
PRIMARY KEY (user_id, variant_type)
);
DROP TABLE IF EXISTS endgames;
CREATE TABLE endgames (
game_id INTEGER REFERENCES games (id),
turn SMALLINT,
/**
* We want to be able to store probabilities for different actions that can be taken.
* Action type can be
0 for play actions
1 for discard actions
2 for clues
*/
action_type SMALLINT CHECK (0 <= action_type AND action_type <= 2),
/**
We store cards as (suit_index, rank) here for uniqueness of representation.
If we want to refer to known trash, we will use (0,0) as representation.
*/
suit_index SMALLINT, /* 0 for clue actions */
rank SMALLINT, /* 0 for clue actions */
enumerator BIGINT NOT NULL CHECK (enumerator >= 0),
denominator BIGINT NOT NULL CHECK (denominator > 0),
chance REAL GENERATED ALWAYS AS (CAST(enumerator AS REAL) / denominator) STORED,
PRIMARY KEY (game_id, turn, action_type, suit_index, rank)
);
/**
We store separately whether we analyzed a certain game already and what the termination reason for the analysis was:
0 if evaluation completed within specified time and memory
1 if evaluation ran into timeout
2 if evaluation was empty because state is unreachable
3 if evaluation ran out of memory
This is also necessary because for some endgames, because in case 2 we will not have data,
simply because the game replay ended too early.
To avoid re-analyzing these seeds, we mark all seeds analyzed in this table.
*/
DROP TABLE IF EXISTS endgames_analyzed;
CREATE TABLE endgames_analyzed (
game_id INTEGER REFERENCES games (id),
termination_reason SMALLINT NOT NULL,
PRIMARY KEY (game_id)
);

View file

@ -48,5 +48,9 @@ USER_HISTORY_CACHE_TIME = 60 * 60
# Fraction of seeds which is assumed to be unwinnable
UNWINNABLE_SEED_FRACTION = 0.02
WEBSITE_OUTPUT_DIRECTORY = 'build'
ENDGAME_MAX_DRAW_PILE_SIZE = 15 # Not interested in game states with more than 15 cards, this should be enough.
ENDGAME_MEMORY_BYTES = 4 * 1024 * 1024 * 1024 # 4 GB of memory
ENDGAME_TIMEOUT_SECONDS = 60 * 15 # 15 Minutes per game by default
ENDGAME_ANALYSIS_QUERY_INTERVAL_MINUTES = 5 # Re-query database every 5 minutes

254
src/endgames.py Normal file
View file

@ -0,0 +1,254 @@
import json
import subprocess
import re
import resource
import time
from typing import List, Dict, Tuple
from dataclasses import dataclass
from pathlib import Path
import platformdirs
import psycopg2.extras
import psycopg2.errors
import hanabi.hanab_game
import hanabi.constants
import hanabi.live.compress
import constants
import games_db_interface
from database import conn_manager
from log_setup import logger
@dataclass
class EndgameAction:
turn: int
action_type: hanabi.hanab_game.ActionType
card: hanabi.hanab_game.DeckCard
enumerator: int
denominator: int
@property
def win_rate(self):
return self.enumerator / self.denominator
def analyze_and_store_game(game_id: int) -> int:
actions, return_code = analyze_game_from_db(game_id)
store_endgame_actions(game_id, actions, return_code)
return return_code
def analyze_game_from_db(game_id: int, refresh_cache=False) -> Tuple[List[EndgameAction], int]:
# In order to pass the game replay to the endgame analyzer (a C++ program), we use the hanab.live json format.
# In order to avoid to need to use the /export endpoint of hanab.live, we create the json replay ourselves from
# the information stored in the database. Note that this is partially lossy, since the GameState class we use here
# does not support stuff like notes or player names, but we don't care about this anyway and can pass the relevant
# information, i.e. deck, number of players and actions.
# We can additionally use the cache dir
filename = Path(platformdirs.user_cache_dir(constants.APP_NAME)) / "replays" / "{}.json".format(game_id)
if not filename.exists() or refresh_cache:
filename.parent.mkdir(exist_ok=True, parents=True)
game, variant_name = games_db_interface.load_game(game_id)
game_json = game.to_json()
game_json["options"]["variant"] = variant_name
with open(filename, 'w') as f:
f.write(json.dumps(game_json))
return analyze_endgame_from_file(str(filename))
def analyze_endgame_from_file(filename: str) -> Tuple[List[EndgameAction], int]:
"""
Analyzes endgame of replay specified in given file using appropriate time and memory limits.
@param filename: Name of file containing replay of game in hanab.live format
@return: List of all evaluated actions and return code why evaluation finished:
0 if evaluation completed within specified time and memory
1 if evaluation ran into timeout
2 if evaluation was empty because state is unreachable
3 if evaluation ran out of memory
No guarantee can be made on what actions are actually evaluated, these might be more or less depending on
timeouts and/or resource limitation.
@raise
"""
raw_output: bytes
return_code = 0
args = [
'./endgame-analyzer', '--file', filename,
'--draw-pile-size', str(constants.ENDGAME_MAX_DRAW_PILE_SIZE), '--list-actions', '--quiet'
]
try:
result = subprocess.run(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=constants.ENDGAME_TIMEOUT_SECONDS,
preexec_fn=set_memory_limit
)
if result.returncode != 0:
# 2 is the return code to report that the specified game state is not reachable
# In this case, there is nothing to analyze, so we will return an empty list and normal program termination.
if result.returncode == 2:
return [], 2
# 3 is the return code used by the subprocess to indicate an out of memory exception
# Since we intentionally limited the memory, this is actually not an exception for us,
# we will simply parse the results we have and report the OOM exception.
if result.returncode == 3:
return_code = 3
else:
raise RuntimeError(
"Abnormal program termination of endgame-analyzer subprocess: Call of\n"
"{}\n"
"resulted in returncode '{}' and stderr\n"
"{}"
.format(
" ".join(args),
result.returncode,
result.stderr.decode('utf-8')
)
)
raw_output = result.stdout
except subprocess.TimeoutExpired as time_err:
return_code = 1
raw_output = time_err.stdout
# It could be that we got no output. In that case, we also cannot parse anything
output = raw_output.decode('utf-8') if raw_output else ""
pattern = r"Turn (?P<turn>\d+), (?P<type>\w+)(?:\s(?P<card>\w\w))?: (?P<enumerator>\d+)/(?P<denominator>\d+)"
return [parse_match(match.groupdict()) for match in re.finditer(pattern, output)], return_code
def set_memory_limit():
resource.setrlimit(resource.RLIMIT_DATA, (constants.ENDGAME_MEMORY_BYTES, constants.ENDGAME_MEMORY_BYTES))
def store_endgame_actions(game_id: int, endgame_actions: List[EndgameAction], result_code) -> None:
values = []
for action in endgame_actions:
values.append((game_id, action.turn, action.action_type.value, action.card.suitIndex, action.card.rank, action.enumerator, action.denominator))
# Remove duplicates (even though we expect none), otherwise this causes errors on insertion.
values = list(set(values))
conn = conn_manager.get_connection()
cur = conn.cursor()
psycopg2.extras.execute_values(
cur,
"INSERT INTO endgames "
"VALUES %s "
"ON CONFLICT (game_id, turn, action_type, suit_index, rank) "
"DO UPDATE "
"SET (enumerator, denominator) = (EXCLUDED.enumerator, EXCLUDED.denominator)",
values
)
# Mark this game as analyzed.
cur.execute(
"INSERT INTO endgames_analyzed "
"VALUES (%s, %s) "
"ON CONFLICT (game_id) "
"DO UPDATE "
"SET termination_reason = EXCLUDED.termination_reason",
(game_id, result_code)
)
conn.commit()
def load_endgame_actions(game_id: int) -> List[EndgameAction]:
cur = conn_manager.get_new_cursor()
cur.execute(
"SELECT turn, action_type, suit_index, rank, enumerator, denominator "
"FROM endgames "
"WHERE game_id = %s "
"ORDER BY turn ASC, action_type ASC, suit_index ASC, rank ASC",
(game_id,)
)
ret = []
for (turn, action_type, suit_index, rank, enumerator, denominator) in cur.fetchall():
ret.append(EndgameAction(
turn,
hanabi.hanab_game.ActionType(action_type),
hanabi.hanab_game.DeckCard(suit_index, rank),
enumerator, denominator)
)
return ret
def parse_match(action: Dict) -> EndgameAction:
turn = action["turn"]
action_type = parse_action_type(action["type"])
card = hanabi.hanab_game.DeckCard(0, 0) if action["card"] is None else parse_card(action["card"])
enumerator = action["enumerator"]
denominator = action["denominator"]
return EndgameAction(turn, action_type, card, enumerator, denominator)
def parse_action_type(action_type: str) -> hanabi.hanab_game.ActionType:
match action_type:
case "play":
return hanabi.hanab_game.ActionType.Play
case "discard":
return hanabi.hanab_game.ActionType.Discard
case "clue":
return hanabi.hanab_game.ActionType.ColorClue
raise ValueError("Failed to parse action type: {}".format(action_type))
def parse_card(card: str) -> hanabi.hanab_game.DeckCard:
assert len(card) == 2
if card == "kt":
return hanabi.hanab_game.DeckCard(0, 0)
else:
rank = int(card[1])
suit = hanabi.constants.COLOR_INITIALS.index(card[0])
assert suit is not None
return hanabi.hanab_game.DeckCard(suit, rank)
def print_action_type(action_type: hanabi.hanab_game.ActionType) -> str:
match action_type:
case hanabi.hanab_game.ActionType.Play:
return "Play"
case hanabi.hanab_game.ActionType.Discard:
return "Discard"
case hanabi.hanab_game.ActionType.RankClue:
return "Clue"
case hanabi.hanab_game.ActionType.ColorClue:
return "Clue"
case _:
return "Unknown Action"
def work_thread():
"""
Will continuously query database to analyze endgames.
@return:
"""
conn = conn_manager.get_connection()
cur = conn.cursor()
while True:
cur.execute(
"SELECT games.id "
"FROM games "
"LEFT OUTER JOIN endgames_analyzed "
" ON endgames_analyzed.game_id = games.id "
"WHERE endgames_analyzed.termination_reason IS NULL "
"ORDER BY games.league_id DESC "
"LIMIT 1",
(False,)
)
res = cur.fetchone()
if res is None:
logger.info("No game found to analyze. Going to sleep for {} Minutes".format(
constants.ENDGAME_ANALYSIS_QUERY_INTERVAL_MINUTES)
)
time.sleep(60 * constants.ENDGAME_ANALYSIS_QUERY_INTERVAL_MINUTES)
else:
(game_id, ) = res
logger.info("Analyzing endgame of game {}".format(game_id))
return_code = analyze_and_store_game(game_id)
print("Finished endgame analysis of {}: Returncode {}".format(game_id, return_code))

View file

@ -81,10 +81,15 @@ def load_deck(seed: str) -> List[hanabi.hanab_game.DeckCard]:
return deck
def load_game(game_id: int) -> Tuple[hanabi.hanab_game.HanabiInstance, List[hanabi.hanab_game.Action]]:
def load_game_parts(game_id: int) -> Tuple[hanabi.hanab_game.HanabiInstance, List[hanabi.hanab_game.Action], str]:
"""
Loads information on game from database
@param game_id: ID of game
@return: Instance (i.e. deck + settings) of game, list of actions, variant name
"""
cur = conn_manager.get_new_cursor()
cur.execute(
"SELECT games.num_players, games.seed, variants.clue_starved "
"SELECT games.num_players, games.seed, variants.clue_starved, variants.name "
"FROM games "
"INNER JOIN variants"
" ON games.variant_id = variants.id "
@ -98,10 +103,19 @@ def load_game(game_id: int) -> Tuple[hanabi.hanab_game.HanabiInstance, List[hana
raise ValueError(err_msg)
# Unpack results now
(num_players, seed, clue_starved) = res
(num_players, seed, clue_starved, variant_name) = res
actions = load_actions(game_id)
deck = load_deck(seed)
instance = hanabi.hanab_game.HanabiInstance(deck, num_players, clue_starved=clue_starved)
return instance, actions
return instance, actions, variant_name
def load_game(game_id: int) -> Tuple[hanabi.hanab_game.GameState, str]:
instance, actions, variant_name = load_game_parts(game_id)
game = hanabi.hanab_game.GameState(instance)
for action in actions:
game.make_action(action)
return game, variant_name

View file

@ -12,16 +12,18 @@ import requests_cache
import platformdirs
import stats
import hanabi.hanab_game
import constants
import config
import utils
from dataclasses import dataclass
import endgames
import games_db_interface
from database import conn_manager
@dataclass
class PlayerEntry:
player_name: str
@ -713,6 +715,111 @@ def build_unique_variants(variant_rows: List[VariantRow]):
return [row for row in variant_rows if row.num_players == config.config_manager.get_config().min_player_count]
def render_game_pages(env: jinja2.Environment, out_dir: Path):
endgames = get_endgame_page_data()
template = env.get_template("game.html")
for game_id, data in endgames.items():
rendered_template = template.render(
total_games_played=get_total_games(),
total_players=get_num_players(),
latest_run=datetime.datetime.now().isoformat(),
game_id=game_id,
data=data
)
output_file = out_dir / 'game' / str(game_id) / 'index.html'
output_file.parent.mkdir(exist_ok=True, parents=True)
with open(output_file, 'w') as f:
f.write(rendered_template)
def format_endgame_action(endgame_action: endgames.EndgameAction):
return "{} {}: {}/{} ~ {}".format(
endgames.print_action_type(endgame_action.action_type),
endgame_action.card,
endgame_action.enumerator,
endgame_action.denominator,
endgame_action.action_type.value
)
@dataclass
class EndgameActionRow:
description: str
enumerator: int
denominator: int
marked: bool = False
@property
def win_rate(self):
return round(100 * self.enumerator / self.denominator, 3)
def convert_endgame_action(endgame_action: endgames.EndgameAction, game: hanabi.hanab_game.GameState, action: hanabi.hanab_game.Action) -> EndgameActionRow:
action_str = endgames.print_action_type(endgame_action.action_type)
target_str: str
if endgame_action.action_type not in [hanabi.hanab_game.ActionType.ColorClue, hanabi.hanab_game.ActionType.RankClue]:
target_str = " {}".format(endgame_action.card)
else:
target_str = ""
description = action_str + target_str
marked = False
# To simplify comparisons, we only work with color clues here. Endgame actions always consist of color clues.
if action.type == hanabi.hanab_game.ActionType.RankClue:
action.type = hanabi.hanab_game.ActionType.ColorClue
if endgame_action.action_type == action.type:
if endgame_action.action_type == hanabi.hanab_game.ActionType.ColorClue:
marked = True
else:
game_target = game.instance.deck[action.target]
if game.is_trash(game_target):
game_target = hanabi.hanab_game.DeckCard(0, 0)
if endgame_action.card == game_target:
marked = True
return EndgameActionRow(description, endgame_action.enumerator, endgame_action.denominator, marked)
def get_endgame_page_data():
cur = conn_manager.get_new_cursor()
cur.execute(
"SELECT games.id, termination_reason "
"FROM games "
"LEFT OUTER JOIN endgames_analyzed "
" ON endgames_analyzed.game_id = games.id "
)
ret = {}
for (game_id, termination_reason) in cur.fetchall():
if termination_reason is not None:
ret[game_id] = []
instance, actions, _ = games_db_interface.load_game_parts(game_id)
game = hanabi.hanab_game.GameState(instance)
endgame_actions = endgames.load_endgame_actions(game_id)
while len(endgame_actions) > 0:
# Move to current turn and update game
cur_turn = endgame_actions[0].turn
# Note the -1 here since turns on hanab.live start to count at 1
while len(game.actions) < cur_turn - 1:
action, *actions = actions
game.make_action(action)
assert len(actions) > 0
actions_this_turn: List[endgames.EndgameAction] = []
while len(endgame_actions) > 0 and endgame_actions[0].turn == cur_turn:
action, *endgame_actions = endgame_actions
actions_this_turn.append(action)
actions_this_turn.sort(key=lambda a: -a.win_rate)
best_action, *other_actions = [convert_endgame_action(a, game, actions[0]) for a in actions_this_turn]
ret[game_id].append(
(cur_turn, best_action, other_actions)
)
else:
ret[game_id] = None
return ret
def render_main_site(env: jinja2.Environment, out_dir: Path):
rating_lists = get_rating_lists()
streak_lists = get_streak_list()
@ -842,7 +949,8 @@ def render_all():
render_main_site(env, out_dir)
render_player_pages(env, out_dir)
render_variant_pages(env, out_dir)
render_game_pages(env, out_dir)
if __name__ == "__main__":
render_all()
render_all()

View file

@ -93,7 +93,7 @@ def analyze_replay(instance: hanab_game.HanabiInstance, actions: List[hanab_game
def analyze_game_and_store_stats(game_id: int):
logger.verbose("Analysing game {} for BDRs and lost crits".format(game_id))
instance, actions = games_db_interface.load_game(game_id)
instance, actions, _ = games_db_interface.load_game_parts(game_id)
analysis = analyze_replay(instance, actions)
cur = conn_manager.get_new_cursor()

73
templates/game.html Normal file
View file

@ -0,0 +1,73 @@
{% extends "layout.html" %}
{% block navbar %}
<nav class="navbar navbar-expand-lg navbar-light bg-light">
<div class="container">
<a class="navbar-brand" href="/">The Hanabi Pro Hunting League</a><a class="navbar-brand" href="#"><small class="text-muted">- Endgame Statistics for {{game_id}}</small></a>
<button class="navbar-toggler" type="button" data-toggle="collapse" data-target="#navbarNav" aria-controls="navbarNav" aria-expanded="false" aria-label="Toggle navigation">
<span class="navbar-toggler-icon"></span>
</button>
<div class="collapse navbar-collapse" id="navbarNav">
<ul class="navbar-nav ml-auto">
<li class="nav-item">
<a class="nav-link" id="Back" href="/">Back</a>
</li>
</ul>
</div>
</div>
</nav>
{% endblock %}
{% block content %}
<div class="tab-content" id="myTabContent">
<div class="tab-pane fade show active" id="overview">
<div class="container my-5">
<h3>
Statistics for game {{game_id}}
</h3>
<ul>
<li>
Replay: <a href="https://hanab.live/replay/{{game_id}}">https://hanab.live/replay/{{game_id}}</a>
</li>
<li>
Shared Replay: <a href="https://hanab.live/shared-replay/{{game_id}}">https://hanab.live/shared-replay/{{game_id}}</a>
</li>
</ul>
<h4>
Endgame Analysis table
</h4>
{% if data %}
<table class="endgame-table">
<tr>
<th>Turn</th>
<th>Action</th>
<th>Fractional Probability</th>
<th>Probability</th>
</tr>
{% for (turn, best_action, other_actions) in data %}
<tr>
<td rowspan="{{ other_actions|length + 1 }}"><a href="https://hanab.live/replay/{{game_id}}#{{turn}}">{{ turn }}</a></td>
<td>{% if best_action.marked %}<b>{% endif %}{{ best_action.description }}{% if best_action.marked %}</b>{% endif %}</td>
<td>{{ best_action.enumerator }}/{{ best_action.denominator }}</td>
<td>{{ best_action.win_rate }}%</td>
</tr>
{% for action in other_actions %}
<tr>
<td>{% if action.marked %}<b>{% endif %}{{ action.description }}{% if action.marked %}</b>{% endif %}</td>
<td>{{ action.enumerator }}/{{ action.denominator }}</td>
<td>{{ action.win_rate }}%</td>
</tr>
{% endfor %}
{% endfor %}
</table>
{% else %}
Currently, there is no endgame analysis available for this game. Since the computation is resource extensive,
this might take a while, also depending on how many other games have been played recently.
<br>
Come back later to check again.
{% endif %}
</div>
</div>
</div>
{% endblock %}

View file

@ -72,7 +72,7 @@ var table_{{div_id}} = new Tabulator("#table-{{div_id}}", {
layout: "fitDataStretch",
columns: [
{title: "Game", field: "game_id", formatter: "link", formatterParams:{
urlPrefix: "https://hanab.live/replay/",
urlPrefix: "/game/",
target:"_blank"
}},
{title: "Played", field: "datetime_finished", formatter: "datetime", formatterParams:{