"""Product-independent, evidence-backed parametric assembly compiler. This candidate does not infer geometry from Chinese part names. AI proposes a graph; this module resolves typed dimensions, derives BOM quantities from real instances and reports geometric contradictions for the AI to repair. It is an isolated compiler, NOT a certificate that the images or design are correct. """ from __future__ import annotations import ast import hashlib import json import math import re import copy from collections import Counter SCHEMA_VERSION = "furniture-assembly-graph/1" # New graphs use one product coordinate system end-to-end: X=width, # Y=depth, Z=height. These templates are role semantics, not Chinese-name # heuristics. They remove the formerly independent ``axis_order`` degree of # freedom that let a model rotate a panel a second time after already writing # product-oriented dimensions. ROLE_PRODUCT_THICKNESS_AXES = { "top": {"z"}, "bottom": {"z"}, "side": {"x"}, "shelf": {"z"}, "door": {"y"}, "back": {"y"}, "drawer_front": {"y"}, "drawer_left": {"x"}, "drawer_right": {"x"}, "drawer_back": {"y"}, "drawer_bottom": {"z"}, "drawer_support": {"z"}, # A divider can be a vertical partition or a horizontal subdivision. # Prefer shelf/drawer_support for horizontal members, but keep both axes # legal so the compiler remains product-independent. "divider": {"x", "z"}, } ROLE_AXIS_SOLVER = "role_axis_templates_v1" def build_source_topology_contract(source_structure: dict, *, min_confidence: float = 0.82) -> dict: """Promote only strongly evidenced source-part counts into a compile gate. This is image-derived evidence, never a company-BOM answer. It prevents a later graph-generation pass from silently omitting a part already counted by an independent multi-view pass. Invisible or low-confidence structure deliberately remains outside the hard contract. """ allowed_roles = { "top", "bottom", "side", "shelf", "door", "back", "drawer_front", "drawer_left", "drawer_right", "drawer_back", "drawer_bottom", "drawer_support", "divider", "leg", "rail", "stile", "frame_member", "hardware", } role_counts = Counter() evidence: dict[str, list[dict]] = {} rows = source_structure.get("parts") if isinstance(source_structure, dict) else [] for row in rows or []: if not isinstance(row, dict): continue role = str(row.get("component_role") or row.get("functional_role") or "") confidence = row.get("confidence") image_ids = [ value for value in (row.get("image_ids") or []) if isinstance(value, int) and not isinstance(value, bool) ] quantity = row.get("quantity") if ( role not in allowed_roles or isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not math.isfinite(confidence) or float(confidence) < float(min_confidence) or not image_ids or isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0 ): continue role_counts[role] += quantity evidence.setdefault(role, []).append({ "source_structure_id": str(row.get("source_structure_id") or ""), "image_ids": sorted(set(image_ids)), "confidence": round(float(confidence), 4), "quantity": quantity, }) return { "source": "independent_multiview_source_structure", "minimum_confidence": float(min_confidence), "role_counts": dict(sorted(role_counts.items())), "evidence": evidence, } class GraphError(ValueError): pass def build_dimension_ledger(labels: list[dict], *, source_image_id: str) -> dict: """Accept only unambiguous overall arrows read from the source drawing. Multiple equal readings are corroboration; materially different readings on the same axis are retained as a conflict and deliberately not converted into a guessed CAD constraint. """ unit_scale = {"mm": 1.0, "cm": 10.0, "m": 1000.0} candidates = {axis: [] for axis in "xyz"} rejected = [] for index, row in enumerate(labels or []): if not isinstance(row, dict): continue axis, unit = row.get("axis"), row.get("unit") confidence = row.get("confidence") value = row.get("value") if ( axis not in candidates or row.get("scope") != "overall" or unit not in unit_scale or isinstance(value, bool) or not isinstance(value, (int, float)) or isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or confidence < 0.8 or value <= 0 ): rejected.append({"index": index, "reason": "not_unambiguous_overall_arrow"}) continue candidates[axis].append({ "value_mm": float(value) * unit_scale[unit], "confidence": min(1.0, float(confidence)), "source_image_ids": [source_image_id], "text": str(row.get("text") or "")[:120], "location": str(row.get("location") or "")[:200], }) ledger = {"conflicts": [], "rejected": rejected} for axis, rows in candidates.items(): if not rows: continue values = [row["value_mm"] for row in rows] tolerance = max(1.0, 0.005 * max(values)) if max(values) - min(values) > tolerance: ledger["conflicts"].append({ "axis": axis, "values_mm": values, "source_image_ids": sorted({source for row in rows for source in row["source_image_ids"]}), }) continue best = max(rows, key=lambda row: row["confidence"]) ledger[axis] = {**best, "value_mm": sum(values) / len(values), "tolerance_mm": tolerance} return ledger def normalize_graph_patch_payload(graph: dict, payload) -> tuple[list[dict], list[dict]]: """Normalize model repair output without weakening patch safety. JSON-capable models sometimes return the requested patch array directly. They can also replace a complete three-axis ``placement`` object even when only one axis changed. Accept the harmless response-shape variant and expand a well-formed complete placement into explicit changed-axis writes. ``apply_graph_patch`` still rejects broad placement writes. """ if isinstance(payload, list): raw_patches = payload response_shape = "patch_list" elif isinstance(payload, dict): raw_patches = payload.get("patches") response_shape = "patch_object" else: raise GraphError("图谱修复响应必须是patches对象或patch数组") if not isinstance(raw_patches, list) or not 1 <= len(raw_patches) <= 64: raise GraphError("图谱局部修改必须包含1到64条操作") instances = graph.get("instances") if isinstance(graph, dict) else None if not isinstance(instances, list): raise GraphError("当前图谱缺少instances数组，无法规范化placement修改") def instance_for(token: str, path: str) -> dict: if re.fullmatch(r"0|[1-9]\d*", token): index = int(token) if not 0 <= index < len(instances) or not isinstance(instances[index], dict): raise GraphError(f"修改路径不存在: {path}") return instances[index] matches = [ row for row in instances if isinstance(row, dict) and str(row.get("id") or "") == token ] if len(matches) != 1: raise GraphError(f"修改路径中的稳定ID不存在或不唯一: {path}") return matches[0] normalized: list[dict] = [] audit: list[dict] = [{ "kind": "response_shape", "source": response_shape, "normalized_to": "patches", }] for index, patch in enumerate(raw_patches): if not isinstance(patch, dict): raise GraphError(f"第{index + 1}条修改操作无效") path = patch.get("path") match = re.fullmatch(r"/instances/([^/]+)/placement", str(path or "")) if patch.get("op") == "replace" and match: value = patch.get("value") if not isinstance(value, dict) or set(value) != set("xyz"): raise GraphError("整段placement规范化要求value恰好包含x、y、z三轴") instance = instance_for(match.group(1), path) current = instance.get("placement") if not isinstance(current, dict) or set(current) != set("xyz"): raise GraphError("当前实例placement必须恰好包含x、y、z三轴") expanded_paths = [] for axis in "xyz": if value[axis] == current[axis]: continue axis_path = f"{path}/{axis}" normalized.append({ "op": "replace", "path": axis_path, "value": copy.deepcopy(value[axis]), }) expanded_paths.append(axis_path) audit.append({ "kind": "placement_expansion", "source_index": index + 1, "source_path": path, "expanded_paths": expanded_paths, }) else: normalized.append(copy.deepcopy(patch)) if not normalized: raise GraphError("图谱修复没有产生任何有效变化") if len(normalized) > 64: raise GraphError("规范化后的图谱局部修改超过64条") return normalized, audit def apply_graph_patch(graph: dict, patches: list[dict]) -> tuple[dict, list[dict]]: """Apply a bounded RFC6902 subset and return an immutable audit trail.""" if not isinstance(patches, list) or not 1 <= len(patches) <= 64: raise GraphError("图谱局部修改必须包含1到64条操作") result = copy.deepcopy(graph) allowed_roots = { "observations", "parameters", "envelope_mm", "reference_bounds_mm", "dimension_scope_overrides", "parts", "instances", "connections", "uncertainties", "constraint_solver", "contact_derivation", "placement_solver", "orientation_solver", "assemblies", "assembly_connections", } audit = [] def list_position(rows, token, path): """Resolve a JSON array index or a stable object ``id``. Model-authored repairs naturally refer to ``/assemblies/drawer-1`` or ``/instances/left-side``. Requiring a volatile numeric array index made otherwise valid repairs fail or target the wrong object after an insertion. Stable IDs are therefore part of the patch contract. """ if isinstance(token, str) and re.fullmatch(r"0|[1-9]\d*", token): position = int(token) if not 0 <= position < len(rows): raise GraphError(f"修改路径不存在: {path}") return position matches = [ index for index, row in enumerate(rows) if isinstance(row, dict) and str(row.get("id") or "") == str(token) ] if len(matches) != 1: raise GraphError(f"修改路径中的稳定ID不存在或不唯一: {path}") return matches[0] def tokens(path): if not isinstance(path, str) or not path.startswith("/"): raise GraphError("局部修改路径必须是JSON Pointer") values = [token.replace("~1", "/").replace("~0", "~") for token in path[1:].split("/")] if not values or values[0] not in allowed_roots or any(token in {"__class__", "__dict__"} for token in values): raise GraphError(f"禁止修改图谱路径: {path}") return values for index, patch in enumerate(patches): if not isinstance(patch, dict) or patch.get("op") not in {"add", "replace", "remove"}: raise GraphError(f"第{index + 1}条修改操作无效") path = patch.get("path") parts = tokens(path) if ( len(parts) == 3 and parts[0] == "instances" and parts[2] == "placement" and patch.get("op") in {"replace", "remove"} ): raise GraphError( "禁止整段覆盖实例placement；必须只修改反馈指出的x/y/z单轴，避免已正确坐标被连带污染" ) parent = result for token in parts[:-1]: try: parent = parent[list_position(parent, token, path)] if isinstance(parent, list) else parent[token] except (KeyError, IndexError, TypeError, ValueError) as exc: raise GraphError(f"修改路径不存在: {path}") from exc leaf = parts[-1] old = None try: if isinstance(parent, list): if leaf == "-": if patch["op"] != "add": raise GraphError(f"只有add允许使用数组末尾路径: {path}") parent.append(copy.deepcopy(patch.get("value"))) else: position = list_position(parent, leaf, path) if patch["op"] == "add": parent.insert(position, copy.deepcopy(patch.get("value"))) else: old = copy.deepcopy(parent[position]) if patch["op"] == "replace": parent[position] = copy.deepcopy(patch.get("value")) else: parent.pop(position) else: if patch["op"] == "add": old = copy.deepcopy(parent.get(leaf)) parent[leaf] = copy.deepcopy(patch.get("value")) else: if leaf not in parent: raise KeyError(leaf) old = copy.deepcopy(parent[leaf]) if patch["op"] == "replace": parent[leaf] = copy.deepcopy(patch.get("value")) else: del parent[leaf] except (KeyError, IndexError, TypeError, ValueError) as exc: if isinstance(exc, GraphError): raise raise GraphError(f"无法执行修改路径: {path}") from exc audit.append({"index": index + 1, "op": patch["op"], "path": path, "old": old, "new": copy.deepcopy(patch.get("value")) if patch["op"] != "remove" else None}) return result, audit class ParameterResolver: def __init__(self, parameters: dict, observations: dict): self.parameters = parameters self.observations = observations self.resolved = {} self.visiting = set() @staticmethod def finite(value): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): raise GraphError("必须为有限数值") return float(value) def parameter(self, key): if key in self.resolved: return self.resolved[key] if key in self.visiting: raise GraphError(f"参数依赖成环: {key}") if key not in self.parameters: raise GraphError(f"参数不存在: {key}") self.visiting.add(key) try: spec = self.parameters[key] kind = spec.get("kind") dimension = 1 if kind == "length" else 0 if kind not in {"length", "count", "scalar"}: raise GraphError(f"参数 {key} 缺少明确类型") origin = spec.get("origin") refs = spec.get("evidence_ids") or [] if origin == "derived": value, actual_dimension, confidence, estimated = self.expression(spec.get("expression")) if actual_dimension != dimension: raise GraphError(f"参数 {key} 派生式量纲错误") elif origin in {"observed", "estimated", "user"}: if origin in {"observed", "user"} and ( not refs or any(ref not in self.observations for ref in refs) ): raise GraphError(f"参数 {key} 缺少可回溯证据") if origin == "observed" and any(self.observations[ref].get("source_type") != "image" for ref in refs): raise GraphError(f"参数 {key} 图像事实引用了非原图证据") if origin == "user" and any(self.observations[ref].get("source_type") != "user" for ref in refs): raise GraphError(f"参数 {key} 不能把AI判断写成用户确认") value = self.finite(spec.get("value")) unit = spec.get("unit") if dimension: if unit not in {"mm", "cm", "m"}: raise GraphError(f"参数 {key} 长度单位不明确") value *= {"mm": 1, "cm": 10, "m": 1000}[unit] elif unit not in {None, "", "1", "件"}: raise GraphError(f"参数 {key} 无量纲参数不应带长度单位") confidence = min([self.finite(spec.get("confidence", 0))] + [self.finite(self.observations[ref].get("confidence", 0)) for ref in refs]) estimated = origin == "estimated" if estimated: confidence = min(confidence, 0.55) else: raise GraphError(f"参数 {key} 缺少来源类型") if kind == "count" and (value < 0 or value != int(value)): raise GraphError(f"参数 {key} 数量必须是非负整数") if not 0 <= confidence <= 1: raise GraphError(f"参数 {key} 置信度超界") result = (value, dimension, confidence, estimated) self.resolved[key] = result return result finally: self.visiting.remove(key) def expression(self, text, *, expected_dimension=None): if not isinstance(text, str) or len(text) > 500: raise GraphError("表达式必须是长度受限的参数算式") try: tree = ast.parse(text, mode="eval") except (SyntaxError, ValueError) as exc: raise GraphError("参数表达式语法错误") from exc if sum(1 for _ in ast.walk(tree)) > 100: raise GraphError("参数表达式过于复杂") def visit(node): if isinstance(node, ast.Constant): return self.finite(node.value), 0, 1.0, False, True if isinstance(node, ast.Name): return (*self.parameter(node.id), False) if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)): value, dimension, confidence, estimated, literal = visit(node.operand) return (-value if isinstance(node.op, ast.USub) else value), dimension, confidence, estimated, literal if not isinstance(node, ast.BinOp) or not isinstance(node.op, (ast.Add, ast.Sub, ast.Mult, ast.Div)): raise GraphError("只允许参数、数字和加减乘除，不执行函数或代码") a, da, ca, ea, la = visit(node.left) b, db, cb, eb, lb = visit(node.right) if isinstance(node.op, (ast.Add, ast.Sub)): # Geometry fields use millimetres as their contextual unit. A # literal offset in ``width-30`` therefore means 30 mm, while # the literal in ``2*thickness`` remains dimensionless. Derived # parameter definitions stay strict unless their caller supplies # an expected length context. if expected_dimension == 1 and da != db: if la and da == 0 and db == 1: da = 1 elif lb and db == 0 and da == 1: db = 1 if da != db: raise GraphError("相加/相减的量纲不同") value, dimension = (a + b if isinstance(node.op, ast.Add) else a - b), da elif isinstance(node.op, ast.Mult): value, dimension = a * b, da + db else: if b == 0: raise GraphError("参数表达式除以零") value, dimension = a / b, da - db self.finite(value) return value, dimension, min(ca, cb), ea or eb, la and lb value, dimension, confidence, estimated, _literal = visit(tree.body) return value, dimension, confidence, estimated def length(self, expression): # Numeric geometry literals are explicitly millimetres under this schema. if isinstance(expression, (int, float)) and not isinstance(expression, bool): return self.finite(expression) if isinstance(expression, str) and re.fullmatch(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)", expression.strip()): return self.finite(float(expression)) value, dimension, _, _ = self.expression(expression, expected_dimension=1) if dimension != 1: raise GraphError("尺寸/位置表达式的结果必须是长度") return value def _solve_orthogonal_contact_normals(instances: dict, connections: list[dict]) -> tuple[list[dict], list[dict]]: """Solve only the normal-axis equation declared by each face mate. The AI owns topology (which faces mate); the solver owns arithmetic. It never guesses tangential alignment or connection type. Every movement is returned for audit, and contradictory cycles remain blocking errors. """ adjacency = {axis: {key: [] for key in instances} for axis in range(3)} errors = [] for index, edge in enumerate(connections): if edge.get("kind") not in {"fixed", "support"}: continue a, b = instances.get(edge.get("a")), instances.get(edge.get("b")) fa, fb = edge.get("face_a"), edge.get("face_b") if not a or not b or fa not in {"x-", "x+", "y-", "y+", "z-", "z+"} or fb not in {"x-", "x+", "y-", "y+", "z-", "z+"}: continue if fa[0] != fb[0] or fa[1] == fb[1]: continue axis = "xyz".index(fa[0]) offset_a = a["size_mm"][axis] if fa[1] == "+" else 0.0 offset_b = b["size_mm"][axis] if fb[1] == "+" else 0.0 delta = offset_a - offset_b # position_b = position_a + delta adjacency[axis][a["id"]].append((b["id"], delta, index)) adjacency[axis][b["id"]].append((a["id"], -delta, index)) adjustments = [] for axis, graph in adjacency.items(): unseen = {key for key, links in graph.items() if links} while unseen: root = next(iter(unseen)) potential = {root: 0.0} pending = [root] component = set() conflict = False while pending: node = pending.pop() component.add(node) for other, delta, edge_index in graph[node]: proposed = potential[node] + delta if other in potential and abs(potential[other] - proposed) > 0.01: errors.append({ "code": "constraint_cycle_conflict", "path": f"/connections/{edge_index}", "detail": f"{('xyz'[axis]).upper()}轴接触约束闭环互相矛盾{abs(potential[other]-proposed):.3f}mm", "repair_owner": "ai", "axis": "xyz"[axis], "instance_ids": sorted(component | {other}), }) conflict = True elif other not in potential: potential[other] = proposed pending.append(other) unseen -= component if conflict: continue grounded = [key for key in component if instances[key].get("grounded") is True] if len(grounded) > 1: translations = [instances[key]["position_mm"][axis] - potential[key] for key in grounded] if max(translations) - min(translations) > 0.01: errors.append({ "code": "grounded_constraint_conflict", "path": "/instances", "detail": f"多个固定实例在{('xyz'[axis]).upper()}轴与接触约束冲突", "repair_owner": "ai", "axis": "xyz"[axis], "instance_ids": sorted(grounded), }) continue anchor = grounded[0] if grounded else min(component, key=lambda key: instances[key]["position_mm"][axis]) translation = instances[anchor]["position_mm"][axis] - potential[anchor] for key in component: old = instances[key]["position_mm"][axis] new = potential[key] + translation if abs(old - new) > 0.01: instances[key]["position_mm"][axis] = new adjustments.append({ "instance_id": key, "axis": "xyz"[axis], "from_mm": old, "to_mm": new, "reason": "orthogonal_contact_normal_constraint", }) return adjustments, errors def _aabb_contact(a: dict, b: dict, *, tolerance_mm: float = 0.6) -> dict | None: """Return one real boundary contact with positive tangential area.""" matches = [] for axis in range(3): a_min = a["position_mm"][axis] a_max = a_min + a["size_mm"][axis] b_min = b["position_mm"][axis] b_max = b_min + b["size_mm"][axis] if abs(a_max - b_min) <= tolerance_mm: matches.append((axis, "+", "-", abs(a_max - b_min))) if abs(b_max - a_min) <= tolerance_mm: matches.append((axis, "-", "+", abs(b_max - a_min))) for axis, sign_a, sign_b, gap in matches: overlaps = [ min(a["position_mm"][other] + a["size_mm"][other], b["position_mm"][other] + b["size_mm"][other]) - max(a["position_mm"][other], b["position_mm"][other]) for other in range(3) if other != axis ] if min(overlaps) <= 0.01: continue return { "a": a["id"], "b": b["id"], "face_a": f"{'xyz'[axis]}{sign_a}", "face_b": f"{'xyz'[axis]}{sign_b}", "gap_mm": gap, "overlap_mm": overlaps, "source": "aabb_positive_area_contact", } return None def _aabb_interference(a: dict, b: dict, *, tolerance_mm: float = 0.6) -> dict | None: """Return a real solid-volume collision, ignoring face contact tolerance. The graph compiler currently models every instance as a closed solid box. Two boxes may share a face, edge or point, but they may not occupy the same volume. This catches shelves cutting through side panels and drawer walls accidentally placed on top of one another. """ overlaps = [ min(a["position_mm"][axis] + a["size_mm"][axis], b["position_mm"][axis] + b["size_mm"][axis]) - max(a["position_mm"][axis], b["position_mm"][axis]) for axis in range(3) ] if min(overlaps) <= tolerance_mm: return None return { "a": a["id"], "b": b["id"], "overlap_mm": overlaps, "overlap_volume_mm3": overlaps[0] * overlaps[1] * overlaps[2], "source": "aabb_positive_volume_intersection", } def _axis_packing_feasibility( moving: dict, instance_rows: list[dict], bounds: list[list[float]], *, axes: list[int] | None = None, ignored_pairs: set[frozenset] | None = None, tolerance_mm: float = 0.6, ) -> list[dict]: """Find every collision-free lower-coordinate interval on selected axes. The other two coordinates remain fixed. This is a diagnostic, not an automatic layout solver: it proves when repeated front/back (or any-axis) shuffling cannot fit a solid inside its permitted envelope. The caller still decides whether the source dimension, envelope semantics, or an explicit machining feature must change. """ ignored_pairs = ignored_pairs or set() selected_axes = list(range(3)) if axes is None else list(axes) answer = [] for axis in selected_axes: axis_name = "xyz"[axis] size = float(moving["size_mm"][axis]) domain_low = float(bounds[axis][0]) domain_high = float(bounds[axis][1]) - size blockers = [] forbidden = [] if domain_high >= domain_low - tolerance_mm: for other in instance_rows: if other["id"] == moving["id"]: continue if frozenset((moving["id"], other["id"])) in ignored_pairs: continue tangent_axes = [value for value in range(3) if value != axis] tangent_overlap = [ min( float(moving["position_mm"][value]) + float(moving["size_mm"][value]), float(other["position_mm"][value]) + float(other["size_mm"][value]), ) - max( float(moving["position_mm"][value]), float(other["position_mm"][value]), ) for value in tangent_axes ] if min(tangent_overlap) <= tolerance_mm: continue other_low = float(other["position_mm"][axis]) other_high = other_low + float(other["size_mm"][axis]) # Positive-volume overlap occurs strictly inside this interval # of possible lower coordinates for ``moving``. forbidden_low = other_low + tolerance_mm - size forbidden_high = other_high - tolerance_mm if forbidden_high <= domain_low or forbidden_low >= domain_high: continue clipped = [ max(domain_low, forbidden_low), min(domain_high, forbidden_high), ] forbidden.append(clipped) blockers.append({ "instance_id": other["id"], "forbidden_lower_position_mm": [round(value, 6) for value in clipped], "tangent_overlap_mm": [round(float(value), 6) for value in tangent_overlap], }) feasible = [[domain_low, domain_high]] if domain_high >= domain_low else [] for forbidden_low, forbidden_high in sorted(forbidden): remaining = [] for segment_low, segment_high in feasible: if forbidden_high <= segment_low or forbidden_low >= segment_high: remaining.append([segment_low, segment_high]) continue if forbidden_low > segment_low: remaining.append([segment_low, min(segment_high, forbidden_low)]) if forbidden_high < segment_high: remaining.append([max(segment_low, forbidden_high), segment_high]) feasible = remaining answer.append({ "moving_instance_id": moving["id"], "axis": axis_name, "allowed_lower_position_domain_mm": [round(domain_low, 6), round(domain_high, 6)], "feasible_lower_position_intervals_mm": [ [round(value, 6) for value in interval] for interval in feasible ], "blocking_instances": blockers, "feasible_without_dimension_or_bounds_change": bool(feasible), "source": "one_axis_exact_aabb_interval_subtraction", }) return answer def _derive_aabb_contacts(instances: dict, *, tolerance_mm: float = 0.6) -> list[dict]: """Derive real positive-area contacts inside each declared assembly. This replaces an unreliable language-model task: the AI declares which physical instance belongs to which assembly, while geometry determines what actually touches. Cross-assembly motion (drawer/door to carcass) remains explicit because a clearance joint should not geometrically touch. """ contacts = [] rows = list(instances.values()) for left_index, a in enumerate(rows): for b in rows[left_index + 1:]: if a.get("assembly_id") != b.get("assembly_id"): continue contact = _aabb_contact(a, b, tolerance_mm=tolerance_mm) if contact: contact_axis = contact["face_a"][0] a_broad_face = a.get("thickness_axis") == contact_axis b_broad_face = b.get("thickness_axis") == contact_axis interface_type = ( "face_to_face" if a_broad_face and b_broad_face else "edge_to_face" if a_broad_face or b_broad_face else "edge_to_edge" ) contacts.append({ **contact, "id": f"derived:{a['id']}:{b['id']}:{contact['face_a'][0]}", "kind": "geometric_contact", "assembly_id": a.get("assembly_id"), "interface_type": interface_type, "broad_face_instance_id": ( a["id"] if a_broad_face and not b_broad_face else b["id"] if b_broad_face and not a_broad_face else None ), "edge_instance_id": ( b["id"] if a_broad_face and not b_broad_face else a["id"] if b_broad_face and not a_broad_face else None ), }) 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, *, resolver: ParameterResolver, envelope_mm: list[float], assembly_bounds: dict | None = None, ) -> tuple[dict, list[dict]]: """Resolve declared design intent into coordinates. The model names an anchor relationship (for example, the right side's maximum X is aligned to the product envelope's maximum X). It never gets to guess the resulting coordinate. References form a dependency graph; cycles and missing axes are rejected instead of being numerically nudged. """ resolved = {} audit = [] visiting = set() anchors = {"min": 0.0, "center": 0.5, "max": 1.0} def length(value): return resolver.length(0 if value is None else value) def solve(instance_id): if instance_id in resolved: return resolved[instance_id] if instance_id in visiting: raise GraphError(f"实例定位依赖成环: {instance_id}") if instance_id not in instance_specs: raise GraphError(f"定位引用了不存在的实例: {instance_id}") visiting.add(instance_id) try: spec = instance_specs[instance_id] placement = spec.get("placement") if not isinstance(placement, dict) or set(placement) != {"x", "y", "z"}: raise GraphError(f"实例 {instance_id} 必须分别声明X/Y/Z三轴定位约束") position = [] for axis, axis_name in enumerate("xyz"): rule = placement[axis_name] if not isinstance(rule, dict): raise GraphError(f"实例 {instance_id} 的{axis_name.upper()}轴定位无效") mode = rule.get("mode") if mode == "absolute": coordinate = length(rule.get("value_mm")) target = "absolute" target_kind = "absolute" elif mode == "align": self_anchor = rule.get("self_anchor") target_anchor = rule.get("target_anchor") if self_anchor not in anchors or target_anchor not in anchors: raise GraphError(f"实例 {instance_id} 的{axis_name.upper()}轴锚点必须为min/center/max") target_id = rule.get("target") if target_id == "envelope": target_position, target_size = 0.0, envelope_mm[axis] target_kind = "envelope" elif target_id in {"assembly", spec.get("assembly_id")}: # A model naturally refers to its declared assembly by # stable ID (for example ``carcass`` or ``drawer-1``). # Treat only the instance's *own* assembly ID as the # explicit ``assembly`` alias. Other assembly IDs do # not silently become cross-assembly coordinates. bounds = (assembly_bounds or {}).get(spec.get("assembly_id")) if not bounds: raise GraphError(f"实例 {instance_id} 引用了没有有效边界的所属总成") target_position = bounds[axis][0] target_size = bounds[axis][1] - bounds[axis][0] target_kind = "assembly" else: target_instance = solve(target_id) target_position = target_instance["position_mm"][axis] target_size = target_instance["size_mm"][axis] target_kind = "instance" coordinate = ( target_position + anchors[target_anchor] * target_size - anchors[self_anchor] * spec["size_mm"][axis] + length(rule.get("offset_mm")) ) target = target_id else: raise GraphError(f"实例 {instance_id} 的{axis_name.upper()}轴只支持absolute/align定位") if not math.isfinite(coordinate): raise GraphError(f"实例 {instance_id} 的{axis_name.upper()}轴定位不是有限数") position.append(coordinate) audit.append({ "instance_id": instance_id, "axis": axis_name, "coordinate_mm": coordinate, "mode": mode, "target": target, "target_kind": target_kind, "source": "declared_anchor_constraint", }) resolved[instance_id] = {**spec, "position_mm": position} return resolved[instance_id] finally: visiting.remove(instance_id) for key in instance_specs: solve(key) return resolved, audit def compile_graph( graph: dict, *, source_image_ids: set | None = None, user_evidence: dict | None = None, dimension_ledger: dict | None = None, dimension_scope_adjudications: dict | None = None, topology_contract: dict | None = None, ) -> dict: """Resolve a graph without repairing it by category-specific heuristics.""" issues = [] def issue(code, path, detail, **context): issues.append({"code": code, "path": path, "detail": detail, "repair_owner": "ai", **context}) if graph.get("schema_version") != SCHEMA_VERSION: issue("schema_version", "/schema_version", "结构图谱版本不匹配") orientation_solver = graph.get("orientation_solver") if orientation_solver not in {None, ROLE_AXIS_SOLVER}: issue( "orientation_solver", "/orientation_solver", f"朝向求解器只支持{ROLE_AXIS_SOLVER}；未知模式不能用于制造图", actual=orientation_solver, expected=ROLE_AXIS_SOLVER, ) role_axis_mode = orientation_solver == ROLE_AXIS_SOLVER orientation_audit = [] observations = {} source_image_ids = source_image_ids or set() user_evidence = user_evidence or {} for index, row in enumerate(graph.get("observations") or []): key = row.get("id") if not key or key in observations or row.get("source_type") not in {"image", "user"}: issue("evidence_identity", f"/observations/{index}", "证据ID缺失、重复或来源不明确") continue if row.get("source_type") == "image" and row.get("image_id") not in source_image_ids: issue("evidence_image", f"/observations/{index}", "图像证据未引用本轮实际提供的原图") continue if row.get("source_type") == "user" and user_evidence.get(key) != row: issue("evidence_user", f"/observations/{index}", "用户确认必须来自会话事实库，不能由模型自行生成") continue observations[key] = row resolver = ParameterResolver(graph.get("parameters") or {}, observations) for key in resolver.parameters: try: resolver.parameter(key) except (GraphError, TypeError, AttributeError) as exc: issue("parameter", f"/parameters/{key}", str(exc)) envelope = [] try: envelope = [resolver.length(value) for value in graph["envelope_mm"]] if len(envelope) != 3 or min(envelope) <= 0: raise GraphError("整体包围尺寸必须有三个正值") except (GraphError, KeyError, TypeError) as exc: issue("envelope", "/envelope_mm", str(exc)) reference_bounds = {} if graph.get("reference_bounds_mm") is not None: try: raw_reference_bounds = graph["reference_bounds_mm"] if not isinstance(raw_reference_bounds, dict) or set(raw_reference_bounds) != {"x", "y", "z"}: raise GraphError("参考包络必须分别声明X/Y/Z最小与最大值") for axis, axis_name in enumerate("xyz"): values = [resolver.length(value) for value in raw_reference_bounds[axis_name]] if len(values) != 2 or values[1] <= values[0]: raise GraphError(f"参考包络{axis_name.upper()}轴上限必须大于下限") if envelope and (values[0] < -0.5 or values[1] > envelope[axis] + 0.5): raise GraphError(f"参考包络{axis_name.upper()}轴必须位于实际占用包络内") reference_bounds[axis_name] = values except (GraphError, KeyError, TypeError) as exc: issue("reference_bounds", "/reference_bounds_mm", str(exc)) parts = {} panel_roles = { "top", "bottom", "side", "shelf", "door", "back", "drawer_front", "drawer_left", "drawer_right", "drawer_back", "drawer_bottom", "drawer_support", "divider", } allowed_roles = panel_roles | {"leg", "rail", "stile", "frame_member", "hardware", "other"} allowed_material_families = { "engineered_panel", "solid_wood", "metal", "plastic", "glass", "other", } for index, part in enumerate(graph.get("parts") or []): key = part.get("id") if not key or key in parts: issue("part_identity", f"/parts/{index}", "零件ID缺失或重复") continue try: size = [resolver.length(value) for value in part["size_mm"]] if len(size) != 3 or min(size) <= 0: raise GraphError("零件三轴尺寸必须大于0") if part.get("shape") != "box": raise GraphError("当前编译器不支持此几何，不得默默替换成长方体") role = part.get("functional_role") if role not in allowed_roles: raise GraphError("零件必须声明稳定的functional_role，不能由中文名称反推结构") material_family = part.get("material_family") if material_family not in {None, "", *allowed_material_families}: raise GraphError("零件material_family不在受支持范围") thickness_axis = part.get("thickness_axis") if role in panel_roles: if thickness_axis not in {"x", "y", "z"}: raise GraphError( "板件必须明确厚度轴thickness_axis；新图谱使用产品X宽/Y深/Z高坐标" if role_axis_mode else "板件必须明确本地厚度轴thickness_axis" ) thickness_index = "xyz".index(thickness_axis) if size[thickness_index] > min(size) + 0.6: raise GraphError("声明的厚度轴不是此板件最薄轴") elif thickness_axis not in {None, "", "x", "y", "z"}: raise GraphError("非板件厚度轴只能为空或x/y/z") parts[key] = {**part, "size_mm": size} if role_axis_mode and role in ROLE_PRODUCT_THICKNESS_AXES: expected_axes = sorted(ROLE_PRODUCT_THICKNESS_AXES[role]) accepted = thickness_axis in ROLE_PRODUCT_THICKNESS_AXES[role] orientation_audit.append({ "part_id": key, "functional_role": role, "product_thickness_axis": thickness_axis, "expected_product_thickness_axes": expected_axes, "accepted": accepted, }) if not accepted: issue( "role_axis_template", f"/parts/{key}/thickness_axis", "板件角色与产品坐标厚度轴矛盾；必须修正产品X/Y/Z尺寸和厚度轴，禁止通过实例二次旋转掩盖", part_id=key, functional_role=role, actual=thickness_axis, expected=expected_axes, ) except (GraphError, KeyError, TypeError) as exc: issue("part_geometry", f"/parts/{index}", str(exc)) assemblies = {} assembly_bounds = {} allowed_assembly_roles = {"carcass", "drawer", "door", "module", "frame", "hardware_group", "other"} if graph.get("assemblies") is not None: for index, assembly in enumerate(graph.get("assemblies") or []): path = f"/assemblies/{index}" key = assembly.get("id") if isinstance(assembly, dict) else None if not key or key in assemblies: issue("assembly_identity", path, "总成ID缺失或重复") continue if assembly.get("role") not in allowed_assembly_roles: issue( "assembly_role", f"{path}/role", "总成必须使用固定结构角色枚举；例如门总成的role是door，不是door_assembly", assembly_id=key, actual=assembly.get("role"), allowed=sorted(allowed_assembly_roles), legal_repair={"op": "replace", "path": f"/assemblies/{key}/role", "value_shape": sorted(allowed_assembly_roles)}, ) continue refs = assembly.get("evidence_ids") or [] if not refs or any(ref not in observations for ref in refs): issue("assembly_evidence", path, "总成划分必须引用本轮原图或真实用户证据") continue try: raw_bounds = assembly["bounds_mm"] if not isinstance(raw_bounds, dict) or set(raw_bounds) != {"x", "y", "z"}: raise GraphError("总成边界必须分别声明X/Y/Z最小与最大值") bounds = [] for axis, axis_name in enumerate("xyz"): values = [resolver.length(value) for value in raw_bounds[axis_name]] if len(values) != 2 or values[1] <= values[0]: raise GraphError(f"总成{axis_name.upper()}轴上限必须大于下限") if envelope and (values[0] < -0.5 or values[1] > envelope[axis] + 0.5): raise GraphError(f"总成{axis_name.upper()}轴边界超出产品外形") bounds.append(values) assembly_bounds[key] = bounds assemblies[key] = {**assembly, "bounds_mm": { axis_name: bounds[axis] for axis, axis_name in enumerate("xyz") }} except (GraphError, KeyError, TypeError) as exc: issue("assembly_bounds", path, str(exc)) instances = {} instance_specs = {} for index, inst in enumerate(graph.get("instances") or []): key = inst.get("id") try: if not key or key in instances or key in instance_specs or inst.get("part_id") not in parts or not inst.get("assembly_id"): raise GraphError("实例ID重复、零件不存在或未声明所属总成") # Legacy graphs may explicitly map local axes. New graphs have # already declared part dimensions in product X/Y/Z and therefore # cannot rotate them again at instance level. axis_order = inst.get("axis_order", [0, 1, 2]) if axis_order not in ([0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]): raise GraphError("只支持明确的正交轴映射，任意角度需后续几何编译器") if role_axis_mode and axis_order != [0, 1, 2]: issue( "role_axis_template", f"/instances/{key}/axis_order", "新图谱的零件尺寸已经是产品X宽/Y深/Z高，实例禁止再次交换坐标轴", instance_id=key, actual=axis_order, expected=[0, 1, 2], ) size = [parts[inst["part_id"]]["size_mm"][a] for a in axis_order] local_thickness_axis = parts[inst["part_id"]].get("thickness_axis") product_thickness_axis = None if local_thickness_axis in {"x", "y", "z"}: product_thickness_axis = ( local_thickness_axis if role_axis_mode else "xyz"[axis_order.index("xyz".index(local_thickness_axis))] ) inherited = { "functional_role": parts[inst["part_id"]].get("functional_role"), "thickness_axis": product_thickness_axis, "material_family": parts[inst["part_id"]].get("material_family"), } if graph.get("placement_solver") == "anchored_axes_v1": instance_specs[key] = {**inst, **inherited, "size_mm": size} else: origin = [resolver.length(value) for value in inst["position_mm"]] if len(origin) != 3: raise GraphError("实例位置必须有三轴") instances[key] = {**inst, **inherited, "position_mm": origin, "size_mm": size} except (GraphError, KeyError, TypeError) as exc: issue("instance_geometry", f"/instances/{index}", str(exc)) placement_audit = [] if graph.get("placement_solver") == "anchored_axes_v1" and envelope: try: instances, placement_audit = _solve_anchored_axis_placements( instance_specs, resolver=resolver, envelope_mm=envelope, assembly_bounds=assembly_bounds, ) 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 } ignored_interference_pairs = { pair for pair, row in valid_mount_pairs.items() if row.get("allows_host_interference") } 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"): low = inst["position_mm"][axis] high = low + inst["size_mm"][axis] if low < bounds[axis][0] - 0.5 or high > bounds[axis][1] + 0.5: outside_axes.append({ "axis": axis_name, "instance_range_mm": [low, high], "assembly_range_mm": bounds[axis], }) if outside_axes: legal_alignment_replacements = [] for row in outside_axes: axis_name = row["axis"] legal_alignment_replacements.extend([ { "op": "replace", "path": f"/instances/{key}/placement/{axis_name}", "value": {"mode": "align", "self_anchor": "min", "target": inst["assembly_id"], "target_anchor": "min", "offset_mm": 0}, }, { "op": "replace", "path": f"/instances/{key}/placement/{axis_name}", "value": {"mode": "align", "self_anchor": "max", "target": inst["assembly_id"], "target_anchor": "max", "offset_mm": 0}, }, ]) issue( "outside_assembly_bounds", f"/instances/{key}", f"部件超出所属总成{inst['assembly_id']}的局部边界；不能借用整件产品坐标掩盖模块错位", instance_id=key, assembly_id=inst["assembly_id"], axes=outside_axes, current_placement=instance_specs.get(key, {}).get("placement"), legal_alignment_replacements=legal_alignment_replacements, correction=( "先核对零件产品轴尺寸，再选择与所属总成min或max面一致的零偏移锚点；" "没有独立图像证据不得给多个部件统一追加经验偏移。" ), ) used = Counter(inst.get("assembly_id") for inst in instances.values()) for key in assemblies: if not used[key]: issue("empty_assembly", f"/assemblies/{key}", "空总成不能作为装配关系的虚拟占位") continue 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 tight_rows), max(row["position_mm"][axis] + row["size_mm"][axis] for row in tight_rows)] for axis in range(3) ] mismatch = [] for axis, axis_name in enumerate("xyz"): declared = assembly_bounds[key][axis] if abs(actual[axis][0] - declared[0]) > 0.6 or abs(actual[axis][1] - declared[1]) > 0.6: mismatch.append({ "axis": axis_name, "declared_mm": declared, "actual_member_union_mm": actual[axis], }) 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": solver_adjustments, solver_errors = _solve_orthogonal_contact_normals( instances, list(graph.get("connections") or []) ) issues.extend(solver_errors) quantities = Counter(inst["part_id"] for inst in instances.values()) for key, part in parts.items(): if not quantities[key]: issue("uninstantiated_part", f"/parts/{key}", "零件没有物理实例，不能计入BOM") expected = part.get("observed_quantity") if expected is not None and expected != quantities[key]: issue("quantity_conflict", f"/parts/{key}", f"图像数量{expected}与实例数量{quantities[key]}不符") role_quantities = Counter( str(inst.get("functional_role") or "") for inst in instances.values() if str(inst.get("functional_role") or "") ) role_contract = (topology_contract or {}).get("role_counts") or {} if isinstance(role_contract, dict): for role, expected in role_contract.items(): if ( not isinstance(role, str) or isinstance(expected, bool) or not isinstance(expected, int) or expected < 0 ): issue("source_topology_contract", "/topology_contract/role_counts", "原图拓扑契约数量格式无效") continue actual = role_quantities.get(role, 0) if actual != expected: issue( "source_topology_count_conflict", f"/topology_contract/role_counts/{role}", f"独立多视图证据计数{role}={expected}，结构图谱生成{actual}；禁止静默漏件或多件", role=role, expected=expected, actual=actual, evidence=((topology_contract or {}).get("evidence") or {}).get(role) or [], ) 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") if not key or key in seen_connections: issue("connection_identity", path, "连接ID缺失或重复") continue seen_connections.add(key) a, b = instances.get(edge.get("a")), instances.get(edge.get("b")) if not a or not b or a["id"] == b["id"]: issue("connection_target", path, "连接两端必须是不同且存在的物理实例") continue if edge.get("kind") not in {"fixed", "sliding", "hinged", "support"}: issue("connection_kind", path, "连接方式不明确") continue neighbours[a["id"]].add(b["id"]) neighbours[b["id"]].add(a["id"]) if edge["kind"] not in {"fixed", "support"}: continue # Motion/hardware templates require a later dedicated validator. faces = [edge.get("face_a"), edge.get("face_b")] if any(face not in {"x-", "x+", "y-", "y+", "z-", "z+"} for face in faces): issue("connection_face", path, "固定连接必须指定两个接触面") continue if faces[0][0] != faces[1][0] or faces[0][1] == faces[1][1]: issue("connection_face", path, "接触面法向必须相反") continue axis = "xyz".index(faces[0][0]) plane_a = a["position_mm"][axis] + (a["size_mm"][axis] if faces[0][1] == "+" else 0) plane_b = b["position_mm"][axis] + (b["size_mm"][axis] if faces[1][1] == "+" else 0) tolerance = float(edge.get("tolerance_mm", 0.5)) if not math.isfinite(tolerance) or not 0 <= tolerance <= 2: issue("connection_tolerance", path, "自动几何接触检查公差必须为0–2mm，不能用大公差掩盖错位") continue overlap = [min(a["position_mm"][d] + a["size_mm"][d], b["position_mm"][d] + b["size_mm"][d]) - max(a["position_mm"][d], b["position_mm"][d]) for d in range(3) if d != axis] if abs(plane_a - plane_b) > tolerance or min(overlap) <= 0: required_b_origin = plane_a - (b["size_mm"][axis] if faces[1][1] == "+" else 0.0) repair_hint = ( "两平面已重合但切向没有正面积交叠；优先回看板件是端面对侧面还是上下叠放，并修改face_a/face_b，不能移动到产品外" if abs(plane_a - plane_b) <= tolerance and min(overlap) <= 0 else "接触法向坐标错误；核对连接面后再修改实例位置或约束" ) issue( "connection_gap", path, f"{a['id']}的{faces[0]}与{b['id']}的{faces[1]}未接触；法向间距{abs(plane_a-plane_b):.3f}mm，切向重叠{overlap}mm", connection_id=key, a=a["id"], b=b["id"], face_a=faces[0], face_b=faces[1], axis="xyz"[axis], plane_a_mm=plane_a, plane_b_mm=plane_b, current_b_origin_mm=b["position_mm"][axis], required_b_origin_mm=required_b_origin, suggested_repair=repair_hint, ) derived_contacts = [] if graph.get("contact_derivation") == "aabb_within_assembly_v1": derived_contacts = _derive_aabb_contacts(instances) for edge in derived_contacts: neighbours[edge["a"]].add(edge["b"]) neighbours[edge["b"]].add(edge["a"]) assembly_connection_audit = [] if assemblies: members = { assembly_id: [inst for inst in instances.values() if inst.get("assembly_id") == assembly_id] for assembly_id in assemblies } # A relation between assemblies cannot hide loose parts inside either # assembly. Every multi-part assembly must first be a real contact # component on its own. internal_neighbours = {key: set() for key in instances} 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 member_ids = {row["id"] for row in rows} reached, pending = set(), [rows[0]["id"]] while pending: current = pending.pop() if current in reached: continue reached.add(current) pending.extend((internal_neighbours[current] & member_ids) - reached) for instance_id in sorted(member_ids - reached): issue( "assembly_internal_disconnected", f"/instances/{instance_id}", f"部件在总成{assembly_id}内没有真实接触路径，总成运动副不能掩盖散落零件", assembly_id=assembly_id, ) seen_assembly_connections = set() assembly_neighbours = {key: set() for key in assemblies} for index, edge in enumerate(graph.get("assembly_connections") or []): path = f"/assembly_connections/{index}" key = edge.get("id") if isinstance(edge, dict) else None if not key or key in seen_assembly_connections: issue("assembly_connection_identity", path, "总成连接ID缺失或重复") continue seen_assembly_connections.add(key) a_id, b_id = edge.get("a"), edge.get("b") instance_endpoints = { endpoint: instances[endpoint].get("assembly_id") for endpoint in (a_id, b_id) if endpoint in instances } if instance_endpoints: moving_hint = None if a_id in instances: current_assembly = instances[a_id].get("assembly_id") moving_hint = { "instance_id": a_id, "current_assembly_id": current_assembly, "required_action": ( "若该实例是运动件，为它建立边界等于实体包围盒的独立总成，" "把实例assembly_id迁入新总成，并让连接a引用新总成ID" ), } issue( "assembly_connection_uses_instance", path, "总成连接错误地引用了零件实例；运动副端点必须是总成ID", instance_endpoint_to_current_assembly=instance_endpoints, moving_endpoint_repair=moving_hint, stationary_endpoint_repair=( f"连接b应引用固定端实例所属总成{instances[b_id].get('assembly_id')}" if b_id in instances else None ), ) continue if a_id not in assemblies or b_id not in assemblies or a_id == b_id: issue("assembly_connection_target", path, "总成连接两端必须是不同且已声明的总成") continue if not members.get(a_id) or not members.get(b_id): issue("assembly_connection_target", path, "总成连接不能引用空总成") continue kind = edge.get("kind") if kind not in {"fixed", "sliding", "hinged"}: issue("assembly_connection_kind", path, "总成关系只能是fixed/sliding/hinged") continue refs = edge.get("evidence_ids") or [] if not refs or any(ref not in observations for ref in refs): issue("assembly_connection_evidence", path, "总成关系必须引用真实证据") continue contact = None if kind == "fixed": for a_inst in members[a_id]: for b_inst in members[b_id]: contact = _aabb_contact(a_inst, b_inst) if contact: break if contact: break if not contact: issue( "fixed_assemblies_not_contacting", path, f"固定总成{a_id}与{b_id}之间没有任何真实正面积接触", ) continue else: if edge.get("axis") not in {"x", "y", "z"}: issue("assembly_motion_axis", path, "滑动/铰接总成必须明确声明运动轴") continue if kind == "sliding": limits = edge.get("travel_limits_mm") try: values = [resolver.length(value) for value in limits] if len(values) != 2 or values[1] <= values[0]: raise GraphError("滑动行程上限必须大于下限") except (GraphError, TypeError) as exc: issue("assembly_motion_limits", path, f"滑动总成需要两个有效行程限位: {exc}") continue else: limits = edge.get("angle_limits_deg") if ( not isinstance(limits, list) or len(limits) != 2 or any(isinstance(value, bool) or not isinstance(value, (int, float)) for value in limits) or limits[1] <= limits[0] ): issue("assembly_motion_limits", path, "铰接总成需要两个有效角度限位") continue if kind == "sliding": motion_axis = "xyz".index(edge["axis"]) moving_bounds = assembly_bounds[a_id] guide_bounds = assembly_bounds[b_id] clearance_errors = [] for axis, axis_name in enumerate("xyz"): if axis == motion_axis: continue if ( moving_bounds[axis][0] < guide_bounds[axis][0] - 0.5 or moving_bounds[axis][1] > guide_bounds[axis][1] + 0.5 ): clearance_errors.append({ "axis": axis_name, "moving_range_mm": moving_bounds[axis], "guide_range_mm": guide_bounds[axis], }) if clearance_errors: issue( "sliding_assembly_outside_guide", path, f"滑动总成{a_id}在非运动轴上超出导向总成{b_id}，无法形成科学的抽拉关系", axes=clearance_errors, ) continue a_rep, b_rep = members[a_id][0], members[b_id][0] neighbours[a_rep["id"]].add(b_rep["id"]) neighbours[b_rep["id"]].add(a_rep["id"]) assembly_neighbours[a_id].add(b_id) assembly_neighbours[b_id].add(a_id) assembly_connection_audit.append({ "id": key, "kind": kind, "a": a_id, "b": b_id, "axis": edge.get("axis"), "verified_contact": contact, }) if len(assemblies) > 1: reached, pending = set(), [next(iter(assemblies))] while pending: key = pending.pop() if key not in reached: reached.add(key) pending.extend(assembly_neighbours[key] - reached) for key in assemblies.keys() - reached: issue("disconnected_assembly", f"/assemblies/{key}", "此总成没有连接到产品的装配关系图") if len(instances) > 1: reached, pending = set(), [next(iter(instances))] while pending: key = pending.pop() if key not in reached: reached.add(key) pending.extend(neighbours[key] - reached) for key in instances.keys() - reached: issue("disconnected_instance", f"/instances/{key}", "此部件没有可追溯到产品整体的连接路径；不能靠省略连接关系跳过几何检查") 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 frozenset((moving_id, other["id"])) not in ignored_interference_pairs 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 minimum_overlap = min(float(value) for value in collision["overlap_mm"]) minimum_axes = [ axis for axis, value in enumerate(collision["overlap_mm"]) if abs(float(value) - minimum_overlap) <= 1e-6 ] packing_feasibility = [] for moving in (a, b): assembly_id = moving.get("assembly_id") structural_rows = structural_members_by_assembly.get(assembly_id) or [] permitted_bounds = assembly_bounds.get(assembly_id) if ( envelope and len(structural_rows) == 1 and structural_rows[0]["id"] == moving["id"] ): permitted_bounds = [[0.0, float(value)] for value in envelope] if not permitted_bounds and envelope: permitted_bounds = [[0.0, float(value)] for value in envelope] if permitted_bounds: packing_feasibility.extend(_axis_packing_feasibility( moving, instance_rows, permitted_bounds, axes=minimum_axes, ignored_pairs=ignored_interference_pairs, )) minimum_axis_infeasible = bool(packing_feasibility) and all( not row["feasible_without_dimension_or_bounds_change"] for row in packing_feasibility ) issue( "solid_interference", f"/instances/{b['id']}", f"实体{a['id']}与{b['id']}发生正体积穿插，不能作为可制造装配", compatible_separation_candidates=collision_separation_candidates(a, b), minimum_separation_axes=["xyz"[axis] for axis in minimum_axes], minimum_axis_packing_feasibility=packing_feasibility, minimum_separation_axis_infeasible=minimum_axis_infeasible, repair_scope=( "source_dimension_or_envelope_semantics_or_explicit_machining_feature" if minimum_axis_infeasible else "placement_or_dimension" ), correction=( ( "最小分离轴在所属边界内没有任何无碰撞位置；禁止在前后障碍之间循环平移。" "必须回看原图核对外形尺寸口径、覆盖/内嵌关系、零件尺寸，或显式建模槽口/避让；" "若高置信原图尺寸互相矛盾，必须输出单一聚焦的尺寸确认问题，不得臆造修改。" if minimum_axis_infeasible else "必须回看原图选择compatible_separation_candidates中与真实结构一致的最小位移；" "候选只保证几何可行，不代替图像证据。" ) ), **collision, ) for key, inst in instances.items(): signature = tuple(round(value, 6) for value in inst["position_mm"] + inst["size_mm"]) if signature in seen_boxes: issue("duplicate_solid", f"/instances/{key}", f"与{seen_boxes[signature]}占据同一实体，禁止重复计费") 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}", "部件超出产品边界；不允许自动裁小来隐藏冲突") # Directly read overall arrows are external evidence, not an AI naming # convention. Compare the compiled X/Y/Z envelope to that ledger so a # model cannot silently turn a 300 mm side-view depth into product width. # Conflicted or low-confidence labels are intentionally omitted by the # upstream ledger builder and therefore remain AI-estimated draft values. dimension_scope_adjudications = dimension_scope_adjudications or {} dimension_scope_audit = [] scope_overrides = graph.get("dimension_scope_overrides") or {} if not isinstance(scope_overrides, dict): issue("dimension_scope_override", "/dimension_scope_overrides", "尺寸口径覆盖必须按X/Y/Z轴声明") scope_overrides = {} unknown_override_axes = sorted(set(scope_overrides) - set("xyz")) if unknown_override_axes: issue( "dimension_scope_override", "/dimension_scope_overrides", "尺寸口径覆盖只允许X/Y/Z轴", unknown_axes=unknown_override_axes, ) if len(envelope) == 3: for axis_index, axis in enumerate("xyz"): fact = (dimension_ledger or {}).get(axis) if not isinstance(fact, dict) or not isinstance(fact.get("value_mm"), (int, float)): continue comparison_mm = float(envelope[axis_index]) comparison_scope = "occupied_overall" override = scope_overrides.get(axis) if override is not None: path = f"/dimension_scope_overrides/{axis}" if not isinstance(override, dict) or override.get("ledger_scope") != "reference": issue( "dimension_scope_override", path, "口径覆盖目前只支持把原图整体尺寸改判为参考包络", ) elif axis not in reference_bounds: issue("dimension_scope_override", path, "引用参考口径前必须完整声明reference_bounds_mm") else: evidence_id = str(override.get("evidence_id") or "") adjudication_id = str(override.get("adjudication_id") or "") user_verified = bool( evidence_id and evidence_id in observations and observations[evidence_id].get("source_type") == "user" ) adjudication = dimension_scope_adjudications.get(adjudication_id) external_verified = bool( adjudication_id and isinstance(adjudication, dict) and adjudication.get("axis") == axis and adjudication.get("ledger_scope") == "reference" and isinstance(adjudication.get("confidence"), (int, float)) and not isinstance(adjudication.get("confidence"), bool) and float(adjudication["confidence"]) >= 0.8 and set(adjudication.get("source_image_ids") or []).issubset(source_image_ids) and bool(adjudication.get("source_image_ids")) ) if not user_verified and not external_verified: issue( "dimension_scope_override_unverified", path, "把原图整体尺寸改判为柜体/开口参考尺寸必须来自独立AI端点裁决或真实用户确认，图谱模型不能自行改口径", evidence_id=evidence_id or None, adjudication_id=adjudication_id or None, ) else: comparison_mm = float(reference_bounds[axis][1] - reference_bounds[axis][0]) comparison_scope = "reference_bounds" dimension_scope_audit.append({ "axis": axis, "ledger_scope": "reference", "comparison_mm": comparison_mm, "verification_source": "user" if user_verified else "independent_source_adjudication", "evidence_id": evidence_id or None, "adjudication_id": adjudication_id or None, }) expected = float(fact["value_mm"]) tolerance = max(1.0, float(fact.get("tolerance_mm") or 1.0)) if abs(comparison_mm - expected) > tolerance: issue( "overall_dimension_conflict", f"/envelope_mm/{axis_index}", f"CAD{comparison_scope}的{axis.upper()}轴{comparison_mm:.3f}mm与原图尺寸箭头{expected:.3f}mm冲突", axis=axis, comparison_scope=comparison_scope, compiled_mm=comparison_mm, source_mm=expected, source_image_ids=list(fact.get("source_image_ids") or []), ) return { "schema_version": SCHEMA_VERSION, "graph_sha256": hashlib.sha256(json.dumps(graph, ensure_ascii=False, sort_keys=True).encode()).hexdigest(), "ok": not issues, "issues": issues, "release_status": "drawing_draft", "visual_validation_required": True, "manufacturing_validation_required": True, "parameters": {key: {"value": value, "length_exponent": dimension, "confidence": confidence, "estimated": estimated} for key, (value, dimension, confidence, estimated) in resolver.resolved.items()}, "orientation_solver": orientation_solver, "orientation_audit": orientation_audit, "instances": list(instances.values()), "solver_adjustments": solver_adjustments, "placement_audit": placement_audit, "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 {}, "occupied_envelope_mm": envelope, "reference_bounds_mm": reference_bounds, "dimension_scope_audit": dimension_scope_audit, "topology_contract": topology_contract or {}, "bom": [{"part_id": key, "name": part.get("name"), "size_mm": part["size_mm"], "quantity": quantities[key], "functional_role": part.get("functional_role"), "material_family": part.get("material_family")} for key, part in parts.items() if quantities[key]], }