Core Implementation: - Created roll_types.py with AbRoll, JumpRoll, FieldingRoll, D20Roll dataclasses - Implemented DiceSystem singleton with cryptographically secure random generation - Added Roll model to db_models.py with JSONB storage for roll history - Implemented save_rolls_batch() and get_rolls_for_game() in database operations Testing: - 27 unit tests for roll type dataclasses (100% passing) - 35 unit tests for dice system (34/35 passing, 1 timing issue) - 16 integration tests for database persistence (uses production DiceSystem) Features: - Unique roll IDs using secrets.token_hex() - League-specific logic (SBA d100 rare plays, PD error-based rare plays) - Automatic derived value calculation (d6_two_total, jump_total, error_total) - Full audit trail with context metadata - Support for batch saving rolls per inning Technical Details: - Fixed dataclass inheritance with kw_only=True for Python 3.13 - Roll data stored as JSONB for flexible querying - Indexed on game_id, roll_type, league_id, team_id for efficient retrieval - Supports filtering by roll type, team, and timestamp ordering Note: Integration tests have async connection pool issue when run together (tests work individually, fixture cleanup needed in follow-up branch) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
348 lines
10 KiB
Python
348 lines
10 KiB
Python
"""
|
|
Cryptographically Secure Dice Rolling System
|
|
|
|
Implements secure random number generation for baseball gameplay with
|
|
support for all roll types: at-bat, jump, fielding, and generic d20.
|
|
"""
|
|
import logging
|
|
import secrets
|
|
from typing import List, Optional, Dict
|
|
from uuid import UUID
|
|
import pendulum
|
|
|
|
from app.core.roll_types import (
|
|
RollType, DiceRoll, AbRoll, JumpRoll, FieldingRoll, D20Roll
|
|
)
|
|
|
|
logger = logging.getLogger(f'{__name__}.DiceSystem')
|
|
|
|
|
|
class DiceSystem:
|
|
"""
|
|
Cryptographically secure dice rolling system for baseball gameplay
|
|
|
|
Uses Python's secrets module for cryptographic randomness.
|
|
Maintains roll history for auditing and game recovery.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._roll_history: List[DiceRoll] = []
|
|
|
|
def _generate_roll_id(self) -> str:
|
|
"""Generate unique cryptographic roll ID"""
|
|
return secrets.token_hex(8)
|
|
|
|
def _roll_d6(self) -> int:
|
|
"""Roll single d6 (1-6)"""
|
|
return secrets.randbelow(6) + 1
|
|
|
|
def _roll_d20(self) -> int:
|
|
"""Roll single d20 (1-20)"""
|
|
return secrets.randbelow(20) + 1
|
|
|
|
def _roll_d100(self) -> int:
|
|
"""Roll single d100 (1-100)"""
|
|
return secrets.randbelow(100) + 1
|
|
|
|
def roll_ab(
|
|
self,
|
|
league_id: str,
|
|
game_id: Optional[UUID] = None
|
|
) -> AbRoll:
|
|
"""
|
|
Roll at-bat dice: 1d6 + 2d6 + 2d20
|
|
|
|
Always rolls all dice. The check_d20 determines usage:
|
|
- check_d20 == 1: Wild pitch check (use resolution_d20 for confirmation)
|
|
- check_d20 == 2: Passed ball check (use resolution_d20 for confirmation)
|
|
- check_d20 >= 3: Normal at-bat (use check_d20 for result, resolution_d20 for splits)
|
|
|
|
Args:
|
|
league_id: 'sba' or 'pd'
|
|
game_id: Optional UUID of game in progress
|
|
|
|
Returns:
|
|
AbRoll with all dice results
|
|
"""
|
|
d6_one = self._roll_d6()
|
|
d6_two_a = self._roll_d6()
|
|
d6_two_b = self._roll_d6()
|
|
check_d20 = self._roll_d20()
|
|
resolution_d20 = self._roll_d20() # Always roll, used for WP/PB or splits
|
|
|
|
roll = AbRoll(
|
|
roll_id=self._generate_roll_id(),
|
|
roll_type=RollType.AB,
|
|
league_id=league_id,
|
|
timestamp=pendulum.now('UTC'),
|
|
game_id=game_id,
|
|
d6_one=d6_one,
|
|
d6_two_a=d6_two_a,
|
|
d6_two_b=d6_two_b,
|
|
check_d20=check_d20,
|
|
resolution_d20=resolution_d20,
|
|
d6_two_total=0, # Calculated in __post_init__
|
|
check_wild_pitch=False,
|
|
check_passed_ball=False
|
|
)
|
|
|
|
self._roll_history.append(roll)
|
|
logger.info(f"AB roll: {roll}", extra={"roll_id": roll.roll_id, "game_id": str(game_id) if game_id else None})
|
|
|
|
return roll
|
|
|
|
def roll_jump(
|
|
self,
|
|
league_id: str,
|
|
game_id: Optional[UUID] = None
|
|
) -> JumpRoll:
|
|
"""
|
|
Roll jump dice for stolen base attempt
|
|
|
|
1d20 check:
|
|
- 1: Pickoff attempt (roll resolution d20)
|
|
- 2: Balk check (roll resolution d20)
|
|
- 3-20: Normal jump (roll 2d6 for jump rating)
|
|
|
|
Args:
|
|
league_id: 'sba' or 'pd'
|
|
game_id: Optional UUID of game in progress
|
|
|
|
Returns:
|
|
JumpRoll with conditional dice based on check_roll
|
|
"""
|
|
check_roll = self._roll_d20()
|
|
|
|
jump_dice_a = None
|
|
jump_dice_b = None
|
|
resolution_roll = None
|
|
|
|
if check_roll == 1 or check_roll == 2:
|
|
# Pickoff or balk - roll resolution die
|
|
resolution_roll = self._roll_d20()
|
|
logger.debug(f"Jump check roll {check_roll}: {'pickoff' if check_roll == 1 else 'balk'}")
|
|
else:
|
|
# Normal jump - roll 2d6
|
|
jump_dice_a = self._roll_d6()
|
|
jump_dice_b = self._roll_d6()
|
|
logger.debug(f"Jump normal: {jump_dice_a} + {jump_dice_b}")
|
|
|
|
roll = JumpRoll(
|
|
roll_id=self._generate_roll_id(),
|
|
roll_type=RollType.JUMP,
|
|
league_id=league_id,
|
|
timestamp=pendulum.now('UTC'),
|
|
game_id=game_id,
|
|
check_roll=check_roll,
|
|
jump_dice_a=jump_dice_a,
|
|
jump_dice_b=jump_dice_b,
|
|
resolution_roll=resolution_roll
|
|
)
|
|
|
|
self._roll_history.append(roll)
|
|
logger.info(f"Jump roll: {roll}", extra={"roll_id": roll.roll_id, "game_id": str(game_id) if game_id else None})
|
|
|
|
return roll
|
|
|
|
def roll_fielding(
|
|
self,
|
|
position: str,
|
|
league_id: str,
|
|
game_id: Optional[UUID] = None
|
|
) -> FieldingRoll:
|
|
"""
|
|
Roll fielding check: 1d20 (range) + 3d6 (error) + 1d100 (rare play)
|
|
|
|
Args:
|
|
position: P, C, 1B, 2B, 3B, SS, LF, CF, RF
|
|
league_id: 'sba' or 'pd'
|
|
game_id: Optional UUID of game in progress
|
|
|
|
Returns:
|
|
FieldingRoll with range, error, and rare play dice
|
|
|
|
Raises:
|
|
ValueError: If position is invalid
|
|
"""
|
|
VALID_POSITIONS = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF']
|
|
if position not in VALID_POSITIONS:
|
|
raise ValueError(f"Invalid position: {position}. Must be one of {VALID_POSITIONS}")
|
|
|
|
d20 = self._roll_d20()
|
|
d6_one = self._roll_d6()
|
|
d6_two = self._roll_d6()
|
|
d6_three = self._roll_d6()
|
|
d100 = self._roll_d100()
|
|
|
|
roll = FieldingRoll(
|
|
roll_id=self._generate_roll_id(),
|
|
roll_type=RollType.FIELDING,
|
|
league_id=league_id,
|
|
timestamp=pendulum.now('UTC'),
|
|
game_id=game_id,
|
|
position=position,
|
|
d20=d20,
|
|
d6_one=d6_one,
|
|
d6_two=d6_two,
|
|
d6_three=d6_three,
|
|
d100=d100,
|
|
error_total=0, # Calculated in __post_init__
|
|
_is_rare_play=False
|
|
)
|
|
|
|
self._roll_history.append(roll)
|
|
logger.info(
|
|
f"Fielding roll ({position}): {roll}",
|
|
extra={
|
|
"roll_id": roll.roll_id,
|
|
"position": position,
|
|
"is_rare": roll.is_rare_play,
|
|
"game_id": str(game_id) if game_id else None
|
|
}
|
|
)
|
|
|
|
return roll
|
|
|
|
def roll_d20(
|
|
self,
|
|
league_id: str,
|
|
game_id: Optional[UUID] = None
|
|
) -> D20Roll:
|
|
"""
|
|
Roll single d20 (modifiers applied to target, not roll)
|
|
|
|
Args:
|
|
league_id: 'sba' or 'pd'
|
|
game_id: Optional UUID of game in progress
|
|
|
|
Returns:
|
|
D20Roll with single die result
|
|
"""
|
|
base_roll = self._roll_d20()
|
|
|
|
roll = D20Roll(
|
|
roll_id=self._generate_roll_id(),
|
|
roll_type=RollType.D20,
|
|
league_id=league_id,
|
|
timestamp=pendulum.now('UTC'),
|
|
game_id=game_id,
|
|
roll=base_roll
|
|
)
|
|
|
|
self._roll_history.append(roll)
|
|
logger.info(f"D20 roll: {roll}", extra={"roll_id": roll.roll_id, "game_id": str(game_id) if game_id else None})
|
|
|
|
return roll
|
|
|
|
def get_roll_history(
|
|
self,
|
|
roll_type: Optional[RollType] = None,
|
|
game_id: Optional[UUID] = None,
|
|
limit: int = 100
|
|
) -> List[DiceRoll]:
|
|
"""
|
|
Get roll history with optional filtering
|
|
|
|
Args:
|
|
roll_type: Filter by specific roll type (AB, JUMP, FIELDING, D20)
|
|
game_id: Filter by game UUID
|
|
limit: Maximum number of rolls to return (most recent)
|
|
|
|
Returns:
|
|
List of DiceRoll objects matching filters
|
|
"""
|
|
filtered = self._roll_history
|
|
|
|
if roll_type:
|
|
filtered = [r for r in filtered if r.roll_type == roll_type]
|
|
|
|
if game_id:
|
|
filtered = [r for r in filtered if r.game_id == game_id]
|
|
|
|
return filtered[-limit:]
|
|
|
|
def get_rolls_since(
|
|
self,
|
|
game_id: UUID,
|
|
since_timestamp: pendulum.DateTime
|
|
) -> List[DiceRoll]:
|
|
"""
|
|
Get all rolls for a game since a specific timestamp
|
|
|
|
Used for batch persistence at end of innings.
|
|
|
|
Args:
|
|
game_id: UUID of game
|
|
since_timestamp: Get rolls after this time
|
|
|
|
Returns:
|
|
List of DiceRoll objects for game since timestamp
|
|
"""
|
|
return [
|
|
roll for roll in self._roll_history
|
|
if roll.game_id == game_id and roll.timestamp >= since_timestamp
|
|
]
|
|
|
|
def verify_roll(self, roll_id: str) -> bool:
|
|
"""
|
|
Verify a roll ID exists in history
|
|
|
|
Args:
|
|
roll_id: Roll ID to verify
|
|
|
|
Returns:
|
|
True if roll exists in history
|
|
"""
|
|
return any(r.roll_id == roll_id for r in self._roll_history)
|
|
|
|
def get_distribution_stats(
|
|
self,
|
|
roll_type: Optional[RollType] = None
|
|
) -> Dict:
|
|
"""
|
|
Get distribution statistics for testing
|
|
|
|
Args:
|
|
roll_type: Optional filter by roll type
|
|
|
|
Returns:
|
|
Dictionary with roll counts by type
|
|
"""
|
|
rolls_to_analyze = self._roll_history
|
|
|
|
if roll_type:
|
|
rolls_to_analyze = [r for r in rolls_to_analyze if r.roll_type == roll_type]
|
|
|
|
if not rolls_to_analyze:
|
|
return {}
|
|
|
|
stats = {
|
|
"total_rolls": len(rolls_to_analyze),
|
|
"by_type": {}
|
|
}
|
|
|
|
# Count by type
|
|
for roll in rolls_to_analyze:
|
|
roll_type_str = roll.roll_type.value
|
|
if roll_type_str not in stats["by_type"]:
|
|
stats["by_type"][roll_type_str] = 0
|
|
stats["by_type"][roll_type_str] += 1
|
|
|
|
return stats
|
|
|
|
def clear_history(self) -> None:
|
|
"""Clear roll history (for testing)"""
|
|
self._roll_history.clear()
|
|
logger.debug("Roll history cleared")
|
|
|
|
def get_stats(self) -> dict:
|
|
"""Get dice system statistics"""
|
|
return {
|
|
"total_rolls": len(self._roll_history),
|
|
"by_type": self.get_distribution_stats()["by_type"] if self._roll_history else {}
|
|
}
|
|
|
|
|
|
# Singleton instance
|
|
dice_system = DiceSystem()
|