327 lines
10 KiB
Python
327 lines
10 KiB
Python
"""
|
|
LocationLoader service for loading location definitions from YAML files.
|
|
|
|
This service reads location configuration files and converts them into Location
|
|
dataclass instances, providing caching for performance. Locations are organized
|
|
by region subdirectories.
|
|
"""
|
|
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
import structlog
|
|
|
|
from app.models.location import Location, Region
|
|
from app.models.enums import LocationType
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class LocationLoader:
|
|
"""
|
|
Loads location definitions from YAML configuration files.
|
|
|
|
Locations are organized in region subdirectories:
|
|
/app/data/locations/
|
|
regions/
|
|
crossville.yaml
|
|
crossville/
|
|
crossville_village.yaml
|
|
crossville_tavern.yaml
|
|
|
|
This allows game designers to define world locations without touching code.
|
|
"""
|
|
|
|
def __init__(self, data_dir: Optional[str] = None):
|
|
"""
|
|
Initialize the location loader.
|
|
|
|
Args:
|
|
data_dir: Path to directory containing location YAML files.
|
|
Defaults to /app/data/locations/
|
|
"""
|
|
if data_dir is None:
|
|
# Default to app/data/locations relative to this file
|
|
current_file = Path(__file__)
|
|
app_dir = current_file.parent.parent # Go up to /app
|
|
data_dir = str(app_dir / "data" / "locations")
|
|
|
|
self.data_dir = Path(data_dir)
|
|
self._location_cache: Dict[str, Location] = {}
|
|
self._region_cache: Dict[str, Region] = {}
|
|
|
|
logger.info("LocationLoader initialized", data_dir=str(self.data_dir))
|
|
|
|
def load_location(self, location_id: str) -> Optional[Location]:
|
|
"""
|
|
Load a single location by ID.
|
|
|
|
Searches all region subdirectories for the location file.
|
|
|
|
Args:
|
|
location_id: Unique location identifier (e.g., "crossville_tavern")
|
|
|
|
Returns:
|
|
Location instance or None if not found
|
|
"""
|
|
# Check cache first
|
|
if location_id in self._location_cache:
|
|
logger.debug("Location loaded from cache", location_id=location_id)
|
|
return self._location_cache[location_id]
|
|
|
|
# Search in region subdirectories
|
|
if not self.data_dir.exists():
|
|
logger.error("Location data directory does not exist", data_dir=str(self.data_dir))
|
|
return None
|
|
|
|
for region_dir in self.data_dir.iterdir():
|
|
# Skip non-directories and the regions folder
|
|
if not region_dir.is_dir() or region_dir.name == "regions":
|
|
continue
|
|
|
|
file_path = region_dir / f"{location_id}.yaml"
|
|
if file_path.exists():
|
|
return self._load_location_file(file_path)
|
|
|
|
logger.warning("Location not found", location_id=location_id)
|
|
return None
|
|
|
|
def _load_location_file(self, file_path: Path) -> Optional[Location]:
|
|
"""
|
|
Load a location from a specific file.
|
|
|
|
Args:
|
|
file_path: Path to the YAML file
|
|
|
|
Returns:
|
|
Location instance or None if loading fails
|
|
"""
|
|
try:
|
|
with open(file_path, 'r') as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
location = self._parse_location_data(data)
|
|
self._location_cache[location.location_id] = location
|
|
|
|
logger.info("Location loaded successfully", location_id=location.location_id)
|
|
return location
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to load location", file=str(file_path), error=str(e))
|
|
return None
|
|
|
|
def _parse_location_data(self, data: Dict) -> Location:
|
|
"""
|
|
Parse YAML data into a Location dataclass.
|
|
|
|
Args:
|
|
data: Dictionary loaded from YAML file
|
|
|
|
Returns:
|
|
Location instance
|
|
|
|
Raises:
|
|
ValueError: If data is invalid or missing required fields
|
|
"""
|
|
# Validate required fields
|
|
required_fields = ["location_id", "name", "region_id", "description"]
|
|
for field in required_fields:
|
|
if field not in data:
|
|
raise ValueError(f"Missing required field: {field}")
|
|
|
|
# Parse location type - default to town
|
|
location_type_str = data.get("location_type", "town")
|
|
try:
|
|
location_type = LocationType(location_type_str)
|
|
except ValueError:
|
|
logger.warning(
|
|
"Invalid location type, defaulting to town",
|
|
location_id=data["location_id"],
|
|
invalid_type=location_type_str
|
|
)
|
|
location_type = LocationType.TOWN
|
|
|
|
return Location(
|
|
location_id=data["location_id"],
|
|
name=data["name"],
|
|
location_type=location_type,
|
|
region_id=data["region_id"],
|
|
description=data["description"],
|
|
lore=data.get("lore"),
|
|
ambient_description=data.get("ambient_description"),
|
|
available_quests=data.get("available_quests", []),
|
|
npc_ids=data.get("npc_ids", []),
|
|
discoverable_locations=data.get("discoverable_locations", []),
|
|
is_starting_location=data.get("is_starting_location", False),
|
|
tags=data.get("tags", []),
|
|
)
|
|
|
|
def load_all_locations(self) -> List[Location]:
|
|
"""
|
|
Load all locations from all region directories.
|
|
|
|
Returns:
|
|
List of Location instances
|
|
"""
|
|
locations = []
|
|
|
|
if not self.data_dir.exists():
|
|
logger.error("Location data directory does not exist", data_dir=str(self.data_dir))
|
|
return locations
|
|
|
|
for region_dir in self.data_dir.iterdir():
|
|
# Skip non-directories and the regions folder
|
|
if not region_dir.is_dir() or region_dir.name == "regions":
|
|
continue
|
|
|
|
for file_path in region_dir.glob("*.yaml"):
|
|
location = self._load_location_file(file_path)
|
|
if location:
|
|
locations.append(location)
|
|
|
|
logger.info("All locations loaded", count=len(locations))
|
|
return locations
|
|
|
|
def load_region(self, region_id: str) -> Optional[Region]:
|
|
"""
|
|
Load a region definition.
|
|
|
|
Args:
|
|
region_id: Unique region identifier (e.g., "crossville")
|
|
|
|
Returns:
|
|
Region instance or None if not found
|
|
"""
|
|
# Check cache first
|
|
if region_id in self._region_cache:
|
|
logger.debug("Region loaded from cache", region_id=region_id)
|
|
return self._region_cache[region_id]
|
|
|
|
file_path = self.data_dir / "regions" / f"{region_id}.yaml"
|
|
|
|
if not file_path.exists():
|
|
logger.warning("Region file not found", region_id=region_id)
|
|
return None
|
|
|
|
try:
|
|
with open(file_path, 'r') as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
region = Region.from_dict(data)
|
|
self._region_cache[region_id] = region
|
|
|
|
logger.info("Region loaded successfully", region_id=region_id)
|
|
return region
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to load region", region_id=region_id, error=str(e))
|
|
return None
|
|
|
|
def get_locations_in_region(self, region_id: str) -> List[Location]:
|
|
"""
|
|
Get all locations belonging to a specific region.
|
|
|
|
Args:
|
|
region_id: Region identifier
|
|
|
|
Returns:
|
|
List of Location instances in this region
|
|
"""
|
|
# Load all locations if cache is empty
|
|
if not self._location_cache:
|
|
self.load_all_locations()
|
|
|
|
return [
|
|
loc for loc in self._location_cache.values()
|
|
if loc.region_id == region_id
|
|
]
|
|
|
|
def get_starting_locations(self) -> List[Location]:
|
|
"""
|
|
Get all locations that can be starting points.
|
|
|
|
Returns:
|
|
List of Location instances marked as starting locations
|
|
"""
|
|
# Load all locations if cache is empty
|
|
if not self._location_cache:
|
|
self.load_all_locations()
|
|
|
|
return [
|
|
loc for loc in self._location_cache.values()
|
|
if loc.is_starting_location
|
|
]
|
|
|
|
def get_location_by_type(self, location_type: LocationType) -> List[Location]:
|
|
"""
|
|
Get all locations of a specific type.
|
|
|
|
Args:
|
|
location_type: Type to filter by
|
|
|
|
Returns:
|
|
List of Location instances of this type
|
|
"""
|
|
# Load all locations if cache is empty
|
|
if not self._location_cache:
|
|
self.load_all_locations()
|
|
|
|
return [
|
|
loc for loc in self._location_cache.values()
|
|
if loc.location_type == location_type
|
|
]
|
|
|
|
def get_all_location_ids(self) -> List[str]:
|
|
"""
|
|
Get a list of all available location IDs.
|
|
|
|
Returns:
|
|
List of location IDs
|
|
"""
|
|
# Load all locations if cache is empty
|
|
if not self._location_cache:
|
|
self.load_all_locations()
|
|
|
|
return list(self._location_cache.keys())
|
|
|
|
def reload_location(self, location_id: str) -> Optional[Location]:
|
|
"""
|
|
Force reload a location from disk, bypassing cache.
|
|
|
|
Useful for development/testing when location definitions change.
|
|
|
|
Args:
|
|
location_id: Unique location identifier
|
|
|
|
Returns:
|
|
Location instance or None if not found
|
|
"""
|
|
# Remove from cache if present
|
|
if location_id in self._location_cache:
|
|
del self._location_cache[location_id]
|
|
|
|
return self.load_location(location_id)
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Clear all cached data. Useful for testing."""
|
|
self._location_cache.clear()
|
|
self._region_cache.clear()
|
|
logger.info("Location cache cleared")
|
|
|
|
|
|
# Global singleton instance
|
|
_loader_instance: Optional[LocationLoader] = None
|
|
|
|
|
|
def get_location_loader() -> LocationLoader:
|
|
"""
|
|
Get the global LocationLoader instance.
|
|
|
|
Returns:
|
|
Singleton LocationLoader instance
|
|
"""
|
|
global _loader_instance
|
|
if _loader_instance is None:
|
|
_loader_instance = LocationLoader()
|
|
return _loader_instance
|