DungeonMaster-AI / src /game /game_state.py
bhupesh-sf's picture
first commit
f8ba6bf verified
"""
DungeonMaster AI - Game State Management
Minimal GameState stub for Phase 1 (MCP Integration).
This provides the interface that tool wrappers need.
Phase 4 will extend this with full implementation.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol, runtime_checkable
@runtime_checkable
class GameStateProtocol(Protocol):
"""
Protocol for game state access.
This interface allows MCP tool wrappers to update game state
without depending on the full GameState implementation.
Phase 4 will provide the complete implementation.
"""
session_id: str
in_combat: bool
party: list[str]
recent_events: list[dict[str, object]]
def add_event(
self,
event_type: str,
description: str,
data: dict[str, object],
) -> None:
"""Add an event to recent events."""
...
def get_character(self, character_id: str) -> dict[str, object] | None:
"""Get character data from cache."""
...
def update_character_cache(
self,
character_id: str,
data: dict[str, object],
) -> None:
"""Update character data in cache."""
...
def set_combat_state(self, combat_state: dict[str, object] | None) -> None:
"""Set or clear combat state."""
...
@dataclass
class GameState:
"""
Minimal game state stub for Phase 1.
Provides the basic interface that MCP tool wrappers need.
Phase 4 will extend this with:
- Full Pydantic models (GameState, CombatState, etc.)
- GameStateManager class
- State persistence (save/load)
- Story context building
"""
# Core state
session_id: str = field(default_factory=lambda: str(uuid.uuid4()))
started_at: datetime = field(default_factory=datetime.now)
system: str = "dnd5e"
# Party management
party: list[str] = field(default_factory=list)
active_character_id: str | None = None
# Location
current_location: str = "Unknown"
current_scene: dict[str, object] = field(default_factory=dict)
# Combat
in_combat: bool = False
_combat_state: dict[str, object] | None = field(default=None, repr=False)
# Events
recent_events: list[dict[str, object]] = field(default_factory=list)
turn_count: int = 0
# Caches
_character_cache: dict[str, dict[str, object]] = field(
default_factory=dict,
repr=False,
)
_known_npcs: dict[str, dict[str, object]] = field(
default_factory=dict,
repr=False,
)
# Story tracking
story_flags: dict[str, object] = field(default_factory=dict)
current_adventure: str | None = None
# Metadata
last_updated: datetime = field(default_factory=datetime.now)
# Configuration
max_recent_events: int = field(default=20, repr=False)
def add_event(
self,
event_type: str,
description: str,
data: dict[str, object],
) -> None:
"""
Add an event to recent events list.
Keeps only the most recent events (default: 20).
Args:
event_type: Type of event (roll, combat, dialogue, etc.)
description: Human-readable description
data: Event-specific data
"""
event = {
"type": event_type,
"description": description,
"data": data,
"timestamp": datetime.now().isoformat(),
"turn": self.turn_count,
}
self.recent_events.append(event)
# Trim to max size
if len(self.recent_events) > self.max_recent_events:
self.recent_events = self.recent_events[-self.max_recent_events :]
self.last_updated = datetime.now()
def get_character(self, character_id: str) -> dict[str, object] | None:
"""
Get character data from cache.
Args:
character_id: Character ID to look up
Returns:
Character data dict or None if not cached
"""
return self._character_cache.get(character_id)
def update_character_cache(
self,
character_id: str,
data: dict[str, object],
) -> None:
"""
Update character data in cache.
Args:
character_id: Character ID
data: Character data to cache
"""
self._character_cache[character_id] = data
self.last_updated = datetime.now()
def set_combat_state(self, combat_state: dict[str, object] | None) -> None:
"""
Set or clear combat state.
Args:
combat_state: Combat state dict, or None to clear
"""
self._combat_state = combat_state
self.in_combat = combat_state is not None
self.last_updated = datetime.now()
# Log combat state change
if combat_state is not None:
self.add_event(
event_type="combat_start",
description="Combat has begun",
data={"combatants": combat_state.get("turn_order", [])},
)
else:
self.add_event(
event_type="combat_end",
description="Combat has ended",
data={},
)
@property
def combat_state(self) -> dict[str, object] | None:
"""Get current combat state."""
return self._combat_state
def add_character_to_party(self, character_id: str) -> None:
"""
Add a character to the party.
Args:
character_id: Character ID to add
"""
if character_id not in self.party:
self.party.append(character_id)
# Set as active if first character
if self.active_character_id is None:
self.active_character_id = character_id
self.last_updated = datetime.now()
def remove_character_from_party(self, character_id: str) -> None:
"""
Remove a character from the party.
Args:
character_id: Character ID to remove
"""
if character_id in self.party:
self.party.remove(character_id)
# Clear active if removed
if self.active_character_id == character_id:
self.active_character_id = self.party[0] if self.party else None
# Clear from cache
self._character_cache.pop(character_id, None)
self.last_updated = datetime.now()
def set_location(
self,
location: str,
scene: dict[str, object] | None = None,
) -> None:
"""
Update current location.
Args:
location: Location name/description
scene: Optional scene details
"""
self.current_location = location
if scene:
self.current_scene = scene
self.add_event(
event_type="movement",
description=f"Moved to {location}",
data={"location": location, "scene": scene or {}},
)
self.last_updated = datetime.now()
def increment_turn(self) -> int:
"""
Increment turn counter.
Returns:
New turn count
"""
self.turn_count += 1
self.last_updated = datetime.now()
return self.turn_count
def set_story_flag(self, flag: str, value: object) -> None:
"""
Set a story/quest flag.
Args:
flag: Flag name
value: Flag value
"""
self.story_flags[flag] = value
self.last_updated = datetime.now()
def get_story_flag(self, flag: str, default: object = None) -> object:
"""
Get a story/quest flag.
Args:
flag: Flag name
default: Default value if not set
Returns:
Flag value or default
"""
return self.story_flags.get(flag, default)
def add_known_npc(self, npc_id: str, data: dict[str, object]) -> None:
"""
Add or update a known NPC.
Args:
npc_id: NPC identifier
data: NPC data
"""
self._known_npcs[npc_id] = data
self.last_updated = datetime.now()
def get_known_npc(self, npc_id: str) -> dict[str, object] | None:
"""
Get known NPC data.
Args:
npc_id: NPC identifier
Returns:
NPC data or None
"""
return self._known_npcs.get(npc_id)
def get_recent_events_by_type(
self,
event_type: str,
limit: int = 10,
) -> list[dict[str, object]]:
"""
Get recent events filtered by type.
Args:
event_type: Event type to filter
limit: Maximum events to return
Returns:
List of matching events
"""
matching = [e for e in self.recent_events if e.get("type") == event_type]
return matching[-limit:]
def to_summary(self) -> dict[str, object]:
"""
Create a summary dict for LLM context.
Returns:
Summary dict with key state information
"""
return {
"session_id": self.session_id,
"system": self.system,
"turn_count": self.turn_count,
"party_size": len(self.party),
"active_character": self.active_character_id,
"location": self.current_location,
"in_combat": self.in_combat,
"combat_round": (
self._combat_state.get("round") if self._combat_state else None
),
"recent_events_count": len(self.recent_events),
"adventure": self.current_adventure,
}
def reset(self) -> None:
"""Reset game state for new game."""
self.session_id = str(uuid.uuid4())
self.started_at = datetime.now()
self.party.clear()
self.active_character_id = None
self.current_location = "Unknown"
self.current_scene.clear()
self.in_combat = False
self._combat_state = None
self.recent_events.clear()
self.turn_count = 0
self._character_cache.clear()
self._known_npcs.clear()
self.story_flags.clear()
self.current_adventure = None
self.last_updated = datetime.now()