From 95cae203f4f967ca6fea8779d71bf4f560685145 Mon Sep 17 00:00:00 2001 From: ganjihong Date: Wed, 9 Sep 2026 14:02:53 +0800 Subject: [PATCH] refactor(cdsl_engine): freeze legacy engine paths under legacy/ Phase 7 of the decoupling refactor (behavior-preserving move): - legacy/llm_compiler.py, legacy/llm_engine.py: frozen build_pack path (unused by run_cdsl_only), moved with git mv for history - legacy/exact_rebuild.py: _run_parameterized / _run_exact / _apply_geometric_compensations moved out of rebuild.py, including the part-specific compensation table - rebuild.py keeps run_rebuild / run_cdsl_only / compare_with_gold and imports the moved functions - llm_compiler.py / llm_engine.py become compatibility shims Frozen zone: new capability belongs in executors/ + schema contracts. The project-specific compensation no longer sits in the main pipeline. --- backend/engine/cdsl_engine/legacy/__init__.py | 13 + .../cdsl_engine/legacy/exact_rebuild.py | 253 ++++++++ .../engine/cdsl_engine/legacy/llm_compiler.py | 287 +++++++++ .../engine/cdsl_engine/legacy/llm_engine.py | 555 +++++++++++++++++ backend/engine/cdsl_engine/llm_compiler.py | 291 +-------- backend/engine/cdsl_engine/llm_engine.py | 559 +----------------- backend/engine/cdsl_engine/rebuild.py | 315 ++-------- 7 files changed, 1167 insertions(+), 1106 deletions(-) create mode 100644 backend/engine/cdsl_engine/legacy/__init__.py create mode 100644 backend/engine/cdsl_engine/legacy/exact_rebuild.py create mode 100644 backend/engine/cdsl_engine/legacy/llm_compiler.py create mode 100644 backend/engine/cdsl_engine/legacy/llm_engine.py diff --git a/backend/engine/cdsl_engine/legacy/__init__.py b/backend/engine/cdsl_engine/legacy/__init__.py new file mode 100644 index 00000000..9e0d2f3c --- /dev/null +++ b/backend/engine/cdsl_engine/legacy/__init__.py @@ -0,0 +1,13 @@ +"""Frozen legacy engine paths. + +``llm_compiler`` (thin CDSL to build_pack) and ``llm_engine`` (build_pack +executor) predate the session-based runtime and are not used by +``run_cdsl_only``. ``exact_rebuild`` hosts the SolidWorks-exact fallback +rebuilds and the project-specific geometric compensations. + +Do not extend these modules; new capability belongs in the session runtime +(``executors/``) or the schema contracts. The top-level ``llm_compiler.py`` +and ``llm_engine.py`` shims keep every historical import path working. +""" + +from __future__ import annotations diff --git a/backend/engine/cdsl_engine/legacy/exact_rebuild.py b/backend/engine/cdsl_engine/legacy/exact_rebuild.py new file mode 100644 index 00000000..db9db6d9 --- /dev/null +++ b/backend/engine/cdsl_engine/legacy/exact_rebuild.py @@ -0,0 +1,253 @@ +"""SolidWorks-exact fallback rebuilds and project-specific compensations. + +``_run_parameterized`` and ``_run_exact`` replay SolidWorks rebuilds through +generated build123d scripts (``compiler_context``). They are the frozen +exact/parameterized legs of ``rebuild.run_rebuild``; the production path is +``run_cdsl_only`` in the session runtime. + +``_apply_geometric_compensations`` is a project-specific workaround for a +feature the SolidWorks export misses. It intentionally lives beside the +legacy paths so a clean checkout of this engine in another project can drop +it without touching the generic code. +""" + +from __future__ import annotations + +import sys +import time +from pathlib import Path +from typing import Any + +from ..sketch_solver import resolve_all_sketches +from ..translator import generate_build123d_code + + +def _run_parameterized(cdsl: dict[str, Any], out_step: Path, gold_step: Path | None = None) -> dict[str, Any]: + """参数化路径: CDSL语义结构 + compiler_context精确数据 → translator生成代码 → 执行 + + 采用双层IR架构： + Learning IR (CDSL) 提供参数化形状、特征结构 + Execution IR (compiler_context) 提供精确坐标 + translator 提供经过充分测试的代码生成 + """ + + import subprocess + import tempfile, os + + t0 = time.time() + + # 1. 获取 compiler_context (Execution IR: 精确坐标) + compiler_context = cdsl.get("compiler_context") or {} + if not compiler_context: + # 从外部文件加载 + ctx_file = out_step.parent / "{}.compiler_context.json".format(cdsl.get("part_id", "")) + if ctx_file.exists(): + import json as _json + with open(ctx_file, "r", encoding="utf-8") as _f: + compiler_context = _json.load(_f) + if not compiler_context: + raise RuntimeError("CDSL缺少 compiler_context，无法重建") + + part_name = str(cdsl.get("part_id") or out_step.stem) + context = dict(compiler_context) + context.setdefault("metadata", {})["part_name"] = part_name + + # 2. 将 compiler_context 的精确实体注入 CDSL 草图 (供 sketch_solver 使用) + # 015133: CDSL (Learning IR) 不含坐标，坐标来自 Execution IR + ctx_sketches_map = {s["id"]: s for s in context.get("sketches", [])} + cdsl_sketches = cdsl.get("geometry", {}).get("sketches", []) + for sk in cdsl_sketches: + ctx_sk = ctx_sketches_map.get(sk["id"]) + if ctx_sk: + # 注入 entities/contour 供 polygon/complex_arc_shape 生成器使用 + if not sk.get("entities"): + sk["entities"] = ctx_sk.get("entities", []) + if not sk.get("contour_edges_mm"): + sk["contour_edges_mm"] = ctx_sk.get("contour_edges_mm", []) + + # 3. 解析 CDSL 的参数化草图 (现在有 entities 可用) + cdsl_resolved = resolve_all_sketches(cdsl) + + # 4. 将 CDSL 解析后的 profile/profile_from 注入 compiler_context + # translator 使用 compiler_context 的精确 entities + CDSL 的 profile 分类 + cdsl_resolved_map = {s["id"]: s for s in cdsl_resolved.get("geometry", {}).get("sketches", [])} + ctx_sketches = list(context.get("sketches", [])) + updated_count = 0 + for i, ctx_sk in enumerate(ctx_sketches): + sk_id = ctx_sk.get("id", "") + cdsl_sk = cdsl_resolved_map.get(sk_id) + if cdsl_sk and cdsl_sk.get("profile"): + ctx_sketches[i] = {**ctx_sk, "profile": cdsl_sk["profile"]} + updated_count += 1 + if cdsl_sk and cdsl_sk.get("profile_from"): + ctx_sketches[i] = {**ctx_sk, "profile_from": cdsl_sk["profile_from"]} + updated_count += 1 + context["sketches"] = ctx_sketches + + # 4. 使用 compiler_context 的原始 operations（保持 translator 兼容性） + + # 5. 读取 gold volume + gold_volume_mm3 = None + if gold_step and gold_step.exists(): + try: + from build123d import import_step + gold_solid = import_step(str(gold_step)) + gold_volume_mm3 = float(gold_solid.volume) + except Exception: + pass + + # 6. 用 translator 生成并执行 + code = generate_build123d_code(context, gold_volume_mm3=gold_volume_mm3) + + # 6b. 应用几何补偿 (SW导出缺失的特征) + part_id = str(cdsl.get("part_id") or "") + code = _apply_geometric_compensations(code, part_id) + + out_step.parent.mkdir(parents=True, exist_ok=True) + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as tf: + tf.write(code) + script_path = tf.name + + try: + r = subprocess.run( + ["python", script_path], + capture_output=True, text=True, encoding="utf-8", timeout=120, + env={**os.environ, "PYTHONIOENCODING": "utf-8"}, + ) + if r.returncode != 0: + raise RuntimeError(f"Build script failed:\n{r.stderr}") + finally: + try: + os.unlink(script_path) + except Exception: + pass + + # 7. 读取重建结果 + out_step.parent.mkdir(parents=True, exist_ok=True) + built_step = Path(part_name + ".step") + if not built_step.exists(): + built_step = Path.cwd() / (part_name + ".step") + if built_step.exists(): + import shutil + shutil.copy2(str(built_step), str(out_step)) + built_step.unlink() + else: + raise RuntimeError(f"No STEP output found: {part_name}.step") + + from build123d import import_step + rebuilt = import_step(str(out_step)) + bbox = rebuilt.bounding_box() + bbox_mm = { + "min": [bbox.min.X, bbox.min.Y, bbox.min.Z], + "max": [bbox.max.X, bbox.max.Y, bbox.max.Z], + } + + elapsed = time.time() - t0 + return { + "out_step": str(out_step), + "volume_mm3": float(rebuilt.volume), + "bbox_mm": bbox_mm, + "log": [f"param: CDSL-informed translator rebuild, {updated_count} sketches updated from CDSL"], + "engine": "parameterized", + "elapsed_s": round(elapsed, 1), + } + + +def _run_exact(cdsl: dict[str, Any], out_step: Path, gold_step: Path | None = None) -> dict[str, Any]: + """精确路径: generate_build123d_code (后备)""" + + import subprocess + + compiler_context = cdsl.get("compiler_context") or {} + part_name = str(cdsl.get("part_id") or out_step.stem) + context = dict(compiler_context) + context.setdefault("metadata", {})["part_name"] = part_name + + # Read gold volume if available, for chamfer/candidate scoring + gold_volume_mm3 = None + if gold_step and gold_step.exists(): + try: + from build123d import import_step + gold_solid = import_step(str(gold_step)) + gold_volume_mm3 = float(gold_solid.volume) + except Exception: + pass + + out_step.parent.mkdir(parents=True, exist_ok=True) + + # Apply geometric compensations FIRST (may return full replacement code) + part_id = str(cdsl.get("part_id") or "") + compensation_code = _apply_geometric_compensations("", part_id) + + if compensation_code and "build123d" in compensation_code and "__main__" in compensation_code: + # 完整替换代码 (跳过generate_build123d_code) + code = compensation_code + else: + code = generate_build123d_code(context, gold_volume_mm3=gold_volume_mm3) + code = _apply_geometric_compensations(code, part_id) + + t0 = time.time() + script_path = out_step.parent / "_tmp" / f"build_{part_name}_{int(time.time())}.py" + script_path.parent.mkdir(exist_ok=True) + script_path.write_text(code, encoding="utf-8") + + completed = subprocess.run( + [sys.executable, str(script_path)], + cwd=out_step.parent, + capture_output=True, + text=True, + timeout=600, + ) + + if completed.returncode != 0: + raise RuntimeError( + f"Exact compiler FAILED (rc={completed.returncode})\n" + f"STDOUT:\n{completed.stdout[-2000:]}\n" + f"STDERR:\n{completed.stderr[-3000:]}" + ) + # Print any warnings from safe_subtract + for line in completed.stdout.split('\n'): + if 'SUBTRACT' in line or 'UNION' in line: + print(f" {line.strip()}") + + from build123d import import_step + # 生成的 build 脚本将 STEP 写到 CWD 下的 "{part_name}.step" + # 移到 out_step 位置以供后续对比 + actual_step = out_step.parent / f"{part_name}.step" + if actual_step.exists(): + import shutil + shutil.copy2(str(actual_step), str(out_step)) + solid = import_step(str(out_step)) + bb = solid.bounding_box() + elapsed = time.time() - t0 + + return { + "out_step": str(out_step), + "volume_mm3": float(solid.volume), + "bbox_mm": {"min": [bb.min.X, bb.min.Y, bb.min.Z], + "max": [bb.max.X, bb.max.Y, bb.max.Z]}, + "engine": "exact", + "elapsed_s": round(elapsed, 1), + } + + +# ═══════════════════════════════════════════════════════════════ +# Geometric compensations（项目特例；拷贝到其他项目时可删） +# ═══════════════════════════════════════════════════════════════ + +def _apply_geometric_compensations(code: str, part_id: str) -> str: + """为SW导出中缺失的特征添加几何补偿切操作""" + if part_id == "113246": + if "export_step(result, " not in code: + return code + comp = ( + " # === COMPENSATION: 侧槽 (SW缺失特征) ===\n" + " with BuildSketch(Plane(origin=(-70.0, -13.0, 10.0), " + "x_dir=(0.0, 1.0, 0.0), z_dir=(1.0, 0.0, 0.0))) as comp_sk:\n" + " Rectangle(10.0, 3.0, align=(Align.MIN, Align.MIN))\n" + " comp_cutter = extrude(comp_sk.sketch, amount=10.0)\n" + " result = safe_subtract(result, comp_cutter)\n" + ) + code = code.replace("export_step(result, ", comp + " export_step(result, ") + return code diff --git a/backend/engine/cdsl_engine/legacy/llm_compiler.py b/backend/engine/cdsl_engine/legacy/llm_compiler.py new file mode 100644 index 00000000..b3875fe7 --- /dev/null +++ b/backend/engine/cdsl_engine/legacy/llm_compiler.py @@ -0,0 +1,287 @@ +"""通用编译器：瘦 CDSL → build_pack；线性阵列在此展开为重复步骤。""" + +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path +from typing import Any + +try: + from ..sketch_solver import resolve_all_sketches +except ImportError: + from sketch_solver import resolve_all_sketches + + +REQUIRED = { + "revolve_add": ["angle_deg", "axis"], + "revolve_cut": ["angle_deg", "axis"], + "extrude_add_blind": ["distance_mm"], + "extrude_add_two_sided": ["distance_mm"], + "extrude_cut_blind": ["distance_mm"], + "hole_blind": ["diameter_mm", "depth_mm"], + "hole_countersink": ["diameter_mm", "depth_mm"], + "hole_counterbore": ["diameter_mm", "depth_mm"], + "sphere_add": ["radius_mm", "center_mm"], +} + + +def _load(path: Path) -> dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8")) + + +def _offset_sketch(sketch: dict[str, Any] | None, dx: float, dy: float, dz: float) -> dict[str, Any] | None: + if sketch is None: + return None + s = deepcopy(sketch) + wp = s.get("workplane") or {} + o = list(wp.get("origin_mm") or [0, 0, 0]) + wp["origin_mm"] = [o[0] + dx, o[1] + dy, o[2] + dz] + s["workplane"] = wp + edges = [] + for e in s.get("contour_edges_mm") or []: + ne = deepcopy(e) + for key in ("start_mm", "end_mm", "center_mm"): + if key in ne: + p = ne[key] + ne[key] = [p[0] + dx, p[1] + dy, p[2] + dz] + edges.append(ne) + if edges: + s["contour_edges_mm"] = edges + # 2D entities: shift in plane if offset has in-plane components only — skip for world offset patterns + return s + + +def _offset_params_positions(params: dict[str, Any], dx: float, dy: float, dz: float) -> dict[str, Any]: + p = deepcopy(params) + if "positions" in p: + for pos in p["positions"]: + mm = pos.get("mm") + if mm: + pos["mm"] = [mm[0] + dx, mm[1] + dy, mm[2] + dz] + if "axis" in p and isinstance(p["axis"], dict): + o = list(p["axis"].get("origin_mm") or [0, 0, 0]) + p["axis"]["origin_mm"] = [o[0] + dx, o[1] + dy, o[2] + dz] + return p + + +def compile_cdsl( + cdsl: dict[str, Any], + atoms_catalog: dict[str, Any] | None = None, + techniques_catalog: dict[str, Any] | None = None, +) -> dict[str, Any]: + allowed = set() + if atoms_catalog: + allowed = {a["atomic_id"] for a in atoms_catalog.get("atoms") or []} + techniques = { + item["technique_id"]: item + for item in (techniques_catalog or {}).get("techniques") or [] + } + + sketches = {s["id"]: s for s in (cdsl.get("geometry") or {}).get("sketches") or []} + + # 参数化轮廓求解：将 profile 字段展开为精确的 entities + contour_edges_mm + cdsl = resolve_all_sketches(cdsl) + sketches = {s["id"]: s for s in (cdsl.get("geometry") or {}).get("sketches") or []} + steps: list[dict[str, Any]] = [] + seen_ids: set[str] = set() + # feature_id -> list of emitted step dicts (for pattern source) + emitted: dict[str, list[dict[str, Any]]] = {} + + def emit(feature: dict[str, Any], params: dict[str, Any], sketch: dict[str, Any] | None, step_id: str) -> dict[str, Any]: + atomic = feature["atomic_id"] + if allowed and atomic not in allowed: + raise ValueError(f"{step_id}: atomic_id {atomic!r} is not admitted by catalog") + for dep in feature.get("depends_on") or []: + if dep not in seen_ids and not any(dep in emitted): + # dependency may be ok if earlier + if dep not in seen_ids: + raise ValueError(f"{step_id}: depends_on {dep} not yet defined") + step = { + "step_id": step_id, + "atomic_id": atomic, + "depends_on": list(feature.get("depends_on") or []), + "params": params, + "sketch": sketch, + "source_name": feature.get("name"), + } + steps.append(step) + seen_ids.add(step_id) + return step + + for feat in cdsl.get("features") or []: + fid = feat["id"] + atomic = feat.get("atomic_id") + technique_id = feat.get("technique_id") + if technique_id: + technique = techniques.get(technique_id) + if technique is None: + raise ValueError(f"{fid}: technique_id {technique_id!r} is not admitted by catalog") + groups = feat.get("params") or {} + expanded: list[dict[str, Any]] = [] + previous_step_id: str | None = None + for index, internal in enumerate(technique.get("internal_steps") or [], start=1): + group_name = internal.get("params_from") + group = deepcopy(groups.get(group_name) or {}) + if not isinstance(group, dict): + raise ValueError(f"{fid}: parameter group {group_name!r} must be an object") + params = deepcopy(group.get("params") if isinstance(group.get("params"), dict) else group) + sketch_id = group.get("sketch_id") or params.pop("sketch_id", None) + sketch = deepcopy(sketches[sketch_id]) if sketch_id and sketch_id in sketches else None + internal_atomic = internal.get("atomic_id") + if not internal_atomic: + raise ValueError(f"{fid}: technique {technique_id!r} has an invalid internal step") + for key in REQUIRED.get(internal_atomic, []): + if params.get(key) is None: + raise ValueError( + f"{fid}: technique {technique_id!r} group {group_name!r} missing {key}" + ) + internal_feature = { + "atomic_id": internal_atomic, + "depends_on": [previous_step_id] if previous_step_id else list(feat.get("depends_on") or []), + "name": f"{feat.get('name') or technique_id}:{group_name or index}", + } + step_id = f"{fid}.t{index}" + expanded.append(emit(internal_feature, params, sketch, step_id)) + previous_step_id = step_id + if len(expanded) < 2: + raise ValueError(f"{fid}: technique {technique_id!r} must expand to at least two steps") + emitted[fid] = expanded + seen_ids.add(fid) + continue + if not atomic: + raise ValueError(f"{fid}: missing atomic_id") + + if atomic == "pattern_linear": + params = feat.get("params") or {} + src_ids = params.get("source_feature_ids") or [] + c1 = int(params.get("pattern_count_1") or 1) + c2 = int(params.get("pattern_count_2") or 1) + s1 = float(params.get("spacing_1_mm") or 0) + s2 = float(params.get("spacing_2_mm") or 0) + d1 = params.get("direction_1") or [1, 0, 0] + d2 = params.get("direction_2") or [0, 1, 0] + if params.get("direction_1_reverse"): + d1 = [-d1[0], -d1[1], -d1[2]] + if params.get("direction_2_reverse"): + d2 = [-d2[0], -d2[1], -d2[2]] + + src_steps: list[dict[str, Any]] = [] + for sid in src_ids: + src_steps.extend(emitted.get(sid) or []) + if not src_steps: + # 无源则跳过并记录 + steps.append( + { + "step_id": fid, + "atomic_id": "noop_pattern", + "depends_on": list(feat.get("depends_on") or []), + "params": params, + "sketch": None, + "note": "pattern source steps missing", + } + ) + seen_ids.add(fid) + continue + + clone_steps = [] + k = 0 + for i in range(c1): + for j in range(c2): + if i == 0 and j == 0: + continue + dx = d1[0] * s1 * i + d2[0] * s2 * j + dy = d1[1] * s1 * i + d2[1] * s2 * j + dz = d1[2] * s1 * i + d2[2] * s2 * j + for src in src_steps: + k += 1 + clone_id = f"{fid}.p{k}" + fake_feat = { + "atomic_id": src["atomic_id"], + "depends_on": [steps[-1]["step_id"]] if steps else [], + "name": f"{src.get('source_name')}_pattern", + } + st = emit( + fake_feat, + _offset_params_positions(src["params"], dx, dy, dz), + _offset_sketch(src.get("sketch"), dx, dy, dz), + clone_id, + ) + clone_steps.append(st) + emitted[fid] = clone_steps + seen_ids.add(fid) + continue + + params = deepcopy(feat.get("params") or {}) + sketch_id = feat.get("sketch_id") or params.get("sketch_id") + sketch = deepcopy(sketches[sketch_id]) if sketch_id and sketch_id in sketches else None + if sketch_id: + params["sketch_id"] = sketch_id + + # Auto-derive revolve axis origin + if "revolve" in atomic and sketch and "axis" in params: + ax = params.get("axis") or {} + # 优先级: from_workplane_origin > from_contour_vertex > origin_mm 裸坐标 + wp = sketch.get("workplane") or {} + wp_origin = wp.get("origin_mm") or [0.0, 0.0, 0.0] + + if ax.get("from_workplane_origin") and "origin_mm" not in ax: + params["axis"] = deepcopy(params["axis"]) + params["axis"]["origin_mm"] = list(wp_origin) + elif "origin_mm" not in ax: + ce = sketch.get("contour_edges_mm") or [] + if ce: + idx = int(ax.get("from_contour_vertex", 0)) + vertex = ce[idx % len(ce)]["start_mm"] + params["axis"] = deepcopy(params["axis"]) + params["axis"]["origin_mm"] = list(vertex) + + for key in REQUIRED.get(atomic, []): + if key == "axis" and "axis" not in params: + raise ValueError(f"{fid}: missing axis") + if key not in ("axis",) and params.get(key) is None and key != "sketch_id": + # positions can be empty temporarily + if key in params: + continue + if key in ("diameter_mm", "depth_mm", "distance_mm", "angle_deg") and params.get(key) is None: + raise ValueError(f"{fid}: missing {key}") + + st = emit(feat, params, sketch, fid) + emitted[fid] = [st] + + # filter noop + steps = [s for s in steps if s.get("atomic_id") != "noop_pattern"] + + return { + "schema": "cad.engine_plan.v1", + "part_id": cdsl.get("part_id"), + "unit": "mm", + "steps": steps, + "compiler_context": deepcopy(cdsl.get("compiler_context")), + "meta": { + "from_cdsl_schema": cdsl.get("schema"), + "compiler": "cad-heard.llm_compiler.v1", + "n_steps": len(steps), + }, + } + + +def main() -> None: + import argparse + + ap = argparse.ArgumentParser() + ap.add_argument("--cdsl", type=Path, required=True) + ap.add_argument("--catalog", type=Path, default=None) + ap.add_argument("--techniques", type=Path, default=None) + ap.add_argument("--out", type=Path, required=True) + args = ap.parse_args() + catalog = _load(args.catalog) if args.catalog else None + techniques = _load(args.techniques) if args.techniques else None + pack = compile_cdsl(_load(args.cdsl), catalog, techniques) + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps(pack, ensure_ascii=False, indent=2), encoding="utf-8") + print(f"wrote {args.out} steps={len(pack['steps'])}") + + +if __name__ == "__main__": + main() diff --git a/backend/engine/cdsl_engine/legacy/llm_engine.py b/backend/engine/cdsl_engine/legacy/llm_engine.py new file mode 100644 index 00000000..bc913eeb --- /dev/null +++ b/backend/engine/cdsl_engine/legacy/llm_engine.py @@ -0,0 +1,555 @@ +"""build123d 绘图引擎：执行 build_pack → STEP。""" + +from __future__ import annotations + +import builtins +import json +import math +import subprocess +import sys +from pathlib import Path +from typing import Any + +# 保留内置 float，防止被 build123d 上下文 shadow +_f = builtins.float + +from build123d import ( # noqa: E402 + Align, + Axis, + BuildPart, + BuildSketch, + Circle, + Cone, + Cylinder, + Edge, + Face, + Location, + Locations, + Mode, + Plane, + Polygon, + Sphere, + Vector, + Wire, + export_step, + extrude, + import_step, + revolve, +) + + +# Keep this in sync with the execution branches in run_engine_plan. The +# agent-facing schema and its parity test prevent unsupported names reaching +# this low-level dispatcher. +SUPPORTED_ATOMIC_IDS = frozenset({ + "extrude_add_blind", + "extrude_add_two_sided", + "extrude_cut_blind", + "revolve_add", + "revolve_cut", + "hole_blind", + "hole_countersink", + "hole_counterbore", + "sphere_add", + "reference_plane", + "reference_axis", +}) + + +def _load(path: Path) -> dict[str, Any]: + return json.loads(path.read_text(encoding="utf-8")) + + +def _plane_from_workplane(wp: dict[str, Any]) -> Plane: + o = wp.get("origin_mm") or [0, 0, 0] + x = wp.get("x_dir") or [1, 0, 0] + n = wp.get("normal") or [0, 0, 1] + return Plane( + origin=Vector(_f(o[0]), _f(o[1]), _f(o[2])), + x_dir=Vector(_f(x[0]), _f(x[1]), _f(x[2])), + z_dir=Vector(_f(n[0]), _f(n[1]), _f(n[2])), + ) + + +def _axis_from_params(axis: dict[str, Any]) -> Axis: + o = axis.get("origin_mm") or [0, 0, 0] + d = axis.get("direction") or [1, 0, 0] + return Axis( + origin=Vector(_f(o[0]), _f(o[1]), _f(o[2])), + direction=Vector(_f(d[0]), _f(d[1]), _f(d[2])), + ) + + +def _ordered_profile_points(sketch: dict[str, Any]) -> list[tuple[float, float]]: + entities = sketch.get("entities") or [] + line_loop = [ + i for i, e in enumerate(entities) if e["type"] == "line" and not e.get("construction") + ] + if not line_loop: + raise ValueError(f"sketch {sketch.get('id')}: no profile lines") + pts: list[tuple[float, float]] = [] + for i in line_loop: + e = entities[i] + s = (_f(e["start"][0]), _f(e["start"][1])) + en = (_f(e["end"][0]), _f(e["end"][1])) + if not pts: + pts.append(s) + if abs(pts[-1][0] - s[0]) + abs(pts[-1][1] - s[1]) > 1e-4: + if abs(pts[-1][0] - en[0]) + abs(pts[-1][1] - en[1]) <= 1e-4: + s, en = en, s + else: + pts.append(s) + pts.append(en) + if abs(pts[0][0] - pts[-1][0]) + abs(pts[0][1] - pts[-1][1]) > 1e-4: + pts.append(pts[0]) + return pts + + +def _arc_midpoint(edge: dict[str, Any], p1: Vector, p2: Vector, center: Vector, radius: float) -> Vector: + """Return a point on the intended directed arc for ``make_three_point_arc``. + + Legacy contour data has no sweep direction and retains its prior shortest + arc behavior. Evidence-v2 analytic contours carry ``clockwise`` so a + major arc or a clockwise arc cannot be silently inverted by the adapter. + """ + v1 = p1 - center + v2 = p2 - center + if v1.length < 1e-9 or v2.length < 1e-9: + return (p1 + p2) / 2 + n = Vector(*(edge.get("normal") or [0, 0, 1])) + if n.length < 1e-9: + n = v1.cross(v2) + if n.length < 1e-9: + n = Vector(0, 0, 1) + n = n.normalized() + v1n = v1.normalized() * radius + if "clockwise" not in edge: + bisector = v1n + v2.normalized() * radius + if bisector.length < 1e-9: + bisector = n.cross(v1n) + return center + bisector.normalized() * radius + sweep = math.atan2(n.dot(v1.cross(v2)), v1.dot(v2)) + if bool(edge["clockwise"]): + if sweep >= 0: + sweep -= math.tau + elif sweep <= 0: + sweep += math.tau + half = sweep / 2 + midpoint_vector = v1n * math.cos(half) + n.cross(v1n) * math.sin(half) + return center + midpoint_vector + + +def _face_from_contour_edges(edges_mm: list[dict[str, Any]], *, desired_normal: list[float] | None = None) -> Face: + b123_edges: list[Edge] = [] + for e in edges_mm: + p1 = Vector(*e["start_mm"]) + p2 = Vector(*e["end_mm"]) + if e.get("type") == "arc" and e.get("center_mm") and e.get("radius_mm") is not None: + center = Vector(*e["center_mm"]) + r = _f(e["radius_mm"]) + v1 = p1 - center + v2 = p2 - center + if v1.length < 1e-9 or v2.length < 1e-9: + b123_edges.append(Edge.make_line(p1, p2)) + continue + mid = _arc_midpoint(e, p1, p2, center, r) + try: + b123_edges.append(Edge.make_three_point_arc(p1, mid, p2)) + except Exception: + b123_edges.append(Edge.make_line(p1, p2)) + else: + b123_edges.append(Edge.make_line(p1, p2)) + face = Face(Wire(b123_edges)) + if desired_normal is not None: + dn = Vector(*desired_normal) + if dn.length > 1e-9: + fn = face.normal_at() + if fn.dot(dn) < 0: + # 重建反转的 Wire：边顺序反转 + 每条边起止点交换 + # 这样法向自然翻转，但每条边的几何方向不变（不同于 Face.Reversed） + rev_edges: list[Edge] = [] + for e in reversed(edges_mm): + p1 = Vector(*e["end_mm"]) + p2 = Vector(*e["start_mm"]) + if e.get("type") == "arc" and e.get("center_mm") and e.get("radius_mm") is not None: + center = Vector(*e["center_mm"]) + r = _f(e["radius_mm"]) + v1 = p1 - center + v2 = p2 - center + if v1.length < 1e-9 or v2.length < 1e-9: + rev_edges.append(Edge.make_line(p1, p2)) + continue + mid = _arc_midpoint(e, p1, p2, center, r) + try: + rev_edges.append(Edge.make_three_point_arc(p1, mid, p2)) + except Exception: + rev_edges.append(Edge.make_line(p1, p2)) + else: + rev_edges.append(Edge.make_line(p1, p2)) + face = Face(Wire(rev_edges)) + return face + + +def _amount(params: dict[str, Any], *, prefer_sign: str | None = None) -> float: + dist = abs(_f(params["distance_mm"])) + if prefer_sign == "plus": + return dist + if prefer_sign == "minus": + return -dist + return -dist if bool(params.get("reverse")) else dist + + +def _build_nested_circle_profiles(circles: list[dict[str, Any]]) -> None: + """Build circular islands and holes from containment parity. + + A circle contained by one larger circle is a hole; a circle contained by + two larger circles is an island again. This preserves annular profiles + without storing the heavy tessellated sketch regions from the SW export. + """ + ordered = sorted(circles, key=lambda item: _f(item["radius_mm"]), reverse=True) + tolerance = 1e-6 + for index, circle in enumerate(ordered): + center = circle["center"] + radius = _f(circle["radius_mm"]) + containing = 0 + for outer in ordered[:index]: + outer_center = outer["center"] + outer_radius = _f(outer["radius_mm"]) + distance = math.hypot( + _f(center[0]) - _f(outer_center[0]), + _f(center[1]) - _f(outer_center[1]), + ) + if distance + radius <= outer_radius + tolerance: + containing += 1 + mode = Mode.ADD if containing % 2 == 0 else Mode.SUBTRACT + with Locations((_f(center[0]), _f(center[1]))): + Circle(radius, mode=mode) + + +def run_engine_plan( + pack: dict[str, Any], + out_step: Path, + *, + cut_sign: str = "from_params", +) -> dict[str, Any]: + log: list[str] = [] + + compiler_context = pack.get("compiler_context") + if isinstance(compiler_context, dict): + # 回退路径：使用本包 translator（不依赖外部 backend.src） + try: + from .translator import generate_build123d_code, get_part_name + except ImportError: + from translator import generate_build123d_code, get_part_name + + context = dict(compiler_context) + context.setdefault("metadata", {})["part_name"] = str(pack.get("part_id") or out_step.stem) + out_step.parent.mkdir(parents=True, exist_ok=True) + completed = subprocess.run( + [sys.executable, "-c", generate_build123d_code(context)], + cwd=out_step.parent, + capture_output=True, + text=True, + timeout=180, + ) + if completed.returncode != 0: + raise RuntimeError( + f"exact compiler execution failed\nSTDOUT:\n{completed.stdout}\nSTDERR:\n{completed.stderr}" + ) + generated_name = get_part_name({"part_name": context["metadata"]["part_name"]}) + generated = out_step.parent / f"{generated_name}.step" + if generated != out_step and generated.exists(): + generated.replace(out_step) + if not out_step.exists(): + raise RuntimeError(f"exact compiler did not generate {out_step}") + solid = import_step(str(out_step)) + bb = solid.bounding_box() + return { + "out_step": str(out_step), + "volume_mm3": _f(solid.volume), + "bbox_mm": { + "min": [bb.min.X, bb.min.Y, bb.min.Z], + "max": [bb.max.X, bb.max.Y, bb.max.Z], + }, + "engine": "translator_fallback", + } + + with BuildPart() as part: + for step in pack.get("steps") or []: + atomic = step["atomic_id"] + params = step["params"] + sketch = step.get("sketch") + sid = step.get("step_id") + + if atomic == "reference_plane": + # Context features deliberately produce no solid. They remain + # executable plan steps so their dependencies are preserved and + # can be registered by the session-based runtime. + plane = _plane_from_workplane(params.get("plane") or {}) + log.append( + f"{sid}: reference_plane origin={tuple(plane.origin)} normal={tuple(plane.z_dir)}" + ) + + elif atomic == "reference_axis": + axis = _axis_from_params(params.get("axis") or {}) + log.append( + f"{sid}: reference_axis origin={tuple(axis.position)} direction={tuple(axis.direction)}" + ) + + elif atomic == "sphere_add": + radius = _f(params.get("radius_mm") or 0) + center = params.get("center_mm") or [0, 0, 0] + if radius <= 0 or len(center) != 3: + raise ValueError(f"{sid}: sphere_add requires a positive radius_mm and center_mm") + with Locations((_f(center[0]), _f(center[1]), _f(center[2]))): + Sphere(radius, mode=Mode.ADD) + log.append(f"{sid}: sphere_add radius={radius}") + + elif atomic in ("extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind"): + if sketch is None: + raise ValueError(f"{sid}: missing sketch") + plane = _plane_from_workplane(sketch.get("workplane") or {}) + mode = Mode.SUBTRACT if "cut" in atomic else Mode.ADD + edges = sketch.get("contour_edges_mm") or [] + regions = sketch.get("contour_regions_mm") or [] + sign = cut_sign if "cut" in atomic else "from_params" + + circles = [ + e + for e in (sketch.get("entities") or []) + if e.get("type") == "circle" and not e.get("construction") + ] + lines = [ + e + for e in (sketch.get("entities") or []) + if e.get("type") == "line" and not e.get("construction") + ] + + # 多区域轮廓（外环 + 孔）：由 shape generator 展开 + if regions: + faces = [] + normal = (sketch.get("workplane") or {}).get("normal") + for reg in regions: + outer_edges = reg.get("outer") or [] + if len(outer_edges) < 2: + continue + face = _face_from_contour_edges(outer_edges, desired_normal=normal) + for hole_edges in reg.get("holes") or []: + if len(hole_edges) < 2: + continue + hole = _face_from_contour_edges(hole_edges, desired_normal=normal) + face = face.cut(hole) + faces.append(face) + if not faces: + raise ValueError(f"{sid}: contour_regions_mm produced no faces") + if atomic == "extrude_add_two_sided": + d = abs(_f(params["distance_mm"])) + for face in faces: + extrude(to_extrude=face, amount=d, both=True, mode=Mode.ADD) + else: + amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) + for face in faces: + extrude(to_extrude=face, amount=amt, mode=mode) + log.append(f"{sid}: {atomic} regions={len(faces)}") + continue + + # 切除：草图常含面外框线+圆孔；优先圆孔，避免误用外框整面切除 + prefer_circles = bool(circles) and atomic.startswith("extrude_cut") + + if prefer_circles: + with BuildSketch(plane): + for e in circles: + with Locations((_f(e["center"][0]), _f(e["center"][1]))): + Circle(_f(e["radius_mm"])) + if atomic == "extrude_add_two_sided": + d = abs(_f(params["distance_mm"])) + extrude(amount=d, both=True, mode=Mode.ADD) + else: + amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) + extrude(amount=amt, mode=mode) + log.append(f"{sid}: {atomic} circle-only n={len(circles)}") + elif len(edges) >= 2: + face = _face_from_contour_edges(edges, desired_normal=sketch.get("workplane", {}).get("normal")) + if atomic == "extrude_add_two_sided": + d = abs(_f(params["distance_mm"])) + extrude(to_extrude=face, amount=d, both=True, mode=Mode.ADD) + log.append(f"{sid}: extrude_two_sided both={d} contour") + else: + amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) + extrude(to_extrude=face, amount=amt, mode=mode) + log.append(f"{sid}: {atomic} amount={amt} contour") + elif circles and not lines: + # 纯圆轮廓：用包含层级区分实体、内孔和孔中岛。 + with BuildSketch(plane): + _build_nested_circle_profiles(circles) + if atomic == "extrude_add_two_sided": + d = abs(_f(params["distance_mm"])) + extrude(amount=d, both=True, mode=Mode.ADD) + else: + amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) + extrude(amount=amt, mode=mode) + log.append(f"{sid}: {atomic} circle-only n={len(circles)}") + else: + with BuildSketch(plane): + pts = _ordered_profile_points(sketch) + poly = pts[:-1] if len(pts) >= 2 and pts[0] == pts[-1] else pts + Polygon(*poly) + for e in circles: + with Locations((_f(e["center"][0]), _f(e["center"][1]))): + Circle(_f(e["radius_mm"]), mode=Mode.SUBTRACT) + if atomic == "extrude_add_two_sided": + d = abs(_f(params["distance_mm"])) + extrude(amount=d, both=True, mode=Mode.ADD) + log.append(f"{sid}: extrude_two_sided both={d} poly") + else: + amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) + extrude(amount=amt, mode=mode) + log.append(f"{sid}: {atomic} amount={amt} poly") + + elif atomic in ("revolve_add", "revolve_cut"): + if sketch is None: + raise ValueError(f"{sid}: missing sketch") + plane = _plane_from_workplane(sketch.get("workplane") or {}) + axis = _axis_from_params(params.get("axis") or {}) + angle = _f(params.get("angle_deg") or 360) + mode = Mode.SUBTRACT if atomic == "revolve_cut" else Mode.ADD + edges = sketch.get("contour_edges_mm") or [] + if len(edges) >= 2: + face = _face_from_contour_edges(edges, desired_normal=sketch.get("workplane", {}).get("normal")) + revolve(profiles=face, axis=axis, revolution_arc=angle, mode=mode) + else: + with BuildSketch(plane): + pts = _ordered_profile_points(sketch) + poly = pts[:-1] if len(pts) >= 2 and pts[0] == pts[-1] else pts + Polygon(*poly) + revolve(axis=axis, revolution_arc=angle, mode=mode) + log.append(f"{sid}: {atomic} angle={angle}") + + elif atomic in ("hole_blind", "hole_countersink", "hole_counterbore"): + dia = _f(params.get("diameter_mm") or 0) + depth = _f(params.get("depth_mm") or 0) + positions = params.get("positions") or [] + if sketch is not None: + plane = _plane_from_workplane(sketch.get("workplane") or {}) + else: + plane = Plane.XY + host_face = params.get("host_face") or {} + frame = host_face.get("frame") or {} + frame_origin = Vector(*(frame.get("origin_mm") or plane.origin.to_tuple())) + frame_x = Vector(*(frame.get("x_dir") or plane.x_dir.to_tuple())) + frame_y = Vector(*(frame.get("y_dir") or plane.y_dir.to_tuple())) + normal = plane.z_dir.normalized() + bb = part.part.bounding_box() + part_center = Vector( + (bb.min.X + bb.max.X) / 2, + (bb.min.Y + bb.max.Y) / 2, + (bb.min.Z + bb.max.Z) / 2, + ) + inward = normal if (part_center - frame_origin).dot(normal) >= 0 else -normal + for pos in positions: + mm = pos.get("mm") or [0, 0, 0] + start = frame_origin + frame_x * _f(mm[0]) + frame_y * _f(mm[1]) + cs_dia = _f(params.get("countersink_diameter_mm") or 0) + cs_angle = _f(params.get("countersink_angle_rad") or 0) + cb_dia = _f(params.get("counterbore_diameter_mm") or 0) + cb_depth = _f(params.get("counterbore_depth_mm") or 0) + cs_depth = ( + ((cs_dia - dia) / 2) / math.tan(cs_angle / 2) + if cs_dia > dia and cs_angle > 0 + else 0 + ) + base_offset = cs_depth + (cb_depth if cb_dia > dia else 0) + main_depth = max(0.001, abs(depth) - base_offset) + main_place = Location(Plane(origin=start + inward * base_offset, z_dir=inward)) + tools = [ + Cylinder( + radius=dia / 2, + height=main_depth, + align=(Align.CENTER, Align.CENTER, Align.MIN), + mode=Mode.PRIVATE, + ).move(main_place) + ] + if cb_dia > dia and cb_depth > 0: + tools.append( + Cylinder( + radius=cb_dia / 2, + height=cb_depth, + align=(Align.CENTER, Align.CENTER, Align.MIN), + mode=Mode.PRIVATE, + ).move(Location(Plane(origin=start, z_dir=inward))) + ) + if cs_depth > 0: + tools.append( + Cone( + bottom_radius=cs_dia / 2, + top_radius=dia / 2, + height=cs_depth, + align=(Align.CENTER, Align.CENTER, Align.MIN), + mode=Mode.PRIVATE, + ).move(Location(Plane(origin=start, z_dir=inward))) + ) + drill_angle = _f(params.get("drill_angle_rad") or 0) + if drill_angle > 0: + tip_depth = (dia / 2) / math.tan(drill_angle / 2) + tools.append( + Cone( + bottom_radius=dia / 2, + top_radius=0, + height=tip_depth, + align=(Align.CENTER, Align.CENTER, Align.MIN), + mode=Mode.PRIVATE, + ).move( + Location( + Plane(origin=start + inward * abs(depth), z_dir=inward) + ) + ) + ) + for tool in tools: + part.part = part.part.cut(tool) + log.append(f"{sid}: {atomic} npos={len(positions)}") + + else: + raise ValueError(f"unsupported atomic_id: {atomic}") + + solid = part.part + + out_step.parent.mkdir(parents=True, exist_ok=True) + export_step(solid, str(out_step)) + bb = solid.bounding_box() + return { + "out_step": str(out_step), + "volume_mm3": _f(solid.volume), + "bbox_mm": { + "min": [bb.min.X, bb.min.Y, bb.min.Z], + "max": [bb.max.X, bb.max.Y, bb.max.Z], + }, + "log": log, + "cut_sign": cut_sign, + } + + +def main() -> None: + import argparse + + ap = argparse.ArgumentParser() + ap.add_argument("--pack", type=Path, required=True) + ap.add_argument("--out-step", type=Path, required=True) + ap.add_argument("--report", type=Path, default=None) + ap.add_argument("--cut-sign", default="from_params", choices=["from_params", "plus", "minus"]) + args = ap.parse_args() + info = run_engine_plan(_load(args.pack), args.out_step, cut_sign=args.cut_sign) + if args.report: + args.report.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8") + print( + json.dumps( + {k: info[k] for k in ("out_step", "volume_mm3", "bbox_mm", "cut_sign", "engine") if k in info}, + ensure_ascii=False, + indent=2, + ) + ) + for line in info.get("log") or []: + print(line) + + +if __name__ == "__main__": + main() diff --git a/backend/engine/cdsl_engine/llm_compiler.py b/backend/engine/cdsl_engine/llm_compiler.py index 0ec2559b..d421c881 100644 --- a/backend/engine/cdsl_engine/llm_compiler.py +++ b/backend/engine/cdsl_engine/llm_compiler.py @@ -1,287 +1,12 @@ -"""通用编译器：瘦 CDSL → build_pack；线性阵列在此展开为重复步骤。""" +"""Compatibility shim: the implementation moved to ``legacy.llm_compiler``. + +``compile_cdsl`` expands a thin CDSL document into a ``build_pack`` plan. +It predates the session-based runtime and is frozen; new capability belongs +in ``executors/`` and the schema contracts. +""" from __future__ import annotations -import json -from copy import deepcopy -from pathlib import Path -from typing import Any +from .legacy.llm_compiler import REQUIRED, compile_cdsl, main -try: - from .sketch_solver import resolve_all_sketches -except ImportError: - from sketch_solver import resolve_all_sketches - - -REQUIRED = { - "revolve_add": ["angle_deg", "axis"], - "revolve_cut": ["angle_deg", "axis"], - "extrude_add_blind": ["distance_mm"], - "extrude_add_two_sided": ["distance_mm"], - "extrude_cut_blind": ["distance_mm"], - "hole_blind": ["diameter_mm", "depth_mm"], - "hole_countersink": ["diameter_mm", "depth_mm"], - "hole_counterbore": ["diameter_mm", "depth_mm"], - "sphere_add": ["radius_mm", "center_mm"], -} - - -def _load(path: Path) -> dict[str, Any]: - return json.loads(path.read_text(encoding="utf-8")) - - -def _offset_sketch(sketch: dict[str, Any] | None, dx: float, dy: float, dz: float) -> dict[str, Any] | None: - if sketch is None: - return None - s = deepcopy(sketch) - wp = s.get("workplane") or {} - o = list(wp.get("origin_mm") or [0, 0, 0]) - wp["origin_mm"] = [o[0] + dx, o[1] + dy, o[2] + dz] - s["workplane"] = wp - edges = [] - for e in s.get("contour_edges_mm") or []: - ne = deepcopy(e) - for key in ("start_mm", "end_mm", "center_mm"): - if key in ne: - p = ne[key] - ne[key] = [p[0] + dx, p[1] + dy, p[2] + dz] - edges.append(ne) - if edges: - s["contour_edges_mm"] = edges - # 2D entities: shift in plane if offset has in-plane components only — skip for world offset patterns - return s - - -def _offset_params_positions(params: dict[str, Any], dx: float, dy: float, dz: float) -> dict[str, Any]: - p = deepcopy(params) - if "positions" in p: - for pos in p["positions"]: - mm = pos.get("mm") - if mm: - pos["mm"] = [mm[0] + dx, mm[1] + dy, mm[2] + dz] - if "axis" in p and isinstance(p["axis"], dict): - o = list(p["axis"].get("origin_mm") or [0, 0, 0]) - p["axis"]["origin_mm"] = [o[0] + dx, o[1] + dy, o[2] + dz] - return p - - -def compile_cdsl( - cdsl: dict[str, Any], - atoms_catalog: dict[str, Any] | None = None, - techniques_catalog: dict[str, Any] | None = None, -) -> dict[str, Any]: - allowed = set() - if atoms_catalog: - allowed = {a["atomic_id"] for a in atoms_catalog.get("atoms") or []} - techniques = { - item["technique_id"]: item - for item in (techniques_catalog or {}).get("techniques") or [] - } - - sketches = {s["id"]: s for s in (cdsl.get("geometry") or {}).get("sketches") or []} - - # 参数化轮廓求解：将 profile 字段展开为精确的 entities + contour_edges_mm - cdsl = resolve_all_sketches(cdsl) - sketches = {s["id"]: s for s in (cdsl.get("geometry") or {}).get("sketches") or []} - steps: list[dict[str, Any]] = [] - seen_ids: set[str] = set() - # feature_id -> list of emitted step dicts (for pattern source) - emitted: dict[str, list[dict[str, Any]]] = {} - - def emit(feature: dict[str, Any], params: dict[str, Any], sketch: dict[str, Any] | None, step_id: str) -> dict[str, Any]: - atomic = feature["atomic_id"] - if allowed and atomic not in allowed: - raise ValueError(f"{step_id}: atomic_id {atomic!r} is not admitted by catalog") - for dep in feature.get("depends_on") or []: - if dep not in seen_ids and not any(dep in emitted): - # dependency may be ok if earlier - if dep not in seen_ids: - raise ValueError(f"{step_id}: depends_on {dep} not yet defined") - step = { - "step_id": step_id, - "atomic_id": atomic, - "depends_on": list(feature.get("depends_on") or []), - "params": params, - "sketch": sketch, - "source_name": feature.get("name"), - } - steps.append(step) - seen_ids.add(step_id) - return step - - for feat in cdsl.get("features") or []: - fid = feat["id"] - atomic = feat.get("atomic_id") - technique_id = feat.get("technique_id") - if technique_id: - technique = techniques.get(technique_id) - if technique is None: - raise ValueError(f"{fid}: technique_id {technique_id!r} is not admitted by catalog") - groups = feat.get("params") or {} - expanded: list[dict[str, Any]] = [] - previous_step_id: str | None = None - for index, internal in enumerate(technique.get("internal_steps") or [], start=1): - group_name = internal.get("params_from") - group = deepcopy(groups.get(group_name) or {}) - if not isinstance(group, dict): - raise ValueError(f"{fid}: parameter group {group_name!r} must be an object") - params = deepcopy(group.get("params") if isinstance(group.get("params"), dict) else group) - sketch_id = group.get("sketch_id") or params.pop("sketch_id", None) - sketch = deepcopy(sketches[sketch_id]) if sketch_id and sketch_id in sketches else None - internal_atomic = internal.get("atomic_id") - if not internal_atomic: - raise ValueError(f"{fid}: technique {technique_id!r} has an invalid internal step") - for key in REQUIRED.get(internal_atomic, []): - if params.get(key) is None: - raise ValueError( - f"{fid}: technique {technique_id!r} group {group_name!r} missing {key}" - ) - internal_feature = { - "atomic_id": internal_atomic, - "depends_on": [previous_step_id] if previous_step_id else list(feat.get("depends_on") or []), - "name": f"{feat.get('name') or technique_id}:{group_name or index}", - } - step_id = f"{fid}.t{index}" - expanded.append(emit(internal_feature, params, sketch, step_id)) - previous_step_id = step_id - if len(expanded) < 2: - raise ValueError(f"{fid}: technique {technique_id!r} must expand to at least two steps") - emitted[fid] = expanded - seen_ids.add(fid) - continue - if not atomic: - raise ValueError(f"{fid}: missing atomic_id") - - if atomic == "pattern_linear": - params = feat.get("params") or {} - src_ids = params.get("source_feature_ids") or [] - c1 = int(params.get("pattern_count_1") or 1) - c2 = int(params.get("pattern_count_2") or 1) - s1 = float(params.get("spacing_1_mm") or 0) - s2 = float(params.get("spacing_2_mm") or 0) - d1 = params.get("direction_1") or [1, 0, 0] - d2 = params.get("direction_2") or [0, 1, 0] - if params.get("direction_1_reverse"): - d1 = [-d1[0], -d1[1], -d1[2]] - if params.get("direction_2_reverse"): - d2 = [-d2[0], -d2[1], -d2[2]] - - src_steps: list[dict[str, Any]] = [] - for sid in src_ids: - src_steps.extend(emitted.get(sid) or []) - if not src_steps: - # 无源则跳过并记录 - steps.append( - { - "step_id": fid, - "atomic_id": "noop_pattern", - "depends_on": list(feat.get("depends_on") or []), - "params": params, - "sketch": None, - "note": "pattern source steps missing", - } - ) - seen_ids.add(fid) - continue - - clone_steps = [] - k = 0 - for i in range(c1): - for j in range(c2): - if i == 0 and j == 0: - continue - dx = d1[0] * s1 * i + d2[0] * s2 * j - dy = d1[1] * s1 * i + d2[1] * s2 * j - dz = d1[2] * s1 * i + d2[2] * s2 * j - for src in src_steps: - k += 1 - clone_id = f"{fid}.p{k}" - fake_feat = { - "atomic_id": src["atomic_id"], - "depends_on": [steps[-1]["step_id"]] if steps else [], - "name": f"{src.get('source_name')}_pattern", - } - st = emit( - fake_feat, - _offset_params_positions(src["params"], dx, dy, dz), - _offset_sketch(src.get("sketch"), dx, dy, dz), - clone_id, - ) - clone_steps.append(st) - emitted[fid] = clone_steps - seen_ids.add(fid) - continue - - params = deepcopy(feat.get("params") or {}) - sketch_id = feat.get("sketch_id") or params.get("sketch_id") - sketch = deepcopy(sketches[sketch_id]) if sketch_id and sketch_id in sketches else None - if sketch_id: - params["sketch_id"] = sketch_id - - # Auto-derive revolve axis origin - if "revolve" in atomic and sketch and "axis" in params: - ax = params.get("axis") or {} - # 优先级: from_workplane_origin > from_contour_vertex > origin_mm 裸坐标 - wp = sketch.get("workplane") or {} - wp_origin = wp.get("origin_mm") or [0.0, 0.0, 0.0] - - if ax.get("from_workplane_origin") and "origin_mm" not in ax: - params["axis"] = deepcopy(params["axis"]) - params["axis"]["origin_mm"] = list(wp_origin) - elif "origin_mm" not in ax: - ce = sketch.get("contour_edges_mm") or [] - if ce: - idx = int(ax.get("from_contour_vertex", 0)) - vertex = ce[idx % len(ce)]["start_mm"] - params["axis"] = deepcopy(params["axis"]) - params["axis"]["origin_mm"] = list(vertex) - - for key in REQUIRED.get(atomic, []): - if key == "axis" and "axis" not in params: - raise ValueError(f"{fid}: missing axis") - if key not in ("axis",) and params.get(key) is None and key != "sketch_id": - # positions can be empty temporarily - if key in params: - continue - if key in ("diameter_mm", "depth_mm", "distance_mm", "angle_deg") and params.get(key) is None: - raise ValueError(f"{fid}: missing {key}") - - st = emit(feat, params, sketch, fid) - emitted[fid] = [st] - - # filter noop - steps = [s for s in steps if s.get("atomic_id") != "noop_pattern"] - - return { - "schema": "cad.engine_plan.v1", - "part_id": cdsl.get("part_id"), - "unit": "mm", - "steps": steps, - "compiler_context": deepcopy(cdsl.get("compiler_context")), - "meta": { - "from_cdsl_schema": cdsl.get("schema"), - "compiler": "cad-heard.llm_compiler.v1", - "n_steps": len(steps), - }, - } - - -def main() -> None: - import argparse - - ap = argparse.ArgumentParser() - ap.add_argument("--cdsl", type=Path, required=True) - ap.add_argument("--catalog", type=Path, default=None) - ap.add_argument("--techniques", type=Path, default=None) - ap.add_argument("--out", type=Path, required=True) - args = ap.parse_args() - catalog = _load(args.catalog) if args.catalog else None - techniques = _load(args.techniques) if args.techniques else None - pack = compile_cdsl(_load(args.cdsl), catalog, techniques) - args.out.parent.mkdir(parents=True, exist_ok=True) - args.out.write_text(json.dumps(pack, ensure_ascii=False, indent=2), encoding="utf-8") - print(f"wrote {args.out} steps={len(pack['steps'])}") - - -if __name__ == "__main__": - main() +__all__ = ["REQUIRED", "compile_cdsl", "main"] diff --git a/backend/engine/cdsl_engine/llm_engine.py b/backend/engine/cdsl_engine/llm_engine.py index bc913eeb..0b4d8652 100644 --- a/backend/engine/cdsl_engine/llm_engine.py +++ b/backend/engine/cdsl_engine/llm_engine.py @@ -1,555 +1,12 @@ -"""build123d 绘图引擎：执行 build_pack → STEP。""" +"""Compatibility shim: the implementation moved to ``legacy.llm_engine``. + +``run_engine_plan`` executes a legacy ``build_pack`` through build123d. +It predates the session-based runtime and is frozen; new capability belongs +in ``executors/`` and the schema contracts. +""" from __future__ import annotations -import builtins -import json -import math -import subprocess -import sys -from pathlib import Path -from typing import Any +from .legacy.llm_engine import SUPPORTED_ATOMIC_IDS, main, run_engine_plan -# 保留内置 float，防止被 build123d 上下文 shadow -_f = builtins.float - -from build123d import ( # noqa: E402 - Align, - Axis, - BuildPart, - BuildSketch, - Circle, - Cone, - Cylinder, - Edge, - Face, - Location, - Locations, - Mode, - Plane, - Polygon, - Sphere, - Vector, - Wire, - export_step, - extrude, - import_step, - revolve, -) - - -# Keep this in sync with the execution branches in run_engine_plan. The -# agent-facing schema and its parity test prevent unsupported names reaching -# this low-level dispatcher. -SUPPORTED_ATOMIC_IDS = frozenset({ - "extrude_add_blind", - "extrude_add_two_sided", - "extrude_cut_blind", - "revolve_add", - "revolve_cut", - "hole_blind", - "hole_countersink", - "hole_counterbore", - "sphere_add", - "reference_plane", - "reference_axis", -}) - - -def _load(path: Path) -> dict[str, Any]: - return json.loads(path.read_text(encoding="utf-8")) - - -def _plane_from_workplane(wp: dict[str, Any]) -> Plane: - o = wp.get("origin_mm") or [0, 0, 0] - x = wp.get("x_dir") or [1, 0, 0] - n = wp.get("normal") or [0, 0, 1] - return Plane( - origin=Vector(_f(o[0]), _f(o[1]), _f(o[2])), - x_dir=Vector(_f(x[0]), _f(x[1]), _f(x[2])), - z_dir=Vector(_f(n[0]), _f(n[1]), _f(n[2])), - ) - - -def _axis_from_params(axis: dict[str, Any]) -> Axis: - o = axis.get("origin_mm") or [0, 0, 0] - d = axis.get("direction") or [1, 0, 0] - return Axis( - origin=Vector(_f(o[0]), _f(o[1]), _f(o[2])), - direction=Vector(_f(d[0]), _f(d[1]), _f(d[2])), - ) - - -def _ordered_profile_points(sketch: dict[str, Any]) -> list[tuple[float, float]]: - entities = sketch.get("entities") or [] - line_loop = [ - i for i, e in enumerate(entities) if e["type"] == "line" and not e.get("construction") - ] - if not line_loop: - raise ValueError(f"sketch {sketch.get('id')}: no profile lines") - pts: list[tuple[float, float]] = [] - for i in line_loop: - e = entities[i] - s = (_f(e["start"][0]), _f(e["start"][1])) - en = (_f(e["end"][0]), _f(e["end"][1])) - if not pts: - pts.append(s) - if abs(pts[-1][0] - s[0]) + abs(pts[-1][1] - s[1]) > 1e-4: - if abs(pts[-1][0] - en[0]) + abs(pts[-1][1] - en[1]) <= 1e-4: - s, en = en, s - else: - pts.append(s) - pts.append(en) - if abs(pts[0][0] - pts[-1][0]) + abs(pts[0][1] - pts[-1][1]) > 1e-4: - pts.append(pts[0]) - return pts - - -def _arc_midpoint(edge: dict[str, Any], p1: Vector, p2: Vector, center: Vector, radius: float) -> Vector: - """Return a point on the intended directed arc for ``make_three_point_arc``. - - Legacy contour data has no sweep direction and retains its prior shortest - arc behavior. Evidence-v2 analytic contours carry ``clockwise`` so a - major arc or a clockwise arc cannot be silently inverted by the adapter. - """ - v1 = p1 - center - v2 = p2 - center - if v1.length < 1e-9 or v2.length < 1e-9: - return (p1 + p2) / 2 - n = Vector(*(edge.get("normal") or [0, 0, 1])) - if n.length < 1e-9: - n = v1.cross(v2) - if n.length < 1e-9: - n = Vector(0, 0, 1) - n = n.normalized() - v1n = v1.normalized() * radius - if "clockwise" not in edge: - bisector = v1n + v2.normalized() * radius - if bisector.length < 1e-9: - bisector = n.cross(v1n) - return center + bisector.normalized() * radius - sweep = math.atan2(n.dot(v1.cross(v2)), v1.dot(v2)) - if bool(edge["clockwise"]): - if sweep >= 0: - sweep -= math.tau - elif sweep <= 0: - sweep += math.tau - half = sweep / 2 - midpoint_vector = v1n * math.cos(half) + n.cross(v1n) * math.sin(half) - return center + midpoint_vector - - -def _face_from_contour_edges(edges_mm: list[dict[str, Any]], *, desired_normal: list[float] | None = None) -> Face: - b123_edges: list[Edge] = [] - for e in edges_mm: - p1 = Vector(*e["start_mm"]) - p2 = Vector(*e["end_mm"]) - if e.get("type") == "arc" and e.get("center_mm") and e.get("radius_mm") is not None: - center = Vector(*e["center_mm"]) - r = _f(e["radius_mm"]) - v1 = p1 - center - v2 = p2 - center - if v1.length < 1e-9 or v2.length < 1e-9: - b123_edges.append(Edge.make_line(p1, p2)) - continue - mid = _arc_midpoint(e, p1, p2, center, r) - try: - b123_edges.append(Edge.make_three_point_arc(p1, mid, p2)) - except Exception: - b123_edges.append(Edge.make_line(p1, p2)) - else: - b123_edges.append(Edge.make_line(p1, p2)) - face = Face(Wire(b123_edges)) - if desired_normal is not None: - dn = Vector(*desired_normal) - if dn.length > 1e-9: - fn = face.normal_at() - if fn.dot(dn) < 0: - # 重建反转的 Wire：边顺序反转 + 每条边起止点交换 - # 这样法向自然翻转，但每条边的几何方向不变（不同于 Face.Reversed） - rev_edges: list[Edge] = [] - for e in reversed(edges_mm): - p1 = Vector(*e["end_mm"]) - p2 = Vector(*e["start_mm"]) - if e.get("type") == "arc" and e.get("center_mm") and e.get("radius_mm") is not None: - center = Vector(*e["center_mm"]) - r = _f(e["radius_mm"]) - v1 = p1 - center - v2 = p2 - center - if v1.length < 1e-9 or v2.length < 1e-9: - rev_edges.append(Edge.make_line(p1, p2)) - continue - mid = _arc_midpoint(e, p1, p2, center, r) - try: - rev_edges.append(Edge.make_three_point_arc(p1, mid, p2)) - except Exception: - rev_edges.append(Edge.make_line(p1, p2)) - else: - rev_edges.append(Edge.make_line(p1, p2)) - face = Face(Wire(rev_edges)) - return face - - -def _amount(params: dict[str, Any], *, prefer_sign: str | None = None) -> float: - dist = abs(_f(params["distance_mm"])) - if prefer_sign == "plus": - return dist - if prefer_sign == "minus": - return -dist - return -dist if bool(params.get("reverse")) else dist - - -def _build_nested_circle_profiles(circles: list[dict[str, Any]]) -> None: - """Build circular islands and holes from containment parity. - - A circle contained by one larger circle is a hole; a circle contained by - two larger circles is an island again. This preserves annular profiles - without storing the heavy tessellated sketch regions from the SW export. - """ - ordered = sorted(circles, key=lambda item: _f(item["radius_mm"]), reverse=True) - tolerance = 1e-6 - for index, circle in enumerate(ordered): - center = circle["center"] - radius = _f(circle["radius_mm"]) - containing = 0 - for outer in ordered[:index]: - outer_center = outer["center"] - outer_radius = _f(outer["radius_mm"]) - distance = math.hypot( - _f(center[0]) - _f(outer_center[0]), - _f(center[1]) - _f(outer_center[1]), - ) - if distance + radius <= outer_radius + tolerance: - containing += 1 - mode = Mode.ADD if containing % 2 == 0 else Mode.SUBTRACT - with Locations((_f(center[0]), _f(center[1]))): - Circle(radius, mode=mode) - - -def run_engine_plan( - pack: dict[str, Any], - out_step: Path, - *, - cut_sign: str = "from_params", -) -> dict[str, Any]: - log: list[str] = [] - - compiler_context = pack.get("compiler_context") - if isinstance(compiler_context, dict): - # 回退路径：使用本包 translator（不依赖外部 backend.src） - try: - from .translator import generate_build123d_code, get_part_name - except ImportError: - from translator import generate_build123d_code, get_part_name - - context = dict(compiler_context) - context.setdefault("metadata", {})["part_name"] = str(pack.get("part_id") or out_step.stem) - out_step.parent.mkdir(parents=True, exist_ok=True) - completed = subprocess.run( - [sys.executable, "-c", generate_build123d_code(context)], - cwd=out_step.parent, - capture_output=True, - text=True, - timeout=180, - ) - if completed.returncode != 0: - raise RuntimeError( - f"exact compiler execution failed\nSTDOUT:\n{completed.stdout}\nSTDERR:\n{completed.stderr}" - ) - generated_name = get_part_name({"part_name": context["metadata"]["part_name"]}) - generated = out_step.parent / f"{generated_name}.step" - if generated != out_step and generated.exists(): - generated.replace(out_step) - if not out_step.exists(): - raise RuntimeError(f"exact compiler did not generate {out_step}") - solid = import_step(str(out_step)) - bb = solid.bounding_box() - return { - "out_step": str(out_step), - "volume_mm3": _f(solid.volume), - "bbox_mm": { - "min": [bb.min.X, bb.min.Y, bb.min.Z], - "max": [bb.max.X, bb.max.Y, bb.max.Z], - }, - "engine": "translator_fallback", - } - - with BuildPart() as part: - for step in pack.get("steps") or []: - atomic = step["atomic_id"] - params = step["params"] - sketch = step.get("sketch") - sid = step.get("step_id") - - if atomic == "reference_plane": - # Context features deliberately produce no solid. They remain - # executable plan steps so their dependencies are preserved and - # can be registered by the session-based runtime. - plane = _plane_from_workplane(params.get("plane") or {}) - log.append( - f"{sid}: reference_plane origin={tuple(plane.origin)} normal={tuple(plane.z_dir)}" - ) - - elif atomic == "reference_axis": - axis = _axis_from_params(params.get("axis") or {}) - log.append( - f"{sid}: reference_axis origin={tuple(axis.position)} direction={tuple(axis.direction)}" - ) - - elif atomic == "sphere_add": - radius = _f(params.get("radius_mm") or 0) - center = params.get("center_mm") or [0, 0, 0] - if radius <= 0 or len(center) != 3: - raise ValueError(f"{sid}: sphere_add requires a positive radius_mm and center_mm") - with Locations((_f(center[0]), _f(center[1]), _f(center[2]))): - Sphere(radius, mode=Mode.ADD) - log.append(f"{sid}: sphere_add radius={radius}") - - elif atomic in ("extrude_add_blind", "extrude_add_two_sided", "extrude_cut_blind"): - if sketch is None: - raise ValueError(f"{sid}: missing sketch") - plane = _plane_from_workplane(sketch.get("workplane") or {}) - mode = Mode.SUBTRACT if "cut" in atomic else Mode.ADD - edges = sketch.get("contour_edges_mm") or [] - regions = sketch.get("contour_regions_mm") or [] - sign = cut_sign if "cut" in atomic else "from_params" - - circles = [ - e - for e in (sketch.get("entities") or []) - if e.get("type") == "circle" and not e.get("construction") - ] - lines = [ - e - for e in (sketch.get("entities") or []) - if e.get("type") == "line" and not e.get("construction") - ] - - # 多区域轮廓（外环 + 孔）：由 shape generator 展开 - if regions: - faces = [] - normal = (sketch.get("workplane") or {}).get("normal") - for reg in regions: - outer_edges = reg.get("outer") or [] - if len(outer_edges) < 2: - continue - face = _face_from_contour_edges(outer_edges, desired_normal=normal) - for hole_edges in reg.get("holes") or []: - if len(hole_edges) < 2: - continue - hole = _face_from_contour_edges(hole_edges, desired_normal=normal) - face = face.cut(hole) - faces.append(face) - if not faces: - raise ValueError(f"{sid}: contour_regions_mm produced no faces") - if atomic == "extrude_add_two_sided": - d = abs(_f(params["distance_mm"])) - for face in faces: - extrude(to_extrude=face, amount=d, both=True, mode=Mode.ADD) - else: - amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) - for face in faces: - extrude(to_extrude=face, amount=amt, mode=mode) - log.append(f"{sid}: {atomic} regions={len(faces)}") - continue - - # 切除：草图常含面外框线+圆孔；优先圆孔，避免误用外框整面切除 - prefer_circles = bool(circles) and atomic.startswith("extrude_cut") - - if prefer_circles: - with BuildSketch(plane): - for e in circles: - with Locations((_f(e["center"][0]), _f(e["center"][1]))): - Circle(_f(e["radius_mm"])) - if atomic == "extrude_add_two_sided": - d = abs(_f(params["distance_mm"])) - extrude(amount=d, both=True, mode=Mode.ADD) - else: - amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) - extrude(amount=amt, mode=mode) - log.append(f"{sid}: {atomic} circle-only n={len(circles)}") - elif len(edges) >= 2: - face = _face_from_contour_edges(edges, desired_normal=sketch.get("workplane", {}).get("normal")) - if atomic == "extrude_add_two_sided": - d = abs(_f(params["distance_mm"])) - extrude(to_extrude=face, amount=d, both=True, mode=Mode.ADD) - log.append(f"{sid}: extrude_two_sided both={d} contour") - else: - amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) - extrude(to_extrude=face, amount=amt, mode=mode) - log.append(f"{sid}: {atomic} amount={amt} contour") - elif circles and not lines: - # 纯圆轮廓：用包含层级区分实体、内孔和孔中岛。 - with BuildSketch(plane): - _build_nested_circle_profiles(circles) - if atomic == "extrude_add_two_sided": - d = abs(_f(params["distance_mm"])) - extrude(amount=d, both=True, mode=Mode.ADD) - else: - amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) - extrude(amount=amt, mode=mode) - log.append(f"{sid}: {atomic} circle-only n={len(circles)}") - else: - with BuildSketch(plane): - pts = _ordered_profile_points(sketch) - poly = pts[:-1] if len(pts) >= 2 and pts[0] == pts[-1] else pts - Polygon(*poly) - for e in circles: - with Locations((_f(e["center"][0]), _f(e["center"][1]))): - Circle(_f(e["radius_mm"]), mode=Mode.SUBTRACT) - if atomic == "extrude_add_two_sided": - d = abs(_f(params["distance_mm"])) - extrude(amount=d, both=True, mode=Mode.ADD) - log.append(f"{sid}: extrude_two_sided both={d} poly") - else: - amt = _amount(params, prefer_sign=None if sign == "from_params" else sign) - extrude(amount=amt, mode=mode) - log.append(f"{sid}: {atomic} amount={amt} poly") - - elif atomic in ("revolve_add", "revolve_cut"): - if sketch is None: - raise ValueError(f"{sid}: missing sketch") - plane = _plane_from_workplane(sketch.get("workplane") or {}) - axis = _axis_from_params(params.get("axis") or {}) - angle = _f(params.get("angle_deg") or 360) - mode = Mode.SUBTRACT if atomic == "revolve_cut" else Mode.ADD - edges = sketch.get("contour_edges_mm") or [] - if len(edges) >= 2: - face = _face_from_contour_edges(edges, desired_normal=sketch.get("workplane", {}).get("normal")) - revolve(profiles=face, axis=axis, revolution_arc=angle, mode=mode) - else: - with BuildSketch(plane): - pts = _ordered_profile_points(sketch) - poly = pts[:-1] if len(pts) >= 2 and pts[0] == pts[-1] else pts - Polygon(*poly) - revolve(axis=axis, revolution_arc=angle, mode=mode) - log.append(f"{sid}: {atomic} angle={angle}") - - elif atomic in ("hole_blind", "hole_countersink", "hole_counterbore"): - dia = _f(params.get("diameter_mm") or 0) - depth = _f(params.get("depth_mm") or 0) - positions = params.get("positions") or [] - if sketch is not None: - plane = _plane_from_workplane(sketch.get("workplane") or {}) - else: - plane = Plane.XY - host_face = params.get("host_face") or {} - frame = host_face.get("frame") or {} - frame_origin = Vector(*(frame.get("origin_mm") or plane.origin.to_tuple())) - frame_x = Vector(*(frame.get("x_dir") or plane.x_dir.to_tuple())) - frame_y = Vector(*(frame.get("y_dir") or plane.y_dir.to_tuple())) - normal = plane.z_dir.normalized() - bb = part.part.bounding_box() - part_center = Vector( - (bb.min.X + bb.max.X) / 2, - (bb.min.Y + bb.max.Y) / 2, - (bb.min.Z + bb.max.Z) / 2, - ) - inward = normal if (part_center - frame_origin).dot(normal) >= 0 else -normal - for pos in positions: - mm = pos.get("mm") or [0, 0, 0] - start = frame_origin + frame_x * _f(mm[0]) + frame_y * _f(mm[1]) - cs_dia = _f(params.get("countersink_diameter_mm") or 0) - cs_angle = _f(params.get("countersink_angle_rad") or 0) - cb_dia = _f(params.get("counterbore_diameter_mm") or 0) - cb_depth = _f(params.get("counterbore_depth_mm") or 0) - cs_depth = ( - ((cs_dia - dia) / 2) / math.tan(cs_angle / 2) - if cs_dia > dia and cs_angle > 0 - else 0 - ) - base_offset = cs_depth + (cb_depth if cb_dia > dia else 0) - main_depth = max(0.001, abs(depth) - base_offset) - main_place = Location(Plane(origin=start + inward * base_offset, z_dir=inward)) - tools = [ - Cylinder( - radius=dia / 2, - height=main_depth, - align=(Align.CENTER, Align.CENTER, Align.MIN), - mode=Mode.PRIVATE, - ).move(main_place) - ] - if cb_dia > dia and cb_depth > 0: - tools.append( - Cylinder( - radius=cb_dia / 2, - height=cb_depth, - align=(Align.CENTER, Align.CENTER, Align.MIN), - mode=Mode.PRIVATE, - ).move(Location(Plane(origin=start, z_dir=inward))) - ) - if cs_depth > 0: - tools.append( - Cone( - bottom_radius=cs_dia / 2, - top_radius=dia / 2, - height=cs_depth, - align=(Align.CENTER, Align.CENTER, Align.MIN), - mode=Mode.PRIVATE, - ).move(Location(Plane(origin=start, z_dir=inward))) - ) - drill_angle = _f(params.get("drill_angle_rad") or 0) - if drill_angle > 0: - tip_depth = (dia / 2) / math.tan(drill_angle / 2) - tools.append( - Cone( - bottom_radius=dia / 2, - top_radius=0, - height=tip_depth, - align=(Align.CENTER, Align.CENTER, Align.MIN), - mode=Mode.PRIVATE, - ).move( - Location( - Plane(origin=start + inward * abs(depth), z_dir=inward) - ) - ) - ) - for tool in tools: - part.part = part.part.cut(tool) - log.append(f"{sid}: {atomic} npos={len(positions)}") - - else: - raise ValueError(f"unsupported atomic_id: {atomic}") - - solid = part.part - - out_step.parent.mkdir(parents=True, exist_ok=True) - export_step(solid, str(out_step)) - bb = solid.bounding_box() - return { - "out_step": str(out_step), - "volume_mm3": _f(solid.volume), - "bbox_mm": { - "min": [bb.min.X, bb.min.Y, bb.min.Z], - "max": [bb.max.X, bb.max.Y, bb.max.Z], - }, - "log": log, - "cut_sign": cut_sign, - } - - -def main() -> None: - import argparse - - ap = argparse.ArgumentParser() - ap.add_argument("--pack", type=Path, required=True) - ap.add_argument("--out-step", type=Path, required=True) - ap.add_argument("--report", type=Path, default=None) - ap.add_argument("--cut-sign", default="from_params", choices=["from_params", "plus", "minus"]) - args = ap.parse_args() - info = run_engine_plan(_load(args.pack), args.out_step, cut_sign=args.cut_sign) - if args.report: - args.report.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8") - print( - json.dumps( - {k: info[k] for k in ("out_step", "volume_mm3", "bbox_mm", "cut_sign", "engine") if k in info}, - ensure_ascii=False, - indent=2, - ) - ) - for line in info.get("log") or []: - print(line) - - -if __name__ == "__main__": - main() +__all__ = ["SUPPORTED_ATOMIC_IDS", "main", "run_engine_plan"] diff --git a/backend/engine/cdsl_engine/rebuild.py b/backend/engine/cdsl_engine/rebuild.py index 5d3815a1..78101205 100644 --- a/backend/engine/cdsl_engine/rebuild.py +++ b/backend/engine/cdsl_engine/rebuild.py @@ -1,7 +1,7 @@ """ CDSL → STEP 重建管道 ==================== -优先: CDSL → capability planner → session runtime → STEP (engine=cdsl_only) +优先: CDSL → capability planner → session runtime → STEP (engine=cdsl_only) 回退: CDSL + compiler_context → translator """ @@ -15,41 +15,43 @@ from pathlib import Path from typing import Any try: - from cdsl_importer.legacy_profile_adapter import lower_legacy_profiles - from .sketch_solver import SHAPE_GENERATORS, resolve_all_sketches + from cdsl_importer.legacy_profile_adapter import lower_legacy_profiles + from .sketch_solver import SHAPE_GENERATORS, resolve_all_sketches from .llm_compiler import compile_cdsl from .llm_engine import run_engine_plan from .translator import generate_build123d_code, normalize_to_ir + from .legacy.exact_rebuild import _apply_geometric_compensations, _run_exact, _run_parameterized except ImportError: # 允许直接 python rebuild.py - from cdsl_importer.legacy_profile_adapter import lower_legacy_profiles - from sketch_solver import SHAPE_GENERATORS, resolve_all_sketches + from cdsl_importer.legacy_profile_adapter import lower_legacy_profiles + from sketch_solver import SHAPE_GENERATORS, resolve_all_sketches from llm_compiler import compile_cdsl from llm_engine import run_engine_plan from translator import generate_build123d_code, normalize_to_ir + from legacy.exact_rebuild import _apply_geometric_compensations, _run_exact, _run_parameterized -def run_rebuild(cdsl: dict[str, Any], out_step: Path, ctx_file: Path | None = None, gold_step: Path | None = None, - force_exact: bool = False) -> dict[str, Any]: +def run_rebuild(cdsl: dict[str, Any], out_step: Path, ctx_file: Path | None = None, gold_step: Path | None = None, + force_exact: bool = False) -> dict[str, Any]: """主重建入口。 优先：纯 CDSL 参数化路径（sketch_solver → llm_compiler → llm_engine），不依赖 compiler_context。 回退：CDSL + compiler_context 的 translator 路径。 """ - # Compatibility entry point only: new CDSL-only runtime calls do not use - # macro profiles and therefore never invoke this adapter. - cdsl = lower_legacy_profiles(cdsl) - sketches = cdsl.get("geometry", {}).get("sketches", []) - # Sketchless parameterized features (for example sphere_add) are fully - # executable by the CDSL-only runtime. ``all([])`` deliberately keeps - # that path available rather than forcing an unavailable legacy fallback. - all_drawable = all(_sketch_is_cdsl_drawable(s) for s in sketches) + # Compatibility entry point only: new CDSL-only runtime calls do not use + # macro profiles and therefore never invoke this adapter. + cdsl = lower_legacy_profiles(cdsl) + sketches = cdsl.get("geometry", {}).get("sketches", []) + # Sketchless parameterized features (for example sphere_add) are fully + # executable by the CDSL-only runtime. ``all([])`` deliberately keeps + # that path available rather than forcing an unavailable legacy fallback. + all_drawable = all(_sketch_is_cdsl_drawable(s) for s in sketches) - cdsl_only_error: Exception | None = None - if all_drawable and not force_exact: - try: - return run_cdsl_only(cdsl, out_step, gold_step=gold_step) - except Exception as e: - cdsl_only_error = e + cdsl_only_error: Exception | None = None + if all_drawable and not force_exact: + try: + return run_cdsl_only(cdsl, out_step, gold_step=gold_step) + except Exception as e: + cdsl_only_error = e # 加载 compiler_context（后备路径） ctx = None @@ -71,13 +73,13 @@ def run_rebuild(cdsl: dict[str, Any], out_step: Path, ctx_file: Path | None = No "references": ir.get("references", []), "validation_hints": ir.get("validation_hints", {}), } - if ctx is None and not (cdsl.get("compiler_context")): - if cdsl_only_error is not None: - raise RuntimeError(f"CDSL-only rebuild failed: {cdsl_only_error}") from cdsl_only_error - raise RuntimeError( - "CDSL-only rebuild is unavailable: every sketch must use a supported " - "self-contained profile." - ) + if ctx is None and not (cdsl.get("compiler_context")): + if cdsl_only_error is not None: + raise RuntimeError(f"CDSL-only rebuild failed: {cdsl_only_error}") from cdsl_only_error + raise RuntimeError( + "CDSL-only rebuild is unavailable: every sketch must use a supported " + "self-contained profile." + ) if ctx is not None: cdsl["compiler_context"] = ctx @@ -112,24 +114,24 @@ def _sketch_is_cdsl_drawable(sketch: dict[str, Any]) -> bool: return ptype in SHAPE_GENERATORS -def run_cdsl_only(cdsl: dict[str, Any], out_step: Path, gold_step: Path | None = None) -> dict[str, Any]: - """Pure semantic CDSL path with no compiler_context fallback.""" - t0 = time.time() - try: - from .runtime import rebuild_cdsl - except ImportError: - from runtime import rebuild_cdsl - slim = {key: value for key, value in cdsl.items() if key != "compiler_context"} - result = rebuild_cdsl(slim, out_step, strict=True) - result["engine"] = "cdsl_only" - result["elapsed_s"] = round(time.time() - t0, 1) - result.setdefault("log", []).append("cdsl_only: capability planner + session runtime (no compiler_context)") +def run_cdsl_only(cdsl: dict[str, Any], out_step: Path, gold_step: Path | None = None) -> dict[str, Any]: + """Pure semantic CDSL path with no compiler_context fallback.""" + t0 = time.time() + try: + from .runtime import rebuild_cdsl + except ImportError: + from runtime import rebuild_cdsl + slim = {key: value for key, value in cdsl.items() if key != "compiler_context"} + result = rebuild_cdsl(slim, out_step, strict=True) + result["engine"] = "cdsl_only" + result["elapsed_s"] = round(time.time() - t0, 1) + result.setdefault("log", []).append("cdsl_only: capability planner + session runtime (no compiler_context)") if gold_step and gold_step.exists(): result["gold_step"] = str(gold_step) return result -def compile_cdsl_to_pack(cdsl: dict[str, Any]) -> dict[str, Any]: +def compile_cdsl_to_pack(cdsl: dict[str, Any]) -> dict[str, Any]: pack = compile_cdsl({k: v for k, v in cdsl.items() if k != "compiler_context"}) pack.pop("compiler_context", None) return pack @@ -139,216 +141,6 @@ def run_engine(pack: dict[str, Any], out_step: Path) -> dict[str, Any]: return run_engine_plan(pack, out_step) -def _run_parameterized(cdsl: dict[str, Any], out_step: Path, gold_step: Path | None = None) -> dict[str, Any]: - """参数化路径: CDSL语义结构 + compiler_context精确数据 → translator生成代码 → 执行 - - 采用双层IR架构： - Learning IR (CDSL) 提供参数化形状、特征结构 - Execution IR (compiler_context) 提供精确坐标 - translator 提供经过充分测试的代码生成 - """ - - import subprocess - import tempfile, os - - t0 = time.time() - - # 1. 获取 compiler_context (Execution IR: 精确坐标) - compiler_context = cdsl.get("compiler_context") or {} - if not compiler_context: - # 从外部文件加载 - ctx_file = out_step.parent / "{}.compiler_context.json".format(cdsl.get("part_id", "")) - if ctx_file.exists(): - import json as _json - with open(ctx_file, "r", encoding="utf-8") as _f: - compiler_context = _json.load(_f) - if not compiler_context: - raise RuntimeError("CDSL缺少 compiler_context，无法重建") - - part_name = str(cdsl.get("part_id") or out_step.stem) - context = dict(compiler_context) - context.setdefault("metadata", {})["part_name"] = part_name - - # 2. 将 compiler_context 的精确实体注入 CDSL 草图 (供 sketch_solver 使用) - # 015133: CDSL (Learning IR) 不含坐标，坐标来自 Execution IR - ctx_sketches_map = {s["id"]: s for s in context.get("sketches", [])} - cdsl_sketches = cdsl.get("geometry", {}).get("sketches", []) - for sk in cdsl_sketches: - ctx_sk = ctx_sketches_map.get(sk["id"]) - if ctx_sk: - # 注入 entities/contour 供 polygon/complex_arc_shape 生成器使用 - if not sk.get("entities"): - sk["entities"] = ctx_sk.get("entities", []) - if not sk.get("contour_edges_mm"): - sk["contour_edges_mm"] = ctx_sk.get("contour_edges_mm", []) - - # 3. 解析 CDSL 的参数化草图 (现在有 entities 可用) - cdsl_resolved = resolve_all_sketches(cdsl) - - # 4. 将 CDSL 解析后的 profile/profile_from 注入 compiler_context - # translator 使用 compiler_context 的精确 entities + CDSL 的 profile 分类 - cdsl_resolved_map = {s["id"]: s for s in cdsl_resolved.get("geometry", {}).get("sketches", [])} - ctx_sketches = list(context.get("sketches", [])) - updated_count = 0 - for i, ctx_sk in enumerate(ctx_sketches): - sk_id = ctx_sk.get("id", "") - cdsl_sk = cdsl_resolved_map.get(sk_id) - if cdsl_sk and cdsl_sk.get("profile"): - ctx_sketches[i] = {**ctx_sk, "profile": cdsl_sk["profile"]} - updated_count += 1 - if cdsl_sk and cdsl_sk.get("profile_from"): - ctx_sketches[i] = {**ctx_sk, "profile_from": cdsl_sk["profile_from"]} - updated_count += 1 - context["sketches"] = ctx_sketches - - # 4. 使用 compiler_context 的原始 operations（保持 translator 兼容性） - - # 5. 读取 gold volume - gold_volume_mm3 = None - if gold_step and gold_step.exists(): - try: - from build123d import import_step - gold_solid = import_step(str(gold_step)) - gold_volume_mm3 = float(gold_solid.volume) - except Exception: - pass - - # 6. 用 translator 生成并执行 - code = generate_build123d_code(context, gold_volume_mm3=gold_volume_mm3) - - # 6b. 应用几何补偿 (SW导出缺失的特征) - part_id = str(cdsl.get("part_id") or "") - code = _apply_geometric_compensations(code, part_id) - - out_step.parent.mkdir(parents=True, exist_ok=True) - - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False, encoding="utf-8") as tf: - tf.write(code) - script_path = tf.name - - try: - r = subprocess.run( - ["python", script_path], - capture_output=True, text=True, encoding="utf-8", timeout=120, - env={**os.environ, "PYTHONIOENCODING": "utf-8"}, - ) - if r.returncode != 0: - raise RuntimeError(f"Build script failed:\n{r.stderr}") - finally: - try: - os.unlink(script_path) - except Exception: - pass - - # 7. 读取重建结果 - out_step.parent.mkdir(parents=True, exist_ok=True) - built_step = Path(part_name + ".step") - if not built_step.exists(): - built_step = Path.cwd() / (part_name + ".step") - if built_step.exists(): - import shutil - shutil.copy2(str(built_step), str(out_step)) - built_step.unlink() - else: - raise RuntimeError(f"No STEP output found: {part_name}.step") - - from build123d import import_step - rebuilt = import_step(str(out_step)) - bbox = rebuilt.bounding_box() - bbox_mm = { - "min": [bbox.min.X, bbox.min.Y, bbox.min.Z], - "max": [bbox.max.X, bbox.max.Y, bbox.max.Z], - } - - elapsed = time.time() - t0 - return { - "out_step": str(out_step), - "volume_mm3": float(rebuilt.volume), - "bbox_mm": bbox_mm, - "log": [f"param: CDSL-informed translator rebuild, {updated_count} sketches updated from CDSL"], - "engine": "parameterized", - "elapsed_s": round(elapsed, 1), - } - - -def _run_exact(cdsl: dict[str, Any], out_step: Path, gold_step: Path | None = None) -> dict[str, Any]: - """精确路径: generate_build123d_code (后备)""" - - import subprocess - - compiler_context = cdsl.get("compiler_context") or {} - part_name = str(cdsl.get("part_id") or out_step.stem) - context = dict(compiler_context) - context.setdefault("metadata", {})["part_name"] = part_name - - # Read gold volume if available, for chamfer/candidate scoring - gold_volume_mm3 = None - if gold_step and gold_step.exists(): - try: - from build123d import import_step - gold_solid = import_step(str(gold_step)) - gold_volume_mm3 = float(gold_solid.volume) - except Exception: - pass - - out_step.parent.mkdir(parents=True, exist_ok=True) - - # Apply geometric compensations FIRST (may return full replacement code) - part_id = str(cdsl.get("part_id") or "") - compensation_code = _apply_geometric_compensations("", part_id) - - if compensation_code and "build123d" in compensation_code and "__main__" in compensation_code: - # 完整替换代码 (跳过generate_build123d_code) - code = compensation_code - else: - code = generate_build123d_code(context, gold_volume_mm3=gold_volume_mm3) - code = _apply_geometric_compensations(code, part_id) - - t0 = time.time() - script_path = out_step.parent / "_tmp" / f"build_{part_name}_{int(time.time())}.py" - script_path.parent.mkdir(exist_ok=True) - script_path.write_text(code, encoding="utf-8") - - completed = subprocess.run( - [sys.executable, str(script_path)], - cwd=out_step.parent, - capture_output=True, - text=True, - timeout=600, - ) - - if completed.returncode != 0: - raise RuntimeError( - f"Exact compiler FAILED (rc={completed.returncode})\n" - f"STDOUT:\n{completed.stdout[-2000:]}\n" - f"STDERR:\n{completed.stderr[-3000:]}" - ) - # Print any warnings from safe_subtract - for line in completed.stdout.split('\n'): - if 'SUBTRACT' in line or 'UNION' in line: - print(f" {line.strip()}") - - from build123d import import_step - # 生成的 build 脚本将 STEP 写到 CWD 下的 "{part_name}.step" - # 移到 out_step 位置以供后续对比 - actual_step = out_step.parent / f"{part_name}.step" - if actual_step.exists(): - import shutil - shutil.copy2(str(actual_step), str(out_step)) - solid = import_step(str(out_step)) - bb = solid.bounding_box() - elapsed = time.time() - t0 - - return { - "out_step": str(out_step), - "volume_mm3": float(solid.volume), - "bbox_mm": {"min": [bb.min.X, bb.min.Y, bb.min.Z], - "max": [bb.max.X, bb.max.Y, bb.max.Z]}, - "engine": "exact", - "elapsed_s": round(elapsed, 1), - } - - def compare_with_gold(gold_step: Path, rebuilt_step: Path) -> dict[str, Any]: from build123d import import_step import math, random, time @@ -493,27 +285,6 @@ def _surface_deviation(gold, rebuilt, n_points: int = 500) -> dict[str, Any]: # (下面的不再需要，新逻辑已在_surface_deviation中实现) -# ═══════════════════════════════════════════════════════════════ -# Geometric compensations（项目特例；拷贝到其他项目时可删） -# ═══════════════════════════════════════════════════════════════ - -def _apply_geometric_compensations(code: str, part_id: str) -> str: - """为SW导出中缺失的特征添加几何补偿切操作""" - if part_id == "113246": - if "export_step(result, " not in code: - return code - comp = ( - " # === COMPENSATION: 侧槽 (SW缺失特征) ===\n" - " with BuildSketch(Plane(origin=(-70.0, -13.0, 10.0), " - "x_dir=(0.0, 1.0, 0.0), z_dir=(1.0, 0.0, 0.0))) as comp_sk:\n" - " Rectangle(10.0, 3.0, align=(Align.MIN, Align.MIN))\n" - " comp_cutter = extrude(comp_sk.sketch, amount=10.0)\n" - " result = safe_subtract(result, comp_cutter)\n" - ) - code = code.replace("export_step(result, ", comp + " export_step(result, ") - return code - - # =========================================================================== # CLI（便携：显式路径，无项目目录假设） # ===========================================================================