379 lines
21 KiB
Python
379 lines
21 KiB
Python
"""Capability analysis and feature planning for semantic CDSL.
|
|
|
|
The analyzer is intentionally independent of the geometry kernel. It treats
|
|
the CDSL ``execution_status`` as provenance, then derives current executable
|
|
state from registered atomic executors, profile support, and complete inputs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
from .runtime_types import CapabilityResult, FeaturePlanNode, HoleSpec, RuntimeDiagnostic
|
|
|
|
|
|
_SELECTOR_REQUIRED = frozenset({"fillet", "chamfer"})
|
|
_SKETCH_ATOM_PREFIXES = ("extrude_", "revolve_")
|
|
_PRIMARY_ATOMICS = frozenset({
|
|
"extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind",
|
|
"revolve_add", "revolve_cut", "hole_blind", "hole_countersink",
|
|
"hole_counterbore", "sphere_add",
|
|
})
|
|
_HOLE_ATOMICS = frozenset({"hole_blind", "hole_countersink", "hole_counterbore", "hole_wizard"})
|
|
_ACTIVE_BODY_REQUIRED = frozenset({
|
|
"extrude_cut_blind", "revolve_cut", *_HOLE_ATOMICS, "fillet", "chamfer",
|
|
})
|
|
_BODY_MUTATING_ATOMICS = frozenset({
|
|
"extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind",
|
|
"revolve_add", "revolve_cut", "sphere_add", *_HOLE_ATOMICS, "fillet", "chamfer",
|
|
})
|
|
# A pattern may replay a previous pattern as well as a direct body mutation.
|
|
# Context-only features have no geometry definition to instance.
|
|
_REPLAYABLE_ATOMICS = _BODY_MUTATING_ATOMICS | frozenset({"pattern_linear", "pattern_mirror"})
|
|
_SUPPORTED_EXTENTS = frozenset({
|
|
"blind", "mid_plane", "through_all", "through_all_both", "through_all_and_blind",
|
|
"up_to_surface", "up_to_vertex", "offset_from_surface", "through_next", "up_to_body",
|
|
})
|
|
_EXTENT_TARGET_KINDS = {
|
|
"up_to_surface": "face",
|
|
"offset_from_surface": "face",
|
|
"up_to_vertex": "vertex",
|
|
"up_to_body": "body",
|
|
}
|
|
|
|
|
|
def _has_explicit_axis(axis: Any) -> bool:
|
|
return isinstance(axis, dict) and axis.get("origin_mm") is not None and axis.get("direction") is not None
|
|
|
|
|
|
def _has_resolvable_axis_selector(node: FeaturePlanNode) -> bool:
|
|
axis = node.params.get("axis") or {}
|
|
selector = axis.get("selector") if isinstance(axis, dict) else None
|
|
if not isinstance(selector, dict):
|
|
selector = next((item for item in node.selectors if item.get("kind") == "axis"), None)
|
|
# A source stable id does not survive SolidWorks -> OCC. An axis selector
|
|
# must therefore name the preceding context feature explicitly.
|
|
return isinstance(selector, dict) and selector.get("kind") == "axis" and bool(selector.get("owner_feature_id"))
|
|
|
|
|
|
def _has_explicit_host_frame(params: dict[str, Any]) -> bool:
|
|
host = params.get("host_face")
|
|
frame = host.get("frame") if isinstance(host, dict) else None
|
|
return isinstance(frame, dict) and all(frame.get(key) is not None for key in ("origin_mm", "x_dir", "normal"))
|
|
|
|
|
|
def pattern_transform_blocker(source: FeaturePlanNode) -> str | None:
|
|
"""Return the selector dependency that cannot be transformed exactly.
|
|
|
|
An explicit host frame is coordinate data, not a topology guess. It can
|
|
be transformed with a patterned instance while preserving local hole
|
|
positions. All topology selectors and selector-dependent extents remain
|
|
blocked until their geometry transform contract is implemented.
|
|
"""
|
|
if source.selectors:
|
|
return "feature selector"
|
|
host = source.params.get("host_face")
|
|
if host is not None and not _has_explicit_host_frame(source.params):
|
|
return "host face selector"
|
|
end_condition = source.params.get("end_condition") or {}
|
|
if isinstance(end_condition, dict) and isinstance(end_condition.get("reference"), dict):
|
|
return "extent target selector"
|
|
return None
|
|
|
|
|
|
def _schema_contract() -> dict[str, dict[str, Any]]:
|
|
path = Path(__file__).with_name("profile_schema.json")
|
|
return json.loads(path.read_text(encoding="utf-8"))["feature_atomic_ids"]
|
|
|
|
|
|
def sketch_ids_required_by_contract(cdsl: dict[str, Any]) -> frozenset[str]:
|
|
"""Return only sketches consumed by a feature with a sketch contract.
|
|
|
|
CAD documents commonly preserve construction or abandoned sketches whose
|
|
contours are incomplete. They are semantic data, but must not make an
|
|
otherwise independent feature history ineligible for execution.
|
|
"""
|
|
contracts = _schema_contract()
|
|
return frozenset(
|
|
str(feature["sketch_id"])
|
|
for feature in cdsl.get("features") or ()
|
|
if feature.get("sketch_id") is not None
|
|
and (contracts.get(str(feature.get("atomic_id") or "")) or {}).get("requires_sketch")
|
|
)
|
|
|
|
|
|
def _has_closed_region(sketch: dict[str, Any]) -> bool:
|
|
"""Mirror the adapter's input contract without importing the geometry kernel."""
|
|
regions = sketch.get("contour_regions_mm") or []
|
|
if any(len(region.get("outer") or []) >= 2 for region in regions if isinstance(region, dict)):
|
|
return True
|
|
if len(sketch.get("contour_edges_mm") or []) >= 2:
|
|
return True
|
|
return any(
|
|
entity.get("type") == "circle" and not entity.get("construction")
|
|
and float(entity.get("radius_mm") or 0.0) > 0
|
|
for entity in sketch.get("entities") or []
|
|
if isinstance(entity, dict)
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CapabilityAnalysis:
|
|
plan: tuple[FeaturePlanNode, ...]
|
|
feature_results: tuple[CapabilityResult, ...]
|
|
document_blockers: tuple[RuntimeDiagnostic, ...] = ()
|
|
|
|
@property
|
|
def runtime_eligible(self) -> bool:
|
|
return not self.document_blockers and all(result.executable for result in self.feature_results)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"runtime_eligible": self.runtime_eligible,
|
|
"feature_results": [result.as_dict() for result in self.feature_results],
|
|
"document_blockers": [blocker.as_dict() for blocker in self.document_blockers],
|
|
}
|
|
|
|
|
|
class CapabilityAnalyzer:
|
|
"""Determine whether a semantic document is executable by this runtime."""
|
|
|
|
def __init__(self, *, atomic_ids: Iterable[str], profile_types: Iterable[str]) -> None:
|
|
self.atomic_ids = frozenset(atomic_ids)
|
|
self.profile_types = frozenset(profile_types)
|
|
self.contracts = _schema_contract()
|
|
|
|
def _blocker(self, feature_id: str, code: str, message: str, **detail: Any) -> RuntimeDiagnostic:
|
|
return RuntimeDiagnostic(code=code, message=message, feature_id=feature_id, detail=detail)
|
|
|
|
def _plan(self, cdsl: dict[str, Any]) -> tuple[FeaturePlanNode, ...]:
|
|
return tuple(
|
|
FeaturePlanNode(
|
|
feature_id=str(feature.get("id") or ""),
|
|
atomic_id=str(feature.get("atomic_id") or ""),
|
|
name=feature.get("name"),
|
|
depends_on=tuple(feature.get("depends_on") or ()),
|
|
params=dict(feature.get("params") or {}),
|
|
selectors=tuple(feature.get("selectors") or ()),
|
|
sketch_id=feature.get("sketch_id"),
|
|
declared_status=feature.get("execution_status"),
|
|
source_feature=feature,
|
|
)
|
|
for feature in cdsl.get("features") or ()
|
|
)
|
|
|
|
def analyze(
|
|
self,
|
|
cdsl: dict[str, Any],
|
|
*,
|
|
sketch_errors: dict[str, str] | None = None,
|
|
) -> CapabilityAnalysis:
|
|
sketches = {str(sketch.get("id")): sketch for sketch in (cdsl.get("geometry") or {}).get("sketches") or ()}
|
|
sketch_errors = sketch_errors or {}
|
|
plan = self._plan(cdsl)
|
|
nodes_by_id = {node.feature_id: node for node in plan}
|
|
results: list[CapabilityResult] = []
|
|
completed: set[str] = set()
|
|
body_available = False
|
|
for node in plan:
|
|
blockers: list[RuntimeDiagnostic] = []
|
|
contract = self.contracts.get(node.atomic_id)
|
|
required = [f"atomic:{node.atomic_id}"]
|
|
if not contract:
|
|
blockers.append(self._blocker(node.feature_id, "unknown_atomic", "The semantic schema has no atomic contract", atomic_id=node.atomic_id))
|
|
elif node.atomic_id not in self.atomic_ids:
|
|
blockers.append(self._blocker(node.feature_id, "unsupported_atomic", "The current runtime has no registered executor", atomic_id=node.atomic_id))
|
|
for unresolved in node.source_feature.get("unresolved") or ():
|
|
blockers.append(self._blocker(node.feature_id, "unresolved_input", str(unresolved)))
|
|
for dependency in node.depends_on:
|
|
if dependency not in completed:
|
|
blockers.append(self._blocker(node.feature_id, "dependency_unavailable", "Feature dependency did not become executable", dependency=dependency))
|
|
if node.atomic_id in _ACTIVE_BODY_REQUIRED:
|
|
required.append("active_body")
|
|
if not body_available:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_active_body",
|
|
"This feature mutates an existing body, but no preceding executable feature created one",
|
|
))
|
|
params = node.params
|
|
if contract:
|
|
for parameter in contract.get("required_params") or ():
|
|
if params.get(parameter) is None:
|
|
blockers.append(self._blocker(node.feature_id, "missing_parameter", "Required parameter is missing", parameter=parameter))
|
|
if contract.get("requires_sketch"):
|
|
if not node.sketch_id or node.sketch_id not in sketches:
|
|
blockers.append(self._blocker(node.feature_id, "missing_sketch", "Feature requires an existing sketch", sketch_id=node.sketch_id))
|
|
else:
|
|
profile_type = str((sketches[node.sketch_id].get("profile") or {}).get("type") or "")
|
|
required.append(f"profile:{profile_type}")
|
|
resolution_error = sketch_errors.get(str(node.sketch_id))
|
|
if resolution_error:
|
|
blockers.append(self._blocker(
|
|
node.feature_id,
|
|
"profile_resolution_failed",
|
|
"The feature's sketch could not be resolved into executable regions",
|
|
sketch_id=node.sketch_id,
|
|
reason=resolution_error,
|
|
))
|
|
if profile_type not in self.profile_types:
|
|
blockers.append(self._blocker(node.feature_id, "unsupported_profile", "The current runtime cannot resolve the sketch profile", profile_type=profile_type))
|
|
elif not resolution_error and not _has_closed_region(sketches[node.sketch_id]):
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "profile_no_closed_region",
|
|
"The resolved sketch contains no closed profile region",
|
|
sketch_id=node.sketch_id,
|
|
))
|
|
if node.atomic_id.startswith(_SKETCH_ATOM_PREFIXES):
|
|
end_condition = params.get("end_condition") or {"type": "blind"}
|
|
end_type = end_condition.get("type")
|
|
required.append(f"extent:{end_type}")
|
|
if end_type not in _SUPPORTED_EXTENTS:
|
|
blockers.append(self._blocker(node.feature_id, "unsupported_extent", "The extent needs a resolved topology selector or is not implemented", extent=end_type))
|
|
target_kind = _EXTENT_TARGET_KINDS.get(end_type)
|
|
if target_kind:
|
|
required.append(f"selector:extent_target:{target_kind}")
|
|
reference = end_condition.get("reference")
|
|
if not isinstance(reference, dict):
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_extent_reference",
|
|
"This end condition requires a captured target selector", extent=end_type,
|
|
))
|
|
elif reference.get("kind") != target_kind:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "unsupported_extent_target",
|
|
"The captured target kind is incompatible with this end condition",
|
|
extent=end_type, expected_kind=target_kind, actual_kind=reference.get("kind"),
|
|
))
|
|
if end_type == "offset_from_surface" and abs(float(params.get("distance_mm") or 0.0)) <= 1e-12:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_offset_distance",
|
|
"Offset-from-surface requires a non-zero captured offset distance",
|
|
))
|
|
if node.atomic_id == "extrude_add_two_sided":
|
|
reverse_condition = params.get("reverse_end_condition") or {"type": "blind"}
|
|
reverse_type = reverse_condition.get("type")
|
|
required.append(f"extent:reverse:{reverse_type}")
|
|
if reverse_type not in _SUPPORTED_EXTENTS:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "unsupported_reverse_extent",
|
|
"The reverse extent is not implemented", extent=reverse_type,
|
|
))
|
|
reverse_target_kind = _EXTENT_TARGET_KINDS.get(reverse_type)
|
|
if reverse_target_kind:
|
|
required.append(f"selector:reverse_extent_target:{reverse_target_kind}")
|
|
reverse_reference = reverse_condition.get("reference")
|
|
if not isinstance(reverse_reference, dict):
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_reverse_extent_reference",
|
|
"This reverse end condition requires a captured target selector", extent=reverse_type,
|
|
))
|
|
elif reverse_reference.get("kind") != reverse_target_kind:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "unsupported_reverse_extent_target",
|
|
"The reverse target kind is incompatible with this end condition",
|
|
extent=reverse_type, expected_kind=reverse_target_kind,
|
|
actual_kind=reverse_reference.get("kind"),
|
|
))
|
|
if reverse_type == "offset_from_surface" and abs(float(params.get("reverse_distance_mm") or 0.0)) <= 1e-12:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_reverse_offset_distance",
|
|
"Reverse offset-from-surface requires a non-zero captured offset distance",
|
|
))
|
|
if node.atomic_id in _SELECTOR_REQUIRED and not node.selectors:
|
|
blockers.append(self._blocker(node.feature_id, "missing_selector", "Dress-up features require an explicit selector"))
|
|
if node.atomic_id in _HOLE_ATOMICS:
|
|
required.append("selector:host_face")
|
|
if not params.get("host_face"):
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_host_face",
|
|
"Hole operations require a host face selector or an explicit host frame",
|
|
))
|
|
try:
|
|
HoleSpec.from_feature(node.atomic_id, params, wizard=node.atomic_id == "hole_wizard")
|
|
except ValueError as error:
|
|
blockers.append(self._blocker(node.feature_id, "invalid_hole_spec", str(error)))
|
|
if node.atomic_id == "hole_wizard":
|
|
if params.get("thread"):
|
|
blockers.append(self._blocker(node.feature_id, "unsupported_hole_subtype", "Threaded Hole Wizard geometry is not represented by the current CDSL runtime"))
|
|
hole_extent = (params.get("end_condition") or {"type": "blind"}).get("type")
|
|
if hole_extent not in {"blind", "through_all", "through_all_both"}:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "unsupported_hole_extent",
|
|
"The current Hole Wizard runtime supports blind and through-all extents only",
|
|
extent=hole_extent,
|
|
))
|
|
if not params.get("positions"):
|
|
blockers.append(self._blocker(node.feature_id, "missing_hole_positions", "Hole Wizard requires captured positions"))
|
|
if node.atomic_id == "reference_plane" and not isinstance(params.get("plane"), dict):
|
|
blockers.append(self._blocker(node.feature_id, "missing_reference_orientation", "Reference plane requires an explicit plane frame"))
|
|
if node.atomic_id == "reference_plane" and isinstance(params.get("plane"), dict) and params["plane"].get("unresolved"):
|
|
blockers.append(self._blocker(node.feature_id, "missing_reference_orientation", "Reference plane orientation was not captured"))
|
|
if node.atomic_id == "reference_axis":
|
|
axis = params.get("axis") or {}
|
|
if not (axis.get("origin_mm") and axis.get("direction")):
|
|
plane_selectors = [selector for selector in node.selectors if selector.get("kind") == "plane"]
|
|
if len(plane_selectors) < 2:
|
|
blockers.append(self._blocker(node.feature_id, "missing_reference_axis_geometry", "Reference axis requires explicit geometry or two reference planes"))
|
|
if node.atomic_id.startswith("revolve_"):
|
|
axis = params.get("axis")
|
|
if not _has_explicit_axis(axis) and not _has_resolvable_axis_selector(node):
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_revolve_axis",
|
|
"Revolve requires an explicit axis or an owner-qualified reference-axis selector",
|
|
))
|
|
if node.atomic_id.startswith("pattern_"):
|
|
required.append("feature_replay")
|
|
sources = params.get("source_feature_ids") or []
|
|
if not sources:
|
|
blockers.append(self._blocker(node.feature_id, "missing_pattern_source", "Pattern has no source features"))
|
|
for source_id in sources:
|
|
source = nodes_by_id.get(str(source_id))
|
|
if source is None:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "missing_pattern_source",
|
|
"Pattern source feature does not exist", source_feature_id=source_id,
|
|
))
|
|
continue
|
|
if source.atomic_id not in _REPLAYABLE_ATOMICS:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "unsupported_pattern_source",
|
|
"Pattern source has no replayable body definition",
|
|
source_feature_id=source_id, atomic_id=source.atomic_id,
|
|
))
|
|
continue
|
|
if source.feature_id not in completed:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "pattern_source_unavailable",
|
|
"Pattern source did not become executable before this pattern",
|
|
source_feature_id=source_id,
|
|
))
|
|
continue
|
|
transform_dependency = pattern_transform_blocker(source) if source else None
|
|
if transform_dependency:
|
|
blockers.append(self._blocker(
|
|
node.feature_id, "unsupported_pattern_selector_transform",
|
|
"Pattern source uses a topology dependency that cannot yet be transformed",
|
|
source_feature_id=source_id, dependency=transform_dependency,
|
|
))
|
|
if node.atomic_id == "pattern_mirror" and not params.get("mirror_plane"):
|
|
blockers.append(self._blocker(node.feature_id, "missing_mirror_plane", "Mirror pattern has no mirror plane"))
|
|
status = "executable" if not blockers else ("unsupported" if any(b.code.startswith("unsupported") or b.code == "unknown_atomic" for b in blockers) else "blocked")
|
|
results.append(CapabilityResult(node.feature_id, node.atomic_id, status, tuple(required), tuple(blockers)))
|
|
if status == "executable":
|
|
completed.add(node.feature_id)
|
|
if node.atomic_id in _BODY_MUTATING_ATOMICS:
|
|
body_available = True
|
|
body_producers = {
|
|
"extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind",
|
|
"revolve_add", "revolve_cut", "sphere_add",
|
|
}
|
|
document_blockers: list[RuntimeDiagnostic] = []
|
|
if not any(node.atomic_id in body_producers for node in plan):
|
|
document_blockers.append(RuntimeDiagnostic(
|
|
"no_solid_feature", "CDSL contains no feature capable of creating a solid body",
|
|
))
|
|
return CapabilityAnalysis(plan=plan, feature_results=tuple(results), document_blockers=tuple(document_blockers))
|