178 lines
6.5 KiB
Python
178 lines
6.5 KiB
Python
"""Validation and normalization for the semantic GenerationSpec contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator
|
|
|
|
|
|
KNOWN_PART_FAMILIES = frozenset({
|
|
"mounting_plate",
|
|
"flange",
|
|
"flange_sleeve",
|
|
"simple_shaft",
|
|
"bearing_housing",
|
|
"mounting_bracket",
|
|
"hex_nut",
|
|
"slotted_plate",
|
|
})
|
|
|
|
KNOWN_FEATURE_KINDS = frozenset({
|
|
"base_extrusion",
|
|
"base_revolve",
|
|
"boss",
|
|
"through_hole",
|
|
"blind_hole",
|
|
"counterbored_hole",
|
|
"countersunk_hole",
|
|
"hole_pattern",
|
|
"counterbored_hole_pattern",
|
|
"obround_cut",
|
|
"pocket",
|
|
"revolve_profile",
|
|
"coaxial_bore",
|
|
"fillet",
|
|
"chamfer",
|
|
})
|
|
|
|
|
|
class GenerationSpecError(ValueError):
|
|
"""A stable, user-repairable GenerationSpec validation error."""
|
|
|
|
def __init__(self, message: str, path: str = "$") -> None:
|
|
super().__init__(message)
|
|
self.path = path
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def generation_spec_schema() -> dict[str, Any]:
|
|
path = Path(__file__).with_name("generation_spec_schema.json")
|
|
schema = json.loads(path.read_text(encoding="utf-8"))
|
|
Draft202012Validator.check_schema(schema)
|
|
return schema
|
|
|
|
|
|
def _schema_error(spec: dict[str, Any]) -> GenerationSpecError | None:
|
|
errors = sorted(
|
|
Draft202012Validator(generation_spec_schema()).iter_errors(spec),
|
|
key=lambda error: (list(error.absolute_path), error.message),
|
|
)
|
|
if not errors:
|
|
return None
|
|
error = errors[0]
|
|
location = "$" + "".join(
|
|
f"[{item}]" if isinstance(item, int) else f".{item}"
|
|
for item in error.absolute_path
|
|
)
|
|
return GenerationSpecError(error.message, location)
|
|
|
|
|
|
def _parameter_value(spec: dict[str, Any], name: str) -> Any:
|
|
value = (spec.get("parameters") or {}).get(name)
|
|
if not isinstance(value, dict):
|
|
raise GenerationSpecError(f"Unknown parameter: {name}", f"$.parameters.{name}")
|
|
return value.get("value")
|
|
|
|
|
|
def _numeric(value: Any, path: str) -> float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise GenerationSpecError("Expected a finite numeric value", path)
|
|
number = float(value)
|
|
if number != number or number in {float("inf"), float("-inf")}:
|
|
raise GenerationSpecError("Expected a finite numeric value", path)
|
|
return number
|
|
|
|
|
|
def _check_constraints(spec: dict[str, Any]) -> None:
|
|
for index, constraint in enumerate(spec.get("constraints") or []):
|
|
path = f"$.constraints[{index}]"
|
|
kind = constraint.get("type")
|
|
if kind in {"less_than", "less_equal", "greater_than", "greater_equal", "equal"}:
|
|
left = _numeric(_parameter_value(spec, str(constraint.get("left") or "")), f"{path}.left")
|
|
right_name = constraint.get("right")
|
|
right = _numeric(_parameter_value(spec, str(right_name)), f"{path}.right") if right_name else _numeric(constraint.get("value"), f"{path}.value")
|
|
ok = {
|
|
"less_than": left < right,
|
|
"less_equal": left <= right,
|
|
"greater_than": left > right,
|
|
"greater_equal": left >= right,
|
|
"equal": abs(left - right) <= 1e-9,
|
|
}[kind]
|
|
if not ok:
|
|
raise GenerationSpecError(constraint.get("message") or f"Constraint {kind} failed", path)
|
|
|
|
|
|
def _check_graph(spec: dict[str, Any]) -> None:
|
|
features = spec.get("features") or []
|
|
ids = [str(item.get("id")) for item in features]
|
|
if len(ids) != len(set(ids)):
|
|
raise GenerationSpecError("Feature ids must be unique", "$.features")
|
|
known: set[str] = set()
|
|
for index, feature in enumerate(features):
|
|
feature_id = str(feature.get("id"))
|
|
kind = str(feature.get("kind"))
|
|
if kind not in KNOWN_FEATURE_KINDS:
|
|
raise GenerationSpecError(f"Unsupported feature kind: {kind}", f"$.features[{index}].kind")
|
|
for dependency in feature.get("depends_on") or []:
|
|
if dependency not in known:
|
|
raise GenerationSpecError(
|
|
f"Feature {feature_id} has a forward or missing dependency: {dependency}",
|
|
f"$.features[{index}].depends_on",
|
|
)
|
|
known.add(feature_id)
|
|
|
|
acceptance_ids = {str(item.get("id")) for item in spec.get("acceptance") or []}
|
|
if len(acceptance_ids) != len(spec.get("acceptance") or []):
|
|
raise GenerationSpecError("Acceptance ids must be unique", "$.acceptance")
|
|
for index, item in enumerate(spec.get("acceptance") or []):
|
|
feature = item.get("feature")
|
|
if feature and feature not in known:
|
|
raise GenerationSpecError(f"Acceptance refers to missing feature: {feature}", f"$.acceptance[{index}].feature")
|
|
|
|
|
|
def normalize_generation_spec(spec: dict[str, Any], *, request: str = "") -> dict[str, Any]:
|
|
if not isinstance(spec, dict):
|
|
raise GenerationSpecError("GenerationSpec must be a JSON object")
|
|
normalized = copy.deepcopy(spec)
|
|
normalized.setdefault("schema", "cad.generation-spec.v1")
|
|
normalized.setdefault("schema_version", "1.0")
|
|
normalized.setdefault("mode", "create")
|
|
normalized.setdefault("base_revision_id", "")
|
|
normalized.setdefault("parameters", {})
|
|
normalized.setdefault("features", [])
|
|
normalized.setdefault("constraints", [])
|
|
normalized.setdefault("acceptance", [])
|
|
normalized.setdefault("assumptions", [])
|
|
normalized.setdefault("approximations", [])
|
|
normalized.setdefault("references", [])
|
|
normalized.setdefault("patch_intent", request)
|
|
return normalized
|
|
|
|
|
|
def validate_generation_spec(
|
|
spec: dict[str, Any],
|
|
*,
|
|
known_families: set[str] | frozenset[str] = KNOWN_PART_FAMILIES,
|
|
) -> dict[str, Any]:
|
|
normalized = normalize_generation_spec(spec)
|
|
error = _schema_error(normalized)
|
|
if error:
|
|
raise error
|
|
family = str(normalized["part"]["family"])
|
|
if family not in known_families:
|
|
raise GenerationSpecError(f"Unsupported part family: {family}", "$.part.family")
|
|
for name, parameter in normalized["parameters"].items():
|
|
source = parameter["source"]
|
|
if source == "user" and not parameter["locked"]:
|
|
raise GenerationSpecError("User parameters must be locked", f"$.parameters.{name}.locked")
|
|
if source == "image_estimate" and not parameter.get("assumption"):
|
|
raise GenerationSpecError("Image estimates require an assumption", f"$.parameters.{name}.assumption")
|
|
_check_graph(normalized)
|
|
_check_constraints(normalized)
|
|
return normalized
|