forked from Hanabi/hanabi-league
Add functions to analyze endgames with external program
This commit is contained in:
parent
25cfd06f1b
commit
5c6c8a6b14
5 changed files with 191 additions and 6 deletions
|
@ -373,6 +373,10 @@ CREATE TABLE endgames (
|
|||
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 INTEGER NOT NULL CHECK (enumerator >= 0),
|
||||
|
|
|
@ -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
|
||||
ENDGAME_MEMORY_BYTES = 4 * 1024 * 1024 * 1024 # 4 GB of memory
|
||||
# In seconds
|
||||
ENDGAME_TIMEOUT = 10
|
||||
|
|
163
src/endgames.py
Normal file
163
src/endgames.py
Normal file
|
@ -0,0 +1,163 @@
|
|||
import json
|
||||
import subprocess
|
||||
import re
|
||||
import resource
|
||||
from typing import List, Dict, Tuple
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import platformdirs
|
||||
import psycopg2.extras
|
||||
|
||||
import hanabi.hanab_game
|
||||
import hanabi.constants
|
||||
import hanabi.live.compress
|
||||
|
||||
import constants
|
||||
import games_db_interface
|
||||
from database import conn_manager
|
||||
|
||||
|
||||
@dataclass
|
||||
class EndgameAction:
|
||||
turn: int
|
||||
action_type: hanabi.hanab_game.ActionType
|
||||
card: hanabi.hanab_game.DeckCard
|
||||
enumerator: int
|
||||
denominator: int
|
||||
|
||||
|
||||
def analyze_game_from_db(game_id: int, refresh_cache=False):
|
||||
# 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 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,
|
||||
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 [], 0
|
||||
# 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 = 2
|
||||
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
|
||||
|
||||
output = raw_output.decode('utf-8')
|
||||
|
||||
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]) -> 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))
|
||||
|
||||
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
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
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)
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
@ -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()
|
||||
|
|
Loading…
Reference in a new issue