likang/cdsl-cad
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(cdsl_engine): freeze legacy engine paths under legacy/

Browse Source 
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.
This commit is contained in:
ganjihong
2026-09-09 14:02:53 +08:00
parent ad88d92ab9
commit 95cae203f4
7 changed files with 1167 additions and 1106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/engine/cdsl_engine/legacy/__init__.py
+13
View File
@@ -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
backend/engine/cdsl_engine/legacy/exact_rebuild.py
+253
View File
@@ -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
backend/engine/cdsl_engine/legacy/llm_compiler.py
+287
View File
@@ -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()
backend/engine/cdsl_engine/legacy/llm_engine.py
+555
View File
@@ -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()
backend/engine/cdsl_engine/llm_compiler.py
+8 -283
View File
@@ -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"]
backend/engine/cdsl_engine/llm_engine.py
+8 -551
View File
@@ -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"]
backend/engine/cdsl_engine/rebuild.py
+43 -272
View File
@@ -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（便携：显式路径，无项目目录假设）
# ===========================================================================