feat(runner): Monte Carlo runner, closed-form metrics and balance CLI
Adds the S6 layer on top of the deterministic combat engine: - metrics.py: closed-form binomial statistics (win rate, standard error, 3-sigma confidence band) with clamped degenerate proportions. - runner.py: EncounterSpec/Side definitions, per-run SeededRng streams, global duplicate-id disambiguation, attrition averages per combatant. - cli.py: pf1e-sim entry point producing a French balance report with win rates, 3-sigma bands, draws, rounds, and attrition. - Movement fix: greedy straight-line heuristic could oscillate at walls; _step_toward now follows the true shortest path via an unbounded Dijkstra cost field from the target (Grid.reachable budget=None). - Regression tests: melee unit routes around a wall and engages; grid unbounded-budget coverage.
This commit is contained in:
@@ -0,0 +1,119 @@
|
||||
"""Command-line balance simulator: load a map and sides, run Monte Carlo, report.
|
||||
|
||||
Usage: pf1e-sim --map MAP.yaml --side NAME FILE.json [FILE.json ...] [--runs N]
|
||||
|
||||
Side names must match the map's `deployment` section (zone letters). Each side
|
||||
may repeat the same monster JSON file; the runner disambiguates the ids.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pf1e_simulator.runner import EncounterReport, EncounterSpec
|
||||
|
||||
from pf1e_simulator.loaders import MonsterLoadError, SheetLoadError, load_monster
|
||||
from pf1e_simulator.map import MapValidationError, load_map
|
||||
from pf1e_simulator.runner import EncounterSpec, Side, run_encounter
|
||||
|
||||
|
||||
class CliError(Exception):
|
||||
"""User-facing CLI error (bad arguments or unusable files)."""
|
||||
|
||||
|
||||
def _build_spec(args: argparse.Namespace) -> EncounterSpec:
|
||||
map_spec = load_map(args.map)
|
||||
provided = [tuple(raw) for raw in args.side]
|
||||
provided_names = {raw[0] for raw in provided}
|
||||
missing = sorted(set(map_spec.deployment) - provided_names)
|
||||
if missing:
|
||||
msg = f"sides in map deployment without --side: {', '.join(missing)}"
|
||||
raise CliError(msg)
|
||||
unknown = sorted(provided_names - set(map_spec.deployment))
|
||||
if unknown:
|
||||
msg = f"--side names not in map deployment: {', '.join(unknown)}"
|
||||
raise CliError(msg)
|
||||
sides: list[Side] = []
|
||||
for raw in provided:
|
||||
name = raw[0]
|
||||
files = raw[1:]
|
||||
if not files:
|
||||
msg = f"side {name!r}: at least one monster JSON file is required"
|
||||
raise CliError(msg)
|
||||
combatants = tuple(load_monster(Path(path)) for path in files)
|
||||
sides.append(Side(name=name, combatants=combatants, zone=map_spec.deployment[name]))
|
||||
return EncounterSpec(map=map_spec, sides=tuple(sides), round_cap=args.round_cap)
|
||||
|
||||
|
||||
def format_report(report: EncounterReport, spec: EncounterSpec) -> str:
|
||||
"""Render a human-readable French balance report."""
|
||||
lines = ["=== Rapport d'équilibrage ===", f"Carte : {spec.map.name}"]
|
||||
lines.extend(f"{side.name} : {len(side.combatants)} combatants" for side in spec.sides)
|
||||
lines.append(
|
||||
f"{report.runs} combats simulés (seed {report.seed}, plafond {spec.round_cap} rounds)"
|
||||
)
|
||||
lines.append("")
|
||||
for side in spec.sides:
|
||||
wins = report.wins.get(side.name, 0)
|
||||
rate = report.win_rate(side.name)
|
||||
low, high = report.win_rate_band(side.name)
|
||||
lines.append(
|
||||
f"Victoires {side.name} : {rate:.1%} ({wins}) — "
|
||||
f"bande 3σ : [{low:.1%}, {high:.1%}]"
|
||||
)
|
||||
lines.append(f"Nuls : {report.draws / report.runs:.1%} ({report.draws})")
|
||||
lines.append(f"Rounds moyens : {report.avg_rounds:.1f}")
|
||||
lines.append("")
|
||||
lines.append("Attrition moyenne par combatant :")
|
||||
lines.extend(
|
||||
f" {cid} : {attrition.hits:.2f} touches · {attrition.crits:.2f} critiques · "
|
||||
f"{attrition.damage_dealt:.2f} dégâts infligés · {attrition.damage_taken:.2f} subis"
|
||||
for cid, attrition in report.attrition.items()
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
"""CLI entry point; returns the process exit code."""
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="pf1e-sim",
|
||||
description="Run Monte Carlo battles on a map and print a balance report.",
|
||||
)
|
||||
parser.add_argument("--map", required=True, type=Path, help="map YAML file")
|
||||
parser.add_argument(
|
||||
"--side",
|
||||
action="append",
|
||||
nargs="+",
|
||||
metavar=("NAME", "FILE"),
|
||||
help="side NAME with one or more monster JSON files (repeatable)",
|
||||
)
|
||||
parser.add_argument("--runs", type=int, default=1000, help="number of battles (default: 1000)")
|
||||
parser.add_argument("--seed", type=int, default=1, help="RNG seed (default: 1)")
|
||||
parser.add_argument(
|
||||
"--round-cap",
|
||||
type=int,
|
||||
default=100,
|
||||
help="max rounds per battle (default: 100)",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
if not args.side:
|
||||
parser.error("at least one --side is required")
|
||||
try:
|
||||
spec = _build_spec(args)
|
||||
except CliError as exc:
|
||||
print(f"pf1e-sim: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
except (MapValidationError, MonsterLoadError, SheetLoadError, OSError) as exc:
|
||||
print(f"pf1e-sim: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
try:
|
||||
report = run_encounter(spec, runs=args.runs, seed=args.seed)
|
||||
except ValueError as exc:
|
||||
print(f"pf1e-sim: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
print(format_report(report, spec))
|
||||
return 0
|
||||
@@ -7,8 +7,8 @@ Phase 0 documented deviations from PF1e (conventions):
|
||||
- DR applies once, after crit multiplication; any bypassing type defeats DR.
|
||||
- Death when hp < min(-10, -CON); hp <= 0 cannot act.
|
||||
- Initiative ties: higher initiative_mod first, then list order (no re-roll).
|
||||
- Movement: greedy single step toward the nearest enemy minimizing
|
||||
accumulated cost + remaining grid distance; ties keep delta order.
|
||||
- Movement: one step per move action toward the nearest enemy, following the
|
||||
true shortest path (Dijkstra cost field from the target); ties keep delta order.
|
||||
- Ranged attacks ignore cover and range penalties in Phase 0.
|
||||
"""
|
||||
|
||||
@@ -301,15 +301,20 @@ class CombatEngine:
|
||||
if speed_cells <= 0:
|
||||
return None
|
||||
blocked = frozenset(s.pos for s in self._states if s is not state and s.active)
|
||||
costs = self._grid.reachable(state.pos, speed_cells, blocked)
|
||||
to_target = self._grid.reachable(target.pos, None, blocked)
|
||||
if state.pos not in to_target:
|
||||
return None
|
||||
best: tuple[int, Pos] | None = None
|
||||
row, col = state.pos
|
||||
for d_row, d_col in _STEP_DELTAS:
|
||||
nxt = (row + d_row, col + d_col)
|
||||
cost = costs.get(nxt)
|
||||
cost = to_target.get(nxt)
|
||||
if cost is None:
|
||||
continue
|
||||
score = cost + self._grid.distance(nxt, target.pos)
|
||||
step = self._grid.step_cost(state.pos, nxt, 0)
|
||||
if step > speed_cells:
|
||||
continue
|
||||
score = step + cost
|
||||
if best is None or score < best[0]:
|
||||
best = (score, nxt)
|
||||
if best is None:
|
||||
|
||||
@@ -87,11 +87,12 @@ class Grid:
|
||||
flank_b = (to[0], frm[1])
|
||||
return self.passable(flank_a) and self.passable(flank_b)
|
||||
|
||||
def reachable(self, start: Pos, budget: int, blocked: frozenset[Pos]) -> dict[Pos, int]:
|
||||
def reachable(self, start: Pos, budget: int | None, blocked: frozenset[Pos]) -> dict[Pos, int]:
|
||||
"""Cheapest movement cost per reachable square, capped at budget.
|
||||
|
||||
`blocked` holds creature-occupied squares: unenterable, but never
|
||||
affecting diagonal_allowed. The start square is included at cost 0.
|
||||
budget=None disables the cap (full cost field).
|
||||
"""
|
||||
best_per_state: dict[tuple[Pos, int], int] = {(start, 0): 0}
|
||||
results: dict[Pos, int] = {start: 0}
|
||||
@@ -109,7 +110,7 @@ class Grid:
|
||||
if is_diagonal and not self.diagonal_allowed(pos, nxt):
|
||||
continue
|
||||
new_cost = cost + self.step_cost(pos, nxt, parity)
|
||||
if new_cost > budget:
|
||||
if budget is not None and new_cost > budget:
|
||||
continue
|
||||
new_parity = (parity + 1) % 2 if is_diagonal else parity
|
||||
state = (nxt, new_parity)
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Closed-form statistics for Monte Carlo balance reports.
|
||||
|
||||
Proportions follow the binomial model: p = wins/n with standard error
|
||||
sqrt(p(1-p)/n). The 3-sigma band is the default confidence interval used in
|
||||
the CLI report; clamp keeps degenerate proportions (0 or 1) exact.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
|
||||
def win_rate(wins: int, runs: int) -> float:
|
||||
"""Fraction of runs won: wins / runs."""
|
||||
if runs <= 0:
|
||||
msg = "runs must be positive"
|
||||
raise ValueError(msg)
|
||||
return wins / runs
|
||||
|
||||
|
||||
def win_rate_sigma(p: float, runs: int) -> float:
|
||||
"""Standard error of a proportion: sqrt(p(1-p)/n)."""
|
||||
if runs <= 0:
|
||||
msg = "runs must be positive"
|
||||
raise ValueError(msg)
|
||||
return math.sqrt(p * (1 - p) / runs)
|
||||
|
||||
|
||||
def win_rate_band(p: float, runs: int, *, sigma_count: float = 3.0) -> tuple[float, float]:
|
||||
"""p +/- sigma_count standard errors, clamped to [0, 1]."""
|
||||
sigma = win_rate_sigma(p, runs)
|
||||
return (max(0.0, p - sigma_count * sigma), min(1.0, p + sigma_count * sigma))
|
||||
@@ -0,0 +1,171 @@
|
||||
"""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,
|
||||
)
|
||||
Reference in New Issue
Block a user