"""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" 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 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", "parts", "instances", "connections", "uncertainties", "constraint_solver", "contact_derivation", "placement_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) 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 _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 _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, 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", "结构图谱版本不匹配") 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)) 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") 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} except (GraphError, KeyError, TypeError) as exc: issue("part_geometry", f"/parts/{index}", str(exc)) assemblies = {} assembly_bounds = {} 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 {"carcass", "drawer", "door", "module", "frame", "hardware_group", "other"}: issue("assembly_role", path, "总成必须声明可审计的结构角色") 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重复、零件不存在或未声明所属总成") # Axis order is explicit. No board-name/longest-edge remapping. if inst.get("axis_order", [0, 1, 2]) not in ([0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]): raise GraphError("只支持明确的正交轴映射，任意角度需后续几何编译器") size = [parts[inst["part_id"]]["size_mm"][a] for a in inst.get("axis_order", [0, 1, 2])] 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 = "xyz"[ inst.get("axis_order", [0, 1, 2]).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)) 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 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: issue( "outside_assembly_bounds", f"/instances/{key}", f"部件超出所属总成{inst['assembly_id']}的局部边界；不能借用整件产品坐标掩盖模块错位", assembly_id=inst["assembly_id"], axes=outside_axes, ) 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 actual = [ [min(row["position_mm"][axis] for row in rows), max(row["position_mm"][axis] + row["size_mm"][axis] for row in 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, ) 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 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 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()) 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) issue( "solid_interference", f"/instances/{b['id']}", f"实体{a['id']}与{b['id']}发生正体积穿插，不能作为可制造装配", **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 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. 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 expected = float(fact["value_mm"]) tolerance = max(1.0, float(fact.get("tolerance_mm") or 1.0)) if abs(envelope[axis_index] - expected) > tolerance: issue( "overall_dimension_conflict", f"/envelope_mm/{axis_index}", f"CAD整体{axis.upper()}轴{envelope[axis_index]:.3f}mm与原图尺寸箭头{expected:.3f}mm冲突", axis=axis, compiled_mm=envelope[axis_index], 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()}, "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, "collision_audit": collision_pairs, "dimension_ledger": dimension_ledger or {}, "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]], }