Skip to content

Add Operator Pydantic model #28

@TimothyWillard

Description

@TimothyWillard

def _normalize_operators(
raw_ops: object,
*,
axis_names: set[str],
) -> list[dict[str, Any]]:
"""Normalize operator metadata and validate axis references.
Returns:
List of normalized operator definitions.
"""
if raw_ops is None:
return []
if not isinstance(raw_ops, list):
_raise_invalid_rhs_spec(detail="operators must be a list")
out: list[dict[str, Any]] = []
for idx, op in enumerate(raw_ops):
op_map = _ensure_mapping(op, name=f"operators[{idx}]")
axis_name = op_map.get("axis")
if not isinstance(axis_name, str) or not axis_name.strip():
_raise_invalid_rhs_spec(
detail=f"operators[{idx}].axis must be a non-empty string"
)
axis_name_s = axis_name.strip()
if axis_name_s not in axis_names:
_raise_invalid_rhs_spec(
detail=f"operators[{idx}] references unknown axis {axis_name_s!r}"
)
op_out = dict(op_map)
op_out["axis"] = axis_name_s
out.append(op_out)
return out

Not clear exactly what these are. Other references:

- Operators: `operators` metadata is normalized and preserved in `NormalizedRhs.meta["operators"]`; wiring into solvers is future-facing (not yet consumed by op_engine).

def test_normalize_rhs_preserves_reserved_blocks_in_meta() -> None:
"""Test that reserved future blocks are preserved in meta."""
spec = {
"kind": "transitions",
"state": ["S", "I", "R"],
"aliases": {"N": "S + I + R"},
"transitions": [{"from": "S", "to": "I", "rate": "beta * I / N"}],
"operators": {"default": {"scheme": "cn"}}, # reserved for future
"sources": {"S": "0.0"}, # reserved for future
}
out = normalize_rhs(spec)
assert out.kind == "transitions"
assert out.meta.get("operators") == {"default": {"scheme": "cn"}}
assert out.meta.get("sources") == {"S": "0.0"}

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions