295 lines
17 KiB
Python
295 lines
17 KiB
Python
"""Validation for the complete CDSL v1.1 semantic contract.
|
|
|
|
The current runtime accepts only a subset of this contract. Keeping this
|
|
validator separate lets import tooling preserve a SolidWorks feature history
|
|
without claiming that every feature can already be rebuilt locally.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from .operation_contracts import materialized_feature_contracts
|
|
|
|
|
|
_ID = re.compile(r"^[A-Za-z0-9_-]{1,80}$")
|
|
|
|
|
|
def _mappings(value: Any):
|
|
"""Yield nested mapping values without treating selector-like data as text."""
|
|
if isinstance(value, dict):
|
|
yield value
|
|
for child in value.values():
|
|
yield from _mappings(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
yield from _mappings(child)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _schema() -> dict[str, Any]:
|
|
path = Path(__file__).with_name("cdsl_schema.json")
|
|
schema = json.loads(path.read_text(encoding="utf-8"))
|
|
Draft202012Validator.check_schema(schema)
|
|
return schema
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _validator() -> Draft202012Validator:
|
|
return Draft202012Validator(_schema())
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _operation_contracts() -> dict[str, dict[str, Any]]:
|
|
path = Path(__file__).with_name("profile_schema.json")
|
|
return materialized_feature_contracts(json.loads(path.read_text(encoding="utf-8")))
|
|
|
|
|
|
def _schema_error(document: dict[str, Any]) -> str | None:
|
|
validator = _validator()
|
|
errors = sorted(validator.iter_errors(document), 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 f"CDSL schema violation at {location}: {error.message}"
|
|
|
|
|
|
def validate_semantic_cdsl(cdsl: dict[str, Any]) -> dict[str, Any]:
|
|
"""Validate a CDSL document without invoking the rebuild compiler.
|
|
|
|
The return value is intentionally serializable so the batch converter can
|
|
write it unchanged into a per-model diagnostic file.
|
|
"""
|
|
if not isinstance(cdsl, dict):
|
|
raise ValueError("CDSL must be a JSON object")
|
|
if cdsl.get("schema") != "cad.cdsl.llm.v1":
|
|
raise ValueError("Unsupported CDSL schema")
|
|
schema_error = _schema_error(cdsl)
|
|
if schema_error:
|
|
raise ValueError(schema_error)
|
|
|
|
version = str(cdsl.get("schema_version") or "1.0.0")
|
|
if not re.fullmatch(r"1\.[0-9]+\.[0-9]+", version):
|
|
raise ValueError("schema_version must be a 1.x.y version")
|
|
version_numbers = tuple(int(component) for component in version.split("."))
|
|
if version_numbers >= (1, 1, 0) and cdsl.get("meta", {}).get("unit") != "mm":
|
|
raise ValueError("CDSL v1.1 requires meta.unit = 'mm'")
|
|
|
|
sketches = (cdsl.get("geometry") or {}).get("sketches") or []
|
|
sketch_ids = {str(sketch.get("id") or "") for sketch in sketches}
|
|
if len(sketch_ids) != len(sketches) or not all(_ID.fullmatch(item) for item in sketch_ids):
|
|
raise ValueError("Sketch ids must be unique valid CDSL identifiers")
|
|
for sketch in sketches:
|
|
profile = sketch.get("profile") or {}
|
|
if profile.get("type") != "planar_imprint":
|
|
continue
|
|
entities = profile.get("source_entities") or []
|
|
entity_ids = [str(entity.get("id") or "") for entity in entities if isinstance(entity, dict)]
|
|
if len(entity_ids) != len(entities) or len(set(entity_ids)) != len(entity_ids) or not all(entity_ids):
|
|
raise ValueError(f"Sketch {sketch.get('id')} planar_imprint source entity ids must be unique")
|
|
known = set(entity_ids)
|
|
for index, selection in enumerate(profile.get("selections") or []):
|
|
source = str(selection.get("source_entity_id") or "")
|
|
fragment = selection.get("fragment") or {}
|
|
anchor = str(fragment.get("anchor_entity_id") or "") if fragment else ""
|
|
if source not in known:
|
|
raise ValueError(f"Sketch {sketch.get('id')} planar_imprint selection {index} has an unknown source entity")
|
|
if fragment and (anchor not in known or anchor == source):
|
|
raise ValueError(f"Sketch {sketch.get('id')} planar_imprint selection {index} has an invalid fragment anchor")
|
|
|
|
feature_ids: set[str] = set()
|
|
preceding_features: dict[str, dict[str, Any]] = {}
|
|
deferred: list[str] = []
|
|
unresolved: list[dict[str, Any]] = []
|
|
contracts = _operation_contracts()
|
|
for feature in cdsl.get("features") or []:
|
|
fid = str(feature.get("id") or "")
|
|
if not _ID.fullmatch(fid) or fid in feature_ids:
|
|
raise ValueError("Feature ids must be unique valid CDSL identifiers")
|
|
for dependency in feature.get("depends_on") or []:
|
|
if dependency not in feature_ids:
|
|
raise ValueError(f"Feature {fid} has a forward or missing dependency: {dependency}")
|
|
sketch_id = feature.get("sketch_id")
|
|
if sketch_id is not None and str(sketch_id) not in sketch_ids:
|
|
raise ValueError(f"Feature {fid} refers to a missing sketch: {sketch_id}")
|
|
if version_numbers >= (1, 1, 0) and feature.get("execution_status") not in {"supported", "deferred"}:
|
|
raise ValueError(f"Feature {fid} must declare execution_status")
|
|
if feature.get("execution_status") == "deferred":
|
|
deferred.append(fid)
|
|
contract = contracts.get(str(feature.get("atomic_id") or "")) or {}
|
|
feature_selectors = feature.get("selectors") or []
|
|
output_role_selector_ids = {id(selector) for selector in feature_selectors if isinstance(selector, dict)}
|
|
for index, selector in enumerate(feature_selectors):
|
|
owner = selector.get("owner_feature_id")
|
|
binding_owner = selector.get("binding_feature_id")
|
|
if owner is not None and owner not in feature_ids and binding_owner not in feature_ids:
|
|
raise ValueError(f"Feature {fid} selector {index} has a forward or missing owner_feature_id")
|
|
if selector.get("output_role") is not None:
|
|
if contract.get("selector_slot") != "feature.selectors" or contract.get("selector_token_kind") != "face":
|
|
raise ValueError(f"Feature {fid} selector {index} cannot consume a feature output role")
|
|
if selector.get("kind") != "face" or not owner or owner not in feature_ids:
|
|
raise ValueError(f"Feature {fid} selector {index} output role requires a preceding face owner_feature_id")
|
|
if selector.get("source") != "runtime_snapshot":
|
|
raise ValueError(f"Feature {fid} selector {index} output role must use runtime_snapshot evidence")
|
|
if any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id")):
|
|
raise ValueError(f"Feature {fid} selector {index} output role cannot mix stable or geometry evidence")
|
|
role_source = selector.get("output_role_source")
|
|
if role_source is not None:
|
|
source_owner = role_source.get("owner_feature_id") if isinstance(role_source, dict) else None
|
|
source_role = role_source.get("output_role") if isinstance(role_source, dict) else None
|
|
source_feature = preceding_features.get(str(source_owner or ""))
|
|
source_params = (source_feature or {}).get("params") or {}
|
|
if not source_owner or source_owner not in feature_ids:
|
|
raise ValueError(f"Feature {fid} selector {index} output role source requires a preceding owner_feature_id")
|
|
if selector.get("output_role") != "shell.offset_face" or feature.get("atomic_id") != "shell":
|
|
raise ValueError(
|
|
f"Feature {fid} selector {index} output role source is only supported for shell.offset_face"
|
|
)
|
|
if source_role not in {"extrude.start", "extrude.end"}:
|
|
raise ValueError(
|
|
f"Feature {fid} selector {index} shell.offset_face source must be an extrude cap role"
|
|
)
|
|
if (
|
|
source_feature is None
|
|
or source_feature.get("atomic_id") != "extrude_add_blind"
|
|
or source_params.get("result_mode") != "new_body"
|
|
or (source_params.get("end_condition") or {}).get("type") != "blind"
|
|
):
|
|
raise ValueError(
|
|
f"Feature {fid} selector {index} shell.offset_face source requires a preceding direct new_body blind extrusion"
|
|
)
|
|
elif selector.get("output_role_source") is not None:
|
|
raise ValueError(f"Feature {fid} selector {index} output role source requires output_role")
|
|
for selector in _mappings(feature):
|
|
# ``output_role_source`` is provenance metadata nested inside a
|
|
# feature selector, not a selector on its own.
|
|
if (
|
|
selector.get("output_role") is not None
|
|
and selector.get("kind") is not None
|
|
and id(selector) not in output_role_selector_ids
|
|
):
|
|
raise ValueError(f"Feature {fid} output role selectors are only supported in feature.selectors")
|
|
if feature.get("atomic_id") == "shell":
|
|
target_feature_id = (feature.get("params") or {}).get("target_feature_id")
|
|
if target_feature_id is not None and target_feature_id not in feature_ids:
|
|
raise ValueError(f"Feature {fid} shell target_feature_id requires a preceding body feature")
|
|
if feature.get("atomic_id") == "transform_bodies":
|
|
params = feature.get("params") or {}
|
|
references = params.get("pattern_instance_refs") or []
|
|
for index, reference in enumerate(references):
|
|
if not isinstance(reference, dict):
|
|
raise ValueError(f"Feature {fid} pattern instance reference {index} is invalid")
|
|
pattern_id = str(reference.get("pattern_feature_id") or "")
|
|
source_id = str(reference.get("source_feature_id") or "")
|
|
pattern = preceding_features.get(pattern_id)
|
|
pattern_atomic_id = pattern.get("atomic_id") if pattern is not None else None
|
|
if pattern is None or pattern_atomic_id not in {"pattern_circular", "pattern_mirror"}:
|
|
raise ValueError(f"Feature {fid} pattern instance reference {index} has a forward or unsupported pattern owner")
|
|
if source_id not in {str(value) for value in (pattern.get("params") or {}).get("source_feature_ids") or ()}:
|
|
raise ValueError(f"Feature {fid} pattern instance reference {index} names a source outside its pattern")
|
|
source = preceding_features.get(source_id)
|
|
if (
|
|
pattern_atomic_id == "pattern_mirror"
|
|
and (source is None or (source.get("params") or {}).get("result_mode") != "new_body")
|
|
):
|
|
raise ValueError(
|
|
f"Feature {fid} pattern instance reference {index} requires a preceding new_body mirror source"
|
|
)
|
|
instance = reference.get("instance_index")
|
|
count = int((pattern.get("params") or {}).get("pattern_count") or 0)
|
|
excluded = {int(value) for value in (pattern.get("params") or {}).get("excluded_instance_indices") or ()}
|
|
if not isinstance(instance, int) or (
|
|
pattern_atomic_id == "pattern_mirror" and instance != 1
|
|
) or (
|
|
pattern_atomic_id == "pattern_circular" and (instance < 1 or instance >= count or instance in excluded)
|
|
):
|
|
raise ValueError(f"Feature {fid} pattern instance reference {index} is not a surviving copy")
|
|
copy_references = params.get("transform_copy_refs") or []
|
|
for index, reference in enumerate(copy_references):
|
|
if not isinstance(reference, dict):
|
|
raise ValueError(f"Feature {fid} transform COPY reference {index} is invalid")
|
|
transform_id = str(reference.get("transform_feature_id") or "")
|
|
source_id = str(reference.get("source_feature_id") or "")
|
|
transform = preceding_features.get(transform_id)
|
|
transform_params = (transform or {}).get("params") or {}
|
|
direct_sources = transform_params.get("source_feature_ids") or []
|
|
if (
|
|
transform is None
|
|
or transform.get("atomic_id") != "transform_bodies"
|
|
or not bool(transform_params.get("make_copy"))
|
|
or not isinstance(direct_sources, list)
|
|
or len(direct_sources) < 2
|
|
):
|
|
raise ValueError(
|
|
f"Feature {fid} transform COPY reference {index} requires a preceding multi-source make_copy transform"
|
|
)
|
|
if source_id not in {str(value) for value in direct_sources}:
|
|
raise ValueError(
|
|
f"Feature {fid} transform COPY reference {index} names a source outside its transform"
|
|
)
|
|
for source_id in params.get("source_feature_ids") or []:
|
|
source = preceding_features.get(str(source_id))
|
|
source_params = (source or {}).get("params") or {}
|
|
source_direct_ids = source_params.get("source_feature_ids") or []
|
|
if (
|
|
source is not None
|
|
and source.get("atomic_id") == "transform_bodies"
|
|
and bool(source_params.get("make_copy"))
|
|
and isinstance(source_direct_ids, list)
|
|
and len(source_direct_ids) > 1
|
|
):
|
|
raise ValueError(
|
|
f"Feature {fid} must use transform_copy_refs to select a source of multi-source COPY {source_id}"
|
|
)
|
|
if feature.get("atomic_id") == "boolean_bodies":
|
|
params = feature.get("params") or {}
|
|
for parameter in ("target_pattern_instance_refs", "tool_pattern_instance_refs"):
|
|
for index, reference in enumerate(params.get(parameter) or []):
|
|
if not isinstance(reference, dict):
|
|
raise ValueError(f"Feature {fid} {parameter} entry {index} is invalid")
|
|
pattern_id = str(reference.get("pattern_feature_id") or "")
|
|
source_id = str(reference.get("source_feature_id") or "")
|
|
pattern = preceding_features.get(pattern_id)
|
|
pattern_atomic_id = pattern.get("atomic_id") if pattern is not None else None
|
|
if pattern is None or pattern_atomic_id not in {"pattern_circular", "pattern_mirror"}:
|
|
raise ValueError(f"Feature {fid} {parameter} entry {index} has a forward or unsupported pattern owner")
|
|
if source_id not in {str(value) for value in (pattern.get("params") or {}).get("source_feature_ids") or ()}:
|
|
raise ValueError(f"Feature {fid} {parameter} entry {index} names a source outside its pattern")
|
|
source = preceding_features.get(source_id)
|
|
if (
|
|
pattern_atomic_id == "pattern_mirror"
|
|
and (source is None or (source.get("params") or {}).get("result_mode") != "new_body")
|
|
):
|
|
raise ValueError(
|
|
f"Feature {fid} {parameter} entry {index} requires a preceding new_body mirror source"
|
|
)
|
|
instance = reference.get("instance_index")
|
|
count = int((pattern.get("params") or {}).get("pattern_count") or 0)
|
|
excluded = {int(value) for value in (pattern.get("params") or {}).get("excluded_instance_indices") or []}
|
|
if not isinstance(instance, int) or (
|
|
pattern_atomic_id == "pattern_mirror" and instance != 1
|
|
) or (
|
|
pattern_atomic_id == "pattern_circular" and (instance < 1 or instance >= count or instance in excluded)
|
|
):
|
|
raise ValueError(f"Feature {fid} {parameter} entry {index} is not a surviving copy")
|
|
if feature.get("unresolved"):
|
|
unresolved.append({"feature_id": fid, "reasons": list(feature["unresolved"])})
|
|
feature_ids.add(fid)
|
|
preceding_features[fid] = feature
|
|
|
|
return {
|
|
"schema_version": version,
|
|
"feature_count": len(feature_ids),
|
|
"sketch_count": len(sketches),
|
|
"deferred_feature_ids": deferred,
|
|
"unresolved": unresolved,
|
|
"future_rebuild_ready": not unresolved,
|
|
}
|