412 lines
14 KiB
Python
412 lines
14 KiB
Python
"""
|
|
Game session data models.
|
|
|
|
This module defines the session-related dataclasses including SessionConfig,
|
|
GameState, and GameSession which manage multiplayer party sessions.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime, timezone
|
|
|
|
from app.models.combat import CombatEncounter
|
|
from app.models.enums import SessionStatus, SessionType
|
|
from app.models.action_prompt import LocationType
|
|
|
|
|
|
@dataclass
|
|
class SessionConfig:
|
|
"""
|
|
Configuration settings for a game session.
|
|
|
|
Attributes:
|
|
min_players: Minimum players required (session ends if below this)
|
|
timeout_minutes: Inactivity timeout in minutes
|
|
auto_save_interval: Turns between automatic saves
|
|
"""
|
|
|
|
min_players: int = 1
|
|
timeout_minutes: int = 30
|
|
auto_save_interval: int = 5
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize configuration to dictionary."""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'SessionConfig':
|
|
"""Deserialize configuration from dictionary."""
|
|
return cls(
|
|
min_players=data.get("min_players", 1),
|
|
timeout_minutes=data.get("timeout_minutes", 30),
|
|
auto_save_interval=data.get("auto_save_interval", 5),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class GameState:
|
|
"""
|
|
Current world/quest state for a game session.
|
|
|
|
Attributes:
|
|
current_location: Current location name/ID
|
|
location_type: Type of current location (town, tavern, wilderness, etc.)
|
|
discovered_locations: All location IDs the party has visited
|
|
active_quests: Quest IDs currently in progress
|
|
world_events: Server-wide events affecting this session
|
|
"""
|
|
|
|
current_location: str = "crossville_village"
|
|
location_type: LocationType = LocationType.TOWN
|
|
discovered_locations: List[str] = field(default_factory=list)
|
|
active_quests: List[str] = field(default_factory=list)
|
|
world_events: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize game state to dictionary."""
|
|
return {
|
|
"current_location": self.current_location,
|
|
"location_type": self.location_type.value,
|
|
"discovered_locations": self.discovered_locations,
|
|
"active_quests": self.active_quests,
|
|
"world_events": self.world_events,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'GameState':
|
|
"""Deserialize game state from dictionary."""
|
|
# Handle location_type as either string or enum
|
|
location_type_value = data.get("location_type", "town")
|
|
if isinstance(location_type_value, str):
|
|
location_type = LocationType(location_type_value)
|
|
else:
|
|
location_type = location_type_value
|
|
|
|
return cls(
|
|
current_location=data.get("current_location", "crossville_village"),
|
|
location_type=location_type,
|
|
discovered_locations=data.get("discovered_locations", []),
|
|
active_quests=data.get("active_quests", []),
|
|
world_events=data.get("world_events", []),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ConversationEntry:
|
|
"""
|
|
Single entry in the conversation history.
|
|
|
|
Attributes:
|
|
turn: Turn number
|
|
character_id: Acting character's ID
|
|
character_name: Acting character's name
|
|
action: Player's action/input text
|
|
dm_response: AI Dungeon Master's response
|
|
timestamp: ISO timestamp of when entry was created
|
|
combat_log: Combat actions if any occurred this turn
|
|
quest_offered: Quest offering info if a quest was offered this turn
|
|
"""
|
|
|
|
turn: int
|
|
character_id: str
|
|
character_name: str
|
|
action: str
|
|
dm_response: str
|
|
timestamp: str = ""
|
|
combat_log: List[Dict[str, Any]] = field(default_factory=list)
|
|
quest_offered: Optional[Dict[str, Any]] = None
|
|
|
|
def __post_init__(self):
|
|
"""Initialize timestamp if not provided."""
|
|
if not self.timestamp:
|
|
self.timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize conversation entry to dictionary."""
|
|
result = {
|
|
"turn": self.turn,
|
|
"character_id": self.character_id,
|
|
"character_name": self.character_name,
|
|
"action": self.action,
|
|
"dm_response": self.dm_response,
|
|
"timestamp": self.timestamp,
|
|
"combat_log": self.combat_log,
|
|
}
|
|
if self.quest_offered:
|
|
result["quest_offered"] = self.quest_offered
|
|
return result
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'ConversationEntry':
|
|
"""Deserialize conversation entry from dictionary."""
|
|
return cls(
|
|
turn=data["turn"],
|
|
character_id=data.get("character_id", ""),
|
|
character_name=data.get("character_name", ""),
|
|
action=data["action"],
|
|
dm_response=data["dm_response"],
|
|
timestamp=data.get("timestamp", ""),
|
|
combat_log=data.get("combat_log", []),
|
|
quest_offered=data.get("quest_offered"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class GameSession:
|
|
"""
|
|
Represents a game session (solo or multiplayer).
|
|
|
|
A session can have one or more players (party) and tracks the entire
|
|
game state including conversation history, combat encounters, and
|
|
turn order.
|
|
|
|
Attributes:
|
|
session_id: Unique identifier
|
|
session_type: Type of session (solo or multiplayer)
|
|
solo_character_id: Character ID for single-player sessions (None for multiplayer)
|
|
user_id: Owner of the session
|
|
party_member_ids: Character IDs in this party (multiplayer only)
|
|
config: Session configuration settings
|
|
combat_encounter: Current combat (None if not in combat)
|
|
conversation_history: Turn-by-turn log of actions and DM responses
|
|
game_state: Current world/quest state
|
|
turn_order: Character turn order
|
|
current_turn: Index in turn_order for current turn
|
|
turn_number: Global turn counter
|
|
created_at: ISO timestamp of session creation
|
|
last_activity: ISO timestamp of last action
|
|
status: Current session status (active, completed, timeout)
|
|
"""
|
|
|
|
session_id: str
|
|
session_type: SessionType = SessionType.SOLO
|
|
solo_character_id: Optional[str] = None
|
|
user_id: str = ""
|
|
party_member_ids: List[str] = field(default_factory=list)
|
|
config: SessionConfig = field(default_factory=SessionConfig)
|
|
combat_encounter: Optional[CombatEncounter] = None
|
|
conversation_history: List[ConversationEntry] = field(default_factory=list)
|
|
game_state: GameState = field(default_factory=GameState)
|
|
turn_order: List[str] = field(default_factory=list)
|
|
current_turn: int = 0
|
|
turn_number: int = 0
|
|
created_at: str = ""
|
|
last_activity: str = ""
|
|
status: SessionStatus = SessionStatus.ACTIVE
|
|
|
|
def __post_init__(self):
|
|
"""Initialize timestamps if not provided."""
|
|
if not self.created_at:
|
|
self.created_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
if not self.last_activity:
|
|
self.last_activity = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
def is_in_combat(self) -> bool:
|
|
"""Check if session is currently in combat."""
|
|
return self.combat_encounter is not None
|
|
|
|
def start_combat(self, encounter: CombatEncounter) -> None:
|
|
"""
|
|
Start a combat encounter.
|
|
|
|
Args:
|
|
encounter: The combat encounter to begin
|
|
"""
|
|
self.combat_encounter = encounter
|
|
self.update_activity()
|
|
|
|
def end_combat(self) -> None:
|
|
"""End the current combat encounter."""
|
|
self.combat_encounter = None
|
|
self.update_activity()
|
|
|
|
def advance_turn(self) -> str:
|
|
"""
|
|
Advance to the next player's turn.
|
|
|
|
Returns:
|
|
Character ID whose turn it now is
|
|
"""
|
|
if not self.turn_order:
|
|
return ""
|
|
|
|
self.current_turn = (self.current_turn + 1) % len(self.turn_order)
|
|
self.turn_number += 1
|
|
self.update_activity()
|
|
|
|
return self.turn_order[self.current_turn]
|
|
|
|
def get_current_character_id(self) -> Optional[str]:
|
|
"""Get the character ID whose turn it currently is."""
|
|
if not self.turn_order:
|
|
return None
|
|
return self.turn_order[self.current_turn]
|
|
|
|
def add_conversation_entry(self, entry: ConversationEntry) -> None:
|
|
"""
|
|
Add an entry to the conversation history.
|
|
|
|
Args:
|
|
entry: Conversation entry to add
|
|
"""
|
|
self.conversation_history.append(entry)
|
|
self.update_activity()
|
|
|
|
def update_activity(self) -> None:
|
|
"""Update the last activity timestamp to now."""
|
|
self.last_activity = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
def add_party_member(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_member_ids:
|
|
self.party_member_ids.append(character_id)
|
|
self.update_activity()
|
|
|
|
def remove_party_member(self, character_id: str) -> None:
|
|
"""
|
|
Remove a character from the party.
|
|
|
|
Args:
|
|
character_id: Character ID to remove
|
|
"""
|
|
if character_id in self.party_member_ids:
|
|
self.party_member_ids.remove(character_id)
|
|
# Also remove from turn order
|
|
if character_id in self.turn_order:
|
|
self.turn_order.remove(character_id)
|
|
self.update_activity()
|
|
|
|
def check_timeout(self) -> bool:
|
|
"""
|
|
Check if session has timed out due to inactivity.
|
|
|
|
Returns:
|
|
True if session should be marked as timed out
|
|
"""
|
|
if self.status != SessionStatus.ACTIVE:
|
|
return False
|
|
|
|
# Calculate time since last activity
|
|
last_activity_str = self.last_activity.replace("Z", "+00:00")
|
|
last_activity_time = datetime.fromisoformat(last_activity_str)
|
|
now = datetime.now(timezone.utc)
|
|
elapsed_minutes = (now - last_activity_time).total_seconds() / 60
|
|
|
|
if elapsed_minutes >= self.config.timeout_minutes:
|
|
self.status = SessionStatus.TIMEOUT
|
|
return True
|
|
|
|
return False
|
|
|
|
def check_min_players(self) -> bool:
|
|
"""
|
|
Check if session still has minimum required players.
|
|
|
|
Returns:
|
|
True if session should continue, False if it should end
|
|
"""
|
|
if len(self.party_member_ids) < self.config.min_players:
|
|
if self.status == SessionStatus.ACTIVE:
|
|
self.status = SessionStatus.COMPLETED
|
|
return False
|
|
return True
|
|
|
|
def is_solo(self) -> bool:
|
|
"""Check if this is a solo session."""
|
|
return self.session_type == SessionType.SOLO
|
|
|
|
def get_character_id(self) -> Optional[str]:
|
|
"""
|
|
Get the primary character ID for the session.
|
|
|
|
For solo sessions, returns solo_character_id.
|
|
For multiplayer, returns the current character in turn order.
|
|
"""
|
|
if self.is_solo():
|
|
return self.solo_character_id
|
|
return self.get_current_character_id()
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize game session to dictionary."""
|
|
return {
|
|
"session_id": self.session_id,
|
|
"session_type": self.session_type.value,
|
|
"solo_character_id": self.solo_character_id,
|
|
"user_id": self.user_id,
|
|
"party_member_ids": self.party_member_ids,
|
|
"config": self.config.to_dict(),
|
|
"combat_encounter": self.combat_encounter.to_dict() if self.combat_encounter else None,
|
|
"conversation_history": [entry.to_dict() for entry in self.conversation_history],
|
|
"game_state": self.game_state.to_dict(),
|
|
"turn_order": self.turn_order,
|
|
"current_turn": self.current_turn,
|
|
"turn_number": self.turn_number,
|
|
"created_at": self.created_at,
|
|
"last_activity": self.last_activity,
|
|
"status": self.status.value,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'GameSession':
|
|
"""Deserialize game session from dictionary."""
|
|
config = SessionConfig.from_dict(data.get("config", {}))
|
|
game_state = GameState.from_dict(data.get("game_state", {}))
|
|
conversation_history = [
|
|
ConversationEntry.from_dict(entry)
|
|
for entry in data.get("conversation_history", [])
|
|
]
|
|
|
|
combat_encounter = None
|
|
if data.get("combat_encounter"):
|
|
combat_encounter = CombatEncounter.from_dict(data["combat_encounter"])
|
|
|
|
status = SessionStatus(data.get("status", "active"))
|
|
|
|
# Handle session_type as either string or enum
|
|
session_type_value = data.get("session_type", "solo")
|
|
if isinstance(session_type_value, str):
|
|
session_type = SessionType(session_type_value)
|
|
else:
|
|
session_type = session_type_value
|
|
|
|
return cls(
|
|
session_id=data["session_id"],
|
|
session_type=session_type,
|
|
solo_character_id=data.get("solo_character_id"),
|
|
user_id=data.get("user_id", ""),
|
|
party_member_ids=data.get("party_member_ids", []),
|
|
config=config,
|
|
combat_encounter=combat_encounter,
|
|
conversation_history=conversation_history,
|
|
game_state=game_state,
|
|
turn_order=data.get("turn_order", []),
|
|
current_turn=data.get("current_turn", 0),
|
|
turn_number=data.get("turn_number", 0),
|
|
created_at=data.get("created_at", ""),
|
|
last_activity=data.get("last_activity", ""),
|
|
status=status,
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
"""String representation of the session."""
|
|
if self.is_solo():
|
|
return (
|
|
f"GameSession({self.session_id}, "
|
|
f"type=solo, "
|
|
f"char={self.solo_character_id}, "
|
|
f"turn={self.turn_number}, "
|
|
f"status={self.status.value})"
|
|
)
|
|
return (
|
|
f"GameSession({self.session_id}, "
|
|
f"type=multiplayer, "
|
|
f"party={len(self.party_member_ids)}, "
|
|
f"turn={self.turn_number}, "
|
|
f"status={self.status.value})"
|
|
)
|