from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List, Mapping, Optional import yaml @dataclass(frozen=True) class Outcome: when: Mapping[str, Any] to: Optional[str] = None emit: Optional[Mapping[str, Any]] = None @dataclass(frozen=True) class Transition: event_id: str stage: str desc: str from_states: List[str] outcomes: List[Outcome] @dataclass(frozen=True) class StateMachineConfig: path: Path raw: Mapping[str, Any] transitions: Mapping[str, Transition] invariants: Mapping[str, Mapping[str, Any]] @staticmethod def load(path: Path) -> "StateMachineConfig": with path.open("r", encoding="utf-8") as f: raw = yaml.safe_load(f) if not isinstance(raw, dict): raise TypeError(f"Invalid state machine yaml: {path}") transitions_raw = raw.get("transitions") if not isinstance(transitions_raw, dict): raise TypeError("'transitions' must be a mapping") parsed: Dict[str, Transition] = {} for event_id, tdef in transitions_raw.items(): if not isinstance(tdef, dict): continue outcomes: List[Outcome] = [] for o in tdef.get("outcomes", []): if not isinstance(o, dict): continue when = o.get("when", {}) if not isinstance(when, dict): continue outcomes.append( Outcome( when=when, to=o.get("to") if isinstance(o.get("to"), str) else None, emit=o.get("emit") if isinstance(o.get("emit"), dict) else None, ) ) parsed[str(event_id)] = Transition( event_id=str(event_id), stage=str(tdef.get("stage", "")), desc=str(tdef.get("desc", "")), from_states=[s for s in tdef.get("from", []) if isinstance(s, str)], outcomes=outcomes, ) invariants = raw.get("invariants") if isinstance(raw.get("invariants"), dict) else {} return StateMachineConfig(path=path, raw=raw, transitions=parsed, invariants=invariants)