237 lines
7.3 KiB
Python
237 lines
7.3 KiB
Python
"""
|
|
OriginService for loading character origin definitions from YAML files.
|
|
|
|
This service reads origin configuration and converts it into Origin
|
|
dataclass instances, providing caching for performance.
|
|
"""
|
|
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
import structlog
|
|
|
|
from app.models.origins import Origin, StartingLocation, StartingBonus
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class OriginService:
|
|
"""
|
|
Loads character origin definitions from YAML configuration.
|
|
|
|
Origins define character backstories, starting locations, and narrative
|
|
hooks that the AI DM uses to create personalized gameplay experiences.
|
|
All origin definitions are stored in /app/data/origins.yaml.
|
|
"""
|
|
|
|
def __init__(self, data_file: Optional[str] = None):
|
|
"""
|
|
Initialize the origin service.
|
|
|
|
Args:
|
|
data_file: Path to origins YAML file.
|
|
Defaults to /app/data/origins.yaml
|
|
"""
|
|
if data_file is None:
|
|
# Default to app/data/origins.yaml relative to this file
|
|
current_file = Path(__file__)
|
|
app_dir = current_file.parent.parent # Go up to /app
|
|
data_file = str(app_dir / "data" / "origins.yaml")
|
|
|
|
self.data_file = Path(data_file)
|
|
self._origins_cache: Dict[str, Origin] = {}
|
|
self._all_origins_loaded = False
|
|
|
|
logger.info("OriginService initialized", data_file=str(self.data_file))
|
|
|
|
def load_origin(self, origin_id: str) -> Optional[Origin]:
|
|
"""
|
|
Load a single origin by ID.
|
|
|
|
Args:
|
|
origin_id: Unique origin identifier (e.g., "soul_revenant")
|
|
|
|
Returns:
|
|
Origin instance or None if not found
|
|
"""
|
|
# Check cache first
|
|
if origin_id in self._origins_cache:
|
|
logger.debug("Origin loaded from cache", origin_id=origin_id)
|
|
return self._origins_cache[origin_id]
|
|
|
|
# Load all origins if not already loaded
|
|
if not self._all_origins_loaded:
|
|
self._load_all_origins()
|
|
|
|
# Return from cache after loading
|
|
origin = self._origins_cache.get(origin_id)
|
|
if origin:
|
|
logger.info("Origin loaded successfully", origin_id=origin_id)
|
|
else:
|
|
logger.warning("Origin not found", origin_id=origin_id)
|
|
|
|
return origin
|
|
|
|
def load_all_origins(self) -> List[Origin]:
|
|
"""
|
|
Load all origins from the data file.
|
|
|
|
Returns:
|
|
List of Origin instances
|
|
"""
|
|
if self._all_origins_loaded and self._origins_cache:
|
|
logger.debug("All origins loaded from cache")
|
|
return list(self._origins_cache.values())
|
|
|
|
return self._load_all_origins()
|
|
|
|
def _load_all_origins(self) -> List[Origin]:
|
|
"""
|
|
Internal method to load all origins from YAML.
|
|
|
|
Returns:
|
|
List of Origin instances
|
|
"""
|
|
if not self.data_file.exists():
|
|
logger.error("Origins data file does not exist", data_file=str(self.data_file))
|
|
return []
|
|
|
|
try:
|
|
# Load YAML file
|
|
with open(self.data_file, 'r') as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
origins_data = data.get("origins", {})
|
|
origins = []
|
|
|
|
# Parse each origin
|
|
for origin_id, origin_data in origins_data.items():
|
|
try:
|
|
origin = self._parse_origin_data(origin_id, origin_data)
|
|
self._origins_cache[origin_id] = origin
|
|
origins.append(origin)
|
|
except Exception as e:
|
|
logger.error("Failed to parse origin", origin_id=origin_id, error=str(e))
|
|
continue
|
|
|
|
self._all_origins_loaded = True
|
|
logger.info("All origins loaded successfully", count=len(origins))
|
|
return origins
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to load origins file", error=str(e))
|
|
return []
|
|
|
|
def get_origin_by_id(self, origin_id: str) -> Optional[Origin]:
|
|
"""
|
|
Get an origin by ID (alias for load_origin).
|
|
|
|
Args:
|
|
origin_id: Unique origin identifier
|
|
|
|
Returns:
|
|
Origin instance or None if not found
|
|
"""
|
|
return self.load_origin(origin_id)
|
|
|
|
def get_all_origin_ids(self) -> List[str]:
|
|
"""
|
|
Get a list of all available origin IDs.
|
|
|
|
Returns:
|
|
List of origin IDs (e.g., ["soul_revenant", "memory_thief"])
|
|
"""
|
|
if not self._all_origins_loaded:
|
|
self._load_all_origins()
|
|
|
|
return list(self._origins_cache.keys())
|
|
|
|
def reload_origins(self) -> List[Origin]:
|
|
"""
|
|
Force reload all origins from disk, bypassing cache.
|
|
|
|
Useful for development/testing when origin definitions change.
|
|
|
|
Returns:
|
|
List of Origin instances
|
|
"""
|
|
self.clear_cache()
|
|
return self._load_all_origins()
|
|
|
|
def clear_cache(self):
|
|
"""Clear the origins cache. Useful for testing."""
|
|
self._origins_cache.clear()
|
|
self._all_origins_loaded = False
|
|
logger.info("Origins cache cleared")
|
|
|
|
def _parse_origin_data(self, origin_id: str, data: Dict) -> Origin:
|
|
"""
|
|
Parse YAML data into an Origin dataclass.
|
|
|
|
Args:
|
|
origin_id: The origin's unique identifier
|
|
data: Dictionary loaded from YAML file
|
|
|
|
Returns:
|
|
Origin instance
|
|
|
|
Raises:
|
|
ValueError: If data is invalid or missing required fields
|
|
"""
|
|
# Validate required fields
|
|
required_fields = ["name", "description", "starting_location"]
|
|
for field in required_fields:
|
|
if field not in data:
|
|
raise ValueError(f"Missing required field in origin '{origin_id}': {field}")
|
|
|
|
# Parse starting location
|
|
location_data = data["starting_location"]
|
|
starting_location = StartingLocation(
|
|
id=location_data.get("id", ""),
|
|
name=location_data.get("name", ""),
|
|
region=location_data.get("region", ""),
|
|
description=location_data.get("description", "")
|
|
)
|
|
|
|
# Parse starting bonus (optional)
|
|
starting_bonus = None
|
|
if "starting_bonus" in data:
|
|
bonus_data = data["starting_bonus"]
|
|
starting_bonus = StartingBonus(
|
|
trait=bonus_data.get("trait", ""),
|
|
description=bonus_data.get("description", ""),
|
|
effect=bonus_data.get("effect", "")
|
|
)
|
|
|
|
# Parse narrative hooks (optional)
|
|
narrative_hooks = data.get("narrative_hooks", [])
|
|
|
|
# Create Origin instance
|
|
origin = Origin(
|
|
id=data.get("id", origin_id), # Use provided ID or fall back to key
|
|
name=data["name"],
|
|
description=data["description"],
|
|
starting_location=starting_location,
|
|
narrative_hooks=narrative_hooks,
|
|
starting_bonus=starting_bonus
|
|
)
|
|
|
|
return origin
|
|
|
|
|
|
# Global instance for convenience
|
|
_service_instance: Optional[OriginService] = None
|
|
|
|
|
|
def get_origin_service() -> OriginService:
|
|
"""
|
|
Get the global OriginService instance.
|
|
|
|
Returns:
|
|
Singleton OriginService instance
|
|
"""
|
|
global _service_instance
|
|
if _service_instance is None:
|
|
_service_instance = OriginService()
|
|
return _service_instance
|