feat: validate CAD hardware mounting geometry
This commit is contained in:
chenpei
parent
272a9aa29f
commit
68766258b9
@ -69,6 +69,15 @@
|
||||
- 局部修改按事务应用到候选副本并重新编译；编译问题数不能高于上一已接受版本。回归候选被完整丢弃，下一轮收到问题数和代码分布，而不是在损坏状态上继续修。
|
||||
- sideboard7 隔离证据显示该机制使编译问题从 57 项总体降到 15 项，并成功消费一次直接数组响应；但剩余总成边界和穿插仍是阻断项。这里的“收敛”只描述验证器问题计数，不等于图片理解正确或可制造。
|
||||
|
||||
## v96 五金安装约束、端到端评分与可恢复探针（2026-09-07）
|
||||
|
||||
- 五金实例不再依赖“碰到某块板就算安装”。`mountings`显式绑定宿主实例、宿主板面、表装/埋入/贯穿类型、嵌入深度和证据；同一铰链可绑定门板与侧板两个宿主。求解器按板厚轴限制可安装板面，并验证法向距离、切向重叠和深度。
|
||||
- 结构总成边界从非五金结构成员计算；已验证安装的五金可形成实际外形突出，但不能反过来把结构总成边界放大。合法埋入只豁免对应五金—宿主这一对碰撞，未声明的碰撞和错误位置仍阻断。
|
||||
- 编译错误返回精确可采用的安装值和最小平移候选，不让模型自由猜板面。若移动实例是其总成唯一成员，候选同时包含紧包围总成边界修改，两个补丁按原子事务应用。
|
||||
- 隔离修复不再只优化编译问题数。每轮以编译、制造、CAD内核和视觉问题总数为唯一接受分数，候选必须严格下降；`--resume-report`会选历史最佳检查点并用当前编译器刷新反馈后续跑，保留原图资格与不读公司BOM的审计链。
|
||||
- 续跑选择器排除任何带`patch_error`的拒绝轮次；编译阶段只有在当前问题数作为下界已经不可能优于上一端到端总分时才提前回滚，避免用旧编译问题数误杀可能显著减少B-rep冲突的候选。专项60项、完整196项回归通过。
|
||||
- 当前仍缺少安装特征实体化：B-rep应依据安装模板在宿主生成孔、杯槽、沉孔或开口，同时把五金拆成可见体和嵌入体，不能永久靠允许碰撞对掩盖。下一阶段还要增加“参考尺寸包络/实际占用包络”语义，识别覆盖门、嵌入背板与净深尺寸，先做尺寸链可行性裁决再进入AI修复。
|
||||
|
||||
## 同行公开资料的可用结论
|
||||
|
||||
Zoo公开说明其代理会操作可编辑的参数化模型，执行几何代码，读取质量/体积等模型信息，并用多视图快照反复检查。我们借鉴“结构表示＋几何执行反馈＋视觉反馈”机制，不能据其宣传推断多张家具图片已经可以稳定得到生产级完整BOM，也不能据此承诺成本误差5%。
|
||||
|
||||
@ -233,3 +233,17 @@
|
||||
- 服务器确认生产 Chat API 活动任务、等待任务、持久队列均为 0 后，仅在 `/tmp/dalongxia-cad-v85-eI8HeF/probe-sideboard7` 运行隔离原图试验；未修改生产文件、未重启服务、未读取公司 BOM。
|
||||
- 同一组四张原图仍正确固定 1200×400×800mm 和 top=1、bottom=1、side=2、door=2、shelf=1、leg=4。五轮编译问题 57→52→35→30→15，CAD B-rep 冲突 11→16→9→12→3；最终仍有门/五金总成边界、一个门把手断联、背板与层板以及左侧板与铰链穿插，状态仍为 `graph_and_cad_kernel_not_passed`。
|
||||
- 该结果证明修复链可消费反馈并总体收敛，但没有通过制造门禁、没有生成可交付报价，也没有完成不同公司产品盲测或冻结后总成本误差≤5%。
|
||||
|
||||
## 2026-09-07 v91-v96 五金安装语义与端到端防回退（候选，未上线）
|
||||
|
||||
- v90仍把旋钮、铰链等五金当作普通长方体。它们只要伸出总成边界就会被判越界，移进板内又会被判穿插；编译器也无法区分“合法安装嵌入”和“误碰撞”。本轮为所有`functional_role=hardware`实例增加显式`mountings`，每条安装声明宿主、安装方式（表装/埋入/贯穿）、合法板面、嵌入深度和原图证据。没有安装关系、宿主不存在、选了板件窄边、切入深度不符或切向没有正面积重叠时均阻断。
|
||||
- 五金安装通过后才进入装配连通图；只有已验证的表装五金允许伸出结构件紧包围盒，只有已验证的埋入/贯穿安装允许与对应宿主发生体积交叠。其他零件碰撞仍按原规则阻断，不以“五金”身份整体豁免。
|
||||
- v91隔离试验暴露“提示模型自己猜安装面”不可行：首轮10条安装关系仅2条有效，之后编译问题50→39后反复回退。该提示路线判定失败并停止。
|
||||
- 编译反馈现按已求解几何给出`compatible_mounting_values`，只列宿主真实板面及对应安装方式/深度；实体冲突反馈给出按最小位移排序的`compatible_separation_candidates`。单成员总成的位移候选会同时给出总成`bounds_mm`补丁，避免只挪零件后制造新的边界错误。
|
||||
- v92隔离试验使10条五金安装全部有效，编译问题48→29→7；但第4轮虽然编译问题继续降到5，B-rep冲突却从11增到15，证明只按编译问题数选修复会接受总体更差的CAD。该“编译单指标单调”路线判定不足。
|
||||
- 修复接受条件改为编译、制造、B-rep与视觉阻塞项的端到端总数必须严格下降。新增`--resume-report`从历史报告选择总阻塞最少且轮次最新的检查点，并用当前编译器重新生成反馈，避免重新读图、继承旧反馈或从错误轮次继续。
|
||||
- v93因早期回归提示错误引用候选问题而停止；v94因续跑检查点沿用旧编译反馈而停止；v95因位移候选先按碰撞数排序，首选出现578mm不合理侧板平移而停止。三次均为隔离systemd单元，修正相应通用机制后才继续，没有覆盖生产文件或用户任务。
|
||||
- v96从v92第3轮检查点恢复，不重新读取原图且保持`company_bom_used_in_inference=false`。端到端阻塞18→12→10；后续模型试图一次改货架、4个铰链和2个旋钮，编译问题由3反弹到20，被事务门拒绝并保留第5轮图谱。报告位于`/tmp/dalongxia-cad-v85-eI8HeF/probe-sideboard13/sideboard/report.json`，最终仍未通过。
|
||||
- v96还暴露两个未解决的真实边界：一是表装/埋入五金在结构编译器合法后，CAD实体仍是完整盒体，B-rep层尚未生成宿主安装孔/槽；二是样例深度链存在语义歧义，门厚20mm、层板深380mm、背板厚9mm与400mm总深无法在当前“所有实体必须包含于同一外包络”的假设下同时无穿插放置。必须先增加参考外形与实际外形/覆盖门语义或尺寸可行性裁决，不能靠移动板件伪造通过。
|
||||
- 最终候选代码专项60项通过（0.063秒），完整回归196项通过（236.111秒）；续跑会跳过所有带`patch_error`的拒绝轮次，编译早退仅在“编译问题下界已不可能优于已接受端到端总分”时触发。测试通过只证明防回退和既有功能未回退，不改变本次真实样例仍失败的结论。
|
||||
- 本轮保持候选状态：五金挂载和修复事务更可靠，但生产级连接特征、真实多产品盲测以及冻结后成本误差≤5%均未完成，禁止标记成功或部署。
|
||||
|
||||
@ -7,6 +7,7 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import copy
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
@ -103,6 +104,52 @@ def structured_kernel_feedback(error: Exception | str) -> list[dict]:
|
||||
}]
|
||||
|
||||
|
||||
def round_blocker_count(row: dict) -> int:
|
||||
"""Count end-to-end blockers for monotonic repair acceptance.
|
||||
|
||||
Compiler issues alone are insufficient: an edit can remove a loose-bounds
|
||||
warning while creating more exact B-rep clashes. A candidate therefore
|
||||
has to improve the complete compile/manufacturing/kernel/visual chain.
|
||||
"""
|
||||
compiled = row.get("compiled") or {}
|
||||
total = len(compiled.get("issues") or [])
|
||||
if compiled.get("ok"):
|
||||
total += len((row.get("manufacturing_plan") or {}).get("issues") or [])
|
||||
kernel_feedback = row.get("cad_kernel_feedback") or []
|
||||
total += len(kernel_feedback)
|
||||
if row.get("cad_kernel_error") and not kernel_feedback:
|
||||
total += 1
|
||||
review = row.get("visual_review")
|
||||
if isinstance(review, dict):
|
||||
review_count = len(review.get("issues") or []) + len(review.get("uncertain") or [])
|
||||
total += review_count
|
||||
if review.get("ok") is not True and review_count == 0:
|
||||
total += 1
|
||||
return total
|
||||
|
||||
|
||||
def compile_lower_bound_can_improve(candidate_compile_issue_count: int, accepted_total_blocker_count) -> bool:
|
||||
"""Whether this candidate can still beat the accepted end-to-end score."""
|
||||
if accepted_total_blocker_count is None:
|
||||
return True
|
||||
return candidate_compile_issue_count < accepted_total_blocker_count
|
||||
|
||||
|
||||
def select_resume_checkpoint(rows: list[dict]) -> dict:
|
||||
"""Select only an accepted, internally consistent audited checkpoint."""
|
||||
candidates = [
|
||||
row for row in rows
|
||||
if (
|
||||
isinstance(row.get("graph"), dict)
|
||||
and isinstance(row.get("compiled"), dict)
|
||||
and not row.get("patch_error")
|
||||
)
|
||||
]
|
||||
if not candidates:
|
||||
raise RuntimeError("resume report has no audited compiled graph checkpoint")
|
||||
return min(candidates, key=lambda row: (round_blocker_count(row), -int(row.get("round") or 0)))
|
||||
|
||||
|
||||
def structured_patch_feedback(graph: dict, patches, error: Exception | str) -> list[dict]:
|
||||
"""Explain a rejected JSON Patch without silently changing its semantics."""
|
||||
issues = []
|
||||
@ -164,7 +211,7 @@ GRAPH_PROMPT = """根据全部原始图片构建可编辑、可验证的家具
|
||||
"envelope_mm":["width","depth","height"],
|
||||
"assemblies":[{"id":"lower_module","role":"carcass","bounds_mm":{"x":[0,"width"],"y":[0,"depth"],"z":[0,"lower_height"]},"evidence_ids":["e1"]},{"id":"drawer_1","role":"drawer","bounds_mm":{"x":["side_thickness","width-side_thickness"],"y":[0,"drawer_depth"],"z":["drawer_z_min","drawer_z_max"]},"evidence_ids":["e1"]}],
|
||||
"parts":[{"id":"part_id","name":"中文部件名","shape":"box","functional_role":"side","thickness_axis":"x","size_mm":["参数或算式","参数或算式","参数或算式"],"observed_quantity":1,"evidence_ids":["e1"]}],
|
||||
"instances":[{"id":"物理实例ID","part_id":"part_id","assembly_id":"明确所属总成ID","placement":{"x":{"mode":"align","self_anchor":"min","target":"envelope","target_anchor":"min","offset_mm":0},"y":{"mode":"absolute","value_mm":0},"z":{"mode":"align","self_anchor":"max","target":"另一实例ID","target_anchor":"min","offset_mm":0}}}],
|
||||
"instances":[{"id":"板件物理实例ID","part_id":"panel_part_id","assembly_id":"明确所属总成ID","placement":{"x":{"mode":"align","self_anchor":"min","target":"envelope","target_anchor":"min","offset_mm":0},"y":{"mode":"absolute","value_mm":0},"z":{"mode":"align","self_anchor":"max","target":"另一实例ID","target_anchor":"min","offset_mm":0}}},{"id":"五金物理实例ID","part_id":"hardware_part_id","assembly_id":"明确所属总成ID","placement":{"x":{"mode":"absolute","value_mm":0},"y":{"mode":"absolute","value_mm":0},"z":{"mode":"absolute","value_mm":0}},"mountings":[{"host_instance_id":"实际安装板件ID","kind":"surface","face":"y-","embed_depth_mm":0,"evidence_ids":["e1"]}]}],
|
||||
"connections":[],
|
||||
"assembly_connections":[{"id":"drawer_slide","kind":"sliding","a":"drawer_1","b":"carcass","axis":"y","travel_limits_mm":[0,250],"evidence_ids":["e1"]}],
|
||||
"uncertainties":[]}
|
||||
@ -188,6 +235,9 @@ assembly.role只允许carcass/drawer/door/module/frame/hardware_group/other；
|
||||
抽屉的前/后/左/右帮板和底板属于同一drawer总成；柜体板属于carcass，不能把“抽屉”运动性质挂在某一块帮板上。层板、底板和抽屉帮板必须按相邻侧板之间的净空生成，不能按产品总宽横穿侧板；任何实体正体积穿插都必须修正。
|
||||
每个会独立运动的抽屉、平开门或移门必须是独立assembly；单块门板也要建立包含一个实例的总成。不同总成之间用assembly_connections，a/b只能引用assemblies中的稳定ID，禁止引用零件实例ID：fixed必须在最终几何中真实接触；sliding必须有axis与travel_limits_mm；hinged必须有axis与angle_limits_deg。connections中不要生成fixed/support。
|
||||
不同规格独立part；相同规格由多个instances引用，BOM数量只等于真实实例数，不额外产生总成板。
|
||||
每个functional_role=hardware的物理实例必须声明mountings，指向真实安装的非五金宿主实例。把手通常一个宿主，铰链可有门板和柜体两个宿主；不得只靠中文名称表示安装关系。
|
||||
mountings.kind只允许surface/mortised/through；face是宿主朝外安装面x-/x+/y-/y+/z-/z+；embed_depth_mm是实际嵌入宿主的深度，surface必须为0。安装件在该面投影上必须与宿主有正面积重叠，法向必须接触或严格符合嵌入深度。五金可经过验证后外凸于结构总成边界，但不能藉此掩盖把手装到错误门板、悬空或非宣称宿主的穿插。
|
||||
总成bounds_mm只由门板、抽屉板、柜体板等结构实体的紧包围盒确定，不应把把手、铰链的外凸纳入边界。五金与宿主的正体积交只有在mortised/through关系且嵌入深度与解算几何一致时才允许。
|
||||
所有运动连接必须引用真实实例和真实总成，不能用大公差掩盖错位。
|
||||
只画现有图像支持的结构；不要根据常见柜子想象层板数量。未标注内部尺寸先由AI按多图比例估算并标记estimated，生成评审草案；
|
||||
只有原图完全不可见且对制造/成本高度敏感、经过局部复读与CAD反向复核仍不能判定的字段，才进入用户确认候选。
|
||||
@ -266,6 +316,10 @@ def main():
|
||||
parser.add_argument("--cases", default="book,slide")
|
||||
parser.add_argument("--model", default="qwen3.5-plus")
|
||||
parser.add_argument("--rounds", type=int, default=2)
|
||||
parser.add_argument(
|
||||
"--resume-report",
|
||||
help="Resume from the lowest-blocker audited graph in a prior isolated report.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--material-family", default="engineered_panel",
|
||||
choices=("engineered_panel", "solid_wood", "metal", "plastic", "glass", "other"),
|
||||
@ -307,29 +361,58 @@ def main():
|
||||
report = {"case": case, "schema_version": SCHEMA_VERSION, "model": args.model,
|
||||
"material_family": args.material_family, "full_chat_api": False,
|
||||
"company_bom_used_in_inference": False, "rounds": [], "status": "not_validated"}
|
||||
print(json.dumps({"case": case, "phase": "ai_source_evidence_qualification"}), flush=True)
|
||||
source_structure, source_dimensions = {}, {}
|
||||
try:
|
||||
source_structure = engine._read_source_structure(source_assets)
|
||||
except Exception as exc:
|
||||
source_structure = {"status": "machine_reread_failed", "error": f"{type(exc).__name__}: {exc}"}
|
||||
try:
|
||||
source_dimensions = engine._read_source_dimension_labels(
|
||||
source_assets[-1], [source_assets[0], source_assets[1]]
|
||||
resume_payload = None
|
||||
if args.resume_report:
|
||||
resume_path = pathlib.Path(args.resume_report)
|
||||
if resume_path.is_dir():
|
||||
resume_path = resume_path / case / "report.json"
|
||||
resume_payload = json.loads(resume_path.read_text(encoding="utf-8"))
|
||||
if isinstance(resume_payload, list):
|
||||
matches = [row for row in resume_payload if row.get("case") == case]
|
||||
if len(matches) != 1:
|
||||
raise RuntimeError(f"resume report does not uniquely contain case: {case}")
|
||||
resume_payload = matches[0]
|
||||
if (
|
||||
not isinstance(resume_payload, dict)
|
||||
or resume_payload.get("case") != case
|
||||
or resume_payload.get("schema_version") != SCHEMA_VERSION
|
||||
or resume_payload.get("company_bom_used_in_inference") is not False
|
||||
):
|
||||
raise RuntimeError("resume report is not a compatible no-company-BOM isolated checkpoint")
|
||||
report["source_qualification"] = copy.deepcopy(resume_payload.get("source_qualification") or {})
|
||||
print(json.dumps({
|
||||
"case": case, "phase": "resume_audited_isolated_checkpoint",
|
||||
"source_report": str(resume_path),
|
||||
}, ensure_ascii=False), flush=True)
|
||||
else:
|
||||
print(json.dumps({"case": case, "phase": "ai_source_evidence_qualification"}), flush=True)
|
||||
source_structure, source_dimensions = {}, {}
|
||||
try:
|
||||
source_structure = engine._read_source_structure(source_assets)
|
||||
except Exception as exc:
|
||||
source_structure = {"status": "machine_reread_failed", "error": f"{type(exc).__name__}: {exc}"}
|
||||
try:
|
||||
source_dimensions = engine._read_source_dimension_labels(
|
||||
source_assets[-1], [source_assets[0], source_assets[1]]
|
||||
)
|
||||
except Exception as exc:
|
||||
source_dimensions = {"status": "machine_reread_failed", "error": f"{type(exc).__name__}: {exc}"}
|
||||
dimension_ledger = build_dimension_ledger(
|
||||
source_dimensions.get("overall_dimension_labels") or [], source_image_id="image4"
|
||||
)
|
||||
except Exception as exc:
|
||||
source_dimensions = {"status": "machine_reread_failed", "error": f"{type(exc).__name__}: {exc}"}
|
||||
dimension_ledger = build_dimension_ledger(
|
||||
source_dimensions.get("overall_dimension_labels") or [], source_image_id="image4"
|
||||
)
|
||||
topology_contract = build_source_topology_contract(source_structure)
|
||||
report["source_qualification"] = {
|
||||
"structure": source_structure,
|
||||
"dimensions": source_dimensions,
|
||||
"dimension_ledger": dimension_ledger,
|
||||
"topology_contract": topology_contract,
|
||||
"policy": "AI先交叉识别和CAD复核；只有不可见且成本敏感的残余冲突才询问用户",
|
||||
}
|
||||
topology_contract = build_source_topology_contract(source_structure)
|
||||
report["source_qualification"] = {
|
||||
"structure": source_structure,
|
||||
"dimensions": source_dimensions,
|
||||
"dimension_ledger": dimension_ledger,
|
||||
"topology_contract": topology_contract,
|
||||
"policy": "AI先交叉识别和CAD复核；只有不可见且成本敏感的残余冲突才询问用户",
|
||||
}
|
||||
source_qualification = report["source_qualification"]
|
||||
source_structure = source_qualification.get("structure") or {}
|
||||
source_dimensions = source_qualification.get("dimensions") or {}
|
||||
dimension_ledger = source_qualification.get("dimension_ledger") or {}
|
||||
topology_contract = source_qualification.get("topology_contract") or {}
|
||||
|
||||
def checkpoint(current_row=None):
|
||||
snapshot = json.loads(json.dumps(report, ensure_ascii=False))
|
||||
@ -370,16 +453,49 @@ def main():
|
||||
)
|
||||
graph, feedback = {}, []
|
||||
accepted_compile_issue_count = None
|
||||
accepted_total_blocker_count = None
|
||||
round_number_base = 0
|
||||
if resume_payload:
|
||||
best = select_resume_checkpoint(resume_payload.get("rounds") or [])
|
||||
graph = copy.deepcopy(best["graph"])
|
||||
refreshed_compiled = compile_graph(
|
||||
graph, source_image_ids=image_ids, dimension_ledger=dimension_ledger,
|
||||
topology_contract=topology_contract,
|
||||
)
|
||||
accepted_compile_issue_count = len(refreshed_compiled.get("issues") or [])
|
||||
prior_kernel_feedback = list(best.get("cad_kernel_feedback") or [])
|
||||
refreshed_checkpoint = {
|
||||
"compiled": refreshed_compiled,
|
||||
"cad_kernel_feedback": prior_kernel_feedback,
|
||||
}
|
||||
accepted_total_blocker_count = round_blocker_count(refreshed_checkpoint)
|
||||
feedback = (
|
||||
list(refreshed_compiled.get("issues") or [])
|
||||
+ prior_kernel_feedback
|
||||
+ list((best.get("visual_review") or {}).get("issues") or [])
|
||||
+ list((best.get("visual_review") or {}).get("uncertain") or [])
|
||||
)
|
||||
round_number_base = int(best.get("round") or 0)
|
||||
report["resumed_from"] = {
|
||||
"source_round": round_number_base,
|
||||
"source_graph_sha256": refreshed_compiled.get("graph_sha256"),
|
||||
"source_end_to_end_blocker_count": accepted_total_blocker_count,
|
||||
"selection": "minimum_end_to_end_blocker_count_then_latest_round",
|
||||
"compiler_feedback_refreshed": True,
|
||||
}
|
||||
for turn in range(max(1, min(args.rounds, 5))):
|
||||
started = time.time()
|
||||
row = {"round": turn + 1}
|
||||
row = {"round": round_number_base + turn + 1}
|
||||
round_base_graph = graph
|
||||
round_base_feedback = list(feedback)
|
||||
round_base_compile_issue_count = accepted_compile_issue_count
|
||||
prompt = GRAPH_PROMPT + evidence_context
|
||||
if graph:
|
||||
prompt = (
|
||||
"你在修正一份已存在的参数化家具结构图谱。必须回看本轮所有原图，并只返回局部JSON修改，不能返回整份图谱。"
|
||||
"每条操作为add/replace/remove和JSON Pointer路径；修复引用、参数、实例位置、接触面或遗漏部件。"
|
||||
"不能通过放大公差、删除有原图证据的部件/连接或改低置信度来逃避问题。"
|
||||
"一次最多24条patch，优先级固定为：原图整体尺寸冲突、总成局部边界错误、实体穿插、已定义却无实例的部件、总成归属/总成关系错误、几何未真实接触、其他外观差异。"
|
||||
"一次最多24条patch，优先级固定为：原图整体尺寸冲突、总成局部边界错误、五金安装宿主/安装面错误、实体穿插、已定义却无实例的部件、总成归属/总成关系错误、几何未真实接触、其他外观差异。"
|
||||
"若零件observed_quantity大于实例数，必须为每个真实物理件补instances，不能只保留零件定义。"
|
||||
"同一总成的固定关系不准手写connections；必须修正部件尺寸、placement锚点或assembly_id，让求解器计算坐标并从真实接触面推导连接。"
|
||||
"anchored_axes_v1下严禁修改不存在的position_mm；必须修改总成bounds_mm、零件size_mm或placement锚点/offset_mm。"
|
||||
@ -389,6 +505,15 @@ def main():
|
||||
"图谱固定使用orientation_solver=role_axis_templates_v1；每个板件必须保留functional_role，size_mm直接表示产品X/Y/Z，thickness_axis也是产品轴。"
|
||||
"严禁新增或修改axis_order；若角色朝向冲突，必须修正part的size_mm与thickness_axis，不能对instance二次旋转。"
|
||||
"编译器会用这些字段区分面贴面、边贴面与边贴边，不允许用名称代替接口语义。"
|
||||
"每个hardware实例必须声明mountings的host_instance_id/kind/face/embed_depth_mm/evidence_ids；"
|
||||
"surface安装嵌入深度为0，mortised/through必须与实际穿入宿主的几何深度一致。"
|
||||
"铰链等跨越两个结构件的五金可声明两个mountings；修复五金时不得改大嵌入深度来掩盖错位。"
|
||||
"hardware_mounting_geometry反馈中若有compatible_mounting_values，必须优先逐字采用其中kind/face/embed_depth_mm；"
|
||||
"该候选由解算几何生成，不得继续猜测其他轴面。候选为空时才修正宿主或placement。"
|
||||
"solid_interference反馈中若有compatible_separation_candidates，必须回看原图后优先选择"
|
||||
"最小必要位移且符合真实装配方向的候选；位移量相同时再选"
|
||||
"remaining_collision_count更少的候选，逐字采用其patches；"
|
||||
"候选中总成bounds修改与唯一结构件位移是一个原子修复，不得只做一半。"
|
||||
"修正定位时优先使用稳定实例ID。若某轴当前没有value_mm或offset_mm，不得replace不存在的叶子字段；"
|
||||
"必须replace整个轴对象，或对align轴用add新增offset_mm。"
|
||||
"合法示例：{\"patches\":[{\"op\":\"replace\",\"path\":\"/instances/door_left_inst/placement/x\","
|
||||
@ -400,7 +525,7 @@ def main():
|
||||
+ "\n编译器及原图视觉复核的结构化反馈：" + json.dumps(feedback, ensure_ascii=False)
|
||||
)
|
||||
phase = "ai_graph_from_original_images" if not graph else "ai_local_graph_repair_from_feedback"
|
||||
print(json.dumps({"case": case, "round": turn + 1, "phase": phase}), flush=True)
|
||||
print(json.dumps({"case": case, "round": row["round"], "phase": phase}), flush=True)
|
||||
try:
|
||||
response = engine._request_model_json({
|
||||
"model": args.model, "messages": [{"role": "user", "content": images + [{"type": "text", "text": prompt}]}],
|
||||
@ -422,15 +547,14 @@ def main():
|
||||
topology_contract=topology_contract,
|
||||
)
|
||||
candidate_issue_count = len(candidate_compiled.get("issues") or [])
|
||||
if (
|
||||
accepted_compile_issue_count is not None
|
||||
and candidate_issue_count > accepted_compile_issue_count
|
||||
if not compile_lower_bound_can_improve(
|
||||
candidate_issue_count, accepted_total_blocker_count
|
||||
):
|
||||
graph = previous_graph
|
||||
row["patch_error"] = "repair_regression_rejected"
|
||||
row["patch_regression"] = {
|
||||
"previous_issue_count": accepted_compile_issue_count,
|
||||
"candidate_issue_count": candidate_issue_count,
|
||||
"accepted_end_to_end_blocker_count": accepted_total_blocker_count,
|
||||
"candidate_compile_issue_lower_bound": candidate_issue_count,
|
||||
"candidate_issue_codes": dict(__import__("collections").Counter(
|
||||
item.get("code") for item in candidate_compiled.get("issues") or []
|
||||
)),
|
||||
@ -438,11 +562,11 @@ def main():
|
||||
feedback = [{
|
||||
"code": "repair_regression_rejected",
|
||||
"detail": (
|
||||
f"本轮局部修改使编译问题从{accepted_compile_issue_count}项增加到"
|
||||
f"{candidate_issue_count}项，已保留上一版图谱；下一轮必须缩小修改范围。"
|
||||
f"本轮仅编译问题已有{candidate_issue_count}项，不可能严格优于上一版"
|
||||
f"端到端总阻塞{accepted_total_blocker_count}项，已保留上一版图谱。"
|
||||
),
|
||||
"correction": "逐条采用编译器给出的稳定ID和单轴路径，禁止整段覆盖placement或引入统一经验偏移。",
|
||||
}] + list(candidate_compiled.get("issues") or [])[:12]
|
||||
}] + round_base_feedback[:12]
|
||||
report["status"] = "regressive_patch_rejected"
|
||||
row["checkpoint_phase"] = "regressive_patch_rejected_graph_preserved"
|
||||
checkpoint(row)
|
||||
@ -477,7 +601,7 @@ def main():
|
||||
row["checkpoint_phase"] = "graph_compiled"
|
||||
checkpoint(row)
|
||||
if compiled["instances"]:
|
||||
build_dir = case_dir / f"round-{turn + 1}"
|
||||
build_dir = case_dir / f"round-{row['round']}"
|
||||
build_dir.mkdir(exist_ok=True)
|
||||
source = build_dir / "assembly.step.py"
|
||||
source.write_text(cad_source(compiled, manufacturing_plan), encoding="utf-8")
|
||||
@ -535,11 +659,43 @@ def main():
|
||||
row["error"] = f"{type(exc).__name__}: {exc}"
|
||||
report["status"] = "failed"
|
||||
feedback = [{"code": "request_or_compile_error", "detail": row["error"]}]
|
||||
if not row.get("patch_error") and isinstance(row.get("compiled"), dict):
|
||||
candidate_blocker_count = round_blocker_count(row)
|
||||
row["end_to_end_blocker_count"] = candidate_blocker_count
|
||||
if (
|
||||
round_base_graph
|
||||
and accepted_total_blocker_count is not None
|
||||
and candidate_blocker_count >= accepted_total_blocker_count
|
||||
):
|
||||
candidate_graph = graph
|
||||
graph = round_base_graph
|
||||
accepted_compile_issue_count = round_base_compile_issue_count
|
||||
row["patch_error"] = "end_to_end_repair_regression_rejected"
|
||||
row["candidate_graph"] = candidate_graph
|
||||
row["end_to_end_regression"] = {
|
||||
"previous_blocker_count": accepted_total_blocker_count,
|
||||
"candidate_blocker_count": candidate_blocker_count,
|
||||
"count_basis": "compile+manufacturing_when_compile_ok+kernel+visual",
|
||||
}
|
||||
feedback = [{
|
||||
"code": "end_to_end_repair_regression_rejected",
|
||||
"detail": (
|
||||
f"本轮修改后综合阻塞数为{candidate_blocker_count}项，未优于上一版"
|
||||
f"{accepted_total_blocker_count}项，已保留上一版图谱。"
|
||||
),
|
||||
"correction": (
|
||||
"同时核对编译器与B-rep内核反馈，只采用不会产生新冲突的单轴候选；"
|
||||
"不得用减少边界提示来换取更多实体穿插。"
|
||||
),
|
||||
}] + round_base_feedback[:12]
|
||||
report["status"] = "regressive_end_to_end_patch_rejected"
|
||||
else:
|
||||
accepted_total_blocker_count = candidate_blocker_count
|
||||
row["elapsed_seconds"] = round(time.time() - started, 2)
|
||||
row.pop("checkpoint_phase", None)
|
||||
report["rounds"].append(row)
|
||||
(case_dir / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(json.dumps({"case": case, "round": turn + 1, "status": report["status"], "elapsed": row["elapsed_seconds"], "feedback": feedback[:8]}, ensure_ascii=False), flush=True)
|
||||
print(json.dumps({"case": case, "round": row["round"], "status": report["status"], "elapsed": row["elapsed_seconds"], "feedback": feedback[:8]}, ensure_ascii=False), flush=True)
|
||||
if report["status"].startswith("candidate_visual_pass"):
|
||||
break
|
||||
reports.append(report)
|
||||
|
||||
@ -629,6 +629,253 @@ def _derive_aabb_contacts(instances: dict, *, tolerance_mm: float = 0.6) -> list
|
||||
return contacts
|
||||
|
||||
|
||||
def _validate_hardware_mountings(
|
||||
instances: dict,
|
||||
*,
|
||||
observations: dict,
|
||||
resolver: "ParameterResolver",
|
||||
tolerance_mm: float = 0.6,
|
||||
) -> tuple[list[dict], list[dict]]:
|
||||
"""Validate explicit hardware-to-host installation geometry.
|
||||
|
||||
Hardware is not an ordinary structural box: a knob may protrude beyond a
|
||||
door's structural envelope and a mortised hinge may occupy a bounded part
|
||||
of its host. Neither case is a licence to ignore collisions. The graph
|
||||
must name every host, mounting face and embed depth; this function then
|
||||
proves contact from solved geometry. It deliberately uses stable roles
|
||||
and IDs, never product names or SKU templates.
|
||||
"""
|
||||
faces = {"x-", "x+", "y-", "y+", "z-", "z+"}
|
||||
allowed_kinds = {"surface", "mortised", "through"}
|
||||
audits: list[dict] = []
|
||||
issues: list[dict] = []
|
||||
|
||||
def mounting_issue(code: str, instance_id: str, index: int | None, detail: str, **context):
|
||||
path = f"/instances/{instance_id}/mountings"
|
||||
if index is not None:
|
||||
path += f"/{index}"
|
||||
issues.append({
|
||||
"code": code,
|
||||
"path": path,
|
||||
"detail": detail,
|
||||
"repair_owner": "ai",
|
||||
"instance_id": instance_id,
|
||||
**context,
|
||||
})
|
||||
|
||||
for instance_id, hardware in instances.items():
|
||||
raw_mountings = hardware.get("mountings")
|
||||
is_hardware = hardware.get("functional_role") == "hardware"
|
||||
if not is_hardware:
|
||||
if raw_mountings not in (None, []):
|
||||
mounting_issue(
|
||||
"mounting_on_non_hardware", instance_id, None,
|
||||
"只有functional_role=hardware的实例可以声明mountings",
|
||||
)
|
||||
continue
|
||||
if not isinstance(raw_mountings, list) or not raw_mountings:
|
||||
mounting_issue(
|
||||
"hardware_mounting_missing", instance_id, None,
|
||||
"五金实例必须声明至少一个可由几何验证的宿主板件安装关系",
|
||||
legal_shape={
|
||||
"host_instance_id": "稳定实例ID", "kind": "surface|mortised|through",
|
||||
"face": "x-|x+|y-|y+|z-|z+", "embed_depth_mm": 0,
|
||||
"evidence_ids": ["真实图像证据ID"],
|
||||
},
|
||||
)
|
||||
continue
|
||||
if len(raw_mountings) > 4:
|
||||
mounting_issue("hardware_mounting_count", instance_id, None, "单个五金实例最多声明4个安装宿主")
|
||||
continue
|
||||
seen_hosts = set()
|
||||
for index, mounting in enumerate(raw_mountings):
|
||||
if not isinstance(mounting, dict):
|
||||
mounting_issue("hardware_mounting_shape", instance_id, index, "安装关系必须是JSON对象")
|
||||
continue
|
||||
host_id = mounting.get("host_instance_id")
|
||||
host = instances.get(host_id)
|
||||
if not host or host_id == instance_id or host.get("functional_role") == "hardware":
|
||||
mounting_issue(
|
||||
"hardware_mounting_host", instance_id, index,
|
||||
"安装宿主必须是存在且非五金的物理实例",
|
||||
host_instance_id=host_id,
|
||||
)
|
||||
continue
|
||||
if host_id in seen_hosts:
|
||||
mounting_issue(
|
||||
"hardware_mounting_duplicate_host", instance_id, index,
|
||||
"同一五金不能重复声明同一安装宿主",
|
||||
host_instance_id=host_id,
|
||||
)
|
||||
continue
|
||||
seen_hosts.add(host_id)
|
||||
kind, face = mounting.get("kind"), mounting.get("face")
|
||||
if kind not in allowed_kinds or face not in faces:
|
||||
mounting_issue(
|
||||
"hardware_mounting_kind", instance_id, index,
|
||||
"安装方式只支持surface/mortised/through且必须声明宿主安装面",
|
||||
actual_kind=kind, actual_face=face,
|
||||
)
|
||||
continue
|
||||
refs = mounting.get("evidence_ids") or []
|
||||
if not refs or any(ref not in observations for ref in refs):
|
||||
mounting_issue(
|
||||
"hardware_mounting_evidence", instance_id, index,
|
||||
"五金安装关系必须引用本轮真实图像或用户证据",
|
||||
host_instance_id=host_id,
|
||||
)
|
||||
continue
|
||||
try:
|
||||
declared_embed = resolver.length(mounting.get("embed_depth_mm", 0))
|
||||
except (GraphError, TypeError, KeyError) as exc:
|
||||
mounting_issue(
|
||||
"hardware_mounting_depth", instance_id, index,
|
||||
f"安装嵌入深度无效: {exc}", host_instance_id=host_id,
|
||||
)
|
||||
continue
|
||||
axis = "xyz".index(face[0])
|
||||
host_low = host["position_mm"][axis]
|
||||
host_high = host_low + host["size_mm"][axis]
|
||||
hardware_low = hardware["position_mm"][axis]
|
||||
hardware_high = hardware_low + hardware["size_mm"][axis]
|
||||
host_plane = host_low if face[1] == "-" else host_high
|
||||
if face[1] == "+":
|
||||
actual_embed = max(0.0, host_plane - hardware_low)
|
||||
gap = max(0.0, hardware_low - host_plane)
|
||||
else:
|
||||
actual_embed = max(0.0, hardware_high - host_plane)
|
||||
gap = max(0.0, host_plane - hardware_high)
|
||||
tangent_axes = [dimension for dimension in range(3) if dimension != axis]
|
||||
tangent_overlaps = [
|
||||
min(
|
||||
host["position_mm"][dimension] + host["size_mm"][dimension],
|
||||
hardware["position_mm"][dimension] + hardware["size_mm"][dimension],
|
||||
) - max(host["position_mm"][dimension], hardware["position_mm"][dimension])
|
||||
for dimension in tangent_axes
|
||||
]
|
||||
host_thickness = host["size_mm"][axis]
|
||||
depth_tolerance = max(tolerance_mm, declared_embed * 0.1)
|
||||
geometric_errors = []
|
||||
host_thickness_axis = host.get("thickness_axis")
|
||||
if host_thickness_axis in {"x", "y", "z"} and face[0] != host_thickness_axis:
|
||||
geometric_errors.append(
|
||||
f"安装面{face}不是宿主板件厚度轴{host_thickness_axis}的广面"
|
||||
)
|
||||
if min(tangent_overlaps) <= tolerance_mm:
|
||||
geometric_errors.append("在安装面上与宿主没有正面积重叠")
|
||||
if gap > tolerance_mm:
|
||||
geometric_errors.append(f"与宿主安装面间隙{gap:.3f}mm超差")
|
||||
if declared_embed < 0 or declared_embed > host_thickness + tolerance_mm:
|
||||
geometric_errors.append("声明嵌入深度超出宿主法向厚度")
|
||||
if kind == "surface":
|
||||
if declared_embed > tolerance_mm:
|
||||
geometric_errors.append("surface安装的embed_depth_mm必须为0")
|
||||
if actual_embed > tolerance_mm:
|
||||
geometric_errors.append(f"表面安装却实际穿入宿主{actual_embed:.3f}mm")
|
||||
elif kind == "mortised":
|
||||
if declared_embed <= tolerance_mm:
|
||||
geometric_errors.append("嵌入式安装必须声明大于公差的嵌入深度")
|
||||
elif abs(actual_embed - declared_embed) > depth_tolerance:
|
||||
geometric_errors.append(
|
||||
f"实际嵌入{actual_embed:.3f}mm与声明{declared_embed:.3f}mm不符"
|
||||
)
|
||||
elif kind == "through":
|
||||
if abs(declared_embed - host_thickness) > tolerance_mm:
|
||||
geometric_errors.append(
|
||||
f"贯穿安装应声明宿主全厚{host_thickness:.3f}mm而不是{declared_embed:.3f}mm"
|
||||
)
|
||||
if actual_embed < host_thickness - tolerance_mm:
|
||||
geometric_errors.append(
|
||||
f"贯穿安装只穿入{actual_embed:.3f}mm，未穿过宿主{host_thickness:.3f}mm厚度"
|
||||
)
|
||||
if geometric_errors:
|
||||
compatible_mounting_values = []
|
||||
for candidate_face in sorted(faces):
|
||||
candidate_axis = "xyz".index(candidate_face[0])
|
||||
if (
|
||||
host_thickness_axis in {"x", "y", "z"}
|
||||
and candidate_face[0] != host_thickness_axis
|
||||
):
|
||||
continue
|
||||
candidate_host_low = host["position_mm"][candidate_axis]
|
||||
candidate_host_high = candidate_host_low + host["size_mm"][candidate_axis]
|
||||
candidate_hardware_low = hardware["position_mm"][candidate_axis]
|
||||
candidate_hardware_high = candidate_hardware_low + hardware["size_mm"][candidate_axis]
|
||||
candidate_plane = (
|
||||
candidate_host_low if candidate_face[1] == "-" else candidate_host_high
|
||||
)
|
||||
if candidate_face[1] == "+":
|
||||
candidate_embed = max(0.0, candidate_plane - candidate_hardware_low)
|
||||
candidate_gap = max(0.0, candidate_hardware_low - candidate_plane)
|
||||
outward_projection = max(0.0, candidate_hardware_high - candidate_plane)
|
||||
else:
|
||||
candidate_embed = max(0.0, candidate_hardware_high - candidate_plane)
|
||||
candidate_gap = max(0.0, candidate_plane - candidate_hardware_high)
|
||||
outward_projection = max(0.0, candidate_plane - candidate_hardware_low)
|
||||
candidate_tangent_axes = [
|
||||
dimension for dimension in range(3) if dimension != candidate_axis
|
||||
]
|
||||
candidate_overlaps = [
|
||||
min(
|
||||
host["position_mm"][dimension] + host["size_mm"][dimension],
|
||||
hardware["position_mm"][dimension] + hardware["size_mm"][dimension],
|
||||
) - max(
|
||||
host["position_mm"][dimension], hardware["position_mm"][dimension]
|
||||
)
|
||||
for dimension in candidate_tangent_axes
|
||||
]
|
||||
if candidate_gap > tolerance_mm or min(candidate_overlaps) <= tolerance_mm:
|
||||
continue
|
||||
candidate_host_thickness = host["size_mm"][candidate_axis]
|
||||
if candidate_embed <= tolerance_mm:
|
||||
candidate_kind, candidate_depth = "surface", 0.0
|
||||
elif candidate_embed < candidate_host_thickness - tolerance_mm:
|
||||
candidate_kind, candidate_depth = "mortised", candidate_embed
|
||||
else:
|
||||
candidate_kind, candidate_depth = "through", candidate_host_thickness
|
||||
compatible_mounting_values.append({
|
||||
"kind": candidate_kind,
|
||||
"face": candidate_face,
|
||||
"embed_depth_mm": round(candidate_depth, 6),
|
||||
"outward_projection_mm": round(outward_projection, 6),
|
||||
"tangent_overlap_mm": [round(value, 6) for value in candidate_overlaps],
|
||||
})
|
||||
compatible_mounting_values.sort(
|
||||
key=lambda row: (
|
||||
row["outward_projection_mm"], min(row["tangent_overlap_mm"])
|
||||
),
|
||||
reverse=True,
|
||||
)
|
||||
mounting_issue(
|
||||
"hardware_mounting_geometry", instance_id, index,
|
||||
"；".join(geometric_errors), host_instance_id=host_id,
|
||||
kind=kind, face=face, declared_embed_depth_mm=declared_embed,
|
||||
actual_embed_depth_mm=actual_embed, gap_mm=gap,
|
||||
tangent_overlap_mm=tangent_overlaps,
|
||||
compatible_mounting_values=compatible_mounting_values[:3],
|
||||
correction=(
|
||||
"先根据compatible_mounting_values修正kind/face/embed_depth_mm；若无候选，"
|
||||
"说明五金与该宿主的位置本身不可成立，再修改实例placement或宿主；"
|
||||
"不得通过改大嵌入深度掩盖错位。"
|
||||
),
|
||||
)
|
||||
continue
|
||||
audits.append({
|
||||
"instance_id": instance_id,
|
||||
"host_instance_id": host_id,
|
||||
"kind": kind,
|
||||
"face": face,
|
||||
"declared_embed_depth_mm": declared_embed,
|
||||
"actual_embed_depth_mm": actual_embed,
|
||||
"gap_mm": gap,
|
||||
"tangent_overlap_mm": tangent_overlaps,
|
||||
"allows_host_interference": kind in {"mortised", "through"},
|
||||
"same_assembly": hardware.get("assembly_id") == host.get("assembly_id"),
|
||||
"source": "declared_mounting_verified_by_aabb",
|
||||
})
|
||||
return audits, issues
|
||||
|
||||
|
||||
def _solve_anchored_axis_placements(
|
||||
instance_specs: dict,
|
||||
*,
|
||||
@ -929,11 +1176,27 @@ def compile_graph(
|
||||
)
|
||||
except (GraphError, KeyError, TypeError) as exc:
|
||||
issue("placement_constraint", "/instances", str(exc))
|
||||
mounting_audit, mounting_issues = _validate_hardware_mountings(
|
||||
instances, observations=observations, resolver=resolver,
|
||||
)
|
||||
issues.extend(mounting_issues)
|
||||
valid_mount_pairs = {
|
||||
frozenset((row["instance_id"], row["host_instance_id"])): row
|
||||
for row in mounting_audit
|
||||
}
|
||||
valid_same_assembly_hardware = {
|
||||
row["instance_id"] for row in mounting_audit if row["same_assembly"]
|
||||
}
|
||||
if graph.get("assemblies") is not None:
|
||||
for key, inst in instances.items():
|
||||
if inst.get("assembly_id") not in assemblies:
|
||||
issue("instance_assembly", f"/instances/{key}/assembly_id", "实例引用了未声明的总成")
|
||||
continue
|
||||
# Assembly bounds define the structural placement frame. A valid
|
||||
# mounted item may protrude from that frame, but only after its
|
||||
# host, face and contact geometry have passed the mounting gate.
|
||||
if inst.get("functional_role") == "hardware" and key in valid_same_assembly_hardware:
|
||||
continue
|
||||
bounds = assembly_bounds[inst["assembly_id"]]
|
||||
outside_axes = []
|
||||
for axis, axis_name in enumerate("xyz"):
|
||||
@ -981,9 +1244,13 @@ def compile_graph(
|
||||
rows = [inst for inst in instances.values() if inst.get("assembly_id") == key]
|
||||
if not rows:
|
||||
continue
|
||||
structural_rows = [row for row in rows if row.get("functional_role") != "hardware"]
|
||||
# A dedicated hardware_group has no structural member. Keep its
|
||||
# declared bounds check meaningful instead of accepting a void.
|
||||
tight_rows = structural_rows or rows
|
||||
actual = [
|
||||
[min(row["position_mm"][axis] for row in rows),
|
||||
max(row["position_mm"][axis] + row["size_mm"][axis] for row in rows)]
|
||||
[min(row["position_mm"][axis] for row in tight_rows),
|
||||
max(row["position_mm"][axis] + row["size_mm"][axis] for row in tight_rows)]
|
||||
for axis in range(3)
|
||||
]
|
||||
mismatch = []
|
||||
@ -997,8 +1264,9 @@ def compile_graph(
|
||||
if mismatch:
|
||||
issue(
|
||||
"assembly_bounds_not_tight", f"/assemblies/{key}/bounds_mm",
|
||||
"总成边界必须等于所属实体的实际包围盒，不能用过大的虚拟空间掩盖错位或漏件",
|
||||
"总成边界必须等于所属结构件的实际包围盒，不能用五金外凸或过大虚拟空间掩盖错位、漏件",
|
||||
assembly_id=key, axes=mismatch,
|
||||
bounds_basis=("structural_members" if structural_rows else "hardware_group_members"),
|
||||
)
|
||||
solver_adjustments = []
|
||||
if graph.get("constraint_solver") == "orthogonal_contacts_v1":
|
||||
@ -1042,6 +1310,9 @@ def compile_graph(
|
||||
)
|
||||
seen_connections = set()
|
||||
neighbours = {key: set() for key in instances}
|
||||
for row in mounting_audit:
|
||||
neighbours[row["instance_id"]].add(row["host_instance_id"])
|
||||
neighbours[row["host_instance_id"]].add(row["instance_id"])
|
||||
for index, edge in enumerate(graph.get("connections") or []):
|
||||
path = f"/connections/{index}"
|
||||
key = edge.get("id")
|
||||
@ -1110,6 +1381,10 @@ def compile_graph(
|
||||
for edge in derived_contacts:
|
||||
internal_neighbours[edge["a"]].add(edge["b"])
|
||||
internal_neighbours[edge["b"]].add(edge["a"])
|
||||
for row in mounting_audit:
|
||||
if row["same_assembly"]:
|
||||
internal_neighbours[row["instance_id"]].add(row["host_instance_id"])
|
||||
internal_neighbours[row["host_instance_id"]].add(row["instance_id"])
|
||||
for assembly_id, rows in members.items():
|
||||
if len(rows) <= 1:
|
||||
continue
|
||||
@ -1269,14 +1544,117 @@ def compile_graph(
|
||||
seen_boxes = {}
|
||||
collision_pairs = []
|
||||
instance_rows = list(instances.values())
|
||||
structural_members_by_assembly = {
|
||||
assembly_id: [
|
||||
row for row in instance_rows
|
||||
if row.get("assembly_id") == assembly_id
|
||||
and row.get("functional_role") != "hardware"
|
||||
]
|
||||
for assembly_id in assemblies
|
||||
}
|
||||
|
||||
def collision_separation_candidates(a: dict, b: dict) -> list[dict]:
|
||||
"""Return exact, bounded one-axis moves; never mutate the graph.
|
||||
|
||||
The language model still decides which move matches the source views.
|
||||
Geometry only removes impossible choices and ranks moves that create
|
||||
the fewest new collisions. A one-member moving assembly receives the
|
||||
matching tight-bounds patch so a valid door/drawer shift does not
|
||||
immediately fail a different invariant.
|
||||
"""
|
||||
candidates = []
|
||||
for moving, stationary in ((a, b), (b, a)):
|
||||
moving_id = moving["id"]
|
||||
for axis, axis_name in enumerate("xyz"):
|
||||
moving_low = moving["position_mm"][axis]
|
||||
moving_high = moving_low + moving["size_mm"][axis]
|
||||
stationary_low = stationary["position_mm"][axis]
|
||||
stationary_high = stationary_low + stationary["size_mm"][axis]
|
||||
deltas = [stationary_low - moving_high, stationary_high - moving_low]
|
||||
for delta in deltas:
|
||||
if abs(delta) <= 0.6:
|
||||
continue
|
||||
new_low, new_high = moving_low + delta, moving_high + delta
|
||||
if envelope and (new_low < -0.5 or new_high > envelope[axis] + 0.5):
|
||||
continue
|
||||
assembly_id = moving.get("assembly_id")
|
||||
assembly_patch = None
|
||||
bounds = assembly_bounds.get(assembly_id)
|
||||
structural_rows = structural_members_by_assembly.get(assembly_id) or []
|
||||
if bounds and (new_low < bounds[axis][0] - 0.5 or new_high > bounds[axis][1] + 0.5):
|
||||
if len(structural_rows) != 1 or structural_rows[0]["id"] != moving_id:
|
||||
continue
|
||||
assembly_patch = {
|
||||
"op": "replace",
|
||||
"path": f"/assemblies/{assembly_id}/bounds_mm/{axis_name}",
|
||||
"value": [round(new_low, 6), round(new_high, 6)],
|
||||
}
|
||||
elif bounds and len(structural_rows) == 1 and structural_rows[0]["id"] == moving_id:
|
||||
# Bounds must remain tight even when the moved member
|
||||
# happens to remain inside the old range.
|
||||
assembly_patch = {
|
||||
"op": "replace",
|
||||
"path": f"/assemblies/{assembly_id}/bounds_mm/{axis_name}",
|
||||
"value": [round(new_low, 6), round(new_high, 6)],
|
||||
}
|
||||
current_rule = (instance_specs.get(moving_id) or {}).get("placement", {}).get(axis_name)
|
||||
if not isinstance(current_rule, dict):
|
||||
continue
|
||||
if current_rule.get("mode") == "align":
|
||||
replacement = copy.deepcopy(current_rule)
|
||||
replacement["offset_mm"] = round(
|
||||
resolver.length(replacement.get("offset_mm", 0)) + delta, 6
|
||||
)
|
||||
else:
|
||||
replacement = {"mode": "absolute", "value_mm": round(new_low, 6)}
|
||||
simulated = {**moving, "position_mm": list(moving["position_mm"])}
|
||||
simulated["position_mm"][axis] = new_low
|
||||
remaining = sum(
|
||||
1 for other in instance_rows
|
||||
if other["id"] != moving_id and _aabb_interference(simulated, other)
|
||||
)
|
||||
patches = [{
|
||||
"op": "replace",
|
||||
"path": f"/instances/{moving_id}/placement/{axis_name}",
|
||||
"value": replacement,
|
||||
}]
|
||||
if assembly_patch:
|
||||
patches.append(assembly_patch)
|
||||
candidates.append({
|
||||
"moving_instance_id": moving_id,
|
||||
"stationary_instance_id": stationary["id"],
|
||||
"axis": axis_name,
|
||||
"translation_mm": round(delta, 6),
|
||||
"new_range_mm": [round(new_low, 6), round(new_high, 6)],
|
||||
"remaining_collision_count": remaining,
|
||||
"patches": patches,
|
||||
})
|
||||
# A minimum separating translation is the physically conservative
|
||||
# default. Moving a 20 mm door inward is preferable to relocating an
|
||||
# entire carcass side by hundreds of millimetres merely because the
|
||||
# latter happens to touch one fewer already-invalid solid.
|
||||
candidates.sort(key=lambda row: (abs(row["translation_mm"]), row["remaining_collision_count"]))
|
||||
return candidates[:6]
|
||||
|
||||
for left_index, a in enumerate(instance_rows):
|
||||
for b in instance_rows[left_index + 1:]:
|
||||
collision = _aabb_interference(a, b)
|
||||
if collision:
|
||||
collision_pairs.append(collision)
|
||||
mounting = valid_mount_pairs.get(frozenset((a["id"], b["id"])))
|
||||
if mounting and mounting["allows_host_interference"]:
|
||||
collision["accepted_as_mounting"] = True
|
||||
collision["mounting_kind"] = mounting["kind"]
|
||||
collision["declared_embed_depth_mm"] = mounting["declared_embed_depth_mm"]
|
||||
continue
|
||||
issue(
|
||||
"solid_interference", f"/instances/{b['id']}",
|
||||
f"实体{a['id']}与{b['id']}发生正体积穿插，不能作为可制造装配",
|
||||
compatible_separation_candidates=collision_separation_candidates(a, b),
|
||||
correction=(
|
||||
"必须回看原图选择compatible_separation_candidates中与真实结构一致的最小位移；"
|
||||
"候选只保证几何可行，不代替图像证据。"
|
||||
),
|
||||
**collision,
|
||||
)
|
||||
for key, inst in instances.items():
|
||||
@ -1286,6 +1664,8 @@ def compile_graph(
|
||||
seen_boxes[signature] = key
|
||||
if envelope:
|
||||
for key, inst in instances.items():
|
||||
if inst.get("functional_role") == "hardware" and key in valid_same_assembly_hardware:
|
||||
continue
|
||||
if any(inst["position_mm"][axis] < -0.5 or
|
||||
inst["position_mm"][axis] + inst["size_mm"][axis] > envelope[axis] + 0.5 for axis in range(3)):
|
||||
issue("outside_envelope", f"/instances/{key}", "部件超出产品边界；不允许自动裁小来隐藏冲突")
|
||||
@ -1325,6 +1705,7 @@ def compile_graph(
|
||||
"derived_contacts": derived_contacts,
|
||||
"assemblies": list(assemblies.values()),
|
||||
"assembly_connection_audit": assembly_connection_audit,
|
||||
"mounting_audit": mounting_audit,
|
||||
"collision_audit": collision_pairs,
|
||||
"dimension_ledger": dimension_ledger or {},
|
||||
"topology_contract": topology_contract or {},
|
||||
|
||||
@ -7,7 +7,8 @@ import unittest
|
||||
|
||||
from manufacturing_oracle import plan_manufacturing_features
|
||||
from probe_structure_graph import (
|
||||
cad_source, source_for_role, structured_kernel_feedback,
|
||||
cad_source, compile_lower_bound_can_improve, round_blocker_count,
|
||||
select_resume_checkpoint, source_for_role, structured_kernel_feedback,
|
||||
structured_patch_feedback,
|
||||
)
|
||||
from structure_graph import SCHEMA_VERSION, compile_graph
|
||||
@ -186,6 +187,43 @@ class ManufacturingOracleTest(unittest.TestCase):
|
||||
self.assertEqual({"x": 12.0, "y": 20.0, "z": 30.0}, feedback[0]["overlap_size_mm"])
|
||||
self.assertIn("/instances/left-door/placement", feedback[0]["legal_patch_roots"])
|
||||
|
||||
def test_repair_score_counts_kernel_regression_after_compile_improvement(self):
|
||||
previous = {
|
||||
"compiled": {"ok": False, "issues": [{}, {}, {}, {}, {}, {}, {}]},
|
||||
"cad_kernel_feedback": [{}, {}, {}],
|
||||
}
|
||||
candidate = {
|
||||
"compiled": {"ok": False, "issues": [{}, {}, {}, {}, {}]},
|
||||
"cad_kernel_feedback": [{}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}],
|
||||
}
|
||||
self.assertEqual(10, round_blocker_count(previous))
|
||||
self.assertEqual(20, round_blocker_count(candidate))
|
||||
self.assertGreater(round_blocker_count(candidate), round_blocker_count(previous))
|
||||
|
||||
compile_pass_with_hidden_review_failure = {
|
||||
"compiled": {"ok": True, "issues": []},
|
||||
"manufacturing_plan": {"issues": []},
|
||||
"visual_review": {"ok": False, "issues": [], "uncertain": []},
|
||||
}
|
||||
self.assertEqual(1, round_blocker_count(compile_pass_with_hidden_review_failure))
|
||||
|
||||
def test_compile_lower_bound_compares_with_end_to_end_score(self):
|
||||
self.assertTrue(compile_lower_bound_can_improve(4, 10))
|
||||
self.assertFalse(compile_lower_bound_can_improve(10, 10))
|
||||
self.assertFalse(compile_lower_bound_can_improve(11, 10))
|
||||
|
||||
def test_resume_skips_rejected_round_and_chooses_latest_tie(self):
|
||||
rows = [
|
||||
{"round": 1, "graph": {"id": "old"}, "compiled": {"issues": [{}]}},
|
||||
{"round": 2, "graph": {"id": "accepted"}, "compiled": {"issues": []},
|
||||
"cad_kernel_feedback": [{"code": "one"}]},
|
||||
{"round": 3, "graph": {"id": "rejected"}, "compiled": {"issues": []},
|
||||
"patch_error": "end_to_end_repair_regression_rejected"},
|
||||
{"round": 4, "graph": {"id": "latest"}, "compiled": {"issues": []},
|
||||
"cad_kernel_feedback": [{"code": "one"}]},
|
||||
]
|
||||
self.assertEqual("latest", select_resume_checkpoint(rows)["graph"]["id"])
|
||||
|
||||
def test_rejected_patch_feedback_uses_stable_id_and_existing_axis_shape(self):
|
||||
graph = graph_for_role()
|
||||
bad_patch = [{
|
||||
|
||||
@ -94,7 +94,107 @@ def anchored_frame_with_drawer_assembly():
|
||||
return graph
|
||||
|
||||
|
||||
def anchored_door_with_hardware(*, kind="surface", embed_depth_mm=0, hardware_x=285):
|
||||
"""A product-independent moving panel with one explicitly mounted item."""
|
||||
hardware_y = 0 if kind == "surface" else 5
|
||||
return {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"placement_solver": "anchored_axes_v1",
|
||||
"orientation_solver": "role_axis_templates_v1",
|
||||
"contact_derivation": "aabb_within_assembly_v1",
|
||||
"observations": [{
|
||||
"id": "view", "source_type": "image", "image_id": "original-front",
|
||||
"confidence": 0.95,
|
||||
}],
|
||||
"parameters": {},
|
||||
"envelope_mm": [700, 100, 700],
|
||||
"assemblies": [{
|
||||
"id": "moving-panel", "role": "door", "evidence_ids": ["view"],
|
||||
"bounds_mm": {"x": [0, 600], "y": [20, 38], "z": [0, 700]},
|
||||
}],
|
||||
"parts": [
|
||||
{"id": "panel", "name": "任意运动板", "shape": "box", "functional_role": "door",
|
||||
"thickness_axis": "y", "size_mm": [600, 18, 700], "observed_quantity": 1},
|
||||
{"id": "fitting", "name": "任意五金", "shape": "box", "functional_role": "hardware",
|
||||
"thickness_axis": None, "size_mm": [30, 20, 30], "observed_quantity": 1},
|
||||
],
|
||||
"instances": [
|
||||
{"id": "panel-instance", "part_id": "panel", "assembly_id": "moving-panel",
|
||||
"placement": {
|
||||
"x": {"mode": "absolute", "value_mm": 0},
|
||||
"y": {"mode": "absolute", "value_mm": 20},
|
||||
"z": {"mode": "absolute", "value_mm": 0},
|
||||
}},
|
||||
{"id": "fitting-instance", "part_id": "fitting", "assembly_id": "moving-panel",
|
||||
"placement": {
|
||||
"x": {"mode": "absolute", "value_mm": hardware_x},
|
||||
"y": {"mode": "absolute", "value_mm": hardware_y},
|
||||
"z": {"mode": "absolute", "value_mm": 335},
|
||||
},
|
||||
"mountings": [{
|
||||
"host_instance_id": "panel-instance", "kind": kind, "face": "y-",
|
||||
"embed_depth_mm": embed_depth_mm, "evidence_ids": ["view"],
|
||||
}]},
|
||||
],
|
||||
"connections": [],
|
||||
"assembly_connections": [],
|
||||
"uncertainties": [],
|
||||
}
|
||||
|
||||
|
||||
class StructureGraphTest(unittest.TestCase):
|
||||
def test_surface_mounted_hardware_can_protrude_beyond_structural_assembly_bounds(self):
|
||||
result = compile_case(anchored_door_with_hardware())
|
||||
self.assertTrue(result["ok"], result["issues"])
|
||||
self.assertEqual(1, len(result["mounting_audit"]))
|
||||
self.assertEqual("surface", result["mounting_audit"][0]["kind"])
|
||||
self.assertEqual(0, result["mounting_audit"][0]["actual_embed_depth_mm"])
|
||||
self.assertNotIn(
|
||||
"assembly_bounds_not_tight", {row["code"] for row in result["issues"]}
|
||||
)
|
||||
|
||||
def test_hardware_on_wrong_panel_position_is_still_blocking(self):
|
||||
result = compile_case(anchored_door_with_hardware(hardware_x=650))
|
||||
codes = {row["code"] for row in result["issues"]}
|
||||
self.assertIn("hardware_mounting_geometry", codes)
|
||||
self.assertIn("assembly_internal_disconnected", codes)
|
||||
self.assertIn("disconnected_instance", codes)
|
||||
self.assertFalse(result["mounting_audit"])
|
||||
|
||||
def test_mortised_hardware_allows_only_declared_host_penetration(self):
|
||||
result = compile_case(anchored_door_with_hardware(kind="mortised", embed_depth_mm=5))
|
||||
self.assertTrue(result["ok"], result["issues"])
|
||||
self.assertEqual(5, result["mounting_audit"][0]["actual_embed_depth_mm"])
|
||||
accepted = [row for row in result["collision_audit"] if row.get("accepted_as_mounting")]
|
||||
self.assertEqual(1, len(accepted))
|
||||
|
||||
too_deep = anchored_door_with_hardware(kind="mortised", embed_depth_mm=2)
|
||||
broken = compile_case(too_deep)
|
||||
self.assertIn("hardware_mounting_geometry", {row["code"] for row in broken["issues"]})
|
||||
self.assertIn("solid_interference", {row["code"] for row in broken["issues"]})
|
||||
|
||||
def test_hardware_mounting_is_required_and_non_hardware_cannot_claim_it(self):
|
||||
missing = anchored_door_with_hardware()
|
||||
missing["instances"][1].pop("mountings")
|
||||
self.assertIn("hardware_mounting_missing", {row["code"] for row in compile_case(missing)["issues"]})
|
||||
|
||||
invalid = anchored_door_with_hardware()
|
||||
invalid["instances"][0]["mountings"] = copy.deepcopy(invalid["instances"][1]["mountings"])
|
||||
self.assertIn("mounting_on_non_hardware", {row["code"] for row in compile_case(invalid)["issues"]})
|
||||
|
||||
def test_panel_hardware_feedback_only_suggests_real_broad_faces(self):
|
||||
graph = anchored_door_with_hardware()
|
||||
graph["instances"][1]["mountings"][0]["face"] = "x+"
|
||||
result = compile_case(graph)
|
||||
problem = next(
|
||||
row for row in result["issues"] if row["code"] == "hardware_mounting_geometry"
|
||||
)
|
||||
self.assertIn("厚度轴y的广面", problem["detail"])
|
||||
self.assertTrue(problem["compatible_mounting_values"])
|
||||
self.assertEqual(
|
||||
{"y-", "y+"}, {row["face"] for row in problem["compatible_mounting_values"]}
|
||||
)
|
||||
|
||||
def test_role_axis_mode_accepts_direct_product_axes_without_instance_mapping(self):
|
||||
graph = anchored_frame()
|
||||
graph["orientation_solver"] = "role_axis_templates_v1"
|
||||
@ -561,6 +661,48 @@ class StructureGraphTest(unittest.TestCase):
|
||||
collision["instances"][2]["placement"]["x"] = {"mode": "absolute", "value_mm": 10}
|
||||
result = compile_case(collision)
|
||||
self.assertIn("solid_interference", {row["code"] for row in result["issues"]})
|
||||
conflict = next(row for row in result["issues"] if row["code"] == "solid_interference")
|
||||
self.assertTrue(conflict["compatible_separation_candidates"])
|
||||
self.assertTrue(all(row["patches"] for row in conflict["compatible_separation_candidates"]))
|
||||
|
||||
def test_collision_candidate_moves_single_member_assembly_and_its_tight_bounds_together(self):
|
||||
graph = anchored_door_with_hardware()
|
||||
graph["assemblies"].append({
|
||||
"id": "stationary", "role": "carcass", "evidence_ids": ["view"],
|
||||
"bounds_mm": {"x": [0, 18], "y": [20, 100], "z": [0, 700]},
|
||||
})
|
||||
graph["parts"].append({
|
||||
"id": "stationary-panel", "name": "静止结构件", "shape": "box",
|
||||
"functional_role": "side", "thickness_axis": "x",
|
||||
"size_mm": [18, 80, 700], "observed_quantity": 1,
|
||||
})
|
||||
graph["instances"].append({
|
||||
"id": "stationary-instance", "part_id": "stationary-panel", "assembly_id": "stationary",
|
||||
"placement": {
|
||||
"x": {"mode": "absolute", "value_mm": 0},
|
||||
"y": {"mode": "absolute", "value_mm": 20},
|
||||
"z": {"mode": "absolute", "value_mm": 0},
|
||||
},
|
||||
})
|
||||
result = compile_case(graph)
|
||||
conflict = next(
|
||||
row for row in result["issues"]
|
||||
if row["code"] == "solid_interference"
|
||||
and {row["a"], row["b"]} == {"panel-instance", "stationary-instance"}
|
||||
)
|
||||
candidate = next(
|
||||
row for row in conflict["compatible_separation_candidates"]
|
||||
if row["moving_instance_id"] == "panel-instance"
|
||||
and row["axis"] == "x" and row["translation_mm"] == 18
|
||||
)
|
||||
self.assertEqual(0, candidate["remaining_collision_count"])
|
||||
self.assertEqual(
|
||||
[
|
||||
"/instances/panel-instance/placement/x",
|
||||
"/assemblies/moving-panel/bounds_mm/x",
|
||||
],
|
||||
[row["path"] for row in candidate["patches"]],
|
||||
)
|
||||
|
||||
def test_anchored_duplicate_instance_id_and_loose_assembly_bounds_are_blocking(self):
|
||||
duplicate = anchored_frame()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user