pf1e-simulator/src/pf1e_simulator/runner.py

"""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.combat import CombatantState, CombatantStats, CombatEngine
from pf1e_simulator.grid import Grid
from pf1e_simulator.map import zone_cells
from pf1e_simulator.metrics import win_rate, win_rate_band, win_rate_sigma
from pf1e_simulator.rng import SeededRng


@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


@dataclass(frozen=True)
class EncounterReport:
    """Aggregated results of a Monte Carlo encounter."""

    runs: int
    seed: int
    wins: dict[str, int]
    draws: int
    avg_rounds: float
    attrition: dict[str, CombatantAttrition]

    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 build_states(spec: EncounterSpec) -> list[CombatantState]:
    """Fresh states for one battle; duplicate combatant ids get -N suffixes.

    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.
    """
    states: list[CombatantState] = []
    seen: dict[str, int] = {}
    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)
        if len(cells) < len(side.combatants):
            msg = (
                f"side {side.name!r}: zone {side.zone!r} has {len(cells)} cells "
                f"for {len(side.combatants)} combatants"
            )
            raise ValueError(msg)
        for combatant, pos in zip(side.combatants, cells, strict=False):
            count = seen.get(combatant.id, 0) + 1
            seen[combatant.id] = count
            cid = combatant.id if count == 1 else f"{combatant.id}-{count}"
            effective = (
                combatant
                if cid == combatant.id
                else combatant.model_copy(update={"id": cid})
            )
            states.append(
                CombatantState(
                    combatant=effective,
                    side=side.name,
                    pos=pos,
                    hp=effective.hp_max,
                )
            )
    return states


def run_encounter(
    spec: EncounterSpec,
    runs: int,
    seed: int,
    *,
    policy: Policy | None = None,
) -> EncounterReport:
    """Run `runs` battles (one SeededRng per run) and aggregate balance metrics."""
    if runs <= 0:
        msg = "runs must be positive"
        raise ValueError(msg)
    grid = Grid.from_spec(spec.map)
    states = build_states(spec)
    ids = [s.combatant.id for s in states]
    wins = {side.name: 0 for side in spec.sides}
    draws = 0
    total_rounds = 0
    totals = {cid: CombatantStats() for cid in ids}
    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).run()
        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[cid]
            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,
            )
    attrition = {
        cid: CombatantAttrition(
            hits=stats.hits / runs,
            crits=stats.crits / runs,
            damage_dealt=stats.damage_dealt / runs,
            damage_taken=stats.damage_taken / runs,
        )
        for cid, stats in totals.items()
    }
    return EncounterReport(
        runs=runs,
        seed=seed,
        wins=wins,
        draws=draws,
        avg_rounds=total_rounds / runs,
        attrition=attrition,
    )