eisterman
4998931cb0
- Remove variant B (Brown cap, +1 for higher beating lower) from rules, engine, bots and simulator; a higher belt beating a lower one gains 0. - Final Exam: the defender fights as a Black belt (RULES.md v0.7). Ties to the defender and best-of-3 stay as simulator switches (`--exam`). - Report stripe income per source as gains and losses. - Add ten one-sidedness measures sampled at turn ends (lead gap, winner margin and lead share, halfway leader, last lead change, contention, rank swaps, leader Kumite losses, Exam challengers, winner pace). - Plan: decisions D28-D33 and the detailed Phase 6. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
598 lines
22 KiB
Python
598 lines
22 KiB
Python
"""Rules engine: one game of Il Labirinto del Dojo.
|
|
|
|
Every choice a player makes is delegated to an ``Agent``; dice and deck order come from a
|
|
``Randomness`` source, so games are reproducible and tests can script every roll.
|
|
Rule references are to RULES.md.
|
|
"""
|
|
|
|
import itertools
|
|
import random
|
|
from collections import deque
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from enum import Enum, auto
|
|
from functools import cache
|
|
from typing import Protocol
|
|
|
|
from dojo.board import QUADRANTS, Board, Pos, Square, is_tunnel
|
|
from dojo.events import (
|
|
CardDrawn,
|
|
Event,
|
|
ExamTaken,
|
|
FightKind,
|
|
GameOver,
|
|
KumiteFought,
|
|
Moved,
|
|
SentToStart,
|
|
Source,
|
|
Teleported,
|
|
TrackChanged,
|
|
TurnSkipped,
|
|
TurnStarted,
|
|
)
|
|
from dojo.movement import Path, walks
|
|
from dojo.rules import BLACK, TECNICHE, Card, RulesConfig
|
|
|
|
MIN_PLAYERS = 2
|
|
MAX_PLAYERS = 8
|
|
SENSEI_OSSERVA_GAIN = 2
|
|
|
|
|
|
class Randomness(Protocol):
|
|
def d6(self) -> int: ...
|
|
|
|
def shuffle(self, cards: list[Card]) -> None: ...
|
|
|
|
|
|
class RandomSource:
|
|
def __init__(self, seed: int | None = None) -> None:
|
|
self._rng = random.Random(seed)
|
|
|
|
def d6(self) -> int:
|
|
return self._rng.randint(1, 6)
|
|
|
|
def shuffle(self, cards: list[Card]) -> None:
|
|
self._rng.shuffle(cards)
|
|
|
|
|
|
@dataclass
|
|
class Player:
|
|
id: int
|
|
corner: int
|
|
position: Pos
|
|
track: int
|
|
tecnica: Card | None = None
|
|
meditating: bool = False
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FightView:
|
|
"""What a fighter sees when deciding whether to play its Tecnica."""
|
|
|
|
attacker: int
|
|
defender: int
|
|
exam: bool
|
|
attacker_total: int
|
|
defender_total: int
|
|
played: tuple[tuple[int, Card], ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlayerView:
|
|
"""What everyone at the table knows about a player: whether they hold a Tecnica, not which."""
|
|
|
|
id: int
|
|
corner: int
|
|
position: Pos
|
|
track: int
|
|
holds_tecnica: bool
|
|
meditating: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GameView:
|
|
"""The game as player ``me`` sees it: public state plus its own Tecnica (RULES.md, D26)."""
|
|
|
|
board: Board
|
|
config: RulesConfig
|
|
round: int
|
|
me: int
|
|
tecnica: Card | None
|
|
players: tuple[PlayerView, ...]
|
|
|
|
def belt(self, player: int) -> int:
|
|
return self.config.belt(self.players[player].track)
|
|
|
|
|
|
class Agent(Protocol):
|
|
def place_pawn(self, view: GameView, squares: Sequence[Pos]) -> Pos: ...
|
|
|
|
def choose_path(self, view: GameView, paths: Sequence[Path]) -> Path: ...
|
|
|
|
def use_tunnel(self, view: GameView, exit: Pos) -> bool: ...
|
|
|
|
def choose_tunnel_destination(self, view: GameView, tunnels: Sequence[Pos]) -> Pos: ...
|
|
|
|
def choose_sfida_target(self, view: GameView, targets: Sequence[int]) -> int: ...
|
|
|
|
def choose_irimi_target(self, view: GameView, targets: Sequence[int]) -> int: ...
|
|
|
|
def choose_exam_defender(self, view: GameView, candidates: Sequence[int]) -> int: ...
|
|
|
|
def keep_tecnica(self, view: GameView, held: Card, drawn: Card) -> Card: ...
|
|
|
|
def play_tecnica(self, view: GameView, fight: FightView) -> bool: ...
|
|
|
|
|
|
class Step(Enum):
|
|
MOVED = auto()
|
|
WON = auto()
|
|
LOST = auto()
|
|
EXAM = auto()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FightResult:
|
|
attacker_won: bool
|
|
kind: FightKind
|
|
attacker_uke: bool
|
|
defender_uke: bool
|
|
|
|
|
|
@cache
|
|
def seat_corners(players: int) -> tuple[int, ...]:
|
|
"""Corner of each seat, in turn order. Corner-mates are never consecutive (cyclically)."""
|
|
if not MIN_PLAYERS <= players <= MAX_PLAYERS:
|
|
raise ValueError(f"{players} players: supported range is {MIN_PLAYERS}-{MAX_PLAYERS}")
|
|
corners = sorted([*range(min(players, QUADRANTS)), *range(max(0, players - QUADRANTS))])
|
|
for order in itertools.permutations(corners):
|
|
if all(order[i] != order[(i + 1) % players] for i in range(players)):
|
|
return order
|
|
raise AssertionError(f"no seating for {players} players")
|
|
|
|
|
|
class Game:
|
|
def __init__(
|
|
self,
|
|
board: Board,
|
|
config: RulesConfig,
|
|
agents: Sequence[Agent],
|
|
randomness: Randomness,
|
|
) -> None:
|
|
self.board = board
|
|
self.config = config
|
|
self.agents = tuple(agents)
|
|
self.events: list[Event] = []
|
|
self.round = 1
|
|
self.current = 0
|
|
self.winner: int | None = None
|
|
self._random = randomness
|
|
self._dojo = board.dojo
|
|
|
|
cards = [card for card, quantity in config.deck for _ in range(quantity)]
|
|
randomness.shuffle(cards)
|
|
self._deck: deque[Card] = deque(cards)
|
|
self._discard: list[Card] = []
|
|
|
|
self.players = [
|
|
Player(seat, corner, min(board.safe_room(corner)), config.start_position)
|
|
for seat, corner in enumerate(seat_corners(len(self.agents)))
|
|
]
|
|
for player in self.players:
|
|
self._place(player)
|
|
|
|
@property
|
|
def finished(self) -> bool:
|
|
return self.winner is not None or self.round > self.config.max_rounds
|
|
|
|
def belt(self, player: int) -> int:
|
|
return self.config.belt(self.players[player].track)
|
|
|
|
def view(self, player: int) -> GameView:
|
|
me = self.players[player]
|
|
return GameView(
|
|
board=self.board,
|
|
config=self.config,
|
|
round=self.round,
|
|
me=player,
|
|
tecnica=me.tecnica,
|
|
players=tuple(
|
|
PlayerView(p.id, p.corner, p.position, p.track, p.tecnica is not None, p.meditating)
|
|
for p in self.players
|
|
),
|
|
)
|
|
|
|
def run(self) -> int | None:
|
|
while not self.finished:
|
|
self.play_turn()
|
|
return self.winner
|
|
|
|
def play_turn(self) -> None:
|
|
if self.finished:
|
|
raise RuntimeError("the game is over")
|
|
player = self.players[self.current]
|
|
if player.meditating:
|
|
player.meditating = False
|
|
self._emit(TurnSkipped(self.round, player.id))
|
|
else:
|
|
roll = self._random.d6() + self._random.d6()
|
|
self._emit(TurnStarted(self.round, player.id, roll))
|
|
self._move(player, roll)
|
|
|
|
if self.winner is not None:
|
|
self._emit(GameOver(self.winner, self.round))
|
|
return
|
|
self.current = (self.current + 1) % len(self.players)
|
|
if self.current == 0:
|
|
self.round += 1
|
|
if self.round > self.config.max_rounds:
|
|
self._emit(GameOver(None, self.config.max_rounds))
|
|
|
|
# Movement -----------------------------------------------------------------------------
|
|
|
|
def _move(self, player: Player, steps: int) -> None:
|
|
origin = player.position
|
|
visited = {origin}
|
|
remaining = steps
|
|
while True:
|
|
paths = self._path_choices(player, remaining, visited)
|
|
if not paths:
|
|
break
|
|
path = self._pick(
|
|
self._agent(player).choose_path(self.view(player.id), paths), paths, "path"
|
|
)
|
|
walked, step = self._walk(player, path, visited)
|
|
if step in (Step.LOST, Step.EXAM):
|
|
return
|
|
if step is not Step.WON or walked == len(path):
|
|
break
|
|
# A won Kumite may have changed the belt (and Dojo access): choose again.
|
|
remaining -= walked
|
|
if self._settle(player, visited):
|
|
self._resolve_square(player, origin)
|
|
|
|
def _walk(self, player: Player, path: Path, visited: set[Pos]) -> tuple[int, Step]:
|
|
"""Walk ``path`` until it ends or a Kumite happens. Returns (squares walked, last step)."""
|
|
step = Step.MOVED
|
|
walked = 0
|
|
for cell in path:
|
|
visited.add(cell)
|
|
walked += 1
|
|
step = self._step(player, cell, previous=player.position)
|
|
if step is not Step.MOVED:
|
|
break
|
|
self._emit(Moved(player.id, path[:walked]))
|
|
return walked, step
|
|
|
|
def _path_choices(self, player: Player, steps: int, visited: set[Pos]) -> list[Path]:
|
|
"""Legal walks, one per distinct outcome (end square and opponents crossed in order)."""
|
|
brown = self.config.is_brown(player.track)
|
|
choices: dict[tuple[Pos, tuple[Pos, ...]], Path] = {}
|
|
for path in walks(self.board, player.position, steps, visited, can_enter_dojo=brown):
|
|
crossed = tuple(cell for cell in path if self._fightable_at(cell, player))
|
|
choices.setdefault((path[-1], crossed), path)
|
|
return list(choices.values())
|
|
|
|
def _step(self, player: Player, cell: Pos, previous: Pos) -> Step:
|
|
if cell in self._dojo:
|
|
self._exam(player, entrance=previous)
|
|
return Step.EXAM
|
|
defender = self._fightable_at(cell, player)
|
|
player.position = cell
|
|
if defender is None:
|
|
return Step.MOVED
|
|
result = self._kumite(player, defender)
|
|
self._apply_fight(player, defender, result, attacker_retreat=previous)
|
|
return Step.WON if result.attacker_won else Step.LOST
|
|
|
|
def _settle(self, player: Player, visited: set[Pos]) -> bool:
|
|
"""Meditazione rule: a move may not end on another pawn's square (outside Safe Rooms).
|
|
|
|
Take one extra step while sharing a square; stay if no step is possible.
|
|
Returns whether the final square's effect should be resolved.
|
|
"""
|
|
while self._shares_square(player):
|
|
brown = self.config.is_brown(player.track)
|
|
options = walks(self.board, player.position, 1, visited, can_enter_dojo=brown)
|
|
if not options:
|
|
return True
|
|
path = self._pick(
|
|
self._agent(player).choose_path(self.view(player.id), options), options, "path"
|
|
)
|
|
_, step = self._walk(player, path, visited)
|
|
if step in (Step.LOST, Step.EXAM):
|
|
return False
|
|
return True
|
|
|
|
# Kumite -------------------------------------------------------------------------------
|
|
|
|
def _kumite(self, attacker: Player, defender: Player, *, exam: bool = False) -> FightResult:
|
|
config = self.config
|
|
defender_belt = config.belt(defender.track)
|
|
if exam and config.exam_defender_black:
|
|
defender_belt = BLACK
|
|
belts = (config.belt(attacker.track), defender_belt)
|
|
bonus = min(abs(belts[0] - belts[1]), config.max_belt_bonus)
|
|
dice = [self._random.d6(), self._random.d6()]
|
|
extra = [bonus if belts[0] > belts[1] else 0, bonus if belts[1] > belts[0] else 0]
|
|
uke = [False, False]
|
|
played: list[tuple[int, Card]] = []
|
|
|
|
for side, fighter in enumerate((attacker, defender)):
|
|
card = fighter.tecnica
|
|
if card is None:
|
|
continue
|
|
view = FightView(
|
|
attacker=attacker.id,
|
|
defender=defender.id,
|
|
exam=exam,
|
|
attacker_total=dice[0] + extra[0],
|
|
defender_total=dice[1] + extra[1],
|
|
played=tuple(played),
|
|
)
|
|
if not self._agent(fighter).play_tecnica(self.view(fighter.id), view):
|
|
continue
|
|
fighter.tecnica = None
|
|
self._discard.append(card)
|
|
played.append((fighter.id, card))
|
|
match card:
|
|
case Card.KIAI:
|
|
extra[side] += config.kiai_bonus
|
|
case Card.KAESHI:
|
|
dice[side] = self._random.d6()
|
|
case Card.UKE:
|
|
uke[side] = True
|
|
|
|
totals = (dice[0] + extra[0], dice[1] + extra[1])
|
|
if exam and config.exam_ties_to_defender:
|
|
attacker_won = totals[0] > totals[1]
|
|
else:
|
|
attacker_won = totals[0] >= totals[1] # Senshu: ties go to the attacker.
|
|
winner_belt, loser_belt = belts if attacker_won else belts[::-1]
|
|
if winner_belt == loser_belt:
|
|
kind = FightKind.EQUAL
|
|
elif winner_belt < loser_belt:
|
|
kind = FightKind.UPSET
|
|
else:
|
|
kind = FightKind.EXPECTED
|
|
self._emit(
|
|
KumiteFought(
|
|
attacker=attacker.id,
|
|
defender=defender.id,
|
|
attacker_belt=belts[0],
|
|
defender_belt=belts[1],
|
|
attacker_total=totals[0],
|
|
defender_total=totals[1],
|
|
attacker_won=attacker_won,
|
|
kind=kind,
|
|
exam=exam,
|
|
cards=tuple(played),
|
|
)
|
|
)
|
|
return FightResult(attacker_won, kind, attacker_uke=uke[0], defender_uke=uke[1])
|
|
|
|
def _apply_fight(
|
|
self, attacker: Player, defender: Player, result: FightResult, attacker_retreat: Pos
|
|
) -> None:
|
|
if result.attacker_won:
|
|
winner, loser, loser_uke = attacker, defender, result.defender_uke
|
|
else:
|
|
winner, loser, loser_uke = defender, attacker, result.attacker_uke
|
|
|
|
stripes = self.config.stripes_per_belt
|
|
if result.kind is not FightKind.EXPECTED:
|
|
self._change_track(winner, stripes, Source.KUMITE)
|
|
self._change_track(loser, -stripes, Source.KUMITE)
|
|
|
|
if not loser_uke:
|
|
self._send_to_start(loser)
|
|
elif loser is attacker:
|
|
self._relocate(attacker, attacker_retreat, "uke")
|
|
|
|
# Final Exam ---------------------------------------------------------------------------
|
|
|
|
def _exam(self, challenger: Player, entrance: Pos) -> None:
|
|
challenger.position = entrance # No pawn ever remains inside the Dojo.
|
|
eligible = [p for p in self.players if p is not challenger and not p.meditating]
|
|
if not eligible:
|
|
return
|
|
top = max(p.track for p in eligible)
|
|
candidates = [p.id for p in eligible if p.track == top]
|
|
if len(candidates) == 1:
|
|
defender_id = candidates[0]
|
|
else:
|
|
choice = self._agent(challenger).choose_exam_defender(
|
|
self.view(challenger.id), candidates
|
|
)
|
|
defender_id = self._pick(choice, candidates, "exam defender")
|
|
|
|
needed = self.config.exam_wins_needed
|
|
won = lost = 0
|
|
uke = False
|
|
while won < needed and lost < needed:
|
|
result = self._kumite(challenger, self.players[defender_id], exam=True)
|
|
uke = uke or result.attacker_uke
|
|
if result.attacker_won:
|
|
won += 1
|
|
else:
|
|
lost += 1
|
|
passed = won == needed
|
|
self._emit(ExamTaken(challenger.id, defender_id, passed))
|
|
if passed:
|
|
self.winner = challenger.id
|
|
return
|
|
self._change_track(challenger, -self.config.stripes_per_belt, Source.EXAM)
|
|
if not uke:
|
|
self._send_to_start(challenger)
|
|
|
|
# Squares ------------------------------------------------------------------------------
|
|
|
|
def _resolve_square(self, player: Player, origin: Pos) -> None:
|
|
code = self.board.code(player.position)
|
|
match code:
|
|
case Square.ALLENAMENTO:
|
|
self._change_track(player, 1, Source.SQUARE)
|
|
case Square.FATICA:
|
|
self._change_track(player, -1, Source.SQUARE)
|
|
case Square.NEBBIA:
|
|
self._draw(player, origin)
|
|
case Square.PASSO_FALSO:
|
|
self._relocate(player, origin, "passo falso")
|
|
case Square.SFIDA:
|
|
self._sfida(player)
|
|
case _ if is_tunnel(code):
|
|
exit_square = self.board.tunnels[player.position]
|
|
if not self._has_unfightable_pawn(exit_square, player) and self._agent(
|
|
player
|
|
).use_tunnel(self.view(player.id), exit_square):
|
|
self._teleport(player, exit_square, "tunnel")
|
|
|
|
def _sfida(self, player: Player) -> None:
|
|
targets = self._targets(player)
|
|
if not targets:
|
|
return
|
|
choice = self._agent(player).choose_sfida_target(self.view(player.id), targets)
|
|
defender = self.players[self._pick(choice, targets, "Sfida target")]
|
|
result = self._kumite(player, defender)
|
|
self._apply_fight(player, defender, result, attacker_retreat=player.position)
|
|
|
|
def _teleport(self, player: Player, destination: Pos, reason: str) -> None:
|
|
"""Arrive on ``destination`` without walking; an opponent there means a Kumite."""
|
|
source = player.position
|
|
defender = self._fightable_at(destination, player)
|
|
self._relocate(player, destination, reason)
|
|
if defender is not None:
|
|
result = self._kumite(player, defender)
|
|
self._apply_fight(player, defender, result, attacker_retreat=source)
|
|
|
|
# Nebbia del Dojo ----------------------------------------------------------------------
|
|
|
|
def _draw(self, player: Player, origin: Pos) -> None:
|
|
if not self._deck and self._discard:
|
|
self._random.shuffle(self._discard)
|
|
self._deck.extend(self._discard)
|
|
self._discard.clear()
|
|
if not self._deck:
|
|
return
|
|
card = self._deck.popleft()
|
|
self._emit(CardDrawn(player.id, card))
|
|
if card in TECNICHE:
|
|
self._take_tecnica(player, card)
|
|
return
|
|
|
|
self._discard.append(card)
|
|
match card:
|
|
case Card.TSUGI_ASHI:
|
|
self._move(player, self._random.d6() + self._random.d6())
|
|
case Card.TUNNEL_SEGRETO:
|
|
tunnels = sorted(
|
|
p
|
|
for p in self.board.tunnels
|
|
if p != player.position and not self._has_unfightable_pawn(p, player)
|
|
)
|
|
if tunnels:
|
|
choice = self._agent(player).choose_tunnel_destination(
|
|
self.view(player.id), tunnels
|
|
)
|
|
self._teleport(player, self._pick(choice, tunnels, "tunnel"), "tunnel segreto")
|
|
case Card.IRIMI:
|
|
self._irimi(player)
|
|
case Card.SFIDA:
|
|
self._sfida(player)
|
|
case Card.SENSEI_OSSERVA:
|
|
lowest = min(p.track for p in self.players)
|
|
for p in [p for p in self.players if p.track == lowest]:
|
|
self._change_track(p, SENSEI_OSSERVA_GAIN, Source.CARD)
|
|
case Card.PESO_DELLA_CINTURA:
|
|
highest = max(p.track for p in self.players)
|
|
for p in [p for p in self.players if p.track == highest]:
|
|
self._change_track(p, -1, Source.CARD)
|
|
case Card.ALLENAMENTO_INTENSIVO:
|
|
self._change_track(player, 1, Source.CARD)
|
|
case Card.COLPO_PROIBITO:
|
|
self._change_track(player, -1, Source.CARD)
|
|
case Card.PASSO_FALSO:
|
|
self._relocate(player, origin, "passo falso")
|
|
case Card.MEDITAZIONE:
|
|
player.meditating = True
|
|
|
|
def _take_tecnica(self, player: Player, drawn: Card) -> None:
|
|
held = player.tecnica
|
|
if held is None:
|
|
player.tecnica = drawn
|
|
return
|
|
choice = self._agent(player).keep_tecnica(self.view(player.id), held, drawn)
|
|
kept = self._pick(choice, (held, drawn), "Tecnica")
|
|
player.tecnica = kept
|
|
self._discard.append(drawn if kept is held else held)
|
|
|
|
def _irimi(self, player: Player) -> None:
|
|
targets = self._targets(player)
|
|
if not targets:
|
|
return
|
|
choice = self._agent(player).choose_irimi_target(self.view(player.id), targets)
|
|
other = self.players[self._pick(choice, targets, "Irimi target")]
|
|
mine, theirs = player.position, other.position
|
|
self._relocate(player, theirs, "irimi")
|
|
self._relocate(other, mine, "irimi")
|
|
|
|
# State helpers ------------------------------------------------------------------------
|
|
|
|
def _change_track(self, player: Player, delta: int, source: Source) -> None:
|
|
after = self.config.shifted(player.track, delta)
|
|
if after != player.track:
|
|
self._emit(TrackChanged(player.id, player.track, after, source))
|
|
player.track = after
|
|
|
|
def _place(self, player: Player) -> None:
|
|
squares = sorted(self.board.safe_room(player.corner))
|
|
choice = self._agent(player).place_pawn(self.view(player.id), squares)
|
|
player.position = self._pick(choice, squares, "start square")
|
|
|
|
def _send_to_start(self, player: Player) -> None:
|
|
self._place(player)
|
|
self._emit(SentToStart(player.id, player.position))
|
|
|
|
def _relocate(self, player: Player, destination: Pos, reason: str) -> None:
|
|
if destination != player.position:
|
|
self._emit(Teleported(player.id, player.position, destination, reason))
|
|
player.position = destination
|
|
|
|
def _in_safe_room(self, square: Pos) -> bool:
|
|
return self.board.code(square) == Square.SAFE
|
|
|
|
def _fightable_at(self, square: Pos, attacker: Player) -> Player | None:
|
|
if self._in_safe_room(square):
|
|
return None
|
|
for p in self.players:
|
|
if p is not attacker and p.position == square and not p.meditating:
|
|
return p
|
|
return None
|
|
|
|
def _has_unfightable_pawn(self, square: Pos, player: Player) -> bool:
|
|
return any(p is not player and p.position == square and p.meditating for p in self.players)
|
|
|
|
def _shares_square(self, player: Player) -> bool:
|
|
if self._in_safe_room(player.position):
|
|
return False
|
|
return any(p is not player and p.position == player.position for p in self.players)
|
|
|
|
def _targets(self, player: Player) -> list[int]:
|
|
return [
|
|
p.id
|
|
for p in self.players
|
|
if p is not player and not p.meditating and not self._in_safe_room(p.position)
|
|
]
|
|
|
|
def _agent(self, player: Player) -> Agent:
|
|
return self.agents[player.id]
|
|
|
|
def _emit(self, event: Event) -> None:
|
|
self.events.append(event)
|
|
|
|
@staticmethod
|
|
def _pick[T](choice: T, options: Sequence[T], what: str) -> T:
|
|
if choice not in options:
|
|
raise ValueError(f"agent chose an illegal {what}: {choice!r}")
|
|
return choice
|