bhupesh-sf's picture
first commit
f8ba6bf verified
"""
DungeonMaster AI - Game State Models
Pydantic models for game entities including events, combat state,
characters, NPCs, scenes, and adventure data.
"""
from __future__ import annotations
import uuid
from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, computed_field
# =============================================================================
# Enums
# =============================================================================
class EventType(str, Enum):
"""Types of game events for logging."""
ROLL = "roll"
COMBAT_START = "combat_start"
COMBAT_END = "combat_end"
COMBAT_ACTION = "combat_action"
DAMAGE = "damage"
HEALING = "healing"
MOVEMENT = "movement"
DIALOGUE = "dialogue"
DISCOVERY = "discovery"
ITEM_ACQUIRED = "item_acquired"
REST = "rest"
LEVEL_UP = "level_up"
DEATH = "death"
STORY_FLAG = "story_flag"
SYSTEM = "system"
class CombatantStatus(str, Enum):
"""Status of a combatant in combat."""
ACTIVE = "active"
UNCONSCIOUS = "unconscious"
DEAD = "dead"
FLED = "fled"
class HPStatus(str, Enum):
"""Health status based on HP percentage."""
HEALTHY = "healthy" # > 50%
WOUNDED = "wounded" # 25-50%
CRITICAL = "critical" # 1-25%
UNCONSCIOUS = "unconscious" # 0
# =============================================================================
# Event Models
# =============================================================================
class SessionEvent(BaseModel):
"""
A single event in the game session.
Events are logged for context building and session history.
"""
event_id: str = Field(
default_factory=lambda: str(uuid.uuid4()),
description="Unique event identifier",
)
event_type: EventType = Field(
description="Type of event",
)
description: str = Field(
description="Human-readable event description",
)
data: dict[str, object] = Field(
default_factory=dict,
description="Event-specific data",
)
timestamp: datetime = Field(
default_factory=datetime.now,
description="When the event occurred",
)
turn: int = Field(
default=0,
description="Game turn when event occurred",
)
is_significant: bool = Field(
default=False,
description="Whether this event is significant for context",
)
# =============================================================================
# Combat Models
# =============================================================================
class Combatant(BaseModel):
"""
A participant in combat.
Tracks initiative, HP, conditions, and status.
"""
combatant_id: str = Field(
description="Unique combatant identifier",
)
name: str = Field(
description="Combatant name",
)
initiative: int = Field(
description="Initiative roll result",
)
is_player: bool = Field(
default=False,
description="Whether this is a player character",
)
hp_current: int = Field(
default=0,
description="Current hit points",
)
hp_max: int = Field(
default=1,
description="Maximum hit points",
)
armor_class: int = Field(
default=10,
description="Armor class",
)
conditions: list[str] = Field(
default_factory=list,
description="Active conditions",
)
status: CombatantStatus = Field(
default=CombatantStatus.ACTIVE,
description="Combat status",
)
@computed_field
@property
def hp_percent(self) -> float:
"""HP as percentage of max."""
if self.hp_max <= 0:
return 0.0
return (self.hp_current / self.hp_max) * 100.0
@computed_field
@property
def hp_status(self) -> HPStatus:
"""Health status based on HP percentage."""
if self.hp_current <= 0:
return HPStatus.UNCONSCIOUS
pct = self.hp_percent
if pct > 50:
return HPStatus.HEALTHY
if pct > 25:
return HPStatus.WOUNDED
return HPStatus.CRITICAL
@computed_field
@property
def is_bloodied(self) -> bool:
"""Whether combatant is at 50% HP or below."""
return self.hp_percent <= 50.0
class CombatState(BaseModel):
"""
State of an active combat encounter.
Tracks round, turn order, and all combatants.
"""
combat_id: str = Field(
default_factory=lambda: str(uuid.uuid4()),
description="Unique combat identifier",
)
round_number: int = Field(
default=1,
description="Current combat round",
)
turn_index: int = Field(
default=0,
description="Index of current combatant in turn order",
)
combatants: list[Combatant] = Field(
default_factory=list,
description="All combatants in initiative order",
)
started_at: datetime = Field(
default_factory=datetime.now,
description="When combat started",
)
@computed_field
@property
def current_combatant(self) -> Optional[Combatant]:
"""Get the combatant whose turn it is."""
if not self.combatants or self.turn_index >= len(self.combatants):
return None
return self.combatants[self.turn_index]
@computed_field
@property
def turn_order(self) -> list[str]:
"""List of combatant names in initiative order."""
return [c.name for c in self.combatants]
@computed_field
@property
def active_combatants(self) -> list[Combatant]:
"""List of combatants still active in combat."""
return [c for c in self.combatants if c.status == CombatantStatus.ACTIVE]
@computed_field
@property
def is_player_turn(self) -> bool:
"""Whether it's currently a player's turn."""
current = self.current_combatant
return current.is_player if current else False
def advance_turn(self) -> Optional[Combatant]:
"""
Advance to the next turn, skipping inactive combatants.
Returns:
The new current combatant, or None if combat should end.
"""
if not self.active_combatants:
return None
# Find next active combatant
start_index = self.turn_index
attempts = 0
max_attempts = len(self.combatants)
while attempts < max_attempts:
self.turn_index = (self.turn_index + 1) % len(self.combatants)
# Check for new round
if self.turn_index == 0:
self.round_number += 1
current = self.combatants[self.turn_index]
if current.status == CombatantStatus.ACTIVE:
return current
attempts += 1
return None
def get_combatant(self, combatant_id: str) -> Optional[Combatant]:
"""Get a combatant by ID."""
for c in self.combatants:
if c.combatant_id == combatant_id:
return c
return None
def update_combatant(self, combatant_id: str, **updates: object) -> bool:
"""
Update a combatant's attributes.
Args:
combatant_id: ID of combatant to update
**updates: Attribute updates
Returns:
True if combatant was found and updated
"""
for i, c in enumerate(self.combatants):
if c.combatant_id == combatant_id:
updated_data = c.model_dump()
updated_data.update(updates)
self.combatants[i] = Combatant.model_validate(updated_data)
return True
return False
# =============================================================================
# Character Models
# =============================================================================
class CharacterSnapshot(BaseModel):
"""
Cached snapshot of character data from MCP.
Used for quick access without hitting MCP on every request.
"""
character_id: str = Field(
description="Character ID from MCP",
)
name: str = Field(
description="Character name",
)
race: str = Field(
default="Unknown",
description="Character race",
)
character_class: str = Field(
default="Unknown",
description="Character class",
)
level: int = Field(
default=1,
description="Character level",
)
hp_current: int = Field(
default=0,
description="Current hit points",
)
hp_max: int = Field(
default=1,
description="Maximum hit points",
)
armor_class: int = Field(
default=10,
description="Armor class",
)
initiative_bonus: int = Field(
default=0,
description="Initiative modifier",
)
speed: int = Field(
default=30,
description="Movement speed in feet",
)
conditions: list[str] = Field(
default_factory=list,
description="Active conditions",
)
ability_scores: dict[str, int] = Field(
default_factory=lambda: {
"strength": 10,
"dexterity": 10,
"constitution": 10,
"intelligence": 10,
"wisdom": 10,
"charisma": 10,
},
description="Ability scores",
)
proficiency_bonus: int = Field(
default=2,
description="Proficiency bonus",
)
cached_at: datetime = Field(
default_factory=datetime.now,
description="When this snapshot was created",
)
@computed_field
@property
def hp_percent(self) -> float:
"""HP as percentage of max."""
if self.hp_max <= 0:
return 0.0
return (self.hp_current / self.hp_max) * 100.0
@computed_field
@property
def hp_status(self) -> HPStatus:
"""Health status based on HP percentage."""
if self.hp_current <= 0:
return HPStatus.UNCONSCIOUS
pct = self.hp_percent
if pct > 50:
return HPStatus.HEALTHY
if pct > 25:
return HPStatus.WOUNDED
return HPStatus.CRITICAL
@computed_field
@property
def is_bloodied(self) -> bool:
"""Whether character is at 50% HP or below."""
return self.hp_percent <= 50.0
@classmethod
def from_mcp_result(cls, data: dict[str, object]) -> CharacterSnapshot:
"""
Create a CharacterSnapshot from MCP get_character result.
Args:
data: Raw result from mcp_get_character
Returns:
CharacterSnapshot instance
"""
# Handle nested character data
char_data = data.get("character", data)
# Extract ability scores
ability_scores = {}
raw_abilities = char_data.get("ability_scores", {})
if isinstance(raw_abilities, dict):
for ability in [
"strength",
"dexterity",
"constitution",
"intelligence",
"wisdom",
"charisma",
]:
ability_scores[ability] = int(raw_abilities.get(ability, 10))
else:
ability_scores = {
"strength": 10,
"dexterity": 10,
"constitution": 10,
"intelligence": 10,
"wisdom": 10,
"charisma": 10,
}
return cls(
character_id=str(char_data.get("id", "")),
name=str(char_data.get("name", "Unknown")),
race=str(char_data.get("race", "Unknown")),
character_class=str(char_data.get("character_class", "Unknown")),
level=int(char_data.get("level", 1)),
hp_current=int(char_data.get("current_hp", char_data.get("hp_current", 0))),
hp_max=int(char_data.get("max_hp", char_data.get("hp_max", 1))),
armor_class=int(char_data.get("armor_class", 10)),
initiative_bonus=int(char_data.get("initiative_bonus", 0)),
speed=int(char_data.get("speed", 30)),
conditions=list(char_data.get("conditions", [])),
ability_scores=ability_scores,
proficiency_bonus=int(char_data.get("proficiency_bonus", 2)),
)
# =============================================================================
# NPC and Scene Models
# =============================================================================
class NPCInfo(BaseModel):
"""
Information about an NPC.
Includes personality, voice, and dialogue hooks.
"""
npc_id: str = Field(
description="Unique NPC identifier",
)
name: str = Field(
description="NPC name",
)
description: str = Field(
default="",
description="Physical and role description",
)
personality: str = Field(
default="",
description="Personality traits",
)
voice_profile: str = Field(
default="dm",
description="Voice profile for TTS",
)
dialogue_hooks: list[str] = Field(
default_factory=list,
description="Sample dialogue lines",
)
monster_stat_block: Optional[str] = Field(
default=None,
description="Monster stat block name if applicable",
)
relationship: str = Field(
default="neutral",
description="Relationship to players (friendly, neutral, hostile)",
)
class SceneInfo(BaseModel):
"""
Information about a location/scene.
Includes description, sensory details, exits, and present NPCs.
"""
scene_id: str = Field(
description="Unique scene identifier",
)
name: str = Field(
description="Scene/location name",
)
description: str = Field(
default="",
description="Scene description",
)
sensory_details: dict[str, str] = Field(
default_factory=dict,
description="Sensory details (sight, sound, smell)",
)
exits: dict[str, str] = Field(
default_factory=dict,
description="Available exits (direction -> destination scene_id)",
)
npcs_present: list[str] = Field(
default_factory=list,
description="NPC IDs present in scene",
)
items: list[dict[str, object]] = Field(
default_factory=list,
description="Items in the scene",
)
encounter_id: Optional[str] = Field(
default=None,
description="Encounter ID if scene has combat",
)
searchable_objects: list[dict[str, object]] = Field(
default_factory=list,
description="Objects that can be searched/investigated",
)
# =============================================================================
# Save/Load Models
# =============================================================================
class GameSaveData(BaseModel):
"""
Complete game state for save/load.
Includes all state needed to restore a game session.
"""
version: str = Field(
default="1.0.0",
description="Save file version",
)
saved_at: datetime = Field(
default_factory=datetime.now,
description="When the game was saved",
)
session_id: str = Field(
description="Game session ID",
)
turn_count: int = Field(
default=0,
description="Current turn count",
)
party_ids: list[str] = Field(
default_factory=list,
description="Character IDs in party",
)
active_character_id: Optional[str] = Field(
default=None,
description="Currently active character ID",
)
character_snapshots: list[CharacterSnapshot] = Field(
default_factory=list,
description="Cached character data",
)
current_location: str = Field(
default="Unknown",
description="Current location name",
)
current_scene: Optional[SceneInfo] = Field(
default=None,
description="Current scene data",
)
in_combat: bool = Field(
default=False,
description="Whether combat is active",
)
combat_state: Optional[CombatState] = Field(
default=None,
description="Combat state if in combat",
)
story_flags: dict[str, object] = Field(
default_factory=dict,
description="Story/quest progress flags",
)
known_npcs: dict[str, NPCInfo] = Field(
default_factory=dict,
description="NPCs encountered (id -> info)",
)
recent_events: list[SessionEvent] = Field(
default_factory=list,
description="Recent session events",
)
adventure_name: Optional[str] = Field(
default=None,
description="Loaded adventure name",
)
conversation_history: list[dict[str, object]] = Field(
default_factory=list,
description="Chat history for restoration",
)
# =============================================================================
# Adventure Models
# =============================================================================
class AdventureMetadata(BaseModel):
"""
Metadata for an adventure module.
"""
name: str = Field(
description="Adventure name",
)
description: str = Field(
default="",
description="Adventure description",
)
difficulty: str = Field(
default="medium",
description="Difficulty level (easy, medium, hard)",
)
estimated_time: str = Field(
default="1-2 hours",
description="Estimated play time",
)
recommended_level: int = Field(
default=1,
description="Recommended character level",
)
tags: list[str] = Field(
default_factory=list,
description="Adventure tags",
)
author: str = Field(
default="DungeonMaster AI",
description="Adventure author",
)
version: str = Field(
default="1.0.0",
description="Adventure version",
)
class EncounterData(BaseModel):
"""
Data for a combat encounter.
"""
encounter_id: str = Field(
description="Unique encounter identifier",
)
name: str = Field(
description="Encounter name",
)
description: str = Field(
default="",
description="Encounter description",
)
enemies: list[dict[str, object]] = Field(
default_factory=list,
description="Enemy definitions [{monster, count, name?}]",
)
difficulty: str = Field(
default="medium",
description="Encounter difficulty",
)
tactics: str = Field(
default="",
description="Enemy tactics description",
)
rewards: dict[str, object] = Field(
default_factory=dict,
description="Rewards (xp, loot)",
)
class AdventureData(BaseModel):
"""
Complete adventure data loaded from JSON.
"""
metadata: AdventureMetadata = Field(
description="Adventure metadata",
)
starting_scene: dict[str, object] = Field(
description="Starting scene configuration",
)
scenes: list[dict[str, object]] = Field(
default_factory=list,
description="All scenes in the adventure",
)
npcs: list[dict[str, object]] = Field(
default_factory=list,
description="NPC definitions",
)
encounters: list[EncounterData] = Field(
default_factory=list,
description="Combat encounters",
)
loot_tables: list[dict[str, object]] = Field(
default_factory=list,
description="Loot table definitions",
)
victory_conditions: dict[str, object] = Field(
default_factory=dict,
description="Win conditions",
)
completion_narrative: str = Field(
default="",
description="Narrative text when adventure is completed",
)
@classmethod
def from_json(cls, data: dict[str, object]) -> AdventureData:
"""
Create AdventureData from raw JSON data.
Args:
data: Parsed JSON adventure data
Returns:
AdventureData instance
"""
# Parse metadata
metadata_raw = data.get("metadata", {})
if isinstance(metadata_raw, dict):
metadata = AdventureMetadata.model_validate(metadata_raw)
else:
metadata = AdventureMetadata(name="Unknown Adventure")
# Parse encounters
encounters_raw = data.get("encounters", [])
encounters = []
if isinstance(encounters_raw, list):
for enc in encounters_raw:
if isinstance(enc, dict):
encounters.append(EncounterData.model_validate(enc))
return cls(
metadata=metadata,
starting_scene=dict(data.get("starting_scene", {})),
scenes=list(data.get("scenes", [])),
npcs=list(data.get("npcs", [])),
encounters=encounters,
loot_tables=list(data.get("loot_tables", [])),
victory_conditions=dict(data.get("victory_conditions", {})),
completion_narrative=str(data.get("completion_narrative", "")),
)
def get_scene(self, scene_id: str) -> Optional[dict[str, object]]:
"""Get a scene by ID."""
for scene in self.scenes:
if isinstance(scene, dict) and scene.get("scene_id") == scene_id:
return scene
return None
def get_npc(self, npc_id: str) -> Optional[dict[str, object]]:
"""Get an NPC by ID."""
for npc in self.npcs:
if isinstance(npc, dict) and npc.get("npc_id") == npc_id:
return npc
return None
def get_encounter(self, encounter_id: str) -> Optional[EncounterData]:
"""Get an encounter by ID."""
for enc in self.encounters:
if enc.encounter_id == encounter_id:
return enc
return None