287 lines
12 KiB
Python
287 lines
12 KiB
Python
"""通用编译器：瘦 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"],
|
|
}
|
|
|
|
|
|
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()
|