79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Load the shared evaluation scenario catalog."""
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import yaml
|
|
|
|
SCENARIO_PATH = Path(__file__).with_name("scenarios.yaml")
|
|
FacetValue = str | bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Scenario:
|
|
"""One golden query and its expected intent facets."""
|
|
|
|
key: str
|
|
chip: str
|
|
query: str
|
|
facets: dict[str, FacetValue]
|
|
after: str | None = None
|
|
|
|
@property
|
|
def is_refinement(self) -> bool:
|
|
"""Return whether this query follows another scenario."""
|
|
return self.after is not None
|
|
|
|
|
|
def load_scenarios(path: Path = SCENARIO_PATH) -> tuple[Scenario, ...]:
|
|
"""Read and validate the shared YAML scenario catalog."""
|
|
payload = yaml.safe_load(path.read_text(encoding="ascii"))
|
|
if not isinstance(payload, dict) or payload.get("version") != 1:
|
|
raise ValueError("Scenario catalog must have version 1")
|
|
entries = payload.get("scenarios")
|
|
if not isinstance(entries, list) or not entries:
|
|
raise ValueError("Scenario catalog must contain scenarios")
|
|
|
|
scenarios = tuple(_parse_scenario(entry) for entry in entries)
|
|
keys = [scenario.key for scenario in scenarios]
|
|
if len(keys) != len(set(keys)):
|
|
raise ValueError("Scenario keys must be unique")
|
|
known_keys = set(keys)
|
|
for scenario in scenarios:
|
|
if scenario.after is not None and scenario.after not in known_keys:
|
|
raise ValueError(f"Unknown parent scenario: {scenario.after}")
|
|
return scenarios
|
|
|
|
|
|
def _parse_scenario(payload: object) -> Scenario:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("Each scenario must be a mapping")
|
|
required = {"key", "chip", "query", "facets"}
|
|
allowed = required | {"after"}
|
|
if set(payload) - allowed or not required <= set(payload):
|
|
raise ValueError("Scenario fields must match the catalog schema")
|
|
facets = payload["facets"]
|
|
if not isinstance(facets, dict) or not facets:
|
|
raise ValueError("Scenario facets must be a non-empty mapping")
|
|
if not all(
|
|
isinstance(key, str) and isinstance(value, (str, bool)) for key, value in facets.items()
|
|
):
|
|
raise ValueError("Scenario facets must contain string or boolean values")
|
|
after = payload.get("after")
|
|
if after is not None and not isinstance(after, str):
|
|
raise ValueError("Scenario after must be a key")
|
|
return Scenario(
|
|
key=_required_string(payload, "key"),
|
|
chip=_required_string(payload, "chip"),
|
|
query=_required_string(payload, "query"),
|
|
facets=cast(dict[str, FacetValue], facets),
|
|
after=after,
|
|
)
|
|
|
|
|
|
def _required_string(payload: dict[object, object], key: str) -> str:
|
|
value = payload.get(key)
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"Scenario {key} must be a non-empty string")
|
|
return value
|