125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Any, Dict, Tuple
|
|
import json
|
|
import os
|
|
|
|
from .config import STATE_PATH
|
|
|
|
|
|
class GameStatus(str):
|
|
RUNNING = "running"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
|
|
|
|
@dataclass
|
|
class Constraints:
|
|
mins: Dict[str, int] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class AttributeStats:
|
|
relative_frequencies: Dict[str, float] = field(default_factory=dict)
|
|
correlations: Dict[str, Dict[str, float]] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class GameRecord:
|
|
game_id: str
|
|
initial_raw: Dict[str, Any] | None = None
|
|
scenario: int | None = None
|
|
status: str = GameStatus.RUNNING
|
|
constraints: Constraints = field(default_factory=Constraints)
|
|
stats: AttributeStats = field(default_factory=AttributeStats)
|
|
admitted_count: int = 0
|
|
rejected_count: int = 0
|
|
next_person: Dict[str, Any] | None = None
|
|
tallies: Dict[str, int] = field(default_factory=dict)
|
|
policy_path: str | None = None # <— path to the user-provided policy script
|
|
|
|
def bump_tallies(self, attributes: Dict[str, bool]) -> None:
|
|
for k, v in attributes.items():
|
|
if v:
|
|
self.tallies[k] = self.tallies.get(k, 0) + 1
|
|
|
|
def raw(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: Dict[str, Any]) -> "GameRecord":
|
|
constraints = Constraints(**(d.get("constraints") or {}))
|
|
stats = AttributeStats(**(d.get("stats") or {}))
|
|
return cls(
|
|
game_id=d["game_id"],
|
|
initial_raw=d.get("initial_raw"),
|
|
scenario=d.get("scenario"),
|
|
status=d.get("status", GameStatus.RUNNING),
|
|
constraints=constraints,
|
|
stats=stats,
|
|
admitted_count=d.get("admitted_count", 0),
|
|
rejected_count=d.get("rejected_count", 0),
|
|
next_person=d.get("next_person"),
|
|
tallies=d.get("tallies", {}) or {},
|
|
policy_path=d.get("policy_path"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class GameState:
|
|
player_id: str | None = None
|
|
current_game_id: str | None = None
|
|
games: Dict[str, GameRecord] = field(default_factory=dict)
|
|
|
|
def save(self) -> None:
|
|
payload = {
|
|
"player_id": self.player_id,
|
|
"current_game_id": self.current_game_id,
|
|
"games": {gid: g.raw() for gid, g in self.games.items()},
|
|
}
|
|
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
|
|
with open(STATE_PATH, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, indent=2)
|
|
|
|
@classmethod
|
|
def load(cls) -> "GameState":
|
|
try:
|
|
with open(STATE_PATH, "r", encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
except FileNotFoundError:
|
|
return cls()
|
|
gs = cls()
|
|
gs.player_id = raw.get("player_id")
|
|
gs.current_game_id = raw.get("current_game_id")
|
|
for gid, graw in (raw.get("games") or {}).items():
|
|
gs.games[gid] = GameRecord.from_dict(graw)
|
|
return gs
|
|
|
|
def add_game(self, rec: GameRecord) -> None:
|
|
self.games[rec.game_id] = rec
|
|
self.current_game_id = rec.game_id
|
|
|
|
def current_game(self) -> GameRecord | None:
|
|
return self.games.get(self.current_game_id) if self.current_game_id else None
|
|
|
|
def set_current_game(self, game_id_prefix: str) -> bool:
|
|
g = self.get_game(game_id_prefix)
|
|
if g:
|
|
self.current_game_id = g.game_id
|
|
return True
|
|
return False
|
|
|
|
def get_game(self, game_id_prefix: str) -> GameRecord | None:
|
|
matches = [gid for gid in self.games if gid.startswith(game_id_prefix)]
|
|
if len(matches) == 1:
|
|
return self.games[matches[0]]
|
|
return None
|
|
|
|
def list_games(self) -> Dict[str, Tuple[str, int | None, int, int]]:
|
|
return {
|
|
gid: (g.status, g.scenario, g.admitted_count, g.rejected_count)
|
|
for gid, g in self.games.items()
|
|
}
|
|
|