386 lines
11 KiB
Python
386 lines
11 KiB
Python
"""
|
|
NPCLoader service for loading NPC definitions from YAML files.
|
|
|
|
This service reads NPC configuration files and converts them into NPC
|
|
dataclass instances, providing caching for performance. NPCs are organized
|
|
by region subdirectories.
|
|
"""
|
|
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
import structlog
|
|
|
|
from app.models.npc import (
|
|
NPC,
|
|
NPCPersonality,
|
|
NPCAppearance,
|
|
NPCKnowledge,
|
|
NPCKnowledgeCondition,
|
|
NPCRelationship,
|
|
NPCInventoryItem,
|
|
NPCDialogueHooks,
|
|
)
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class NPCLoader:
|
|
"""
|
|
Loads NPC definitions from YAML configuration files.
|
|
|
|
NPCs are organized in region subdirectories:
|
|
/app/data/npcs/
|
|
crossville/
|
|
npc_grom_001.yaml
|
|
npc_mira_001.yaml
|
|
|
|
This allows game designers to define NPCs without touching code.
|
|
"""
|
|
|
|
def __init__(self, data_dir: Optional[str] = None):
|
|
"""
|
|
Initialize the NPC loader.
|
|
|
|
Args:
|
|
data_dir: Path to directory containing NPC YAML files.
|
|
Defaults to /app/data/npcs/
|
|
"""
|
|
if data_dir is None:
|
|
# Default to app/data/npcs relative to this file
|
|
current_file = Path(__file__)
|
|
app_dir = current_file.parent.parent # Go up to /app
|
|
data_dir = str(app_dir / "data" / "npcs")
|
|
|
|
self.data_dir = Path(data_dir)
|
|
self._npc_cache: Dict[str, NPC] = {}
|
|
self._location_npc_cache: Dict[str, List[str]] = {}
|
|
|
|
logger.info("NPCLoader initialized", data_dir=str(self.data_dir))
|
|
|
|
def load_npc(self, npc_id: str) -> Optional[NPC]:
|
|
"""
|
|
Load a single NPC by ID.
|
|
|
|
Searches all region subdirectories for the NPC file.
|
|
|
|
Args:
|
|
npc_id: Unique NPC identifier (e.g., "npc_grom_001")
|
|
|
|
Returns:
|
|
NPC instance or None if not found
|
|
"""
|
|
# Check cache first
|
|
if npc_id in self._npc_cache:
|
|
logger.debug("NPC loaded from cache", npc_id=npc_id)
|
|
return self._npc_cache[npc_id]
|
|
|
|
# Search in region subdirectories
|
|
if not self.data_dir.exists():
|
|
logger.error("NPC data directory does not exist", data_dir=str(self.data_dir))
|
|
return None
|
|
|
|
for region_dir in self.data_dir.iterdir():
|
|
if not region_dir.is_dir():
|
|
continue
|
|
|
|
file_path = region_dir / f"{npc_id}.yaml"
|
|
if file_path.exists():
|
|
return self._load_npc_file(file_path)
|
|
|
|
logger.warning("NPC not found", npc_id=npc_id)
|
|
return None
|
|
|
|
def _load_npc_file(self, file_path: Path) -> Optional[NPC]:
|
|
"""
|
|
Load an NPC from a specific file.
|
|
|
|
Args:
|
|
file_path: Path to the YAML file
|
|
|
|
Returns:
|
|
NPC instance or None if loading fails
|
|
"""
|
|
try:
|
|
with open(file_path, 'r') as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
npc = self._parse_npc_data(data)
|
|
self._npc_cache[npc.npc_id] = npc
|
|
|
|
# Update location cache
|
|
if npc.location_id not in self._location_npc_cache:
|
|
self._location_npc_cache[npc.location_id] = []
|
|
if npc.npc_id not in self._location_npc_cache[npc.location_id]:
|
|
self._location_npc_cache[npc.location_id].append(npc.npc_id)
|
|
|
|
logger.info("NPC loaded successfully", npc_id=npc.npc_id)
|
|
return npc
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to load NPC", file=str(file_path), error=str(e))
|
|
return None
|
|
|
|
def _parse_npc_data(self, data: Dict) -> NPC:
|
|
"""
|
|
Parse YAML data into an NPC dataclass.
|
|
|
|
Args:
|
|
data: Dictionary loaded from YAML file
|
|
|
|
Returns:
|
|
NPC instance
|
|
|
|
Raises:
|
|
ValueError: If data is invalid or missing required fields
|
|
"""
|
|
# Validate required fields
|
|
required_fields = ["npc_id", "name", "role", "location_id"]
|
|
for field in required_fields:
|
|
if field not in data:
|
|
raise ValueError(f"Missing required field: {field}")
|
|
|
|
# Parse personality
|
|
personality_data = data.get("personality", {})
|
|
personality = NPCPersonality(
|
|
traits=personality_data.get("traits", []),
|
|
speech_style=personality_data.get("speech_style", ""),
|
|
quirks=personality_data.get("quirks", []),
|
|
)
|
|
|
|
# Parse appearance
|
|
appearance_data = data.get("appearance", {})
|
|
if isinstance(appearance_data, str):
|
|
appearance = NPCAppearance(brief=appearance_data)
|
|
else:
|
|
appearance = NPCAppearance(
|
|
brief=appearance_data.get("brief", ""),
|
|
detailed=appearance_data.get("detailed"),
|
|
)
|
|
|
|
# Parse knowledge (optional)
|
|
knowledge = None
|
|
if data.get("knowledge"):
|
|
knowledge_data = data["knowledge"]
|
|
conditions = [
|
|
NPCKnowledgeCondition(
|
|
condition=c.get("condition", ""),
|
|
reveals=c.get("reveals", ""),
|
|
)
|
|
for c in knowledge_data.get("will_share_if", [])
|
|
]
|
|
knowledge = NPCKnowledge(
|
|
public=knowledge_data.get("public", []),
|
|
secret=knowledge_data.get("secret", []),
|
|
will_share_if=conditions,
|
|
)
|
|
|
|
# Parse relationships
|
|
relationships = [
|
|
NPCRelationship(
|
|
npc_id=r["npc_id"],
|
|
attitude=r["attitude"],
|
|
reason=r.get("reason"),
|
|
)
|
|
for r in data.get("relationships", [])
|
|
]
|
|
|
|
# Parse inventory
|
|
inventory = []
|
|
for item_data in data.get("inventory_for_sale", []):
|
|
# Handle shorthand format: { item: "ale", price: 2 }
|
|
item_id = item_data.get("item_id") or item_data.get("item", "")
|
|
inventory.append(NPCInventoryItem(
|
|
item_id=item_id,
|
|
price=item_data.get("price", 0),
|
|
quantity=item_data.get("quantity"),
|
|
))
|
|
|
|
# Parse dialogue hooks (optional)
|
|
dialogue_hooks = None
|
|
if data.get("dialogue_hooks"):
|
|
hooks_data = data["dialogue_hooks"]
|
|
dialogue_hooks = NPCDialogueHooks(
|
|
greeting=hooks_data.get("greeting"),
|
|
farewell=hooks_data.get("farewell"),
|
|
busy=hooks_data.get("busy"),
|
|
quest_complete=hooks_data.get("quest_complete"),
|
|
)
|
|
|
|
return NPC(
|
|
npc_id=data["npc_id"],
|
|
name=data["name"],
|
|
role=data["role"],
|
|
location_id=data["location_id"],
|
|
personality=personality,
|
|
appearance=appearance,
|
|
knowledge=knowledge,
|
|
relationships=relationships,
|
|
inventory_for_sale=inventory,
|
|
dialogue_hooks=dialogue_hooks,
|
|
quest_giver_for=data.get("quest_giver_for", []),
|
|
reveals_locations=data.get("reveals_locations", []),
|
|
tags=data.get("tags", []),
|
|
)
|
|
|
|
def load_all_npcs(self) -> List[NPC]:
|
|
"""
|
|
Load all NPCs from all region directories.
|
|
|
|
Returns:
|
|
List of NPC instances
|
|
"""
|
|
npcs = []
|
|
|
|
if not self.data_dir.exists():
|
|
logger.error("NPC data directory does not exist", data_dir=str(self.data_dir))
|
|
return npcs
|
|
|
|
for region_dir in self.data_dir.iterdir():
|
|
if not region_dir.is_dir():
|
|
continue
|
|
|
|
for file_path in region_dir.glob("*.yaml"):
|
|
npc = self._load_npc_file(file_path)
|
|
if npc:
|
|
npcs.append(npc)
|
|
|
|
logger.info("All NPCs loaded", count=len(npcs))
|
|
return npcs
|
|
|
|
def get_npcs_at_location(self, location_id: str) -> List[NPC]:
|
|
"""
|
|
Get all NPCs at a specific location.
|
|
|
|
Args:
|
|
location_id: Location identifier
|
|
|
|
Returns:
|
|
List of NPC instances at this location
|
|
"""
|
|
# Ensure all NPCs are loaded
|
|
if not self._npc_cache:
|
|
self.load_all_npcs()
|
|
|
|
npc_ids = self._location_npc_cache.get(location_id, [])
|
|
return [
|
|
self._npc_cache[npc_id]
|
|
for npc_id in npc_ids
|
|
if npc_id in self._npc_cache
|
|
]
|
|
|
|
def get_npc_ids_at_location(self, location_id: str) -> List[str]:
|
|
"""
|
|
Get NPC IDs at a specific location.
|
|
|
|
Args:
|
|
location_id: Location identifier
|
|
|
|
Returns:
|
|
List of NPC IDs at this location
|
|
"""
|
|
# Ensure all NPCs are loaded
|
|
if not self._npc_cache:
|
|
self.load_all_npcs()
|
|
|
|
return self._location_npc_cache.get(location_id, [])
|
|
|
|
def get_npcs_by_tag(self, tag: str) -> List[NPC]:
|
|
"""
|
|
Get all NPCs with a specific tag.
|
|
|
|
Args:
|
|
tag: Tag to filter by (e.g., "merchant", "quest_giver")
|
|
|
|
Returns:
|
|
List of NPC instances with this tag
|
|
"""
|
|
# Ensure all NPCs are loaded
|
|
if not self._npc_cache:
|
|
self.load_all_npcs()
|
|
|
|
return [
|
|
npc for npc in self._npc_cache.values()
|
|
if tag in npc.tags
|
|
]
|
|
|
|
def get_quest_givers(self, quest_id: str) -> List[NPC]:
|
|
"""
|
|
Get all NPCs that can give a specific quest.
|
|
|
|
Args:
|
|
quest_id: Quest identifier
|
|
|
|
Returns:
|
|
List of NPC instances that give this quest
|
|
"""
|
|
# Ensure all NPCs are loaded
|
|
if not self._npc_cache:
|
|
self.load_all_npcs()
|
|
|
|
return [
|
|
npc for npc in self._npc_cache.values()
|
|
if quest_id in npc.quest_giver_for
|
|
]
|
|
|
|
def get_all_npc_ids(self) -> List[str]:
|
|
"""
|
|
Get a list of all available NPC IDs.
|
|
|
|
Returns:
|
|
List of NPC IDs
|
|
"""
|
|
# Ensure all NPCs are loaded
|
|
if not self._npc_cache:
|
|
self.load_all_npcs()
|
|
|
|
return list(self._npc_cache.keys())
|
|
|
|
def reload_npc(self, npc_id: str) -> Optional[NPC]:
|
|
"""
|
|
Force reload an NPC from disk, bypassing cache.
|
|
|
|
Useful for development/testing when NPC definitions change.
|
|
|
|
Args:
|
|
npc_id: Unique NPC identifier
|
|
|
|
Returns:
|
|
NPC instance or None if not found
|
|
"""
|
|
# Remove from caches if present
|
|
if npc_id in self._npc_cache:
|
|
old_npc = self._npc_cache[npc_id]
|
|
# Remove from location cache
|
|
if old_npc.location_id in self._location_npc_cache:
|
|
self._location_npc_cache[old_npc.location_id] = [
|
|
n for n in self._location_npc_cache[old_npc.location_id]
|
|
if n != npc_id
|
|
]
|
|
del self._npc_cache[npc_id]
|
|
|
|
return self.load_npc(npc_id)
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Clear all cached data. Useful for testing."""
|
|
self._npc_cache.clear()
|
|
self._location_npc_cache.clear()
|
|
logger.info("NPC cache cleared")
|
|
|
|
|
|
# Global singleton instance
|
|
_loader_instance: Optional[NPCLoader] = None
|
|
|
|
|
|
def get_npc_loader() -> NPCLoader:
|
|
"""
|
|
Get the global NPCLoader instance.
|
|
|
|
Returns:
|
|
Singleton NPCLoader instance
|
|
"""
|
|
global _loader_instance
|
|
if _loader_instance is None:
|
|
_loader_instance = NPCLoader()
|
|
return _loader_instance
|