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", "loft", "sweep", "draft", "thicken", "split", "booleanBodies", "circularPattern", "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]] 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} 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 _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 _plane_from_query(value: Any, feature_frames: dict[str, dict[str, Any]]) -> 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) frame = feature_frames.get(query.owner_feature or "") if frame: return dict(frame["start" if query.is_start is not False else "end"]) raise ValueError("unsupported or unresolved sketch workplane") 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"] 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]) -> 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"])) 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, construction = _contours(segments); construction.extend(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 _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 lower_model(model: ModelIR, provenance: dict[str, Any]) -> LoweringResult: diagnostics: list[dict[str, Any]] = []; history = [] sketches: list[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] = [] 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) lowered, entities = _lower_sketch(step, plane); sketches.append(lowered); sketch_by_source[step.feature_id] = lowered; entity_by_sketch[step.feature_id] = entities except Exception as exc: 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 == "cPlane": base = _plane_from_query(p.get("entities"), feature_frames); offset = _number(p.get("offset", 0), True); plane = _shift_plane(base, offset) feature = {"id": fid, "name": item.feature_id, "atomic_id": "reference_plane", "depends_on": depends, "params": {"plane": plane, "offset_mm": offset}, "execution_status": "supported"} feature_frames[item.feature_id] = {"start": plane, "end": plane} elif item.operation == "extrude": 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") depth = _number(p.get("depth"), True); operation = str(p.get("operationType") or "NEW").upper(); reverse = _bool(p.get("oppositeDirection")) atomic = "extrude_cut_blind" if any(x in operation for x in ("REMOVE", "CUT")) else "extrude_add_two_sided" if _bool(p.get("hasSecondDirection")) else "extrude_add_blind" params = {"distance_mm": depth, "reverse": reverse} if atomic == "extrude_add_two_sided": params["reverse_distance_mm"] = _number(p.get("secondDirectionDepth", depth), True) 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"]; feature_frames[item.feature_id] = {"start": plane, "end": _shift_plane(plane, -depth if reverse else depth)} elif item.operation == "revolve": 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, "axis": {"origin_mm": start, "direction": direction}}, "execution_status": "supported"} elif item.operation in {"fillet", "chamfer"}: key = "radius" if item.operation == "fillet" else "width"; amount = _number(p.get(key), 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") 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 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": # OCC/build123d commonly splits a closed circular edge into four # quarter-circle records. Bind all four deterministic arc centres. radius = source_entity["radius_mm"]; center = source_entity["center"] for quadrant, (sx, sy) in enumerate(((1, 1), (-1, 1), (-1, -1), (1, -1))): local = [center[0] + sx*radius/math.sqrt(2), center[1] + sy*radius/math.sqrt(2)] selectors.append({"kind": "edge", "owner_feature_id": f"f_{owner}", "stable_id": f"cadfs_{fid}_{index}_{quadrant}", "source": "runtime_snapshot", "confidence": 1.0, "geometry": {"curve_type": "circle", "center_mm": _global(cap, local)}}) 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") selectors.append({"kind": "edge", "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")) 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 hole_params: dict[str, Any] = {"hole_type": style, "diameter_mm": _number(p.get("holeDiameter"), True), "depth_mm": depth, "end_condition": {"type": condition, "solidworks_code": 1}, "positions": positions, "host_face": {"frame": frame}} if "COUNTERSINK" in style.upper(): hole_params["countersink"] = {"diameter_mm": _number(p.get("countersinkDiameter") or p.get("majorDiameter"), True), "angle_rad": math.radians(_number(p.get("countersinkAngle") or 90.0))} if "COUNTERBORE" in style.upper(): hole_params["counterbore"] = {"diameter_mm": _number(p.get("counterboreDiameter") or p.get("majorDiameter"), True), "depth_mm": _number(p.get("counterboreDepth"), 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 == "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); previous.append(fid) 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)