"""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 from .operation_contracts import materialized_feature_contracts _SELECTOR_REQUIRED = frozenset({"fillet", "chamfer"}) _SKETCH_ATOM_PREFIXES = ("extrude_", "revolve_") # 开放轮廓（closed=false / role=open）只有"刀具截面补槽口边闭合后作切除"的 # 物理意义：仅 extrude 直切类原子支持；add/回转对开放轮廓会造出无意义的封块。 _OPEN_PROFILE_ATOMICS = frozenset({"extrude_cut_blind", "extrude_cut_through"}) _PRIMARY_ATOMICS = frozenset({ "extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind", "extrude_cut_through", "revolve_add", "revolve_cut", "hole_blind", "hole_countersink", "hole_counterbore", "sphere_add", "box_add", "cylinder_add", }) _HOLE_ATOMICS = frozenset({"hole_blind", "hole_countersink", "hole_counterbore", "hole_wizard"}) _ACTIVE_BODY_REQUIRED = frozenset({ "extrude_cut_blind", "extrude_cut_through", "revolve_cut", *_HOLE_ATOMICS, "fillet", "chamfer", # thread_cut 是 cut 型特征：必须在已有主体（宿主）上做布尔差，不能凭空 # 造实体；无宿主时按 active_body 前置阻止而非让 executor 在 None 上崩溃。 "thread_cut", }) _BODY_MUTATING_ATOMICS = frozenset({ "extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind", "extrude_cut_through", "revolve_add", "revolve_cut", "sphere_add", "box_add", "cylinder_add", "thread_add", "thread_cut", *_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. thread_add # and thread_cut are excluded: pattern translation does not yet move their # parametric axis, so a replayed thread would silently re-run at the original # location. _REPLAYABLE_ATOMICS = ( _BODY_MUTATING_ATOMICS - frozenset({"thread_add", "thread_cut"}) ) | frozenset({"pattern_linear", "pattern_mirror", "pattern_circular"}) _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 _is_host_face_source(source: FeaturePlanNode) -> bool: """Whether a ``face`` selector on the source is a fixed host face. A patterned hole keeps its host face: the instance positions travel with the pattern (``_translated_node`` shifts ``positions``) while the face itself is resolved unchanged, so the face selector can stay untouched (issue #6). A face selector on anything else (for example a fillet selecting a face) would need per-instance edge geometry and must stay blocked. """ return source.atomic_id in _HOLE_ATOMICS or ( isinstance(source.params.get("host_face"), dict) and source.params.get("host_face", {}).get("kind") == "face" ) def pattern_transform_blocker(source: FeaturePlanNode) -> str | None: """Return the selector dependency that cannot be transformed exactly. Selectors fully captured by explicit coordinate data (a revolve axis with origin/direction, or a host face with a complete world frame) are transformed together with each patterned instance and are not blockers. A patterned hole's host face and an up_to_surface extrusion's target face are *fixed* body faces rather than instance geometry: positions and profiles travel with the instance while the face itself is resolved unchanged (issue #6). Selectors that must follow the instance but cannot be translated (edges, vertices, …) stay blocked until their geometry transform contract is implemented. """ for selector in source.selectors or (): if selector.get("kind") == "axis" and _has_explicit_axis(source.params.get("axis")): # The axis is coordinate data; _translated_node shifts its origin. continue if selector.get("kind") == "face" and _is_host_face_source(source): # 孔宿主面：主体上的固定面，不随实例平移；实例位置由 positions # 平移决定（_translated_node），selector 原样保留即可正确 resolve。 continue if selector.get("kind") == "plane" and source.atomic_id == "pattern_mirror" and selector.get("owner_feature_id"): # #6 pattern 引用重解析：pattern_mirror 的镜像面是对 # reference_plane 的 plane 引用（带 owner_feature_id），重放时 # 由运行时从 owner 解析显式 frame 并随实例变换 # （_translated_node/_mirrored_node 内联 frame），放行。 continue return "feature selector" host = source.params.get("host_face") if host is not None and not _has_explicit_host_frame(source.params): # 无 frame 的孔：宿主面以 face selector 形式给出（host_face 自身或 # selectors 列表）→ 上面的 face 分支已放行；其它形态（无 frame 也 # 非 face selector）仍阻塞。 if not (isinstance(host, dict) and host.get("kind") == "face"): return "host face selector" end_condition = source.params.get("end_condition") or {} if isinstance(end_condition, dict) and isinstance(end_condition.get("reference"), dict): # up_to_surface / offset_from_surface 的终止面是主体上的固定面： # 不随实例平移，reference 原样保留即可正确 resolve（#5 裁剪已支持 # 非均匀相交）。顶点/主体目标无法构造"固定终止面"，仍显式阻塞。 if end_condition["reference"].get("kind") == "face": return None return "extent target selector" return None def _schema_contract() -> 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 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, )) elif ( not resolution_error and sketches[node.sketch_id].get("_open_contour") and node.atomic_id not in _OPEN_PROFILE_ATOMICS ): blockers.append(self._blocker( node.feature_id, "unsupported_open_profile", "Open profiles are only supported for straight extruded cut features", sketch_id=node.sketch_id, atomic_id=node.atomic_id, )) if node.atomic_id.startswith(_SKETCH_ATOM_PREFIXES): # #2 draft：extrudeParams.draft 在 cdsl_schema.json 中被允许， # 但 runtime 的拉伸执行器（build123d Solid.extrude）没有锥形 # 拉伸能力，人读契约 profile_schema.json 也未声明该参数。 # 若 importer 把 SolidWorks 的 draft_angle_rad 写进 CDSL， # 当前 runtime 会静默产出无拔模角的直壁实体。这里把它从 # "静默忽略"改为"显式拒绝"（与 unsupported_extent 同模式）。 if params.get("draft"): blockers.append(self._blocker( node.feature_id, "unsupported_draft", "Extrude draft/taper is not implemented; the runtime would silently ignore it", )) 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": # #9 hole thread：SolidWorks 螺纹孔的 thread 是装饰信息（无螺距、 # 不进实体几何，STEP 导出即光滑孔）。HoleSpec.from_feature 只读 # 直径/深度/位置/沉头/沉孔，thread 天然不参与几何计算 → 孔特征 # 直接按光滑圆柱孔执行，runtime 侧记录 thread_decoration_ignored # info 诊断便于追溯（见 _execute_hole）。不再报 # unsupported_hole_subtype，使 ≈712 个带 thread 的孔恢复可执行。 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")) if node.atomic_id == "pattern_circular": if not _has_explicit_axis(params.get("axis")): blockers.append(self._blocker( node.feature_id, "missing_circular_axis", "Circular pattern requires an explicit axis with origin_mm and direction", )) pattern_count = params.get("pattern_count") if pattern_count is None or int(pattern_count) < 1: blockers.append(self._blocker( node.feature_id, "invalid_pattern_count", "Circular pattern requires pattern_count >= 1", )) 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", "box_add", "cylinder_add", "thread_add", # thread_cut 与 extrude_cut_blind/revolve_cut 一致：无宿主时由 # active_body 前置阻止，文档含该类特征即视为携带可执行几何。 "thread_cut", } 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))