Add match replay and JSONL export
This commit is contained in:
@@ -37,6 +37,7 @@ the workflow document for deciding where new documentation belongs.
|
||||
## Engine / Performance
|
||||
|
||||
- [Classic game port architecture](classic-port-notes.md) — Describes the standalone Cython classic engine as the stable rules layer for all consumers. (2026-05-08)
|
||||
- [Lost Cities match record v1](lost-cities-match-record-v1.md) — Defines the versioned JSONL format for complete hidden-state replay and post-game analysis. (2026-07-12)
|
||||
- [Fast engine optimization architecture](fast-engine-next-optimizations.md) — Prioritizes C-level APIs, contiguous allocation, and zero-copy extraction for high-throughput RL. (2026-05-08)
|
||||
- [Optimization sequencing](optimization_sequencing.md) — Orders runtime, traversal, model-scale, and inference optimizations to avoid invalidating experiments. (2026-05-07)
|
||||
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
# Lost Cities Match Record v1
|
||||
|
||||
Last verified: 2026-07-12
|
||||
|
||||
`coolrl.lost-cities.match.v1` is the canonical portable record for one complete
|
||||
or in-progress classic Lost Cities match. Files use UTF-8 JSON Lines (`.jsonl`)
|
||||
so metadata can be inspected without loading the full replay and steps can be
|
||||
streamed in order.
|
||||
|
||||
## Row 1: metadata
|
||||
|
||||
The first row has `type: "metadata"` and `format:
|
||||
"coolrl.lost-cities.match.v1"`. It records creation time, classic rules,
|
||||
initial seed, human seat, opponent identity, and completion status. Producers
|
||||
may add fields; v1 readers must ignore unknown metadata fields.
|
||||
|
||||
## Rows 2+: steps
|
||||
|
||||
Every remaining row has `type: "step"` and a contiguous zero-based `index`.
|
||||
Step 0 is the initial position and has null `actor`, `phase_before`, and
|
||||
`action_id`. Later steps contain:
|
||||
|
||||
- `actor`: player that took the action (`0` or `1`).
|
||||
- `phase_before`: `card` or `draw`.
|
||||
- `action_id`: the classic engine unified action ID.
|
||||
- `state`: complete `GameState.to_snapshot()` output after the action.
|
||||
- `public_hands`: cards known to be in each player's hand because they were
|
||||
drawn from a discard pile. Entries use `{color, value, count}`; wagers have
|
||||
value 0 and numbered cards use their printed value.
|
||||
|
||||
Full state snapshots intentionally include hidden hands and deck order. A match
|
||||
record is therefore suitable for post-game analysis and deterministic replay,
|
||||
but must not be exposed to a player during a live match.
|
||||
|
||||
## Compatibility
|
||||
|
||||
Readers must reject an unknown `format` value rather than guessing. New
|
||||
optional fields may be added within v1. Any incompatible change to action IDs,
|
||||
state encoding, or required row semantics requires `v2`.
|
||||
|
||||
The reference reader, writer, and validator are in
|
||||
`src/coolrl_lost_cities/games/classic/match_record.py`.
|
||||
@@ -0,0 +1,146 @@
|
||||
"""Versioned, replayable Lost Cities match records."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections import Counter
|
||||
from collections.abc import Iterable, Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .game import GameState
|
||||
|
||||
FORMAT_NAME = "coolrl.lost-cities.match"
|
||||
FORMAT_VERSION = 1
|
||||
FORMAT_ID = f"{FORMAT_NAME}.v{FORMAT_VERSION}"
|
||||
|
||||
|
||||
def encode_public_hands(
|
||||
counts: Sequence[Mapping[tuple[int, int], int]],
|
||||
) -> list[list[dict[str, int]]]:
|
||||
return [
|
||||
[
|
||||
{"color": color, "value": value, "count": count}
|
||||
for (color, value), count in sorted(player.items())
|
||||
if count > 0
|
||||
]
|
||||
for player in counts
|
||||
]
|
||||
|
||||
|
||||
def decode_public_hands(
|
||||
payload: Sequence[Sequence[Mapping[str, Any]]],
|
||||
) -> list[Counter[tuple[int, int]]]:
|
||||
if len(payload) != 2:
|
||||
raise ValueError("public_hands must contain exactly two players")
|
||||
result: list[Counter[tuple[int, int]]] = []
|
||||
for cards in payload:
|
||||
counter: Counter[tuple[int, int]] = Counter()
|
||||
for card in cards:
|
||||
color = int(card["color"])
|
||||
value = int(card["value"])
|
||||
count = int(card["count"])
|
||||
if color < 0 or value < 0 or count <= 0:
|
||||
raise ValueError("invalid public hand entry")
|
||||
counter[(color, value)] += count
|
||||
result.append(counter)
|
||||
return result
|
||||
|
||||
|
||||
def make_step(
|
||||
*,
|
||||
index: int,
|
||||
state: GameState,
|
||||
public_hands: Sequence[Mapping[tuple[int, int], int]],
|
||||
actor: int | None,
|
||||
phase_before: str | None,
|
||||
action_id: int | None,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "step",
|
||||
"index": index,
|
||||
"actor": actor,
|
||||
"phase_before": phase_before,
|
||||
"action_id": action_id,
|
||||
"state": state.to_snapshot(),
|
||||
"public_hands": encode_public_hands(public_hands),
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatchRecord:
|
||||
metadata: dict[str, Any]
|
||||
steps: list[dict[str, Any]]
|
||||
|
||||
def validate(self) -> None:
|
||||
if self.metadata.get("type") != "metadata":
|
||||
raise ValueError("first record must be metadata")
|
||||
if self.metadata.get("format") != FORMAT_ID:
|
||||
raise ValueError(f"unsupported match format: {self.metadata.get('format')!r}")
|
||||
if not self.steps:
|
||||
raise ValueError("match record must contain at least the initial step")
|
||||
for expected, step in enumerate(self.steps):
|
||||
if step.get("type") != "step" or step.get("index") != expected:
|
||||
raise ValueError("steps must be contiguous and zero-indexed")
|
||||
state = step.get("state")
|
||||
if not isinstance(state, dict):
|
||||
raise ValueError(f"step {expected} has no state snapshot")
|
||||
GameState.from_snapshot(state)
|
||||
decode_public_hands(step.get("public_hands", []))
|
||||
if expected == 0 and any(
|
||||
step.get(key) is not None for key in ("actor", "phase_before", "action_id")
|
||||
):
|
||||
raise ValueError("initial step cannot contain an action")
|
||||
if expected > 0:
|
||||
previous = GameState.from_snapshot(self.steps[expected - 1]["state"])
|
||||
if step.get("actor") != previous.current_player:
|
||||
raise ValueError(f"step {expected} actor does not match previous state")
|
||||
if step.get("phase_before") != previous.phase:
|
||||
raise ValueError(f"step {expected} phase does not match previous state")
|
||||
try:
|
||||
previous.apply_unified_action(int(step["action_id"]))
|
||||
except (KeyError, TypeError, ValueError) as exc:
|
||||
raise ValueError(f"step {expected} has an invalid action") from exc
|
||||
if previous.to_snapshot() != state:
|
||||
raise ValueError(f"step {expected} state does not follow its action")
|
||||
|
||||
def write_jsonl(self, path: str | Path) -> Path:
|
||||
self.validate()
|
||||
destination = Path(path)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
with destination.open("w", encoding="utf-8") as handle:
|
||||
for row in (self.metadata, *self.steps):
|
||||
handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n")
|
||||
return destination
|
||||
|
||||
@classmethod
|
||||
def read_jsonl(cls, path: str | Path) -> MatchRecord:
|
||||
with Path(path).open(encoding="utf-8") as handle:
|
||||
rows = [json.loads(line) for line in handle if line.strip()]
|
||||
if not rows:
|
||||
raise ValueError("empty match record")
|
||||
record = cls(metadata=rows[0], steps=rows[1:])
|
||||
record.validate()
|
||||
return record
|
||||
|
||||
|
||||
def record_from_rows(rows: Iterable[dict[str, Any]]) -> MatchRecord:
|
||||
materialized = list(rows)
|
||||
if not materialized:
|
||||
raise ValueError("empty match record")
|
||||
record = MatchRecord(materialized[0], materialized[1:])
|
||||
record.validate()
|
||||
return record
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FORMAT_ID",
|
||||
"FORMAT_NAME",
|
||||
"FORMAT_VERSION",
|
||||
"MatchRecord",
|
||||
"decode_public_hands",
|
||||
"encode_public_hands",
|
||||
"make_step",
|
||||
"record_from_rows",
|
||||
]
|
||||
@@ -27,6 +27,7 @@ import pygame
|
||||
|
||||
from .bots import build_bot
|
||||
from .game import Card, GameState, classic_config, score_expedition
|
||||
from .match_record import FORMAT_ID, MatchRecord, decode_public_hands, make_step
|
||||
|
||||
LOGGER = logging.getLogger("coolrl_lost_cities.games.classic.pygame_table")
|
||||
|
||||
@@ -549,7 +550,9 @@ class Layout:
|
||||
col_gap = int(self.board_w * 0.52)
|
||||
self.col_w = self.board_w + col_gap
|
||||
total = self.col_w * n_colors
|
||||
cx = (w - deck_zone) // 2
|
||||
# The five expedition lanes are the table's visual anchor. Center them
|
||||
# on the window; the deck occupies independent space on the far right.
|
||||
cx = w // 2
|
||||
self.col_x = [cx - total // 2 + self.col_w * i + col_gap // 2 for i in range(n_colors)]
|
||||
self.discard_cy = (self.board_top + self.board_bottom) // 2
|
||||
self.deck_rect = pygame.Rect(0, 0, self.board_w, self.board_h)
|
||||
@@ -711,6 +714,7 @@ class TableApp:
|
||||
height: int = 1000,
|
||||
offline: bool = False,
|
||||
headless: bool = False,
|
||||
export_dir: str | Path = "exports",
|
||||
) -> None:
|
||||
if headless:
|
||||
import os
|
||||
@@ -726,6 +730,7 @@ class TableApp:
|
||||
|
||||
self.rng = random.Random(seed)
|
||||
self.seed = seed
|
||||
self.export_dir = Path(export_dir)
|
||||
self.config = classic_config()
|
||||
self.min_rank = self.config.min_rank
|
||||
self.opponent = Opponent(
|
||||
@@ -751,9 +756,15 @@ class TableApp:
|
||||
self.ai_min_reveal_at = 0
|
||||
self.prompt_text = ""
|
||||
self.prompt_since = 0
|
||||
self.end_overlay_at: int | None = None
|
||||
self.end_backdrop: pygame.Surface | None = None
|
||||
self.play_again_rect: pygame.Rect | None = None
|
||||
self.menu_open = False
|
||||
self.menu_button_rect = pygame.Rect(0, 0, 1, 1)
|
||||
self.menu_items: list[tuple[str, pygame.Rect]] = []
|
||||
self.review_index: int | None = None
|
||||
self.match_metadata: dict[str, Any] = {}
|
||||
self.match_steps: list[dict[str, Any]] = []
|
||||
self.timeline_cursor = 0
|
||||
self.toast_text = ""
|
||||
self.toast_until = 0
|
||||
self.shake_until = 0
|
||||
self.shake_rect: pygame.Rect | None = None
|
||||
self.running = False
|
||||
@@ -786,6 +797,9 @@ class TableApp:
|
||||
self.history = []
|
||||
self.future = []
|
||||
self.public_hand_counts = [Counter(), Counter()]
|
||||
self.review_index = None
|
||||
self.menu_open = False
|
||||
self.toast_text = ""
|
||||
|
||||
n = self.config.n_colors
|
||||
self.sprites = []
|
||||
@@ -794,10 +808,33 @@ class TableApp:
|
||||
self.discard_zones = [[] for _ in range(n)]
|
||||
self.selected = None
|
||||
self.hovered = None
|
||||
self.end_overlay_at = None
|
||||
self.end_backdrop = None
|
||||
self.ai_next_at = 0
|
||||
|
||||
self.match_metadata = {
|
||||
"type": "metadata",
|
||||
"format": FORMAT_ID,
|
||||
"created_at": datetime.now().astimezone().isoformat(),
|
||||
"complete": False,
|
||||
"seed": game_seed,
|
||||
"human_seat": self.human_seat,
|
||||
"opponent": {
|
||||
"kind": "heuristic" if self.opponent.offline else "jax_ppo",
|
||||
"checkpoint": (str(self.opponent.checkpoint) if self.opponent.checkpoint else None),
|
||||
},
|
||||
"config": self.config.to_snapshot(),
|
||||
}
|
||||
self.match_steps = [
|
||||
make_step(
|
||||
index=0,
|
||||
state=self.state,
|
||||
public_hands=self.public_hand_counts,
|
||||
actor=None,
|
||||
phase_before=None,
|
||||
action_id=None,
|
||||
)
|
||||
]
|
||||
self.timeline_cursor = 0
|
||||
|
||||
deck_pos = self.layout.deck_rect.topleft
|
||||
deal_order: list[tuple[int, Sprite]] = []
|
||||
for slot in range(self.config.hand_size):
|
||||
@@ -841,10 +878,13 @@ class TableApp:
|
||||
|
||||
def apply_unified(self, action_id: int) -> None:
|
||||
"""Apply an engine action and move the matching sprites."""
|
||||
if self.timeline_cursor < len(self.match_steps) - 1:
|
||||
self.match_steps = self.match_steps[: self.timeline_cursor + 1]
|
||||
self.history.append((self.state.clone(), self._copy_public_counts()))
|
||||
self.future.clear()
|
||||
state = self.state
|
||||
seat = state.current_player
|
||||
phase_before = state.phase
|
||||
is_human = seat == self.human_seat
|
||||
if state.phase == "card":
|
||||
slot, place = divmod(action_id, 2)
|
||||
@@ -888,8 +928,26 @@ class TableApp:
|
||||
if is_human:
|
||||
self._sort_human_hand()
|
||||
self.selected = None
|
||||
if state.terminal and self.end_overlay_at is None:
|
||||
self.end_overlay_at = pygame.time.get_ticks() + 1300
|
||||
self.match_steps.append(
|
||||
make_step(
|
||||
index=len(self.match_steps),
|
||||
state=state,
|
||||
public_hands=self.public_hand_counts,
|
||||
actor=seat,
|
||||
phase_before=phase_before,
|
||||
action_id=action_id,
|
||||
)
|
||||
)
|
||||
self.timeline_cursor += 1
|
||||
if state.terminal and not self.match_metadata.get("complete"):
|
||||
self.match_metadata.update(
|
||||
{
|
||||
"complete": True,
|
||||
"completed_at": datetime.now().astimezone().isoformat(),
|
||||
"scores": [state.total_score(0), state.total_score(1)],
|
||||
}
|
||||
)
|
||||
self.enter_review(len(self.match_steps) - 1)
|
||||
|
||||
# -- undo / redo -------------------------------------------------------------
|
||||
|
||||
@@ -905,28 +963,30 @@ class TableApp:
|
||||
def undo(self) -> None:
|
||||
"""Step back to the most recent point the human had agency, skipping
|
||||
over the rival's automatic moves in between."""
|
||||
if not self.history:
|
||||
if self.review_index is not None or not self.history:
|
||||
return
|
||||
while self.history:
|
||||
self.future.append((self.state.clone(), self._copy_public_counts()))
|
||||
self.state, counts = self.history.pop()
|
||||
self.public_hand_counts = [Counter(values) for values in counts]
|
||||
self.timeline_cursor = max(0, self.timeline_cursor - 1)
|
||||
if self._at_human_decision():
|
||||
break
|
||||
self._sync_view_to_state()
|
||||
|
||||
def redo(self) -> None:
|
||||
if not self.future:
|
||||
if self.review_index is not None or not self.future:
|
||||
return
|
||||
while self.future:
|
||||
self.history.append((self.state.clone(), self._copy_public_counts()))
|
||||
self.state, counts = self.future.pop()
|
||||
self.public_hand_counts = [Counter(values) for values in counts]
|
||||
self.timeline_cursor = min(len(self.match_steps) - 1, self.timeline_cursor + 1)
|
||||
if self._at_human_decision():
|
||||
break
|
||||
self._sync_view_to_state()
|
||||
|
||||
def _sync_view_to_state(self) -> None:
|
||||
def _sync_view_to_state(self, *, reveal_all: bool = False) -> None:
|
||||
"""Rebuild every sprite/zone from self.state — used after an undo/redo
|
||||
jump, since animating backward through discarded history isn't sound."""
|
||||
now = pygame.time.get_ticks()
|
||||
@@ -940,7 +1000,7 @@ class TableApp:
|
||||
|
||||
deck_pos = self.layout.deck_rect.topleft
|
||||
for seat in range(2):
|
||||
face_up = seat == self.human_seat
|
||||
face_up = seat == self.human_seat or reveal_all
|
||||
public_remaining = Counter(self.public_hand_counts[seat])
|
||||
for card in self.state.hands[seat]:
|
||||
face = card_key(card, self.min_rank)
|
||||
@@ -982,8 +1042,40 @@ class TableApp:
|
||||
self.ai_next_at = 0
|
||||
self.ai_min_reveal_at = 0
|
||||
self.shake_until = 0
|
||||
self.end_backdrop = None
|
||||
self.end_overlay_at = now if self.state.terminal else None
|
||||
|
||||
# -- replay / export -------------------------------------------------------
|
||||
|
||||
def enter_review(self, index: int | None = None) -> None:
|
||||
if not self.match_steps or not self.match_metadata.get("complete"):
|
||||
return
|
||||
target = len(self.match_steps) - 1 if index is None else index
|
||||
target = max(0, min(target, len(self.match_steps) - 1))
|
||||
step_row = self.match_steps[target]
|
||||
self.state = GameState.from_snapshot(step_row["state"])
|
||||
self.public_hand_counts = decode_public_hands(step_row["public_hands"])
|
||||
self.review_index = target
|
||||
self._sync_view_to_state(reveal_all=True)
|
||||
|
||||
def review_step(self, delta: int) -> None:
|
||||
if self.review_index is None:
|
||||
self.enter_review()
|
||||
return
|
||||
self.enter_review(self.review_index + delta)
|
||||
|
||||
def current_match_record(self) -> MatchRecord:
|
||||
return MatchRecord(
|
||||
dict(self.match_metadata),
|
||||
list(self.match_steps[: self.timeline_cursor + 1]),
|
||||
)
|
||||
|
||||
def export_match(self) -> Path:
|
||||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
||||
path = self.export_dir / f"lost-cities-match-{stamp}.jsonl"
|
||||
written = self.current_match_record().write_jsonl(path)
|
||||
self.toast_text = f"EXPORTED {written}"
|
||||
self.toast_until = pygame.time.get_ticks() + 4000
|
||||
LOGGER.info("대국 기록 내보내기: %s", written)
|
||||
return written
|
||||
|
||||
# -- helpers ---------------------------------------------------------------
|
||||
|
||||
@@ -992,7 +1084,11 @@ class TableApp:
|
||||
return 1 - self.human_seat
|
||||
|
||||
def human_turn(self) -> bool:
|
||||
return not self.state.terminal and self.state.current_player == self.human_seat
|
||||
return (
|
||||
self.review_index is None
|
||||
and not self.state.terminal
|
||||
and self.state.current_player == self.human_seat
|
||||
)
|
||||
|
||||
def _legal(self, action_id: int) -> bool:
|
||||
mask = self.state.unified_legal_mask()
|
||||
@@ -1036,7 +1132,6 @@ class TableApp:
|
||||
self.screen = pygame.display.set_mode((w, h), pygame.RESIZABLE)
|
||||
self.layout = Layout(w, h, self.config.n_colors)
|
||||
self.background = self._make_background(w, h)
|
||||
self.end_backdrop = None
|
||||
elif event.type == pygame.KEYDOWN:
|
||||
mod = getattr(event, "mod", 0)
|
||||
ctrl = bool(mod & pygame.KMOD_CTRL)
|
||||
@@ -1044,9 +1139,22 @@ class TableApp:
|
||||
if event.key == pygame.K_n:
|
||||
self.new_game()
|
||||
elif event.key == pygame.K_ESCAPE:
|
||||
if self.menu_open:
|
||||
self.menu_open = False
|
||||
else:
|
||||
self.selected = None
|
||||
elif event.key in (pygame.K_RETURN, pygame.K_SPACE) and self.state.terminal:
|
||||
self.new_game()
|
||||
elif event.key == pygame.K_m:
|
||||
self.menu_open = not self.menu_open
|
||||
elif event.key in (pygame.K_LEFT, pygame.K_PAGEUP):
|
||||
self.review_step(-1)
|
||||
elif event.key in (pygame.K_RIGHT, pygame.K_PAGEDOWN):
|
||||
self.review_step(1)
|
||||
elif event.key == pygame.K_HOME:
|
||||
self.enter_review(0)
|
||||
elif event.key == pygame.K_END:
|
||||
self.enter_review(len(self.match_steps) - 1)
|
||||
elif event.key == pygame.K_e and self.review_index is not None:
|
||||
self.export_match()
|
||||
elif event.key == pygame.K_F12:
|
||||
self.save_screenshot()
|
||||
elif ctrl and event.key == pygame.K_z and shift:
|
||||
@@ -1062,13 +1170,9 @@ class TableApp:
|
||||
|
||||
def on_click(self, pos: tuple[int, int]) -> None:
|
||||
now = pygame.time.get_ticks()
|
||||
if self.state.terminal:
|
||||
if (
|
||||
self.end_overlay_at is not None
|
||||
and now >= self.end_overlay_at
|
||||
and (self.play_again_rect and self.play_again_rect.collidepoint(pos))
|
||||
):
|
||||
self.new_game()
|
||||
if self._handle_chrome_click(pos):
|
||||
return
|
||||
if self.review_index is not None or self.state.terminal:
|
||||
return
|
||||
if now < self.input_locked_until or not self.human_turn():
|
||||
return
|
||||
@@ -1077,6 +1181,33 @@ class TableApp:
|
||||
else:
|
||||
self._click_draw_phase(pos)
|
||||
|
||||
def _handle_chrome_click(self, pos: tuple[int, int]) -> bool:
|
||||
if self.menu_button_rect.collidepoint(pos):
|
||||
self.menu_open = not self.menu_open
|
||||
return True
|
||||
for action, rect in self.menu_items:
|
||||
if not rect.collidepoint(pos):
|
||||
continue
|
||||
if action == "review":
|
||||
self.enter_review()
|
||||
self.menu_open = False
|
||||
elif action == "export":
|
||||
self.export_match()
|
||||
self.menu_open = False
|
||||
elif action == "new":
|
||||
self.new_game()
|
||||
elif action == "close":
|
||||
self.running = False
|
||||
elif action == "previous":
|
||||
self.review_step(-1)
|
||||
elif action == "next":
|
||||
self.review_step(1)
|
||||
return True
|
||||
if self.menu_open:
|
||||
self.menu_open = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def _click_card_phase(self, pos: tuple[int, int]) -> None:
|
||||
hand = self.hand_zones[self.human_seat]
|
||||
for sprite in reversed(hand):
|
||||
@@ -1127,7 +1258,11 @@ class TableApp:
|
||||
for sprite in self.sprites:
|
||||
sprite.update(dt, now)
|
||||
self._update_hover()
|
||||
if not self.state.terminal and self.state.current_player == self.ai_seat:
|
||||
if (
|
||||
self.review_index is None
|
||||
and not self.state.terminal
|
||||
and self.state.current_player == self.ai_seat
|
||||
):
|
||||
self._update_ai(now)
|
||||
self._update_prompt(now)
|
||||
if pygame.display.get_active():
|
||||
@@ -1194,8 +1329,12 @@ class TableApp:
|
||||
|
||||
def _cursor_interactive(self) -> bool:
|
||||
pos = pygame.mouse.get_pos()
|
||||
if self.menu_button_rect.collidepoint(pos) or any(
|
||||
rect.collidepoint(pos) for _, rect in self.menu_items
|
||||
):
|
||||
return True
|
||||
if self.state.terminal:
|
||||
return bool(self.play_again_rect and self.play_again_rect.collidepoint(pos))
|
||||
return False
|
||||
if not self.human_turn():
|
||||
return False
|
||||
if self.hovered is not None:
|
||||
@@ -1216,7 +1355,7 @@ class TableApp:
|
||||
return False
|
||||
|
||||
def _update_prompt(self, now: int) -> None:
|
||||
if self.state.terminal:
|
||||
if self.review_index is not None or self.state.terminal:
|
||||
text = ""
|
||||
elif not self.human_turn():
|
||||
if self.opponent.load_error is not None:
|
||||
@@ -1289,11 +1428,9 @@ class TableApp:
|
||||
for sprite in self.sprites:
|
||||
if sprite.flying:
|
||||
sprite.draw(screen, self.art, sprite.width or self.layout.board_w)
|
||||
if self.state.terminal and self.end_overlay_at is not None:
|
||||
if now >= self.end_overlay_at:
|
||||
self._draw_end_overlay(now)
|
||||
else:
|
||||
self.play_again_rect = None
|
||||
self.menu_items = []
|
||||
self._draw_review_controls()
|
||||
self._draw_menu(now)
|
||||
|
||||
def _pulse(self, now: int, speed: float = 2.4) -> float:
|
||||
return 0.5 + 0.5 * math.sin(now / 1000.0 * speed * math.tau / 2)
|
||||
@@ -1490,7 +1627,13 @@ class TableApp:
|
||||
self._draw_plaque(
|
||||
pygame.Rect(lay.margin, 14, 236, 64),
|
||||
"THE RIVAL",
|
||||
self.opponent.label if self.opponent.ready else "loading model…",
|
||||
(
|
||||
"FINAL PPO"
|
||||
if self.opponent.ready and self.opponent.load_error is None
|
||||
else self.opponent.label
|
||||
if self.opponent.ready
|
||||
else "LOADING"
|
||||
),
|
||||
self.total_score(self.ai_seat),
|
||||
active=ai_active,
|
||||
thinking=ai_active and self.opponent.load_error is None,
|
||||
@@ -1499,33 +1642,75 @@ class TableApp:
|
||||
self._draw_plaque(
|
||||
pygame.Rect(lay.margin, lay.h - 78, 236, 64),
|
||||
"YOU",
|
||||
"expedition leader",
|
||||
"",
|
||||
self.total_score(self.human_seat),
|
||||
active=human_active,
|
||||
thinking=False,
|
||||
now=now,
|
||||
)
|
||||
self._draw_hints(lay)
|
||||
title_font = self.fonts.serif(20)
|
||||
title = render_tracked(title_font, "LOST CITIES", GOLD_DIM, 6)
|
||||
self.screen.blit(title, title.get_rect(topright=(lay.w - lay.margin, 20)))
|
||||
# The table itself carries the identity; keep chrome intentionally sparse.
|
||||
|
||||
def _draw_hints(self, lay: Layout) -> None:
|
||||
disabled = (72, 72, 74)
|
||||
parts = [
|
||||
("N NEW GAME", True),
|
||||
("CTRL+Y REDO", self.can_redo()),
|
||||
("CTRL+Z UNDO", self.can_undo()),
|
||||
]
|
||||
x = lay.w - lay.margin
|
||||
y = lay.h - 16
|
||||
for text, enabled in parts:
|
||||
surf = render_tracked(
|
||||
self.fonts.sans(11, bold=True), text, MUTED if enabled else disabled, 2
|
||||
def _draw_review_controls(self) -> None:
|
||||
if self.review_index is None:
|
||||
return
|
||||
lay = self.layout
|
||||
y = lay.prompt_y - 8
|
||||
label = f"STEP {self.review_index} / {len(self.match_steps) - 1}"
|
||||
label_surface = render_tracked(self.fonts.sans(12, bold=True), label, INK, 2)
|
||||
label_rect = label_surface.get_rect(center=(lay.w // 2, y))
|
||||
self.screen.blit(label_surface, label_rect)
|
||||
|
||||
prev_rect = pygame.Rect(label_rect.left - 66, y - 18, 44, 36)
|
||||
next_rect = pygame.Rect(label_rect.right + 22, y - 18, 44, 36)
|
||||
self.menu_items.extend([("previous", prev_rect), ("next", next_rect)])
|
||||
for text, rect, enabled in (
|
||||
("‹", prev_rect, self.review_index > 0),
|
||||
("›", next_rect, self.review_index < len(self.match_steps) - 1),
|
||||
):
|
||||
color = GOLD if enabled else (72, 72, 74)
|
||||
pygame.draw.rect(self.screen, color, rect, width=1, border_radius=9)
|
||||
glyph = self.fonts.serif(25).render(text, True, color)
|
||||
self.screen.blit(glyph, glyph.get_rect(center=rect.center))
|
||||
|
||||
def _draw_menu(self, now: int) -> None:
|
||||
lay = self.layout
|
||||
self.menu_button_rect = pygame.Rect(lay.w - lay.margin - 92, 16, 92, 38)
|
||||
pygame.draw.rect(self.screen, (9, 17, 14), self.menu_button_rect, border_radius=10)
|
||||
pygame.draw.rect(self.screen, GOLD_DIM, self.menu_button_rect, width=1, border_radius=10)
|
||||
menu_text = render_tracked(self.fonts.sans(11, bold=True), "MENU", INK, 2)
|
||||
self.screen.blit(menu_text, menu_text.get_rect(center=self.menu_button_rect.center))
|
||||
|
||||
if self.toast_text and now < self.toast_until:
|
||||
toast = self.fonts.sans(11).render(self.toast_text, True, GOLD)
|
||||
self.screen.blit(
|
||||
toast,
|
||||
toast.get_rect(
|
||||
topright=(self.menu_button_rect.right, self.menu_button_rect.bottom + 8)
|
||||
),
|
||||
)
|
||||
rect = surf.get_rect(bottomright=(x, y))
|
||||
self.screen.blit(surf, rect)
|
||||
x = rect.left - 22
|
||||
if not self.menu_open:
|
||||
return
|
||||
|
||||
labels: list[tuple[str, str]] = []
|
||||
if self.match_metadata.get("complete"):
|
||||
labels.append(("review", "REVIEW MATCH"))
|
||||
labels.extend(
|
||||
[
|
||||
("export", "EXPORT JSONL"),
|
||||
("new", "NEW GAME"),
|
||||
("close", "QUIT"),
|
||||
]
|
||||
)
|
||||
panel = pygame.Rect(self.menu_button_rect.right - 238, 62, 238, 18 + 48 * len(labels))
|
||||
pygame.draw.rect(self.screen, (8, 16, 13), panel, border_radius=12)
|
||||
pygame.draw.rect(self.screen, GOLD_DIM, panel, width=1, border_radius=12)
|
||||
for index, (action, text) in enumerate(labels):
|
||||
rect = pygame.Rect(panel.x + 10, panel.y + 9 + index * 48, panel.w - 20, 38)
|
||||
self.menu_items.append((action, rect))
|
||||
if rect.collidepoint(pygame.mouse.get_pos()):
|
||||
pygame.draw.rect(self.screen, (*GOLD, 35), rect, border_radius=8)
|
||||
label = render_tracked(self.fonts.sans(11, bold=True), text, INK, 2)
|
||||
self.screen.blit(label, label.get_rect(midleft=(rect.x + 12, rect.centery)))
|
||||
|
||||
def _draw_plaque(
|
||||
self,
|
||||
@@ -1552,6 +1737,7 @@ class TableApp:
|
||||
dots = "·" * (1 + (now // 350) % 3)
|
||||
sub_text = f"{sub} {dots}"
|
||||
sub_surf = self.fonts.sans(11).render(sub_text, True, MUTED)
|
||||
if sub_text:
|
||||
self.screen.blit(sub_surf, (rect.x + 16, rect.y + 34))
|
||||
score_surf = self.fonts.serif(26).render(str(score), True, GOLD if active else INK)
|
||||
self.screen.blit(score_surf, score_surf.get_rect(midright=(rect.right - 16, rect.centery)))
|
||||
@@ -1571,91 +1757,6 @@ class TableApp:
|
||||
surf.set_alpha(alpha)
|
||||
self.screen.blit(surf, surf.get_rect(center=(self.layout.w // 2, self.layout.prompt_y)))
|
||||
|
||||
# -- end overlay -------------------------------------------------------------
|
||||
|
||||
def _draw_end_overlay(self, now: int) -> None:
|
||||
lay = self.layout
|
||||
if self.end_backdrop is None or self.end_backdrop.get_size() != self.screen.get_size():
|
||||
snapshot = self.screen.copy()
|
||||
blurred = _blur(snapshot, 6)
|
||||
scrim = pygame.Surface(blurred.get_size(), pygame.SRCALPHA)
|
||||
scrim.fill((4, 9, 7, 175))
|
||||
blurred.blit(scrim, (0, 0))
|
||||
self.end_backdrop = blurred
|
||||
reveal = ease_out_cubic((now - self.end_overlay_at) / 420)
|
||||
self.screen.blit(self.end_backdrop, (0, 0))
|
||||
|
||||
me = self.total_score(self.human_seat)
|
||||
rival = self.total_score(self.ai_seat)
|
||||
if me > rival:
|
||||
verdict, vcolor = "YOU PREVAIL", GOLD
|
||||
elif me < rival:
|
||||
verdict, vcolor = "THE RIVAL PREVAILS", (206, 130, 116)
|
||||
else:
|
||||
verdict, vcolor = "DEAD HEAT", INK
|
||||
|
||||
panel_w, panel_h = 620, 420
|
||||
panel = pygame.Rect(0, 0, panel_w, panel_h)
|
||||
panel.center = (lay.w // 2, int(lay.h * (0.52 - 0.03 * (1 - reveal))))
|
||||
surface = pygame.Surface(panel.size, pygame.SRCALPHA)
|
||||
pygame.draw.rect(surface, (10, 19, 16, 235), surface.get_rect(), border_radius=18)
|
||||
pygame.draw.rect(surface, (*GOLD_DIM, 220), surface.get_rect(), width=1, border_radius=18)
|
||||
|
||||
header = render_tracked(self.fonts.sans(13, bold=True), "EXPEDITION COMPLETE", MUTED, 4)
|
||||
surface.blit(header, header.get_rect(midtop=(panel_w // 2, 34)))
|
||||
title = render_tracked(self.fonts.serif(34), verdict, vcolor, 2)
|
||||
surface.blit(title, title.get_rect(midtop=(panel_w // 2, 62)))
|
||||
score_font = self.fonts.serif(26)
|
||||
dash = score_font.render("—", True, FAINT)
|
||||
surface.blit(dash, dash.get_rect(center=(panel_w // 2, 132)))
|
||||
mine_score = score_font.render(str(me), True, INK)
|
||||
rival_score = score_font.render(str(rival), True, INK)
|
||||
surface.blit(mine_score, mine_score.get_rect(center=(panel_w // 2 - 90, 132)))
|
||||
surface.blit(rival_score, rival_score.get_rect(center=(panel_w // 2 + 90, 132)))
|
||||
for text, dx in (("YOU", -90), ("RIVAL", 90)):
|
||||
label = render_tracked(self.fonts.sans(10, bold=True), text, FAINT, 2)
|
||||
surface.blit(label, label.get_rect(center=(panel_w // 2 + dx, 158)))
|
||||
|
||||
# per-suit breakdown
|
||||
table_y = 196
|
||||
col_step = 104
|
||||
x0 = panel_w // 2 - col_step * 2
|
||||
for color in range(self.config.n_colors):
|
||||
cx = x0 + color * col_step
|
||||
glyph = self.art.glyph(color, 30)
|
||||
surface.blit(glyph, glyph.get_rect(center=(cx, table_y)))
|
||||
mine = self.expedition_score(self.human_seat, color)
|
||||
theirs = self.expedition_score(self.ai_seat, color)
|
||||
mine_s = self.fonts.sans(15, bold=True).render(
|
||||
f"{mine:+d}" if self.state.expeditions[self.human_seat][color] else "—",
|
||||
True,
|
||||
INK if mine >= 0 else (222, 148, 134),
|
||||
)
|
||||
their_s = self.fonts.sans(13).render(
|
||||
f"{theirs:+d}" if self.state.expeditions[self.ai_seat][color] else "—", True, MUTED
|
||||
)
|
||||
surface.blit(mine_s, mine_s.get_rect(center=(cx, table_y + 36)))
|
||||
surface.blit(their_s, their_s.get_rect(center=(cx, table_y + 60)))
|
||||
row_label = self.fonts.sans(11).render("you", True, FAINT)
|
||||
surface.blit(row_label, row_label.get_rect(midright=(x0 - 64, table_y + 36)))
|
||||
row_label2 = self.fonts.sans(11).render("rival", True, FAINT)
|
||||
surface.blit(row_label2, row_label2.get_rect(midright=(x0 - 64, table_y + 60)))
|
||||
|
||||
button = pygame.Rect(0, 0, 220, 46)
|
||||
button.midtop = (panel_w // 2, 306)
|
||||
mouse = pygame.mouse.get_pos()
|
||||
hovered = button.move(panel.topleft).collidepoint(mouse)
|
||||
pygame.draw.rect(surface, (*GOLD, 40 if hovered else 0), button, border_radius=12)
|
||||
pygame.draw.rect(surface, GOLD, button, width=1, border_radius=12)
|
||||
btn_text = render_tracked(self.fonts.sans(14, bold=True), "PLAY AGAIN", GOLD, 3)
|
||||
surface.blit(btn_text, btn_text.get_rect(center=button.center))
|
||||
hint = self.fonts.sans(11).render("enter / space / n", True, FAINT)
|
||||
surface.blit(hint, hint.get_rect(midtop=(panel_w // 2, 362)))
|
||||
|
||||
surface.set_alpha(int(255 * reveal))
|
||||
self.screen.blit(surface, panel)
|
||||
self.play_again_rect = button.move(panel.topleft)
|
||||
|
||||
# -- misc ---------------------------------------------------------------------
|
||||
|
||||
def save_screenshot(self) -> Path:
|
||||
@@ -1687,6 +1788,11 @@ def build_argparser() -> argparse.ArgumentParser:
|
||||
parser.add_argument("--seed", type=int, default=None)
|
||||
parser.add_argument("--width", type=int, default=1600)
|
||||
parser.add_argument("--height", type=int, default=1000)
|
||||
parser.add_argument(
|
||||
"--export-dir",
|
||||
default="exports",
|
||||
help="Directory for versioned match JSONL exports",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--offline",
|
||||
action="store_true",
|
||||
@@ -1709,6 +1815,7 @@ def main(argv: list[str] | None = None) -> None:
|
||||
width=args.width,
|
||||
height=args.height,
|
||||
offline=args.offline,
|
||||
export_dir=args.export_dir,
|
||||
)
|
||||
app.run()
|
||||
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections import Counter
|
||||
|
||||
import pytest
|
||||
from coolrl_lost_cities.games.classic.game import GameState, classic_config
|
||||
|
||||
from coolrl_lost_cities.games.classic.match_record import (
|
||||
FORMAT_ID,
|
||||
MatchRecord,
|
||||
decode_public_hands,
|
||||
encode_public_hands,
|
||||
make_step,
|
||||
)
|
||||
|
||||
|
||||
def _record() -> MatchRecord:
|
||||
state = GameState.new_game(classic_config(), seed=17)
|
||||
public = [Counter(), Counter()]
|
||||
steps = [
|
||||
make_step(
|
||||
index=0,
|
||||
state=state,
|
||||
public_hands=public,
|
||||
actor=None,
|
||||
phase_before=None,
|
||||
action_id=None,
|
||||
)
|
||||
]
|
||||
before_player, before_phase = state.current_player, state.phase
|
||||
action = state.unified_legal_actions()[0]
|
||||
state.apply_unified_action(action)
|
||||
steps.append(
|
||||
make_step(
|
||||
index=1,
|
||||
state=state,
|
||||
public_hands=public,
|
||||
actor=before_player,
|
||||
phase_before=before_phase,
|
||||
action_id=action,
|
||||
)
|
||||
)
|
||||
return MatchRecord(
|
||||
metadata={"type": "metadata", "format": FORMAT_ID, "complete": False},
|
||||
steps=steps,
|
||||
)
|
||||
|
||||
|
||||
def test_match_record_jsonl_round_trip(tmp_path) -> None:
|
||||
record = _record()
|
||||
path = record.write_jsonl(tmp_path / "match.jsonl")
|
||||
|
||||
restored = MatchRecord.read_jsonl(path)
|
||||
|
||||
assert restored.metadata == record.metadata
|
||||
assert restored.steps == record.steps
|
||||
assert len(path.read_text().splitlines()) == 3
|
||||
|
||||
|
||||
def test_public_hand_encoding_is_stable_and_counted() -> None:
|
||||
counts = [Counter({(2, 7): 1, (0, 0): 2}), Counter()]
|
||||
encoded = encode_public_hands(counts)
|
||||
|
||||
assert encoded[0] == [
|
||||
{"color": 0, "value": 0, "count": 2},
|
||||
{"color": 2, "value": 7, "count": 1},
|
||||
]
|
||||
assert decode_public_hands(encoded) == counts
|
||||
|
||||
|
||||
def test_match_record_rejects_unknown_format(tmp_path) -> None:
|
||||
path = tmp_path / "future.jsonl"
|
||||
rows = [
|
||||
{"type": "metadata", "format": "coolrl.lost-cities.match.v2"},
|
||||
_record().steps[0],
|
||||
]
|
||||
path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
|
||||
|
||||
with pytest.raises(ValueError, match="unsupported match format"):
|
||||
MatchRecord.read_jsonl(path)
|
||||
|
||||
|
||||
def test_match_record_rejects_noncontiguous_steps() -> None:
|
||||
record = _record()
|
||||
record.steps[1]["index"] = 3
|
||||
|
||||
with pytest.raises(ValueError, match="contiguous"):
|
||||
record.validate()
|
||||
|
||||
|
||||
def test_match_record_rejects_state_not_produced_by_action() -> None:
|
||||
record = _record()
|
||||
record.steps[1]["action_id"] = record.steps[1]["action_id"] + 1
|
||||
|
||||
with pytest.raises(ValueError, match="state does not follow|invalid action"):
|
||||
record.validate()
|
||||
@@ -16,6 +16,7 @@ def test_argparser_defaults_to_final_candidate() -> None:
|
||||
assert args.width == 1600
|
||||
assert args.height == 1000
|
||||
assert args.offline is False
|
||||
assert args.export_dir == "exports"
|
||||
|
||||
|
||||
def test_argparser_accepts_overrides() -> None:
|
||||
@@ -56,6 +57,8 @@ def test_layout_scales_between_window_sizes() -> None:
|
||||
assert len(slots) == 8
|
||||
assert slots[0][0] >= 0
|
||||
assert slots[-1][0] + layout.hand_w <= layout.w
|
||||
lane_center = (layout.col_x[0] + layout.col_x[-1] + layout.board_w) / 2
|
||||
assert abs(lane_center - layout.w / 2) <= 1
|
||||
assert large.hand_w >= small.hand_w
|
||||
|
||||
|
||||
@@ -128,3 +131,82 @@ def test_offline_mode_explicitly_uses_heuristic() -> None:
|
||||
assert "offline" in opponent.label
|
||||
finally:
|
||||
opponent.shutdown()
|
||||
|
||||
|
||||
def test_finished_match_can_be_replayed_and_exported(tmp_path) -> None:
|
||||
app = pygame_table.TableApp(
|
||||
seed=29,
|
||||
offline=True,
|
||||
headless=True,
|
||||
export_dir=tmp_path,
|
||||
)
|
||||
try:
|
||||
while not app.state.terminal:
|
||||
app.apply_unified(app.state.unified_legal_actions()[0])
|
||||
|
||||
final_index = len(app.match_steps) - 1
|
||||
assert app.review_index == final_index
|
||||
assert all(sprite.face_up for sprite in app.hand_zones[app.ai_seat])
|
||||
|
||||
app.review_step(-1)
|
||||
assert app.review_index == final_index - 1
|
||||
app.enter_review(0)
|
||||
assert app.review_index == 0
|
||||
assert app.state.turn_count == 0
|
||||
assert all(sprite.face_up for sprite in app.hand_zones[app.ai_seat])
|
||||
|
||||
path = app.export_match()
|
||||
restored = pygame_table.MatchRecord.read_jsonl(path)
|
||||
assert restored.metadata["complete"] is True
|
||||
assert len(restored.steps) == len(app.match_steps)
|
||||
assert restored.steps[-1]["state"]["terminal"] is True
|
||||
finally:
|
||||
app.opponent.shutdown()
|
||||
pygame_table.pygame.quit()
|
||||
|
||||
|
||||
def test_undo_then_new_action_replaces_export_timeline() -> None:
|
||||
app = pygame_table.TableApp(seed=37, offline=True, headless=True)
|
||||
try:
|
||||
first = app.state.unified_legal_actions()[0]
|
||||
app.apply_unified(first)
|
||||
app.apply_unified(app.state.unified_legal_actions()[0])
|
||||
assert app.timeline_cursor == 2
|
||||
|
||||
app.undo()
|
||||
assert app.timeline_cursor < len(app.match_steps) - 1
|
||||
replacement = app.state.unified_legal_actions()[-1]
|
||||
app.apply_unified(replacement)
|
||||
|
||||
record = app.current_match_record()
|
||||
record.validate()
|
||||
assert len(record.steps) == app.timeline_cursor + 1
|
||||
assert record.steps[-1]["action_id"] == replacement
|
||||
finally:
|
||||
app.opponent.shutdown()
|
||||
pygame_table.pygame.quit()
|
||||
|
||||
|
||||
def test_menu_exports_current_match(tmp_path) -> None:
|
||||
app = pygame_table.TableApp(
|
||||
seed=43,
|
||||
offline=True,
|
||||
headless=True,
|
||||
export_dir=tmp_path,
|
||||
)
|
||||
try:
|
||||
app.draw()
|
||||
app.on_click(app.menu_button_rect.center)
|
||||
assert app.menu_open is True
|
||||
app.draw()
|
||||
export_rect = next(rect for action, rect in app.menu_items if action == "export")
|
||||
app.on_click(export_rect.center)
|
||||
|
||||
exported = list(tmp_path.glob("lost-cities-match-*.jsonl"))
|
||||
assert len(exported) == 1
|
||||
restored = pygame_table.MatchRecord.read_jsonl(exported[0])
|
||||
assert restored.metadata["complete"] is False
|
||||
assert len(restored.steps) == 1
|
||||
finally:
|
||||
app.opponent.shutdown()
|
||||
pygame_table.pygame.quit()
|
||||
|
||||
Reference in New Issue
Block a user