377 lines
13 KiB
Python
377 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
from enum import Enum
|
||
from uuid import UUID, uuid4
|
||
from datetime import datetime, timezone
|
||
from dataclasses import dataclass, field, asdict
|
||
from typing import Callable, List, Optional, Dict, Any, Iterable, Mapping, TYPE_CHECKING
|
||
|
||
from app.models.enums import HeroClass, Race
|
||
from app.models.primitives import Attributes, Resources, Collections
|
||
from app.utils.merging import merge_catalogs, merge_sheets
|
||
|
||
if TYPE_CHECKING:
|
||
from app.utils.hero_catalog import HeroDataRegistry
|
||
from app.utils.race_catalog import RaceDataRegistry
|
||
|
||
# -----------------------------
|
||
# Small utilities
|
||
# -----------------------------
|
||
|
||
def _utcnow_iso() -> str:
|
||
"""Return current UTC time as ISO 8601 string."""
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
def _clamp(value: int, lo: int, hi: int) -> int:
|
||
"""Clamp integer between lo and hi."""
|
||
return max(lo, min(hi, value))
|
||
|
||
# -----------------------------
|
||
# Component dataclasses
|
||
# -----------------------------
|
||
|
||
@dataclass
|
||
class Progress:
|
||
"""Economy, experience, level, and other progression flags."""
|
||
level: int = 1
|
||
gold: int = 50
|
||
xp: int = 0
|
||
xp_next_level: int = 100 # will be recomputed on init
|
||
needs_sleep: int = 0
|
||
fame: int = 0
|
||
alignment: int = 0 # -100..+100 works nicely
|
||
battlecount: int = 0
|
||
|
||
def gain_xp(self, amount: int, xp_curve: Callable[[int, int], int]) -> int:
|
||
"""
|
||
Add XP and compute how many levels are gained using the provided xp curve.
|
||
Returns the number of levels gained.
|
||
"""
|
||
amount = max(0, int(amount))
|
||
if amount == 0:
|
||
return 0
|
||
|
||
self.xp += amount
|
||
levels_gained = 0
|
||
|
||
# Level up loop (handles large XP chunks)
|
||
while self.xp >= self.xp_next_level:
|
||
self.level += 1
|
||
levels_gained += 1
|
||
self.xp_next_level = xp_curve(self.xp, self.level)
|
||
|
||
return levels_gained
|
||
|
||
@dataclass
|
||
class Location:
|
||
current_town: str = "Brindlemark"
|
||
current_nation: str = "Brindlemark"
|
||
current_continent: str = "Brindlemark"
|
||
|
||
@dataclass
|
||
class Equipment:
|
||
"""References to equipped items (IDs or simple names for now)."""
|
||
weapon: Optional[str] = None
|
||
armor: Optional[str] = None
|
||
|
||
# -----------------------------
|
||
# Main model
|
||
# -----------------------------
|
||
|
||
@dataclass
|
||
class Hero:
|
||
# Identity
|
||
hero_id: UUID = field(default_factory=uuid4)
|
||
name: str = "Unnamed"
|
||
|
||
# Build
|
||
race: Optional[Race] = None
|
||
hero_class: Optional[HeroClass] = None
|
||
|
||
# Stats & state
|
||
attributes: Attributes = field(default_factory=Attributes)
|
||
resources: Resources = field(default_factory=Resources)
|
||
progress: Progress = field(default_factory=Progress)
|
||
location: Location = field(default_factory=Location)
|
||
equipment: Equipment = field(default_factory=Equipment)
|
||
collections: Collections = field(default_factory=Collections)
|
||
|
||
# RP bits
|
||
origin_story: str = ""
|
||
|
||
# Auditing & versioning
|
||
version: int = 1
|
||
created_at: str = field(default_factory=_utcnow_iso)
|
||
updated_at: str = field(default_factory=_utcnow_iso)
|
||
|
||
# -------------------------
|
||
# Lifecycle helpers
|
||
# -------------------------
|
||
|
||
def initialize(self, xp_curve: Callable[[int, int], int]) -> None:
|
||
"""
|
||
Call once after creation or after loading if you need to recompute derived fields.
|
||
For example: set xp_next_level using your curve.
|
||
"""
|
||
self.progress.xp_next_level = xp_curve(self.progress.xp, self.progress.level)
|
||
self.touch()
|
||
|
||
def touch(self) -> None:
|
||
"""Update the `updated_at` timestamp."""
|
||
self.updated_at = _utcnow_iso()
|
||
|
||
@classmethod
|
||
def from_race_and_class(
|
||
cls,
|
||
name: str,
|
||
race: Race | None,
|
||
hero_class: HeroClass | None,
|
||
class_registry: HeroDataRegistry | None = None,
|
||
race_registry: RaceDataRegistry | None = None,
|
||
*,
|
||
# Optional global registries (if you’ve split catalogs):
|
||
skill_registry: Any | None = None,
|
||
spell_registry: Any | None = None,
|
||
) -> Hero:
|
||
"""
|
||
Factory that builds a Hero and seeds starting skills/spells from race/class.
|
||
Validates starting IDs against global registries when provided, otherwise
|
||
falls back to merged local catalogs from the race/class sheets.
|
||
"""
|
||
# Always define locals so later checks are safe
|
||
skills_catalog: Dict[str, Any] = {}
|
||
spells_catalog: Dict[str, Any] = {}
|
||
race_sheet = None
|
||
cls_sheet = None
|
||
|
||
if class_registry and race_registry and hero_class and race:
|
||
cls_sheet = class_registry.for_class(hero_class)
|
||
race_sheet = race_registry.for_race(race)
|
||
# Merge any local catalogs for fallback validation
|
||
skills_catalog, spells_catalog = merge_catalogs(
|
||
race_sheet.skills, race_sheet.spells,
|
||
cls_sheet.skills, cls_sheet.spells
|
||
)
|
||
|
||
starting_skills: list[str] = []
|
||
starting_spells: list[str] = []
|
||
if race_sheet:
|
||
starting_skills.extend(race_sheet.starting_skills or [])
|
||
starting_spells.extend(race_sheet.starting_spells or [])
|
||
if cls_sheet:
|
||
starting_skills.extend(cls_sheet.starting_skills or [])
|
||
starting_spells.extend(cls_sheet.starting_spells or [])
|
||
|
||
# Dedup while preserving order
|
||
def _dedup(seq: list[str]) -> list[str]:
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for x in seq:
|
||
if x not in seen:
|
||
out.append(x)
|
||
seen.add(x)
|
||
return out
|
||
|
||
starting_skills = _dedup(starting_skills)
|
||
starting_spells = _dedup(starting_spells)
|
||
|
||
# Validate against global registries first, fallback to locals
|
||
_validate_ids_exist(
|
||
starting_skills, skills_catalog, "skill",
|
||
owner=f"{race.value if race else 'unknown'}+{hero_class.value if hero_class else 'unknown'}",
|
||
registry=skill_registry
|
||
)
|
||
_validate_ids_exist(
|
||
starting_spells, spells_catalog, "spell",
|
||
owner=f"{race.value if race else 'unknown'}+{hero_class.value if hero_class else 'unknown'}",
|
||
registry=spell_registry
|
||
)
|
||
|
||
hero = cls(name=name, race=race, hero_class=hero_class)
|
||
hero.collections.skills = starting_skills
|
||
hero.collections.spells = starting_spells
|
||
hero.touch()
|
||
return hero
|
||
|
||
# -------------------------
|
||
# Progression
|
||
# -------------------------
|
||
|
||
def gain_xp(
|
||
self,
|
||
amount: int,
|
||
xp_curve: Callable[[int, int], int],
|
||
class_registry: HeroDataRegistry | None = None,
|
||
race_registry: RaceDataRegistry | None = None,
|
||
*,
|
||
# Optional global registries (preferred for validation)
|
||
skill_registry: Any | None = None,
|
||
spell_registry: Any | None = None,
|
||
) -> int:
|
||
"""
|
||
Gain XP and apply level-ups via the provided xp_curve.
|
||
If registries are provided, also grant any class/race skills & spells
|
||
unlocked by the newly reached level(s).
|
||
Returns number of levels gained.
|
||
"""
|
||
prev_level = self.progress.level
|
||
levels = self.progress.gain_xp(amount, xp_curve)
|
||
|
||
# Always prepare empty local catalogs; may be filled below
|
||
skills_catalog: Dict[str, Any] = {}
|
||
spells_catalog: Dict[str, Any] = {}
|
||
|
||
if levels > 0 and class_registry and race_registry and self.hero_class and self.race:
|
||
cls_sheet = class_registry.for_class(self.hero_class)
|
||
race_sheet = race_registry.for_race(self.race)
|
||
|
||
# Optional: merge local catalogs for fallback validation
|
||
skills_catalog, spells_catalog = merge_catalogs(
|
||
race_sheet.skills, race_sheet.spells,
|
||
cls_sheet.skills, cls_sheet.spells
|
||
)
|
||
|
||
unlocked_skills: list[str] = []
|
||
unlocked_spells: list[str] = []
|
||
now_lvl = self.progress.level
|
||
|
||
# Unlock skills whose min_level is newly reached
|
||
for nodes in (cls_sheet.skill_trees or {}).values():
|
||
for ref in nodes:
|
||
if ref.min_level is not None and prev_level < ref.min_level <= now_lvl:
|
||
unlocked_skills.append(ref.id)
|
||
|
||
# Unlock spells by each level reached in the jump (handles 1→6, etc.)
|
||
for L in range(prev_level + 1, now_lvl + 1):
|
||
unlocked_spells.extend((cls_sheet.spells_by_level.get(L) or []))
|
||
|
||
# Validate using global registries when provided, fallback to locals
|
||
owner = f"{self.race.value}+{self.hero_class.value}"
|
||
_validate_ids_exist(unlocked_skills, skills_catalog, "skill", owner, registry=skill_registry)
|
||
_validate_ids_exist(unlocked_spells, spells_catalog, "spell", owner, registry=spell_registry)
|
||
|
||
# Grant only new ones (dedupe)
|
||
_unique_extend(self.collections.skills, unlocked_skills)
|
||
_unique_extend(self.collections.spells, unlocked_spells)
|
||
|
||
# Your existing resource gains
|
||
hp_gain = 8 * levels
|
||
mp_gain = 6 * levels
|
||
self.resources.maxhp += hp_gain
|
||
self.resources.maxmp += mp_gain
|
||
self.resources.hp = self.resources.maxhp
|
||
self.resources.mp = self.resources.maxmp
|
||
|
||
self.touch()
|
||
return levels
|
||
|
||
# -------------------------
|
||
# (De)serialization helpers
|
||
# -------------------------
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""
|
||
Convert to a pure-JSON-friendly dict (Enums → value, UUID → str).
|
||
"""
|
||
def _enum_to_value(obj: Any) -> Any:
|
||
if isinstance(obj, Enum):
|
||
return obj.value
|
||
if isinstance(obj, UUID):
|
||
return str(obj)
|
||
return obj
|
||
|
||
raw = asdict(self)
|
||
# Fix enum & UUID fields:
|
||
raw["hero_id"] = str(self.hero_id)
|
||
raw["hero_class"] = self.hero_class.value if self.hero_class else None
|
||
return _walk_map(raw, _enum_to_value)
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict[str, Any]) -> Hero:
|
||
"""
|
||
Build Hero from a dict (reverse of to_dict).
|
||
Accepts hero_class as string or None; hero_id as str or UUID.
|
||
"""
|
||
data = dict(data) # shallow copy
|
||
|
||
# Parse UUID
|
||
hero_id_val = data.get("hero_id")
|
||
if hero_id_val and not isinstance(hero_id_val, UUID):
|
||
data["hero_id"] = UUID(hero_id_val)
|
||
|
||
# Parse enum
|
||
hc = data.get("hero_class")
|
||
if isinstance(hc, str) and hc:
|
||
data["hero_class"] = HeroClass(hc)
|
||
elif hc is None:
|
||
data["hero_class"] = None
|
||
|
||
# Rebuild nested dataclasses if passed as dicts
|
||
for k, typ in {
|
||
"attributes": Attributes,
|
||
"resources": Resources,
|
||
"progress": Progress,
|
||
"location": Location,
|
||
"equipment": Equipment,
|
||
"collections": Collections,
|
||
}.items():
|
||
if isinstance(data.get(k), dict):
|
||
data[k] = typ(**data[k])
|
||
|
||
hero: Hero = cls(**data) # type: ignore[arg-type]
|
||
return hero
|
||
|
||
# -----------------------------
|
||
# Helpers
|
||
# -----------------------------
|
||
|
||
def _walk_map(obj: Any, f) -> Any:
|
||
"""Recursively map Python structures with function f applied to leaf values as needed."""
|
||
if isinstance(obj, dict):
|
||
return {k: _walk_map(v, f) for k, v in obj.items()}
|
||
if isinstance(obj, list):
|
||
return [_walk_map(x, f) for x in obj]
|
||
return f(obj)
|
||
|
||
def _unique_extend(target: list[str], new_items: Iterable[str]) -> list[str]:
|
||
"""Append only items not already present; return the list for convenience."""
|
||
seen = set(target)
|
||
for x in new_items:
|
||
if x not in seen:
|
||
target.append(x)
|
||
seen.add(x)
|
||
return target
|
||
|
||
def default_xp_curve(current_xp: int, level: int) -> int:
|
||
"""
|
||
Replace this with your real XP curve (e.g., g_utils.get_xp_for_next_level).
|
||
Simple example: mild exponential growth.
|
||
"""
|
||
base = 100
|
||
return base + int(50 * (level ** 1.5)) + (level * 25)
|
||
|
||
def _validate_ids_exist(
|
||
ids: Iterable[str],
|
||
local_catalog: Mapping[str, Any] | None,
|
||
kind: str,
|
||
owner: str,
|
||
*,
|
||
registry: Any | None = None, # e.g., SkillRegistry or SpellRegistry with .get()
|
||
) -> None:
|
||
"""
|
||
Validate that each id is present either in the provided local_catalog OR
|
||
resolvable via the optional global registry (registry.get(id) not None).
|
||
"""
|
||
local_catalog = local_catalog or {}
|
||
reg_get = getattr(registry, "get", None) if registry is not None else None
|
||
|
||
missing: list[str] = []
|
||
for _id in ids:
|
||
in_local = _id in local_catalog
|
||
in_global = bool(reg_get(_id)) if reg_get else False
|
||
if not (in_local or in_global):
|
||
missing.append(_id)
|
||
|
||
if missing:
|
||
raise ValueError(f"{owner}: undefined {kind} id(s): {', '.join(missing)}")
|