Files
cdsl-cad/cadfs_to_cdsl/lowering.py
T
likang 038d38ed98 feat(cadfs): 补齐核心建模能力并建立代表性回归
- 新增 loft、双向切除、through-all/up-to-next 等 CADFS lowering 与 engine 支持
- 支持多种 reference plane、B-spline profile 和 circular pattern replay
- 保留 transform 历史,并烘焙安全的单源平移/旋转变换
- 改进 selector 绑定、拓扑快照和 pattern 变换处理
- 建立 17 个代表样本的转换、重建与比较回归工具链
- 补充 schema、author guidance、运行时和几何回归测试
2026-09-07 18:21:07 +08:00

798 lines
51 KiB
Python

from __future__ import annotations
import math
from dataclasses import dataclass
from typing import Any
from .featurescript_parser import symbolic_string
from .ir import Call, FeatureIR, ModelIR, SketchIR
from .query_parser import parse_query, walk_calls
UNSUPPORTED = {"shell", "sweep", "draft", "thicken", "split", "booleanBodies", "moveFace", "replaceFace", "deleteFace", "import", "derive"}
PLANES = {
"Top": {"origin_mm": [0., 0., 0.], "x_dir": [1., 0., 0.], "normal": [0., 0., 1.]},
"Front": {"origin_mm": [0., 0., 0.], "x_dir": [1., 0., 0.], "normal": [0., -1., 0.]},
"Right": {"origin_mm": [0., 0., 0.], "x_dir": [0., 1., 0.], "normal": [1., 0., 0.]},
}
@dataclass
class LoweringResult:
cdsl: dict[str, Any] | None
status: str
diagnostics: list[dict[str, Any]]
history: list[dict[str, Any]]
class UnsupportedCapability(ValueError):
def __init__(self, capability: str, message: str):
super().__init__(message); self.capability = capability
def plain(value: Any) -> Any:
if isinstance(value, Call): return {"call": value.name, "args": [plain(arg) for arg in value.args], "line": value.line}
if isinstance(value, list): return [plain(item) for item in value]
if isinstance(value, dict): return {key: plain(item) for key, item in value.items()}
return value
def _bool(value: Any) -> bool:
return value is True or (isinstance(value, str) and value.lower() == "true")
def _number(value: Any, units: bool = False) -> float:
if isinstance(value, (float, int)): return float(value)
if isinstance(value, str):
constants = {"mm": 1., "millimeter": 1., "cm": 10., "m": 1000., "inch": 25.4, "in": 25.4, "ft": 304.8, "degree": 1.}
if value in constants: return constants[value]
return float(value)
if isinstance(value, Call) and value.name == "__binary__":
left, op, right = value.args; a, b = _number(left, units), _number(right, units)
return {"+": a + b, "-": a - b, "*": a * b, "/": a / b}[str(op)]
raise ValueError(f"not a constant number: {plain(value)!r}")
def _point(value: Any) -> list[float]:
if isinstance(value, Call) and value.name == "__binary__" and value.args[1] == "*":
scale = _number(value.args[2], True); point = _point(value.args[0]); return [v * scale for v in point]
if isinstance(value, Call) and value.name in {"v", "vector"} and len(value.args) >= 2:
return [_number(value.args[0]), _number(value.args[1])]
if isinstance(value, list) and len(value) >= 2: return [_number(value[0]), _number(value[1])]
raise ValueError(f"not a 2D point: {plain(value)!r}")
def _cross(a: list[float], b: list[float]) -> list[float]:
return [a[1]*b[2]-a[2]*b[1], a[2]*b[0]-a[0]*b[2], a[0]*b[1]-a[1]*b[0]]
def _dot(a: list[float], b: list[float]) -> float: return sum(left * right for left, right in zip(a, b))
def _unit(value: list[float], message: str) -> list[float]:
length = math.sqrt(_dot(value, value))
if length <= 1e-9: raise ValueError(message)
return [component / length for component in value]
def _sub(a: list[float], b: list[float]) -> list[float]: return [a[index] - b[index] for index in range(3)]
def _rotate(value: list[float], axis: list[float], angle_rad: float) -> list[float]:
axis = _unit(axis, "rotation axis is degenerate")
cosine, sine = math.cos(angle_rad), math.sin(angle_rad)
cross = _cross(axis, value); projection = _dot(axis, value) * (1.0 - cosine)
return [value[index] * cosine + cross[index] * sine + axis[index] * projection for index in range(3)]
def _translate_frame(frame: dict[str, Any], offset: list[float]) -> dict[str, Any]:
return {**frame, "origin_mm": [frame["origin_mm"][index] + offset[index] for index in range(3)]}
def _rotate_point(point: list[float], axis: dict[str, list[float]], angle_rad: float) -> list[float]:
relative = _rotate(_sub(point, axis["origin_mm"]), axis["direction"], angle_rad)
return [axis["origin_mm"][index] + relative[index] for index in range(3)]
def _rotate_frame(frame: dict[str, Any], axis: dict[str, list[float]], angle_rad: float) -> dict[str, Any]:
return {
**frame,
"origin_mm": _rotate_point(frame["origin_mm"], axis, angle_rad),
"x_dir": _rotate(frame["x_dir"], axis["direction"], angle_rad),
"normal": _rotate(frame["normal"], axis["direction"], angle_rad),
}
def _y_dir(plane: dict[str, Any]) -> list[float]: return _cross(plane["normal"], plane["x_dir"])
def _global(plane: dict[str, Any], point: list[float]) -> list[float]:
y = _y_dir(plane); return [plane["origin_mm"][i] + plane["x_dir"][i]*point[0] + y[i]*point[1] for i in range(3)]
def _shift_plane(plane: dict[str, Any], distance: float) -> dict[str, Any]:
return {**plane, "origin_mm": [plane["origin_mm"][i] + plane["normal"][i]*distance for i in range(3)]}
def _oriented_plane(plane: dict[str, Any], normal_sign: float, distance: float = 0.0) -> dict[str, Any]:
shifted = _shift_plane(plane, distance)
return {**shifted, "normal": [normal_sign * value for value in plane["normal"]]}
def _frame(origin: list[float], x_dir: list[float], normal: list[float]) -> dict[str, list[float]]:
normal = _unit(normal, "reference plane normal is degenerate")
x_dir = _sub(x_dir, [normal[index] * _dot(x_dir, normal) for index in range(3)])
return {"origin_mm": origin, "x_dir": _unit(x_dir, "reference plane x direction is degenerate"), "normal": normal}
def _plane_from_query(value: Any, feature_frames: dict[str, dict[str, Any]], sketch_by_source: dict[str, dict[str, Any]] | None = None) -> dict[str, Any]:
for call in walk_calls(value):
if call.name in {"makeId", "qCreatedBy"}:
text = " ".join(symbolic_string(arg) for arg in call.args)
for name, plane in PLANES.items():
if f"{name}.planeOp" in text: return dict(plane)
query = parse_query(value)
if query.topology_type == "IMPRINT" and sketch_by_source and query.source_sketch in sketch_by_source:
return dict(sketch_by_source[query.source_sketch]["workplane"])
frame = feature_frames.get(query.owner_feature or "")
if frame and query.topology_type == "CAP_FACE":
return dict(frame["start" if query.is_start is not False else "end"])
if frame and frame.get("start") == frame.get("end") and "qCreatedBy" in query.calls:
return dict(frame["start"])
raise ValueError("unsupported or unresolved sketch workplane")
def _bound_name(value: Any) -> str:
return str(value or "BLIND").split(".")[-1].upper()
def _end_condition(value: Any) -> dict[str, Any]:
name = _bound_name(value)
if name == "BLIND": return {"type": "blind", "solidworks_code": 0}
if name == "SYMMETRIC": return {"type": "mid_plane", "solidworks_code": 8}
if name == "THROUGH_ALL": return {"type": "through_all", "solidworks_code": 1}
if name == "UP_TO_NEXT": return {"type": "through_next", "solidworks_code": 4}
raise UnsupportedCapability(f"extrude_extent:{name.lower()}", f"current CDSL atomic set has no exact extrusion operation for {name}")
def _arc(start: list[float], mid: list[float], end: list[float]) -> dict[str, Any]:
ax, ay = start; bx, by = mid; cx, cy = end
d = 2 * (ax*(by-cy) + bx*(cy-ay) + cx*(ay-by))
if abs(d) < 1e-9: raise ValueError("collinear arc points")
ux = ((ax*ax+ay*ay)*(by-cy)+(bx*bx+by*by)*(cy-ay)+(cx*cx+cy*cy)*(ay-by))/d
uy = ((ax*ax+ay*ay)*(cx-bx)+(bx*bx+by*by)*(ax-cx)+(cx*cx+cy*cy)*(bx-ax))/d
cross = (mid[0]-start[0])*(end[1]-mid[1])-(mid[1]-start[1])*(end[0]-mid[0])
return {"type": "arc", "start": start, "end": end, "center": [ux, uy], "radius_mm": math.hypot(ax-ux, ay-uy), "clockwise": cross < 0}
def _endpoint(segment: dict[str, Any], end: bool = False) -> tuple[int, int]:
point = segment["end" if end else "start"]; return round(point[0]*1e5), round(point[1]*1e5)
def _contours(segments: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
circles = [{"role": "unknown", "closed": True, "segments": [item]} for item in segments if item["type"] == "circle"]
edges = [item for item in segments if item["type"] != "circle"]; unused = set(range(len(edges))); contours = []; construction = []
while unused:
idx = unused.pop(); contour = [edges[idx]]; first = _endpoint(contour[0]); tail = _endpoint(contour[-1], True)
while tail != first:
match = next((j for j in unused if _endpoint(edges[j]) == tail or _endpoint(edges[j], True) == tail), None)
if match is None:
construction.extend(contour); break
unused.remove(match); item = dict(edges[match])
if _endpoint(item, True) == tail:
item["start"], item["end"] = item["end"], item["start"]
if item["type"] == "arc": item["clockwise"] = not item["clockwise"]
elif item["type"] == "bspline": item["points"] = list(reversed(item["points"]))
contour.append(item); tail = _endpoint(item, True)
if tail == first: contours.append({"role": "unknown", "closed": True, "segments": contour})
return contours + circles, construction
def _lower_sketch(sketch: SketchIR, plane: dict[str, Any], allow_open: bool = False) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
segments: list[dict[str, Any]] = []; explicit_construction: list[dict[str, Any]] = []; entities: dict[str, dict[str, Any]] = {}; unsupported = []
for entity in sketch.entities:
p = entity.params
if entity.operation == "skPoint": entities[entity.feature_id] = {"type": "point", "point": _point(p["position"])}; continue
if entity.operation == "skLineSegment": item = {"type": "line", "start": _point(p["start"]), "end": _point(p["end"])}
elif entity.operation == "skCircle": item = {"type": "circle", "center": _point(p["center"]), "radius_mm": _number(p["radius"], True)}
elif entity.operation == "skArc": item = _arc(_point(p["start"]), _point(p["mid"]), _point(p["end"]))
elif entity.operation == "skFitSpline":
points = [_point(point) for point in p.get("points") or []]
if len(points) < 3: raise ValueError("fit spline needs at least 3 points")
item = {"type": "bspline", "start": points[0], "end": points[-1], "points": points}
else: unsupported.append(entity.operation); continue
(explicit_construction if _bool(p.get("construction")) else segments).append(item); entities[entity.feature_id] = item
if unsupported: raise ValueError("unsupported sketch entities: " + ",".join(sorted(set(unsupported))))
if not segments:
profile: dict[str, Any] = {"type": "analytic_contours", "contours": []}
if explicit_construction: profile["construction"] = explicit_construction
return {"id": f"sketch_{sketch.feature_id}", "name": sketch.feature_id, "workplane": plane, "profile": profile, "role": "reference"}, entities
if len(segments) == 1 and segments[0]["type"] == "circle" and not explicit_construction:
profile = {"type": "circle", "center": segments[0]["center"], "radius_mm": segments[0]["radius_mm"]}
else:
contours, open_segments = _contours(segments)
if open_segments:
if not allow_open: raise ValueError(f"sketch has {len(open_segments)} open non-construction segment(s)")
profile = {"type": "analytic_contours", "contours": [], "construction": explicit_construction + open_segments}
return {"id": f"sketch_{sketch.feature_id}", "name": sketch.feature_id, "workplane": plane, "profile": profile, "role": "reference"}, entities
construction = list(explicit_construction)
if not contours:
profile = {"type": "analytic_contours", "contours": [], "construction": construction}
return {"id": f"sketch_{sketch.feature_id}", "name": sketch.feature_id, "workplane": plane, "profile": profile, "role": "reference"}, entities
profile = {"type": "analytic_contours", "contours": contours}
if construction: profile["construction"] = construction
return {"id": f"sketch_{sketch.feature_id}", "name": sketch.feature_id, "workplane": plane, "profile": profile}, entities
def _queries(value: Any) -> list[Any]:
if isinstance(value, Call) and value.name == "qUnion" and value.args and isinstance(value.args[0], list): return value.args[0]
return [value]
def _source_refs(value: Any) -> list[tuple[str, str]]:
refs = []
for call in walk_calls(value):
if call.name in {"sQuery", "sketchEntityQuery"} and len(call.args) >= 3:
refs.append((symbolic_string(call.args[0]).split(".", 1)[0], str(call.args[2])))
return refs
def _source_sketch(params: dict[str, Any]) -> str | None:
for key in ("entities", "sheetProfilesArray"):
if key in params:
query = parse_query(params[key])
if query.source_sketch: return query.source_sketch
return None
def _pattern_source_features(value: Any, previous: list[str]) -> list[str]:
sources = []
for call in walk_calls(value):
if call.name != "makeQuery" or not call.args:
continue
owner = symbolic_string(call.args[0])
if "F" not in owner:
continue
source = "f_" + owner[owner.find("F"):].split(".", 1)[0]
if source in previous and source not in sources:
sources.append(source)
if not sources:
raise ValueError("pattern source features are unresolved")
return sources
def _transform_source_features(value: Any, previous: list[str]) -> list[str]:
sources = []
for call in walk_calls(value):
if call.name not in {"makeQuery", "qCreatedBy"} or not call.args:
continue
owner = symbolic_string(call.args[0])
if "F" not in owner:
continue
source = "f_" + owner[owner.find("F"):].split(".", 1)[0]
if source in previous and source not in sources:
sources.append(source)
if not sources:
raise ValueError("transform source features are unresolved")
return sources
def _circular_pattern_axis(
value: Any,
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> dict[str, list[float]]:
query = parse_query(value)
entity = (entity_by_sketch.get(query.source_sketch or "") or {}).get(query.source_entity or "")
if entity is None or query.source_sketch not in sketch_by_source:
raise ValueError("circular pattern axis is unresolved")
plane = sketch_by_source[query.source_sketch]["workplane"]
if entity["type"] == "line":
start = _global(plane, entity["start"]); end = _global(plane, entity["end"])
direction = [end[index] - start[index] for index in range(3)]
norm = math.sqrt(sum(component * component for component in direction))
if norm <= 1e-9:
raise ValueError("circular pattern axis line is degenerate")
return {"origin_mm": start, "direction": [component / norm for component in direction]}
if entity["type"] == "circle":
frame = feature_frames.get(query.owner_feature or "")
if frame and query.is_start is not None:
plane = frame["start" if query.is_start else "end"]
return {"origin_mm": _global(plane, entity["center"]), "direction": list(plane["normal"])}
raise ValueError("circular pattern axis must be a sketch line or circular edge")
def _loft_profile_sketches(params: dict[str, Any]) -> list[str]:
# CADFS loft 的 profile 是草图 IMPRINT 面;几何仍来自原始闭合草图,
# 保留草图 source,不能把前序实体的选中面近似为新的放样轮廓。
profiles = params.get("sheetProfilesArray")
if not isinstance(profiles, list):
raise ValueError("loft sheetProfilesArray is unresolved")
sources: list[str] = []
for profile in profiles:
query_value = profile.get("sheetProfileEntities") if isinstance(profile, dict) else profile
query = parse_query(query_value)
if query.topology_type and query.topology_type != "IMPRINT":
raise UnsupportedCapability(
f"loft_profile_topology:{query.topology_type.lower()}",
f"current CDSL loft only supports sketch-imprint profiles, not {query.topology_type}",
)
if not query.source_sketch:
raise ValueError("loft profile sketch query is unresolved")
sources.append(query.source_sketch)
if len(sources) < 2:
raise ValueError("loft requires at least two profile sketches")
if len(set(sources)) != len(sources):
raise ValueError("loft profile sketches must be distinct")
return sources
def _profile_query_kind(params: dict[str, Any]) -> str | None:
for key in ("entities", "sheetProfilesArray"):
if key in params:
return parse_query(params[key]).topology_type
return None
def _profile_executable(sketch: dict[str, Any]) -> bool:
profile = sketch.get("profile") or {}
if profile.get("type") == "circle": return True
if profile.get("type") == "polygon": return len(profile.get("vertices") or []) >= 3
return bool(profile.get("contours"))
def _default_plane(value: Any) -> dict[str, Any] | None:
for call in walk_calls(value):
text = " ".join(symbolic_string(arg) for arg in call.args)
for name, plane in PLANES.items():
if f"{name}.planeOp" in text: return dict(plane)
return None
def _entity_from_query(
query: Any,
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> tuple[dict[str, Any], dict[str, Any], str]:
info = parse_query(query); source = info.source_sketch or ""; token = info.source_entity or ""
available = entity_by_sketch.get(source) or {}
entity = available.get(token)
if entity is None:
entity_id = max((key for key in available if token.startswith(key + ".")), key=len, default="")
entity = available.get(entity_id)
sketch = sketch_by_source.get(source)
if entity is None or sketch is None:
raise ValueError("reference geometry source is unresolved")
return entity, sketch["workplane"], token
def _entity_point(entity: dict[str, Any], plane: dict[str, Any], token: str) -> list[float]:
if entity["type"] == "point": return _global(plane, entity["point"])
if entity["type"] == "line":
local = entity["end"] if ".end" in token else entity["start"]
return _global(plane, local)
if entity["type"] == "bspline":
index = next((int(part) - 1 for part in token.split(".") if part.isdigit()), 0)
points = entity.get("points") or []
if not points: raise ValueError("B-spline reference point is unresolved")
return _global(plane, points[max(0, min(index, len(points) - 1))])
raise ValueError("reference entity does not define a point")
def _entity_line(entity: dict[str, Any], plane: dict[str, Any]) -> tuple[list[float], list[float]]:
if entity["type"] != "line": raise ValueError("reference entity is not a line")
return _global(plane, entity["start"]), _global(plane, entity["end"])
def _query_plane(
query: Any,
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> dict[str, Any]:
try:
return _plane_from_query(query, feature_frames, sketch_by_source)
except ValueError:
info = parse_query(query)
if info.topology_type != "SWEPT_FACE" or not info.owner_feature:
raise
entity, source_plane, _ = _entity_from_query(query, sketch_by_source, entity_by_sketch)
start, end = _entity_line(entity, source_plane); frame = feature_frames.get(info.owner_feature)
if frame is None: raise ValueError("swept face owner frame is unresolved")
direction = _unit(_sub(end, start), "swept face source line is degenerate")
normal = _cross(direction, frame["end"]["normal"])
return _frame(start, direction, normal)
def _query_line(
query: Any,
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> tuple[list[float], list[float]]:
info = parse_query(query)
entity, plane, _ = _entity_from_query(query, sketch_by_source, entity_by_sketch)
if info.topology_type == "CAP_EDGE" and info.owner_feature in feature_frames:
frame = feature_frames[info.owner_feature]["start" if info.is_start else "end"]
plane = frame
return _entity_line(entity, plane)
def _transform_axis(
query: Any,
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> dict[str, list[float]]:
start, end = _query_line(query, feature_frames, sketch_by_source, entity_by_sketch)
return {"origin_mm": start, "direction": _unit(_sub(end, start), "transform axis is degenerate")}
def _bake_transform(
params: dict[str, Any],
previous: list[str],
feature_by_id: dict[str, dict[str, Any]],
feature_source_by_id: dict[str, str],
sketches_by_id: dict[str, dict[str, Any]],
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> None:
if _bool(params.get("makeCopy")):
raise UnsupportedCapability("transform", "current CDSL engine cannot exactly copy transformed CADFS source bodies")
sources = _transform_source_features(params.get("entities"), previous)
if len(sources) != 1:
raise UnsupportedCapability("transform", "current CDSL engine cannot exactly transform multiple selected CADFS source bodies")
source_id = sources[0]; source = feature_by_id.get(source_id)
if source is None or source.get("atomic_id") not in {"extrude_add_blind", "extrude_add_two_sided", "revolve_add"}:
raise UnsupportedCapability("transform", "current CDSL engine can only bake a direct additive extrusion or revolve transform")
sketch = sketches_by_id.get(str(source.get("sketch_id") or ""))
source_feature_id = feature_source_by_id.get(source_id)
if sketch is None or source_feature_id is None:
raise UnsupportedCapability("transform", "current CDSL engine cannot resolve the transformed source feature geometry")
transform_type = str(params.get("transformType") or "").split(".")[-1].upper()
if transform_type == "TRANSLATION_3D":
offset = [_number(params.get(key, 0.0), True) for key in ("dx", "dy", "dz")]
transform_frame = lambda frame: _translate_frame(frame, offset)
transform_axis = lambda axis: {**axis, "origin_mm": [axis["origin_mm"][index] + offset[index] for index in range(3)]}
elif transform_type == "ROTATION":
axis = _transform_axis(params.get("transformAxis"), feature_frames, sketch_by_source, entity_by_sketch)
angle_rad = math.radians(_number(params.get("angle"), True))
transform_frame = lambda frame: _rotate_frame(frame, axis, angle_rad)
transform_axis = lambda value: {
**value,
"origin_mm": _rotate_point(value["origin_mm"], axis, angle_rad),
"direction": _rotate(value["direction"], axis["direction"], angle_rad),
}
else:
raise UnsupportedCapability("transform", f"current CDSL engine cannot exactly bake {transform_type or 'unknown'} transform")
sketch["workplane"] = transform_frame(sketch["workplane"])
frame = feature_frames.get(source_feature_id)
if frame is not None:
feature_frames[source_feature_id] = {key: transform_frame(value) for key, value in frame.items()}
source_axis = source.get("params", {}).get("axis")
if isinstance(source_axis, dict) and source_axis.get("origin_mm") and source_axis.get("direction"):
source["params"]["axis"] = transform_axis(source_axis)
def _query_point(
query: Any,
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> list[float]:
info = parse_query(query)
if info.topology_type == "CAP_VERTEX" and info.owner_feature in feature_frames:
references = _source_refs(query)
if len(references) >= 2:
plane = feature_frames[info.owner_feature]["start" if info.is_start else "end"]
lines = []
for source, token in references:
available = entity_by_sketch.get(source) or {}
entity = available.get(token)
if entity is None:
entity_id = max((key for key in available if token.startswith(key + ".")), key=len, default="")
entity = available.get(entity_id)
if entity and entity.get("type") == "line": lines.append(_entity_line(entity, plane))
if len(lines) >= 2:
pairs = [(math.dist(left, right), left) for left in lines[0] for right in lines[1]]
distance, point = min(pairs, key=lambda item: item[0])
if distance <= 1e-5: return point
entity, plane, token = _entity_from_query(query, sketch_by_source, entity_by_sketch)
if info.topology_type == "CAP_VERTEX" and info.owner_feature in feature_frames:
plane = feature_frames[info.owner_feature]["start" if info.is_start else "end"]
return _entity_point(entity, plane, token)
def _cplane(
params: dict[str, Any],
feature_frames: dict[str, dict[str, Any]],
sketch_by_source: dict[str, dict[str, Any]],
entity_by_sketch: dict[str, dict[str, dict[str, Any]]],
) -> dict[str, Any]:
plane_type = str(params.get("cplaneType") or "OFFSET").split(".")[-1].upper()
entities = _queries(params.get("entities"))
if plane_type == "OFFSET":
return _shift_plane(_query_plane(entities[0], feature_frames, sketch_by_source, entity_by_sketch), _number(params.get("offset", 0), True))
if plane_type == "LINE_ANGLE":
line_query = next((item for item in entities if parse_query(item).source_entity), None)
if line_query is None: raise ValueError("line-angle reference line is unresolved")
base_query = next((item for item in entities if item is not line_query), line_query)
try:
base = _query_plane(base_query, feature_frames, sketch_by_source, entity_by_sketch)
except ValueError:
_, base, _ = _entity_from_query(line_query, sketch_by_source, entity_by_sketch)
start, end = _query_line(line_query, feature_frames, sketch_by_source, entity_by_sketch)
axis = _sub(end, start); angle = _number(params.get("angle", 0.0))
if _bool(params.get("oppositeDirection")): angle = -angle
return _frame(start, _rotate(base["x_dir"], axis, math.radians(angle)), _rotate(base["normal"], axis, math.radians(angle)))
if plane_type == "PLANE_POINT":
base_query = next((item for item in entities if _default_plane(item) or "qCreatedBy" in parse_query(item).calls), None)
point_query = next((item for item in entities if item is not base_query), None)
if base_query is None or point_query is None: raise ValueError("plane-point references are unresolved")
base = _query_plane(base_query, feature_frames, sketch_by_source, entity_by_sketch)
return _frame(_query_point(point_query, feature_frames, sketch_by_source, entity_by_sketch), base["x_dir"], base["normal"])
if plane_type == "CURVE_POINT":
point_query = next((item for item in entities if parse_query(item).kind and "vertex" in parse_query(item).kind), None)
curve_query = next((item for item in entities if item is not point_query), None)
if point_query is None or curve_query is None: raise ValueError("curve-point references are unresolved")
point = _query_point(point_query, feature_frames, sketch_by_source, entity_by_sketch)
start, end = _query_line(curve_query, feature_frames, sketch_by_source, entity_by_sketch)
_, source_plane, _ = _entity_from_query(curve_query, sketch_by_source, entity_by_sketch)
return _frame(point, source_plane["normal"], _sub(end, start))
if plane_type == "THREE_POINT":
if len(entities) != 3: raise ValueError("three-point plane requires exactly three points")
first, second, third = [_query_point(item, feature_frames, sketch_by_source, entity_by_sketch) for item in entities]
normal = _cross(_sub(second, first), _sub(third, first))
if _bool(params.get("oppositeDirection")): normal = [-value for value in normal]
return _frame(first, _sub(second, first), normal)
if plane_type == "LINE_POINT":
line_query = next((item for item in entities if "edge" in (parse_query(item).kind or "")), None)
point_query = next((item for item in entities if item is not line_query), None)
if line_query is None or point_query is None: raise ValueError("line-point references are unresolved")
start, end = _query_line(line_query, feature_frames, sketch_by_source, entity_by_sketch)
point = _query_point(point_query, feature_frames, sketch_by_source, entity_by_sketch)
direction = _sub(end, start); normal = _cross(direction, _sub(point, start))
if _bool(params.get("oppositeDirection")): normal = [-value for value in normal]
return _frame(start, direction, normal)
if plane_type == "MID_PLANE":
if len(entities) != 2: raise ValueError("mid-plane requires exactly two reference planes")
first, second = [_query_plane(item, feature_frames, sketch_by_source, entity_by_sketch) for item in entities]
alignment = 1.0 if _dot(first["normal"], second["normal"]) >= 0 else -1.0
offset = _dot(_sub(second["origin_mm"], first["origin_mm"]), first["normal"]) * alignment
return _shift_plane(first, offset / 2.0)
raise UnsupportedCapability(f"reference_plane:{plane_type.lower()}", f"current converter has no exact {plane_type} reference plane")
def lower_model(model: ModelIR, provenance: dict[str, Any]) -> LoweringResult:
diagnostics: list[dict[str, Any]] = []; history = []
sketches: list[dict[str, Any]] = []; sketches_by_id: dict[str, dict[str, Any]] = {}; sketch_by_source: dict[str, dict[str, Any]] = {}; entity_by_sketch: dict[str, dict[str, dict[str, Any]]] = {}
feature_frames: dict[str, dict[str, Any]] = {}
features: list[dict[str, Any]] = []; complete = True; previous: list[str] = []
feature_by_id: dict[str, dict[str, Any]] = {}; feature_source_by_id: dict[str, str] = {}
for step in model.steps:
if isinstance(step, SketchIR):
history.append({"feature_id": step.feature_id, "operation": "newSketch", "parameters": {"sketchPlane": plain(step.workplane)}, "entities": [{"entity_id": e.feature_id, "operation": e.operation, "parameters": plain(e.params)} for e in step.entities]})
try:
plane = _plane_from_query(step.workplane, feature_frames, sketch_by_source)
lowered, entities = _lower_sketch(step, plane); sketches.append(lowered); sketches_by_id[lowered["id"]] = lowered; sketch_by_source[step.feature_id] = lowered; entity_by_sketch[step.feature_id] = entities
except Exception as exc:
# 开放草图不能作为实体 profile,但其几何仍可能是后续基准面、
# 阵列轴或旋转轴的精确引用。保留为 reference 草图,后续实体
# 特征仍由 _profile_executable 明确拒绝,不能静默把开放轮廓实体化。
try:
plane = _plane_from_query(step.workplane, feature_frames, sketch_by_source)
lowered, entities = _lower_sketch(step, plane, allow_open=True)
sketches.append(lowered); sketches_by_id[lowered["id"]] = lowered; sketch_by_source[step.feature_id] = lowered; entity_by_sketch[step.feature_id] = entities
except Exception:
pass
diagnostics.append({"code": "sketch_deferred", "feature_id": step.feature_id, "message": str(exc)}); complete = False
continue
item = step
history.append({"feature_id": item.feature_id, "operation": item.operation, "source_span": {"line_start": item.line_start, "line_end": item.line_end or item.line_start}, "parameters": plain(item.params), "raw_source": item.raw_source})
if item.operation in UNSUPPORTED:
diagnostics.append({"code": "unsupported_operation", "feature_id": item.feature_id, "operation": item.operation}); complete = False; continue
try:
fid = f"f_{item.feature_id}"; depends = list(previous[-1:]); p = item.params; feature: dict[str, Any]
if item.operation == "transform":
# 仅将单一、直接的原始实体变换烘焙回其输入几何。不能移动当前
# 聚合主体:CADFS transform 可能只选择 pattern copy 或多 body。
_bake_transform(p, previous, feature_by_id, feature_source_by_id, sketches_by_id, feature_frames, sketch_by_source, entity_by_sketch)
continue
elif item.operation == "cPlane":
plane = _cplane(p, feature_frames, sketch_by_source, entity_by_sketch)
feature = {"id": fid, "name": item.feature_id, "atomic_id": "reference_plane", "depends_on": depends, "params": {"plane": plane}, "execution_status": "supported"}
feature_frames[item.feature_id] = {"start": plane, "end": plane}
elif item.operation == "extrude":
if p.get("surfaceOperationType") is not None:
raise UnsupportedCapability("extrude_surface_or_mixed", "current CDSL engine has no exact surface or mixed solid/surface extrusion operation")
profile_kind = _profile_query_kind(p)
if profile_kind and profile_kind not in {"IMPRINT"}:
raise UnsupportedCapability(f"extrude_profile_topology:{profile_kind.lower()}", f"current CDSL engine cannot exactly replay an extrude profile selected from {profile_kind}")
source = _source_sketch(p)
if not source or source not in sketch_by_source: raise ValueError("extrude sketch query is unresolved")
if not _profile_executable(sketch_by_source[source]): raise ValueError("extrude sketch has no closed profile")
end = _end_condition("SYMMETRIC" if _bool(p.get("symmetric")) else p.get("endBound"))
depth_value = p.get("depth"); depth = _number(depth_value, True) if depth_value is not None else 1.0
operation = str(p.get("operationType") or "NEW").upper(); reverse = _bool(p.get("oppositeDirection")); cutting = any(x in operation for x in ("REMOVE", "CUT"))
second = _bool(p.get("hasSecondDirection"))
if cutting and not second and end["type"] not in {"blind", "mid_plane", "through_all", "through_next"}:
capability = f"extrude_cut_{end['type']}"
raise UnsupportedCapability(capability, f"current CDSL atomic set has no exact {capability} operation")
if not cutting and not second and end["type"] not in {"blind", "mid_plane", "through_all", "through_next"}:
capability = f"extrude_add_{end['type']}"
raise UnsupportedCapability(capability, f"current CDSL atomic set has no exact {capability} operation")
if second:
atomic = "extrude_cut_two_sided" if cutting else "extrude_add_two_sided"
params = {"distance_mm": depth, "reverse": reverse, "end_condition": end}
reverse_depth = _number(p.get("secondDirectionDepth", depth), True)
params.update({"reverse_distance_mm": reverse_depth, "reverse_end_condition": _end_condition(p.get("secondDirectionBound"))})
elif end["type"] == "mid_plane":
blind = _end_condition("BLIND")
atomic = "extrude_cut_two_sided" if cutting else "extrude_add_two_sided"
params = {"distance_mm": depth / 2, "reverse_distance_mm": depth / 2, "reverse": reverse, "end_condition": blind, "reverse_end_condition": dict(blind)}
else:
atomic = "extrude_cut_blind" if cutting else "extrude_add_blind"
params = {"distance_mm": depth, "reverse": reverse, "end_condition": end}
feature = {"id": fid, "name": item.feature_id, "atomic_id": atomic, "depends_on": depends, "sketch_id": sketch_by_source[source]["id"], "params": params, "execution_status": "supported"}
plane = sketch_by_source[source]["workplane"]
if end["type"] == "blind" and not second:
direction = -1 if reverse else 1
feature_frames[item.feature_id] = {"start": dict(plane), "end": _shift_plane(plane, direction * depth)}
elif end["type"] == "mid_plane":
direction = -1 if reverse else 1
feature_frames[item.feature_id] = {"start": _shift_plane(plane, -direction * depth / 2), "end": _shift_plane(plane, direction * depth / 2)}
elif item.operation == "loft":
sources = _loft_profile_sketches(p)
missing = [source for source in sources if source not in sketch_by_source]
if missing:
raise ValueError("loft profile sketches are unresolved: " + ", ".join(missing))
non_closed = [source for source in sources if not _profile_executable(sketch_by_source[source])]
if non_closed:
raise ValueError("loft profile sketches have no closed profile: " + ", ".join(non_closed))
feature = {
"id": fid,
"name": item.feature_id,
"atomic_id": "loft_add",
"depends_on": depends,
"params": {"profile_sketch_ids": [sketch_by_source[source]["id"] for source in sources]},
"execution_status": "supported",
}
elif item.operation == "revolve":
if p.get("surfaceOperationType") is not None and p.get("operationType") is None:
raise UnsupportedCapability("revolve_surface", "current CDSL engine has no exact surface-revolve operation")
source = _source_sketch(p)
if not source or source not in sketch_by_source: raise ValueError("revolve sketch query is unresolved")
if not _profile_executable(sketch_by_source[source]): raise ValueError("revolve sketch has no closed profile")
axis_q = parse_query(p.get("axis")); axis_entity = (entity_by_sketch.get(axis_q.source_sketch or "") or {}).get(axis_q.source_entity or "")
if not axis_entity or axis_entity.get("type") != "line": raise ValueError("revolve axis is unresolved")
plane = sketch_by_source[axis_q.source_sketch]["workplane"]; start, end = _global(plane, axis_entity["start"]), _global(plane, axis_entity["end"])
direction = [end[i]-start[i] for i in range(3)]; norm = math.sqrt(sum(x*x for x in direction)); direction = [x/norm for x in direction]
operation = str(p.get("operationType") or p.get("surfaceOperationType") or "NEW").upper(); atomic = "revolve_cut" if "REMOVE" in operation else "revolve_add"
full = "FULL" in str(p.get("revolveType") or "FULL").upper(); angle = 360.0 if full else _number(p.get("angle", 360.0))
feature = {"id": fid, "name": item.feature_id, "atomic_id": atomic, "depends_on": depends, "sketch_id": sketch_by_source[source]["id"], "params": {"angle_deg": angle, "reverse": _bool(p.get("oppositeDirection")), "axis": {"origin_mm": start, "direction": direction}}, "execution_status": "supported"}
elif item.operation in {"fillet", "chamfer"}:
key = "radius" if item.operation == "fillet" else "width"
chamfer_type = str(p.get("chamferType") or "EQUAL_OFFSETS").split(".")[-1].upper()
amount_value = p.get(key)
if item.operation == "chamfer" and chamfer_type == "TWO_OFFSETS": amount_value = p.get("width1")
amount = _number(amount_value, True); selectors = []
for index, query_value in enumerate(_queries(p.get("entities"))):
query = parse_query(query_value); owner = query.owner_feature
if not owner: raise ValueError("selector owner is unresolved")
selector_kind = "face" if query.kind in {"face", "entitytype.face"} or query.topology_type in {"CAP_FACE", "SWEPT_FACE"} else "edge"
refs = _source_refs(query_value)
source_entity = (entity_by_sketch.get(query.source_sketch or "") or {}).get(query.source_entity or "")
geometry: dict[str, Any] = {}
frame = feature_frames.get(owner); cap = frame and frame["start" if query.is_start else "end"]
if selector_kind == "face" and query.topology_type == "CAP_FACE" and cap:
geometry = {"normal": cap["normal"], "plane_offset_mm": sum(cap["normal"][i] * cap["origin_mm"][i] for i in range(3))}
if selector_kind == "edge" and source_entity and cap:
if query.topology_type == "SWEPT_EDGE" and frame:
local_point = source_entity.get("point") if source_entity["type"] == "point" else None
if len(refs) >= 2:
left = (entity_by_sketch.get(refs[0][0]) or {}).get(refs[0][1]); right = (entity_by_sketch.get(refs[1][0]) or {}).get(refs[1][1])
if left and right and left.get("type") == right.get("type") == "line":
local_point = next((a for a in (left["start"], left["end"]) for b in (right["start"], right["end"]) if math.dist(a, b) <= 1e-5), None)
if local_point is None: raise ValueError("swept edge source intersection is unresolved")
start, end = _global(frame["start"], local_point), _global(frame["end"], local_point)
geometry = {"curve_type": "line", "bbox_mm": [min(start[i], end[i]) for i in range(3)] + [max(start[i], end[i]) for i in range(3)]}
elif source_entity["type"] == "circle":
# Keep the source circle signature until the prefix has been
# rebuilt. OCC may expose it as one edge or several arcs.
selectors.append({"kind": selector_kind, "owner_feature_id": f"f_{owner}", "stable_id": f"cadfs_{fid}_{index}", "source": "runtime_snapshot", "confidence": 1.0, "geometry": geometry or {"curve_type": "circle", "source_circle_center_mm": _global(cap, source_entity["center"]), "source_circle_radius_mm": source_entity["radius_mm"], "source_plane_normal": cap["normal"]}})
continue
elif source_entity["type"] == "line":
start, end = _global(cap, source_entity["start"]), _global(cap, source_entity["end"])
geometry = {"curve_type": "line", "bbox_mm": [min(start[i], end[i]) for i in range(3)] + [max(start[i], end[i]) for i in range(3)]}
if not geometry: raise ValueError("selector geometry is unresolved")
if selector_kind == "face" and not geometry:
raise ValueError(f"{query.topology_type or 'face'} selector geometry is unresolved")
selectors.append({"kind": selector_kind, "owner_feature_id": f"f_{owner}", "stable_id": f"cadfs_{fid}_{index}", "source": "runtime_snapshot", "confidence": 1.0, "geometry": geometry})
params = {"radius_mm" if item.operation == "fillet" else "distance_mm": amount}
if item.operation == "fillet": params["tangent_propagation"] = _bool(p.get("tangentPropagation"))
elif chamfer_type == "TWO_OFFSETS":
second = _number(p.get("width2"), True)
if _bool(p.get("oppositeDirection")): params["distance_mm"], second = second, params["distance_mm"]
params["distance_2_mm"] = second
elif chamfer_type == "OFFSET_ANGLE":
angle = math.radians(_number(p.get("angle")))
if _bool(p.get("oppositeDirection")):
second = amount * math.tan(angle); params["distance_mm"] = second; params["distance_2_mm"] = amount
else: params["angle_rad"] = angle
feature = {"id": fid, "name": item.feature_id, "atomic_id": item.operation, "depends_on": depends, "params": params, "selectors": selectors, "execution_status": "supported"}
elif item.operation == "hole":
locations = _queries(p.get("locations")); positions = []; host_plane = None
for location in locations:
query = parse_query(location); source = query.source_sketch
entity = (entity_by_sketch.get(source or "") or {}).get(query.source_entity or "")
if not source or source not in sketch_by_source or not entity or entity.get("type") != "point": raise ValueError("hole location is unresolved")
positions.append({"mm": [entity["point"][0], entity["point"][1], 0.0]}); host_plane = sketch_by_source[source]["workplane"]
if not positions or host_plane is None: raise ValueError("hole has no resolved locations")
frame = {**host_plane, "y_dir": _y_dir(host_plane)}
if _bool(p.get("oppositeDirection")): frame = {**frame, "normal": [-v for v in frame["normal"]]}
style = str(p.get("style") or "SIMPLE").split(".")[-1].lower(); end = str(p.get("endStyle") or "BLIND").upper()
condition = "through_all_both" if "BOTH" in end else "through_all" if "THROUGH" in end else "blind"
depth_value = p.get("holeDepth") or p.get("tappedDepth")
if condition == "blind" and depth_value is None: raise ValueError("blind hole depth is unresolved")
depth = _number(depth_value, True) if depth_value is not None else 1.0
condition_code = {"blind": 0, "through_all": 1, "through_all_both": 2}[condition]
hole_params: dict[str, Any] = {"hole_type": style, "diameter_mm": _number(p.get("holeDiameter"), True), "depth_mm": depth, "end_condition": {"type": condition, "solidworks_code": condition_code}, "positions": positions, "host_face": {"frame": frame}}
if style.upper() in {"COUNTERSINK", "C_SINK"}:
hole_params["countersink"] = {"diameter_mm": _number(p.get("countersinkDiameter") or p.get("cSinkDiameter") or p.get("majorDiameter"), True), "angle_rad": math.radians(_number(p.get("countersinkAngle") or p.get("cSinkAngle") or 90.0))}
if style.upper() in {"COUNTERBORE", "C_BORE"}:
hole_params["counterbore"] = {"diameter_mm": _number(p.get("counterboreDiameter") or p.get("cBoreDiameter") or p.get("majorDiameter"), True), "depth_mm": _number(p.get("counterboreDepth") or p.get("cBoreDepth"), True)}
if _bool(p.get("isTappedThrough")) or p.get("tapSize") is not None: hole_params["thread"] = {"source": "CADFS", "decorative": True}
feature = {"id": fid, "name": item.feature_id, "atomic_id": "hole_wizard", "depends_on": depends, "params": hole_params, "execution_status": "supported"}
elif item.operation == "circularPattern":
sources = _pattern_source_features(p.get("entities"), previous)
axis = _circular_pattern_axis(p.get("axis"), feature_frames, sketch_by_source, entity_by_sketch)
count = int(_number(p.get("instanceCount")))
if count < 1:
raise ValueError("circular pattern instanceCount must be positive")
feature = {
"id": fid,
"name": item.feature_id,
"atomic_id": "pattern_circular",
"depends_on": list(dict.fromkeys(sources + depends)),
"params": {
"source_feature_ids": sources,
"axis": axis,
"pattern_count": count,
"sweep_angle_deg": _number(p.get("angle", 360.0)),
},
"execution_status": "supported",
}
elif item.operation == "mirror":
owners = []
for call in walk_calls(p.get("entities")):
if call.name == "makeQuery" and call.args:
owner = symbolic_string(call.args[0]);
if "F" in owner:
source_id = "f_" + owner[owner.find("F"):].split(".", 1)[0]
if source_id in previous and source_id not in owners: owners.append(source_id)
if not owners: raise ValueError("mirror source features are unresolved")
plane_query = p.get("mirrorPlane"); plane_info = parse_query(plane_query); plane_owner = f"f_{plane_info.owner_feature}" if plane_info.owner_feature else None
if plane_owner and any(existing["id"] == plane_owner and existing["atomic_id"] == "reference_plane" for existing in features):
mirror_plane = {"kind": "plane", "owner_feature_id": plane_owner, "stable_id": f"cadfs_{fid}_plane", "source": "runtime_snapshot", "confidence": 1.0}
else:
plane = _default_plane(plane_query)
if plane is None: raise ValueError("mirror plane is not a default or reference plane")
plane_owner = f"{fid}_plane"
features.append({"id": plane_owner, "name": f"{item.feature_id} plane", "atomic_id": "reference_plane", "depends_on": depends, "params": {"plane": plane}, "execution_status": "supported"})
previous.append(plane_owner)
mirror_plane = {"kind": "plane", "owner_feature_id": plane_owner, "stable_id": f"cadfs_{fid}_plane", "source": "runtime_snapshot", "confidence": 1.0}
feature = {"id": fid, "name": item.feature_id, "atomic_id": "pattern_mirror", "depends_on": list(dict.fromkeys(owners + [plane_owner])), "params": {"source_feature_ids": owners, "mirror_plane": mirror_plane}, "selectors": [mirror_plane], "execution_status": "supported"}
else:
raise ValueError(f"operation mapping not implemented: {item.operation}")
features.append(feature); feature_by_id[fid] = feature; feature_source_by_id[fid] = item.feature_id; previous.append(fid)
except UnsupportedCapability as exc:
diagnostics.append({"code": "unsupported_engine_capability", "capability": exc.capability, "feature_id": item.feature_id, "operation": item.operation, "message": str(exc)}); complete = False
except Exception as exc:
diagnostics.append({"code": "feature_deferred", "feature_id": item.feature_id, "operation": item.operation, "message": str(exc)}); complete = False
if not features: return LoweringResult(None, "deferred_no_executable_feature", diagnostics, history)
cdsl = {"schema": "cad.cdsl.llm.v1", "schema_version": "1.1.0", "kind": "part", "part_id": model.sample_id,
"meta": {"unit": "mm", "source": "CADFS", "provenance": provenance, "capability_gaps": sorted({d.get("operation") for d in diagnostics if d.get("operation")})},
"geometry": {"sketches": sketches}, "features": features}
return LoweringResult(cdsl, "converted_complete" if complete else "converted_partial", diagnostics, history)