평가 스키마와 게임 undo 기반 정리

This commit is contained in:
2026-05-06 21:28:15 +09:00
parent 1843d93df2
commit 771c0ec821
5 changed files with 276 additions and 13 deletions
@@ -8,7 +8,9 @@ from .bots import (
from .env import LostCitiesEnv
from .evaluation import (
GameResult,
MatchEvalRecord,
MatchResult,
TimingResult,
evaluate_bot,
make_bot_factory,
play_game_for_evaluation,
@@ -29,8 +31,10 @@ __all__ = [
"LostCitiesBot",
"LostCitiesConfig",
"LostCitiesEnv",
"MatchEvalRecord",
"MatchResult",
"Snapshot",
"TimingResult",
"available_bot_names",
"build_bot",
"classic_config",
@@ -14,6 +14,7 @@ from .game import GameState, LostCitiesConfig, classic_config
from .interfaces import LostCitiesBot
BotFactory = Callable[[int | None], LostCitiesBot]
MATCH_EVAL_RECORD_TYPE = "lost_cities.classic.eval.match.v1"
@dataclass(frozen=True)
@@ -66,6 +67,72 @@ class MatchResult:
"steps_per_second": self.steps_per_second,
}
def result_dict(self) -> dict[str, float | int]:
return {
"games": self.games,
"wins0": self.wins0,
"wins1": self.wins1,
"draws": self.draws,
"win_rate0": self.win_rate0,
"win_rate1": self.win_rate1,
"avg_score0": self.avg_score0,
"avg_score1": self.avg_score1,
"avg_score_diff0": self.avg_score_diff0,
"avg_game_length": self.avg_game_length,
"max_step_timeouts": self.max_step_timeouts,
}
def timing(self) -> TimingResult:
return TimingResult(
elapsed_seconds=self.elapsed_seconds,
games_per_second=self.games_per_second,
steps_per_second=self.steps_per_second,
)
@dataclass(frozen=True)
class TimingResult:
elapsed_seconds: float
games_per_second: float
steps_per_second: float
def to_dict(self) -> dict[str, float]:
return {
"elapsed_seconds": self.elapsed_seconds,
"games_per_second": self.games_per_second,
"steps_per_second": self.steps_per_second,
}
@dataclass(frozen=True)
class MatchEvalRecord:
bot0: str
bot1: str
config: LostCitiesConfig
seed: int
alternate_seats: bool
max_steps: int
result: MatchResult
record_type: str = MATCH_EVAL_RECORD_TYPE
def to_dict(self) -> dict[str, Any]:
return {
"type": self.record_type,
"bots": {
"bot0": self.bot0,
"bot1": self.bot1,
},
"settings": {
"games": self.result.games,
"seed": self.seed,
"alternate_seats": self.alternate_seats,
"max_steps": self.max_steps,
},
"config": self.config.to_snapshot(),
"result": self.result.result_dict(),
"timing": self.result.timing().to_dict(),
}
def make_bot_factory(name: str) -> BotFactory:
canonical = _canonical_bot_name(name)
@@ -221,6 +288,7 @@ def main(argv: list[str] | None = None) -> None:
args = parser.parse_args(argv)
config = classic_config()
alternate_seats = not args.no_alternate_seats
result = play_match(
make_bot_factory(args.bot0),
make_bot_factory(args.bot1),
@@ -228,17 +296,20 @@ def main(argv: list[str] | None = None) -> None:
games=args.games,
seed=args.seed,
max_steps=args.max_steps,
alternate_seats=not args.no_alternate_seats,
alternate_seats=alternate_seats,
)
record = MatchEvalRecord(
bot0=args.bot0,
bot1=args.bot1,
config=config,
seed=args.seed,
alternate_seats=alternate_seats,
max_steps=args.max_steps,
result=result,
)
payload: dict[str, Any] = {
"bot0": args.bot0,
"bot1": args.bot1,
"benchmark": bool(args.benchmark),
**result.to_dict(),
}
if args.json:
print(json.dumps(payload, indent=2, sort_keys=True))
print(json.dumps(record.to_dict(), indent=2, sort_keys=True))
return
print(f"{args.bot0} vs {args.bot1}: {result.games} games")
+104 -4
View File
@@ -260,7 +260,6 @@ cdef class GameState:
for _ in range(config.hand_size):
for player in range(2):
state.hands[player].append(state.deck.pop())
state.sort_hands()
state.validate_invariants()
return state
@@ -550,6 +549,70 @@ cdef class GameState:
cpdef apply_unified_action(self, int action_id):
self.apply_action(self.from_unified_action(action_id))
cpdef object apply_action_with_undo(self, int action_id):
if self.terminal:
raise IllegalMoveError("game is already terminal")
cdef list mask = self.legal_mask()
if action_id < 0 or action_id >= len(mask) or not mask[action_id]:
raise IllegalMoveError(
f"illegal action {action_id} in phase {self.phase} "
f"for player {self.current_player}"
)
cdef object undo
if self.phase == "card":
undo = self._card_action_undo(action_id)
self._apply_card_action(action_id)
else:
undo = self._draw_action_undo(action_id)
self._apply_draw_action(action_id)
return undo
cpdef object apply_unified_action_with_undo(self, int action_id):
return self.apply_action_with_undo(self.from_unified_action(action_id))
cpdef undo_action(self, object undo):
cdef str phase = undo[0]
if phase == "card":
self._undo_card_action(undo)
return
if phase == "draw":
self._undo_draw_action(undo)
return
raise ValueError(f"invalid undo phase: {phase!r}")
cdef object _card_action_undo(self, int action_id):
cdef int slot = action_id // 2
cdef bint play = action_id % 2 == 0
cdef Card card = <Card>self.hands[self.current_player][slot]
return (
"card",
self.current_player,
action_id,
self.pending_discarded_color,
self.terminal,
slot,
play,
card,
)
cdef object _draw_action_undo(self, int action_id):
cdef Card card
cdef list source
if action_id == 0:
source = self.deck
else:
source = self.discards[action_id - 1]
card = <Card>source[len(source) - 1]
return (
"draw",
self.current_player,
action_id,
self.pending_discarded_color,
self.terminal,
self.turn_count,
card,
)
cdef void _apply_card_action(self, int action_id) except *:
cdef int slot = action_id // 2
cdef bint play = action_id % 2 == 0
@@ -581,7 +644,6 @@ cdef class GameState:
color = action_id - 1
card = <Card>self.discards[color].pop()
self.hands[self.current_player].append(card)
self.sort_hand(self.current_player)
self.pending_discarded_color = None
self.turn_count += 1
if len(self.deck) == 0:
@@ -590,6 +652,46 @@ cdef class GameState:
self.current_player = 1 - self.current_player
self.phase = "card"
cdef void _undo_card_action(self, object undo) except *:
cdef int player = <int>undo[1]
cdef object pending_before = undo[3]
cdef bint terminal_before = <bint>undo[4]
cdef int slot = <int>undo[5]
cdef bint play = <bint>undo[6]
cdef Card card = <Card>undo[7]
cdef Card moved
if play:
moved = <Card>self.expeditions[player][card.color].pop()
else:
moved = <Card>self.discards[card.color].pop()
if moved != card:
raise ValueError("undo card mismatch")
self.hands[player].insert(slot, card)
self.current_player = player
self.phase = "card"
self.pending_discarded_color = pending_before
self.terminal = terminal_before
cdef void _undo_draw_action(self, object undo) except *:
cdef int player = <int>undo[1]
cdef int action_id = <int>undo[2]
cdef object pending_before = undo[3]
cdef bint terminal_before = <bint>undo[4]
cdef int turn_count_before = <int>undo[5]
cdef Card card = <Card>undo[6]
cdef Card moved = <Card>self.hands[player].pop()
if moved != card:
raise ValueError("undo draw mismatch")
if action_id == 0:
self.deck.append(card)
else:
self.discards[action_id - 1].append(card)
self.current_player = player
self.phase = "draw"
self.pending_discarded_color = pending_before
self.turn_count = turn_count_before
self.terminal = terminal_before
cpdef int expedition_score(self, int player, int color):
return score_expedition(self.expeditions[player][color], self.config)
@@ -623,8 +725,6 @@ cdef class GameState:
for player, hand in enumerate(self.hands):
if len(hand) > self.config.hand_size:
raise ValueError(f"hand {player} exceeds hand_size")
if hand != sorted(hand, key=_card_sort_key):
raise ValueError(f"hand {player} is not sorted")
all_cards.extend(hand)
for player, expeditions in enumerate(self.expeditions):
@@ -40,6 +40,35 @@ def test_new_game_from_deck_uses_explicit_internal_deck_order() -> None:
state.validate_invariants()
def test_new_game_preserves_dealt_hand_order() -> None:
config = LostCitiesConfig(
n_colors=2,
n_ranks=3,
min_rank=1,
n_handshakes=0,
hand_size=2,
expedition_penalty=0,
bonus_threshold=99,
bonus_amount=0,
)
state = GameState.new_game_from_deck(
[
Card(0, 1),
Card(0, 2),
Card(0, 3),
Card(1, 1),
Card(1, 2),
Card(1, 3),
],
config,
)
assert state.hands[0] == [Card(1, 3), Card(1, 1)]
assert state.hands[1] == [Card(1, 2), Card(0, 3)]
assert state.deck == [Card(0, 1), Card(0, 2)]
state.validate_invariants()
def test_snapshot_roundtrip_preserves_json_state() -> None:
state = GameState.new_game(LostCitiesConfig(seed=5))
first_action = next(index for index, legal in enumerate(state.unified_legal_mask()) if legal)
@@ -110,6 +139,60 @@ def test_random_games_preserve_python_core_invariants() -> None:
state.validate_invariants()
def test_apply_action_with_undo_restores_every_legal_action() -> None:
config = LostCitiesConfig(
n_colors=3,
n_ranks=5,
min_rank=2,
n_handshakes=1,
hand_size=5,
)
for seed in range(32):
state = GameState.new_game(config, seed=seed)
rng = random.Random(seed ^ 0xA11CE)
steps = 0
while not state.terminal:
legal = [index for index, is_legal in enumerate(state.unified_legal_mask()) if is_legal]
for unified_action in legal:
candidate = state.clone()
before = candidate.to_snapshot()
undo = candidate.apply_unified_action_with_undo(unified_action)
candidate.undo_action(undo)
assert candidate.to_snapshot() == before
candidate.validate_invariants()
state.apply_unified_action(rng.choice(legal))
steps += 1
assert steps < 1000
def test_apply_action_with_undo_matches_apply_action_result() -> None:
config = LostCitiesConfig(
n_colors=3,
n_ranks=5,
min_rank=2,
n_handshakes=1,
hand_size=5,
)
for seed in range(32):
state = GameState.new_game(config, seed=seed)
rng = random.Random(seed ^ 0xC0FFEE)
steps = 0
while not state.terminal:
legal = [index for index, is_legal in enumerate(state.unified_legal_mask()) if is_legal]
action = rng.choice(legal)
left = state.clone()
right = state.clone()
left.apply_unified_action(action)
right.apply_unified_action_with_undo(action)
assert right.to_snapshot() == left.to_snapshot()
state = left
steps += 1
assert steps < 1000
def test_same_seed_and_action_sequence_are_deterministic() -> None:
config = LostCitiesConfig(seed=1234)
left = GameState.new_game(config)
+6 -1
View File
@@ -5,7 +5,7 @@ from coolrl_lost_cities.games.classic import (
play_game_for_evaluation,
play_match,
)
from coolrl_lost_cities.games.classic.evaluation import main
from coolrl_lost_cities.games.classic.evaluation import MATCH_EVAL_RECORD_TYPE, main
def test_play_game_for_evaluation_finishes_small_match() -> None:
@@ -60,5 +60,10 @@ def test_evaluation_cli_smoke_json(capsys) -> None:
)
captured = capsys.readouterr()
assert f'"type": "{MATCH_EVAL_RECORD_TYPE}"' in captured.out
assert '"bots": {' in captured.out
assert '"settings": {' in captured.out
assert '"result": {' in captured.out
assert '"timing": {' in captured.out
assert '"games": 2' in captured.out
assert '"win_rate0"' in captured.out