feat(dice): seeded RNG protocol and dice notation parser
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
@@ -42,6 +42,8 @@ ignore = [
|
||||
[tool.ruff.lint.per-file-ignores]
|
||||
# Tests: assert is expected, magic numbers are fine, annotations are noisy
|
||||
"tests/**" = ["S101", "PLR2004", "ANN"]
|
||||
# rng.py: `random.Random` is used for reproducible Monte Carlo streams, not cryptography
|
||||
"src/pf1e_simulator/rng.py" = ["S311"]
|
||||
|
||||
# ── Basedpyright ──────────────────────────────────────────────────────────────
|
||||
[tool.basedpyright]
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Dice notation parser and expression evaluator.
|
||||
|
||||
Design: `DiceExpr` is a frozen Pydantic model representing a dice expression.
|
||||
Normal dice have count >= 1, sides >= 2. Flat bonuses (e.g. parse_dice("7"))
|
||||
are represented with count=0, sides=0 as a sentinel — this is the ONLY path
|
||||
where count=0 is valid, produced exclusively by `DiceExpr.flat()` or
|
||||
`parse_dice()` on a plain integer.
|
||||
|
||||
The `roll` method returns the raw sum + bonus (no minimum floor).
|
||||
Damage floors are the caller's concern.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, model_validator
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pf1e_simulator.rng import Rng
|
||||
|
||||
_MIN_SIDES = 2
|
||||
|
||||
_DICE_RE = re.compile(r"^(\d*)d(\d+)([+-]\d+)?$")
|
||||
|
||||
|
||||
class DiceParseError(Exception):
|
||||
"""Raised when dice notation cannot be parsed.
|
||||
|
||||
Carries the offending text in its message.
|
||||
"""
|
||||
|
||||
def __init__(self, text: str) -> None:
|
||||
super().__init__(f"invalid dice notation: {text!r}")
|
||||
self.text = text
|
||||
|
||||
|
||||
class DiceExpr(BaseModel):
|
||||
"""A parsed dice expression.
|
||||
|
||||
Invariants (enforced by model_validator):
|
||||
- Normal dice: count >= 1, sides >= 2
|
||||
- Flat bonus sentinel: count == 0, sides == 0 (via .flat() or parse_dice)
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
count: int
|
||||
sides: int
|
||||
bonus: int = 0
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_invariants(self) -> DiceExpr:
|
||||
if self.count == 0 and self.sides == 0:
|
||||
return self
|
||||
if self.count < 1:
|
||||
msg = f"count must be >= 1 for dice expressions, got {self.count}"
|
||||
raise ValueError(msg)
|
||||
if self.sides < _MIN_SIDES:
|
||||
msg = f"sides must be >= {_MIN_SIDES} for dice expressions, got {self.sides}"
|
||||
raise ValueError(msg)
|
||||
return self
|
||||
|
||||
@classmethod
|
||||
def flat(cls, n: int) -> DiceExpr:
|
||||
"""Create a flat-bonus expression (no dice rolled)."""
|
||||
return cls(count=0, sides=0, bonus=n)
|
||||
|
||||
def roll(self, rng: Rng) -> int:
|
||||
"""Roll the expression using the given RNG.
|
||||
|
||||
Returns sum of `count` rolls of `sides` + `bonus`.
|
||||
For flat expressions (count=0), returns `bonus` without calling rng.
|
||||
"""
|
||||
if self.count == 0:
|
||||
return self.bonus
|
||||
total = sum(rng.roll_die(self.sides) for _ in range(self.count))
|
||||
return total + self.bonus
|
||||
|
||||
def mean(self) -> float:
|
||||
"""Expected value: count * (sides + 1) / 2 + bonus."""
|
||||
if self.count == 0:
|
||||
return float(self.bonus)
|
||||
return self.count * (self.sides + 1) / 2 + self.bonus
|
||||
|
||||
|
||||
def parse_dice(text: str) -> DiceExpr:
|
||||
"""Parse a dice notation string into a DiceExpr.
|
||||
|
||||
Accepted forms:
|
||||
- "2d6+3", "2d6-1", "1d4", "d20" (standard notation)
|
||||
- "7", "-5", "0" (flat integers)
|
||||
|
||||
Raises DiceParseError on invalid input.
|
||||
"""
|
||||
stripped = text.strip()
|
||||
if not stripped:
|
||||
raise DiceParseError(text)
|
||||
|
||||
m = _DICE_RE.match(stripped)
|
||||
if m:
|
||||
count_str, sides_str, bonus_str = m.groups()
|
||||
if not sides_str:
|
||||
raise DiceParseError(text)
|
||||
count = int(count_str) if count_str else 1
|
||||
sides = int(sides_str)
|
||||
bonus = int(bonus_str) if bonus_str else 0
|
||||
|
||||
if count == 0:
|
||||
raise DiceParseError(text)
|
||||
if sides < 1:
|
||||
raise DiceParseError(text)
|
||||
|
||||
return DiceExpr(count=count, sides=sides, bonus=bonus)
|
||||
|
||||
try:
|
||||
n = int(stripped)
|
||||
except ValueError:
|
||||
raise DiceParseError(text) from None
|
||||
|
||||
return DiceExpr.flat(n)
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Seeded and scripted RNG protocol for deterministic Monte Carlo simulation.
|
||||
|
||||
Design: every die roll in the engine flows through the `Rng` protocol.
|
||||
`SeededRng` wraps `random.Random(seed)` for reproducible streams.
|
||||
`ScriptedRng` pops from a fixed queue — used in tests to script exact outcomes.
|
||||
|
||||
Security note: `random.Random` is NOT cryptographically secure.
|
||||
This is by design — the simulator needs reproducibility, not security.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from collections import deque
|
||||
from typing import Protocol
|
||||
|
||||
_EXHAUSTED_MSG = "ScriptedRng queue exhausted"
|
||||
|
||||
|
||||
class RngExhaustedError(Exception):
|
||||
"""Raised when a ScriptedRng queue is exhausted."""
|
||||
|
||||
|
||||
class Rng(Protocol):
|
||||
"""Protocol for deterministic die-roll sources."""
|
||||
|
||||
def roll_die(self, sides: int) -> int:
|
||||
"""Return a uniform integer in 1..=sides."""
|
||||
...
|
||||
|
||||
def d20(self) -> int:
|
||||
"""Return roll_die(20)."""
|
||||
...
|
||||
|
||||
def choice_index(self, n: int) -> int:
|
||||
"""Return a uniform integer in 0..n-1 for deterministic tie-breaks."""
|
||||
...
|
||||
|
||||
|
||||
class SeededRng:
|
||||
"""Rng backed by random.Random(seed) for reproducible streams."""
|
||||
|
||||
def __init__(self, seed: int) -> None:
|
||||
self._rng = random.Random(seed)
|
||||
|
||||
def roll_die(self, sides: int) -> int:
|
||||
return self._rng.randint(1, sides)
|
||||
|
||||
def d20(self) -> int:
|
||||
return self.roll_die(20)
|
||||
|
||||
def choice_index(self, n: int) -> int:
|
||||
return self._rng.randrange(n)
|
||||
|
||||
|
||||
class ScriptedRng:
|
||||
"""Rng that pops predetermined values from a queue.
|
||||
|
||||
Each call to roll_die, d20, or choice_index consumes one value.
|
||||
Raises RngExhaustedError when the queue is empty.
|
||||
"""
|
||||
|
||||
def __init__(self, queue: list[int]) -> None:
|
||||
self._queue: deque[int] = deque(queue)
|
||||
|
||||
def roll_die(self, _sides: int) -> int:
|
||||
if not self._queue:
|
||||
raise RngExhaustedError(_EXHAUSTED_MSG)
|
||||
return self._queue.popleft()
|
||||
|
||||
def d20(self) -> int:
|
||||
return self.roll_die(20)
|
||||
|
||||
def choice_index(self, _n: int) -> int:
|
||||
if not self._queue:
|
||||
raise RngExhaustedError(_EXHAUSTED_MSG)
|
||||
return self._queue.popleft()
|
||||
@@ -0,0 +1,196 @@
|
||||
"""Tests for dice notation parsing and DiceExpr evaluation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
|
||||
import pytest
|
||||
|
||||
from pf1e_simulator.dice import DiceExpr, DiceParseError, parse_dice
|
||||
from pf1e_simulator.rng import ScriptedRng
|
||||
|
||||
# ── Parsing: accepted forms ──────────────────────────────────────────────────
|
||||
|
||||
class TestParseDiceAccepted:
|
||||
"""Given: valid dice notation strings
|
||||
When: parse_dice is called
|
||||
Then: returns a correct DiceExpr."""
|
||||
|
||||
def test_2d6_plus_3(self) -> None:
|
||||
expr = parse_dice("2d6+3")
|
||||
assert expr.count == 2
|
||||
assert expr.sides == 6
|
||||
assert expr.bonus == 3
|
||||
|
||||
def test_2d6_minus_1(self) -> None:
|
||||
expr = parse_dice("2d6-1")
|
||||
assert expr.count == 2
|
||||
assert expr.sides == 6
|
||||
assert expr.bonus == -1
|
||||
|
||||
def test_1d4_no_bonus(self) -> None:
|
||||
expr = parse_dice("1d4")
|
||||
assert expr.count == 1
|
||||
assert expr.sides == 4
|
||||
assert expr.bonus == 0
|
||||
|
||||
def test_d20_shorthand(self) -> None:
|
||||
expr = parse_dice("d20")
|
||||
assert expr.count == 1
|
||||
assert expr.sides == 20
|
||||
assert expr.bonus == 0
|
||||
|
||||
def test_flat_integer(self) -> None:
|
||||
expr = parse_dice("7")
|
||||
assert expr.count == 0
|
||||
assert expr.sides == 0
|
||||
assert expr.bonus == 7
|
||||
|
||||
def test_flat_negative(self) -> None:
|
||||
expr = parse_dice("-5")
|
||||
assert expr.count == 0
|
||||
assert expr.sides == 0
|
||||
assert expr.bonus == -5
|
||||
|
||||
def test_flat_zero(self) -> None:
|
||||
expr = parse_dice("0")
|
||||
assert expr.count == 0
|
||||
assert expr.sides == 0
|
||||
assert expr.bonus == 0
|
||||
|
||||
def test_large_dice(self) -> None:
|
||||
expr = parse_dice("10d100+50")
|
||||
assert expr.count == 10
|
||||
assert expr.sides == 100
|
||||
assert expr.bonus == 50
|
||||
|
||||
def test_whitespace_stripped(self) -> None:
|
||||
expr = parse_dice(" 2d6+3 ")
|
||||
assert expr.count == 2
|
||||
assert expr.sides == 6
|
||||
assert expr.bonus == 3
|
||||
|
||||
|
||||
# ── Parsing: rejected forms ──────────────────────────────────────────────────
|
||||
|
||||
class TestParseDiceRejected:
|
||||
"""Given: invalid dice notation strings
|
||||
When: parse_dice is called
|
||||
Then: raises DiceParseError carrying the offending text."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"text",
|
||||
[
|
||||
"",
|
||||
"sizeRoll(1, 8, @size)",
|
||||
"2x6",
|
||||
"1d",
|
||||
"d",
|
||||
"2d0",
|
||||
"0d6",
|
||||
"2d 6",
|
||||
"abc",
|
||||
"d6+",
|
||||
],
|
||||
)
|
||||
def test_rejects_invalid(self, text: str) -> None:
|
||||
with pytest.raises(DiceParseError) as exc_info:
|
||||
parse_dice(text)
|
||||
assert text in str(exc_info.value)
|
||||
|
||||
|
||||
# ── Mean calculation ─────────────────────────────────────────────────────────
|
||||
|
||||
class TestDiceExprMean:
|
||||
"""Given: a DiceExpr
|
||||
When: mean() is called
|
||||
Then: returns count * (sides + 1) / 2 + bonus."""
|
||||
|
||||
def test_2d6_plus_3_mean(self) -> None:
|
||||
assert DiceExpr(count=2, sides=6, bonus=3).mean() == 10.0
|
||||
|
||||
def test_1d4_mean(self) -> None:
|
||||
assert DiceExpr(count=1, sides=4, bonus=0).mean() == 2.5
|
||||
|
||||
def test_d20_mean(self) -> None:
|
||||
assert DiceExpr(count=1, sides=20, bonus=0).mean() == 10.5
|
||||
|
||||
def test_flat_bonus_mean(self) -> None:
|
||||
assert DiceExpr(count=0, sides=0, bonus=7).mean() == 7.0
|
||||
|
||||
def test_negative_bonus_mean(self) -> None:
|
||||
assert DiceExpr(count=1, sides=6, bonus=-2).mean() == 1.5
|
||||
|
||||
|
||||
# ── Roll: uses RNG correctly ────────────────────────────────────────────────
|
||||
|
||||
class TestDiceExprRoll:
|
||||
"""Given: a DiceExpr and a ScriptedRng
|
||||
When: roll(rng) is called
|
||||
Then: consumes exactly `count` values from the RNG."""
|
||||
|
||||
def test_2d6_plus_3_uses_two_rolls(self) -> None:
|
||||
rng = ScriptedRng([3, 4])
|
||||
expr = DiceExpr(count=2, sides=6, bonus=3)
|
||||
assert expr.roll(rng) == 3 + 4 + 3 # 10
|
||||
|
||||
def test_1d20_uses_one_roll(self) -> None:
|
||||
rng = ScriptedRng([15])
|
||||
expr = DiceExpr(count=1, sides=20, bonus=0)
|
||||
assert expr.roll(rng) == 15
|
||||
|
||||
def test_flat_bonus_uses_no_rolls(self) -> None:
|
||||
rng = ScriptedRng([])
|
||||
expr = DiceExpr(count=0, sides=0, bonus=7)
|
||||
assert expr.roll(rng) == 7
|
||||
|
||||
def test_negative_bonus(self) -> None:
|
||||
rng = ScriptedRng([6, 6])
|
||||
expr = DiceExpr(count=2, sides=6, bonus=-1)
|
||||
assert expr.roll(rng) == 6 + 6 - 1 # 11
|
||||
|
||||
|
||||
# ── Distribution: exhaustive 2d6+3 ──────────────────────────────────────────
|
||||
|
||||
class TestDiceExprDistribution:
|
||||
"""Given: 2d6+3 enumerated over all 36 face pairs
|
||||
When: probabilities are computed
|
||||
Then: they match the theoretical distribution."""
|
||||
|
||||
def test_2d6_plus_3_distribution(self) -> None:
|
||||
totals: Counter[int] = Counter()
|
||||
for d1 in range(1, 7):
|
||||
for d2 in range(1, 7):
|
||||
rng = ScriptedRng([d1, d2])
|
||||
expr = DiceExpr(count=2, sides=6, bonus=3)
|
||||
totals[expr.roll(rng)] += 1
|
||||
|
||||
assert totals[5] == 1 # (1,1) → 1+1+3=5
|
||||
assert totals[7] == 3 # (1,3),(2,2),(3,1) → sum=4+3=7
|
||||
assert totals[10] == 6 # sum=7: (1,6)..(6,1) → 7+3=10
|
||||
assert totals[15] == 1 # (6,6) → 12+3=15
|
||||
assert sum(totals.values()) == 36
|
||||
|
||||
|
||||
# ── DiceExpr.flat factory ────────────────────────────────────────────────────
|
||||
|
||||
class TestDiceExprFlat:
|
||||
"""Given: DiceExpr.flat(n)
|
||||
When: called with any integer
|
||||
Then: returns count=0, sides=0, bonus=n."""
|
||||
|
||||
def test_flat_positive(self) -> None:
|
||||
expr = DiceExpr.flat(10)
|
||||
assert expr.count == 0
|
||||
assert expr.sides == 0
|
||||
assert expr.bonus == 10
|
||||
assert expr.mean() == 10.0
|
||||
|
||||
def test_flat_negative(self) -> None:
|
||||
expr = DiceExpr.flat(-3)
|
||||
assert expr.bonus == -3
|
||||
|
||||
def test_flat_roll_returns_bonus(self) -> None:
|
||||
rng = ScriptedRng([])
|
||||
expr = DiceExpr.flat(5)
|
||||
assert expr.roll(rng) == 5
|
||||
@@ -0,0 +1,115 @@
|
||||
"""Tests for the RNG protocol, SeededRng, and ScriptedRng."""
|
||||
|
||||
import pytest
|
||||
|
||||
from pf1e_simulator.rng import RngExhaustedError, ScriptedRng, SeededRng
|
||||
|
||||
|
||||
class TestSeededRngRollDie:
|
||||
"""Given: a SeededRng with a fixed seed
|
||||
When: roll_die is called
|
||||
Then: returns values in 1..=sides deterministically."""
|
||||
|
||||
def test_roll_die_returns_in_range(self) -> None:
|
||||
rng = SeededRng(42)
|
||||
for _ in range(100):
|
||||
result = rng.roll_die(6)
|
||||
assert 1 <= result <= 6
|
||||
|
||||
def test_roll_die_d20_in_range(self) -> None:
|
||||
rng = SeededRng(42)
|
||||
for _ in range(100):
|
||||
result = rng.d20()
|
||||
assert 1 <= result <= 20
|
||||
|
||||
def test_d20_is_roll_die_20(self) -> None:
|
||||
rng1 = SeededRng(123)
|
||||
rng2 = SeededRng(123)
|
||||
assert rng1.d20() == rng2.roll_die(20)
|
||||
|
||||
|
||||
class TestSeededRngChoiceIndex:
|
||||
"""Given: a SeededRng
|
||||
When: choice_index(n) is called
|
||||
Then: returns values in 0..n-1."""
|
||||
|
||||
def test_choice_index_in_range(self) -> None:
|
||||
rng = SeededRng(42)
|
||||
for _ in range(100):
|
||||
result = rng.choice_index(5)
|
||||
assert 0 <= result <= 4
|
||||
|
||||
|
||||
class TestSeededRngReproducibility:
|
||||
"""Given: two SeededRng instances with the same seed
|
||||
When: each produces 1000 rolls
|
||||
Then: the streams are identical."""
|
||||
|
||||
def test_identical_d20_streams(self) -> None:
|
||||
rng1 = SeededRng(42)
|
||||
rng2 = SeededRng(42)
|
||||
rolls1 = [rng1.d20() for _ in range(1000)]
|
||||
rolls2 = [rng2.d20() for _ in range(1000)]
|
||||
assert rolls1 == rolls2
|
||||
|
||||
def test_identical_choice_index_streams(self) -> None:
|
||||
rng1 = SeededRng(42)
|
||||
rng2 = SeededRng(42)
|
||||
idx1 = [rng1.choice_index(8) for _ in range(1000)]
|
||||
idx2 = [rng2.choice_index(8) for _ in range(1000)]
|
||||
assert idx1 == idx2
|
||||
|
||||
def test_different_seeds_produce_different_streams(self) -> None:
|
||||
rng1 = SeededRng(1)
|
||||
rng2 = SeededRng(2)
|
||||
rolls1 = [rng1.d20() for _ in range(100)]
|
||||
rolls2 = [rng2.d20() for _ in range(100)]
|
||||
assert rolls1 != rolls2
|
||||
|
||||
|
||||
class TestScriptedRng:
|
||||
"""Given: a ScriptedRng with a queue of ints
|
||||
When: roll_die / choice_index is called
|
||||
Then: returns queued values in order."""
|
||||
|
||||
def test_roll_die_returns_queued_values(self) -> None:
|
||||
rng = ScriptedRng([3, 1, 4, 1, 5])
|
||||
assert rng.roll_die(6) == 3
|
||||
assert rng.roll_die(6) == 1
|
||||
assert rng.roll_die(6) == 4
|
||||
|
||||
def test_d20_returns_queued_values(self) -> None:
|
||||
rng = ScriptedRng([20, 1, 13])
|
||||
assert rng.d20() == 20
|
||||
assert rng.d20() == 1
|
||||
assert rng.d20() == 13
|
||||
|
||||
def test_choice_index_returns_queued_values(self) -> None:
|
||||
rng = ScriptedRng([0, 2, 1])
|
||||
assert rng.choice_index(5) == 0
|
||||
assert rng.choice_index(5) == 2
|
||||
assert rng.choice_index(5) == 1
|
||||
|
||||
def test_raises_on_exhaustion_roll_die(self) -> None:
|
||||
rng = ScriptedRng([5])
|
||||
rng.roll_die(6)
|
||||
with pytest.raises(RngExhaustedError):
|
||||
rng.roll_die(6)
|
||||
|
||||
def test_raises_on_exhaustion_d20(self) -> None:
|
||||
rng = ScriptedRng([10])
|
||||
rng.d20()
|
||||
with pytest.raises(RngExhaustedError):
|
||||
rng.d20()
|
||||
|
||||
def test_raises_on_exhaustion_choice_index(self) -> None:
|
||||
rng = ScriptedRng([0])
|
||||
rng.choice_index(3)
|
||||
with pytest.raises(RngExhaustedError):
|
||||
rng.choice_index(3)
|
||||
|
||||
def test_exhaustion_error_carries_message(self) -> None:
|
||||
rng = ScriptedRng([1])
|
||||
rng.roll_die(6)
|
||||
with pytest.raises(RngExhaustedError, match="exhausted"):
|
||||
rng.roll_die(6)
|
||||
Reference in New Issue
Block a user