264 lines
9.2 KiB
Python
264 lines
9.2 KiB
Python
"""Monte Carlo runner: many deterministic battles aggregated into balance metrics.
|
|
|
|
Each run builds fresh combatant states (duplicate ids get -N suffixes), places
|
|
each side inside its deployment zone, and plays one battle with a dedicated
|
|
SeededRng stream derived from the encounter seed. Results are aggregated into
|
|
an EncounterReport: wins per side, draws, average rounds, and per-combatant
|
|
attrition averages. All statistical helpers live in metrics.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from pf1e_simulator.combat import Policy
|
|
from pf1e_simulator.grid import Grid
|
|
from pf1e_simulator.map import MapSpec
|
|
from pf1e_simulator.models import Combatant
|
|
from pf1e_simulator.spells import SpellSpec
|
|
|
|
from pf1e_simulator.combat import (
|
|
SIZE_CELLS,
|
|
CombatantState,
|
|
CombatantStats,
|
|
CombatEngine,
|
|
)
|
|
from pf1e_simulator.grid import Grid
|
|
from pf1e_simulator.map import Pos, zone_cells
|
|
from pf1e_simulator.metrics import win_rate, win_rate_band, win_rate_sigma
|
|
from pf1e_simulator.rng import SeededRng
|
|
from pf1e_simulator.spells import SPELLS_DIR, load_spell_registry
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Side:
|
|
"""One faction in an encounter: combatants deployed inside one zone letter."""
|
|
|
|
name: str
|
|
combatants: tuple[Combatant, ...]
|
|
zone: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EncounterSpec:
|
|
"""A full encounter definition: map, sides, and a battle round cap."""
|
|
|
|
map: MapSpec
|
|
sides: tuple[Side, ...]
|
|
round_cap: int = 100
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CombatantAttrition:
|
|
"""Average per-combatant outcomes over a Monte Carlo run."""
|
|
|
|
hits: float
|
|
crits: float
|
|
damage_dealt: float
|
|
damage_taken: float
|
|
healing_done: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EncounterReport:
|
|
"""Aggregated results of a Monte Carlo encounter.
|
|
|
|
`transcript` holds the turn-by-turn log of the first battle when
|
|
`run_encounter` is called with ``capture_first_transcript=True``; it is
|
|
None otherwise (the default keeps the report cheap for pure Monte Carlo).
|
|
"""
|
|
|
|
runs: int
|
|
seed: int
|
|
wins: dict[str, int]
|
|
draws: int
|
|
avg_rounds: float
|
|
attrition: dict[str, CombatantAttrition]
|
|
transcript: tuple[str, ...] | None = None
|
|
|
|
def win_rate(self, side: str) -> float:
|
|
"""Fraction of runs won by `side` (closed form, see metrics)."""
|
|
return win_rate(self.wins.get(side, 0), self.runs)
|
|
|
|
def win_rate_sigma(self, side: str) -> float:
|
|
"""Standard error of the side's win rate: sqrt(p(1-p)/n)."""
|
|
return win_rate_sigma(self.win_rate(side), self.runs)
|
|
|
|
def win_rate_band(self, side: str) -> tuple[float, float]:
|
|
"""3-sigma confidence band for the side's win rate."""
|
|
return win_rate_band(self.win_rate(side), self.runs)
|
|
|
|
|
|
def _first_fit_anchor(
|
|
cells: list[Pos], placed: set[Pos], n: int, grid: Grid
|
|
) -> Pos | None:
|
|
"""First zone cell whose n x n footprint is in bounds, passable, and unplaced."""
|
|
for anchor in cells:
|
|
r, c = anchor
|
|
footprint = {(r + dr, c + dc) for dr in range(n) for dc in range(n)}
|
|
if any(not grid.in_bounds(p) or not grid.passable(p) for p in footprint):
|
|
continue
|
|
if footprint & placed:
|
|
continue
|
|
return anchor
|
|
return None
|
|
|
|
|
|
def build_states(spec: EncounterSpec) -> list[CombatantState]:
|
|
"""Fresh states for one battle; duplicate combatant ids get -N suffixes.
|
|
|
|
The suffix keeps incrementing until the candidate id is not already
|
|
taken, so an explicit id like ``goblin-2`` never collides with a
|
|
disambiguated duplicate of ``goblin`` (the engine keys stats by id).
|
|
|
|
Combatants of a side fill the zone's cells in (row, col) order. A side with
|
|
more combatants than zone cells, or a zone with no cells, is a spec error.
|
|
Creatures larger than 1x1 occupy an n x n footprint anchored at their cell.
|
|
"""
|
|
states: list[CombatantState] = []
|
|
taken: set[str] = set()
|
|
grid = Grid.from_spec(spec.map)
|
|
for side in spec.sides:
|
|
cells = zone_cells(spec.map, side.zone)
|
|
if not cells:
|
|
msg = f"side {side.name!r}: zone {side.zone!r} has no cells"
|
|
raise ValueError(msg)
|
|
placed: set[Pos] = set()
|
|
for combatant in side.combatants:
|
|
n = SIZE_CELLS.get(combatant.size, 1)
|
|
anchor = _first_fit_anchor(cells, placed, n, grid)
|
|
if anchor is None:
|
|
if n == 1:
|
|
msg = (
|
|
f"side {side.name!r}: zone {side.zone!r} has {len(cells)} cells "
|
|
f"for {len(side.combatants)} combatants"
|
|
)
|
|
else:
|
|
msg = (
|
|
f"side {side.name!r}: zone {side.zone!r} has no room for "
|
|
f"{combatant.id!r} ({combatant.size}, {n}x{n})"
|
|
)
|
|
raise ValueError(msg)
|
|
cid = combatant.id
|
|
suffix = 2
|
|
while cid in taken:
|
|
cid = f"{combatant.id}-{suffix}"
|
|
suffix += 1
|
|
taken.add(cid)
|
|
effective = (
|
|
combatant
|
|
if cid == combatant.id
|
|
else combatant.model_copy(update={"id": cid})
|
|
)
|
|
placed |= {
|
|
(anchor[0] + dr, anchor[1] + dc)
|
|
for dr in range(n)
|
|
for dc in range(n)
|
|
}
|
|
states.append(
|
|
CombatantState(
|
|
combatant=effective,
|
|
side=side.name,
|
|
pos=anchor,
|
|
hp=effective.hp_max,
|
|
resources_left=dict(effective.resources),
|
|
)
|
|
)
|
|
return states
|
|
|
|
|
|
def run_encounter(
|
|
spec: EncounterSpec,
|
|
runs: int,
|
|
seed: int,
|
|
*,
|
|
policy: Policy | None = None,
|
|
spell_registry: dict[str, SpellSpec] | None = None,
|
|
capture_first_transcript: bool = False,
|
|
monsters_die_at_zero: bool = False,
|
|
) -> EncounterReport:
|
|
"""Run `runs` battles (one SeededRng per run) and aggregate balance metrics.
|
|
|
|
The spell registry is loaded once (from ``data/spells/`` unless overridden)
|
|
and passed into every ``CombatEngine`` so caster PCs actually cast.
|
|
|
|
When ``capture_first_transcript`` is True, the turn-by-turn log of the
|
|
first battle (``seed + 0``) is stored in ``report.transcript``.
|
|
|
|
``monsters_die_at_zero`` enables the optional simple monster-death rule
|
|
(CLI ``--simple-monster-death``): monsters die at hp <= 0; PCs keep the
|
|
PF1e hp <= -CON rule. Default False keeps behavior bit-identical.
|
|
"""
|
|
if runs <= 0:
|
|
msg = "runs must be positive"
|
|
raise ValueError(msg)
|
|
grid = Grid.from_spec(spec.map)
|
|
states = build_states(spec)
|
|
if spell_registry is None:
|
|
spell_registry = load_spell_registry(SPELLS_DIR)
|
|
wins = {side.name: 0 for side in spec.sides}
|
|
draws = 0
|
|
total_rounds = 0
|
|
# Accumulate whatever ids the engine reports: summoned creatures act in the
|
|
# battle and appear in result.stats, so seeding from the initial states
|
|
# only would KeyError on them. Ids seen for the first time get a fresh
|
|
# accumulator. Duplicate summons of the same creature are disambiguated by
|
|
# the engine the same way build_states does (celestial_wolf,
|
|
# celestial_wolf-2, ...), so each summoned creature owns its own slot and
|
|
# their attrition is split, not merged.
|
|
totals: dict[str, CombatantStats] = {}
|
|
transcript: tuple[str, ...] | None = None
|
|
for index in range(runs):
|
|
if index > 0:
|
|
states = build_states(spec)
|
|
rng = SeededRng(seed + index)
|
|
result = CombatEngine(
|
|
rng,
|
|
grid,
|
|
states,
|
|
round_cap=spec.round_cap,
|
|
policy=policy,
|
|
spell_registry=spell_registry,
|
|
monsters_die_at_zero=monsters_die_at_zero,
|
|
).run()
|
|
if capture_first_transcript and index == 0:
|
|
transcript = result.transcript
|
|
if result.winner is None:
|
|
draws += 1
|
|
else:
|
|
wins[result.winner] = wins.get(result.winner, 0) + 1
|
|
total_rounds += result.rounds
|
|
for cid, stats in result.stats.items():
|
|
current = totals.get(cid)
|
|
if current is None:
|
|
totals[cid] = stats
|
|
continue
|
|
totals[cid] = CombatantStats(
|
|
hits=current.hits + stats.hits,
|
|
crits=current.crits + stats.crits,
|
|
damage_dealt=current.damage_dealt + stats.damage_dealt,
|
|
damage_taken=current.damage_taken + stats.damage_taken,
|
|
healing_done=current.healing_done + stats.healing_done,
|
|
)
|
|
attrition = {
|
|
cid: CombatantAttrition(
|
|
hits=stats.hits / runs,
|
|
crits=stats.crits / runs,
|
|
damage_dealt=stats.damage_dealt / runs,
|
|
damage_taken=stats.damage_taken / runs,
|
|
healing_done=stats.healing_done / runs,
|
|
)
|
|
for cid, stats in totals.items()
|
|
}
|
|
return EncounterReport(
|
|
runs=runs,
|
|
seed=seed,
|
|
wins=wins,
|
|
draws=draws,
|
|
avg_rounds=total_rounds / runs,
|
|
attrition=attrition,
|
|
transcript=transcript,
|
|
)
|