Files
cdsl-cad/backend/engine/cdsl_engine/capabilities.py
T
2026-09-07 19:30:56 +08:00

479 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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", "bend_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. thread_add,
# thread_cut and bend_add are excluded: pattern translation does not yet move
# their parametric axis/frame, so a replayed instance would silently re-run at
# the original location.
_REPLAYABLE_ATOMICS = (
_BODY_MUTATING_ATOMICS - frozenset({"thread_add", "thread_cut", "bend_add"})
) | 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 draftextrudeParams.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 or "")
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 or "")
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 threadSolidWorks 螺纹孔的 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", "bend_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))