Praxis CI
813bd586d6
v0.3 milestone merged to main. Mastery scoring + competency rubrics + verifiable credentials (formative-tier) shipped. 13/13 REQ-IDs covered. Next milestone: v0.4 (operator tier — cohort dashboard + auth + Postgres). ---ci--- project: praxis phase: 2 milestone: v0.3 status: complete milestone_complete: true milestone_merged_to_main: true ---/ci---
91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
"""Scenario loader — YAML → Pydantic Scenario (D-018).
|
|
|
|
Loads a scenario by id from the scenarios/ directory, validates it against the
|
|
Pydantic schema, and returns a typed Scenario object. Used by the pipeline
|
|
(TASK-03-07) and the e2e smoke test.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from server.scenarios.schema import Scenario, ValidationError
|
|
|
|
_DEFAULT_SCENARIOS_DIR = Path(__file__).resolve().parent.parent.parent / "scenarios"
|
|
|
|
|
|
def _find_yaml(scenario_id: str, base: Path) -> Path | None:
|
|
"""Resolve a scenario id to its YAML path.
|
|
|
|
Searches the scenarios root and any one-level subdirectory (e.g.
|
|
customer_service/). Supports two alias forms for backward compatibility:
|
|
- cs_<id> -> customer_service_<id>.yaml (v0.1 call sites used the long form)
|
|
- customer_service_<id> -> cs_<id>.yaml (reverse, for the renamed v01 file)
|
|
"""
|
|
primary = base / f"{scenario_id}.yaml"
|
|
if primary.exists():
|
|
return primary
|
|
cs_alias = base / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml"
|
|
if cs_alias.exists():
|
|
return cs_alias
|
|
long_alias = base / f"{scenario_id.replace('customer_service_', 'cs_')}.yaml"
|
|
if long_alias.exists():
|
|
return long_alias
|
|
# One-level subdirectory walk (subdir named by skill, e.g. customer_service/).
|
|
for d in sorted(base.glob("*/")):
|
|
if not d.is_dir():
|
|
continue
|
|
for cand in (
|
|
d / f"{scenario_id}.yaml",
|
|
d / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml",
|
|
d / f"{scenario_id.replace('customer_service_', 'cs_')}.yaml",
|
|
):
|
|
if cand.exists():
|
|
return cand
|
|
return None
|
|
|
|
|
|
def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario:
|
|
"""Load and validate a scenario by id.
|
|
|
|
Args:
|
|
scenario_id: e.g. 'cs_refund_ca_v01' (the YAML filename stem).
|
|
scenarios_dir: override the scenarios directory (default: repo /scenarios).
|
|
|
|
Returns:
|
|
A validated Scenario object.
|
|
|
|
Raises:
|
|
FileNotFoundError: if the YAML file doesn't exist.
|
|
ValidationError: if the YAML fails schema validation (typed Pydantic error).
|
|
"""
|
|
base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
|
|
path = _find_yaml(scenario_id, base)
|
|
if path is None:
|
|
raise FileNotFoundError(f"Scenario YAML not found: {scenario_id} in {base}")
|
|
|
|
with path.open("r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
|
|
return Scenario.model_validate(raw)
|
|
|
|
|
|
def load_all(scenarios_dir: Path | None = None) -> list[Scenario]:
|
|
"""Load all scenarios in the directory tree (root + one-level subdirs)."""
|
|
base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
|
|
out: list[Scenario] = []
|
|
paths = sorted(base.glob("*.yaml")) + sorted(base.glob("*/**/*.yaml"))
|
|
seen: set[Path] = set()
|
|
for p in paths:
|
|
if p in seen or p.name == "index.yaml" or p.name == "cost_rates.yaml":
|
|
continue
|
|
seen.add(p)
|
|
with p.open("r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
out.append(Scenario.model_validate(raw))
|
|
return out
|
|
|
|
|
|
__all__ = ["load", "load_all", "ValidationError"] |