909 lines
38 KiB
Python
909 lines
38 KiB
Python
from __future__ import annotations
|
||
|
||
"""Drawing understanding and engineering-review evidence.
|
||
|
||
This module consumes the stable inspect_dxf output and, when the source
|
||
file is available, re-opens the DXF for geometry and text clustering. It
|
||
never writes master data: every inferred field is a PENDING_REVIEW candidate
|
||
with confidence and evidence.
|
||
"""
|
||
|
||
import re
|
||
from datetime import UTC, datetime
|
||
from hashlib import sha256
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
UNDERSTANDING_VERSION = "drawing-understanding.v1"
|
||
LINK_EVIDENCE_VERSION = "drawing-link-evidence.v1"
|
||
DIFF_VERSION = "drawing-version-diff.v1"
|
||
|
||
_DRAWING_NAME_RE = re.compile(
|
||
r"^(?P<drawing_no>[A-Za-z0-9]+(?:-[A-Za-z0-9]+)+)-(?P<revision>[A-Za-z])(?:\(\d+\))?$"
|
||
)
|
||
_ITEM_REF_RE = re.compile(r"^(?:[A-Z]?\d{2,}[.]\d{2,}[A-Z]?|\d{6,}(?:-\d+)?)$", re.IGNORECASE)
|
||
_HEADER_TOKEN_RE = re.compile(
|
||
r"^(?:POS(?:ITION)?|NR|NO\.?|BENENNUNG|DESIGNATION|TITLE|SHEET|BLATT|"
|
||
r"MENGE|QTY|QUANTITY|ANZAHL|STUECK(?:LISTE)?|MATERIAL|WERKSTOFF|"
|
||
r"SACHNR|PART(?:S)?|TEILE)$",
|
||
re.IGNORECASE,
|
||
)
|
||
_POSITION_HEADER_RE = re.compile(r"^(?:POS(?:ITION)?|NR|NO\.?)$", re.IGNORECASE)
|
||
_QUANTITY_HEADER_RE = re.compile(r"^(?:MENGE|QTY|QUANTITY|ANZAHL)$", re.IGNORECASE)
|
||
_SHEET_HEADER_RE = re.compile(r"^(?:SHEET|BLATT)$", re.IGNORECASE)
|
||
_TABLE_LAYER_RE = re.compile(r"(?:LISTE|STUECK|BOM|POS|TAB|PARTS|TEILE)", re.IGNORECASE)
|
||
_TABLE_BLOCK_RE = re.compile(r"(?:POS|STUECK|BOM|LISTE|TAB|PARTS|TEILE)", re.IGNORECASE)
|
||
_FRAME_BLOCK_RE = re.compile(r"^(?:STD|RAHMEN|FRAME|BORDER)_?", re.IGNORECASE)
|
||
_TITLE_BLOCK_RE = re.compile(r"^(?:SF_STD|TITLE|TITEL|SCHRIFTFELD|ZEICHNUNGSRAHMEN)", re.IGNORECASE)
|
||
_THREAD_SPEC_RE = re.compile(r"\b(?:M|G)\d+(?:[xX*]\d+(?:\.\d+)?)?(?!\s*[:：]\s*\d)")
|
||
_MATERIAL_WORD_RE = re.compile(
|
||
r"\b(?:STAHL|STEEL|ALU(?:MINIUM)?|KUPFER|COPPER|EDELSTAHL|STAINLESS|MESSING|BRASS)\b",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
_TITLE_DRAWING_KEYS = (
|
||
"ZEICHNUNGSNUMMER",
|
||
"DRAWING_NO",
|
||
"DRAWINGNUMBER",
|
||
"DWG_NO",
|
||
"DRAWING_NUMBER",
|
||
)
|
||
_TITLE_REVISION_KEYS = ("B_FREI", "REVISION", "REV", "REV_LTR", "REVISION_LETTER")
|
||
_TITLE_SHEET_KEYS = ("BLATTNUMMER", "SHEET", "SHEET_NO", "SHEETNUMBER")
|
||
_TITLE_NAME_KEYS = ("BENENNUNG1", "BENENNUNG2", "BENENNUNG3", "TITLE", "DRAWING_TITLE")
|
||
_TITLE_SCALE_KEYS = ("GEN-TITLE-SCA{5.42}", "SCALE", "MASSSTAB")
|
||
_TITLE_MATERIAL_KEYS = ("WERKSTOFF", "MATERIAL", "MATERIAL_SPEC", "MATERIAL_SPECIFICATION")
|
||
_TITLE_SURFACE_KEYS = ("OBERFLAECHE", "SURFACE", "SURFACE_SPEC")
|
||
_TITLE_MASS_KEYS = ("MASSE", "MASS", "WEIGHT")
|
||
_TITLE_FORMAT_KEYS = ("FORMAT", "SIZE", "FORMAT_SIZE")
|
||
_TITLE_HALBZEUG_KEYS = ("HALBZEUG", "SEMIFINISHED", "SEMI_FINISHED")
|
||
|
||
def utc_now() -> str:
|
||
return datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
||
|
||
|
||
def _stable_id(kind: str, digest: str, suffix: str = "") -> str:
|
||
seed = f"{kind}:{digest}:{suffix}".encode()
|
||
return f"{kind}_{sha256(seed).hexdigest()[:20]}"
|
||
|
||
|
||
def _clean_text(value: Any) -> str:
|
||
text = str(value or "")
|
||
text = text.replace("\\P", "\n").replace("%%d", "°")
|
||
text = re.sub(r"[{}]", "", text)
|
||
return "\n".join(part.strip() for part in text.splitlines() if part.strip())
|
||
|
||
|
||
def _iter_text_lines(value: Any):
|
||
for line in _clean_text(value).splitlines():
|
||
token = line.strip()
|
||
if token:
|
||
yield token
|
||
|
||
|
||
def _to_number(value: str) -> int | float | str | None:
|
||
try:
|
||
return int(value)
|
||
except ValueError:
|
||
try:
|
||
return float(value)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _bbox_payload(min_x: float, min_y: float, max_x: float, max_y: float) -> dict[str, Any]:
|
||
return {
|
||
"min": [round(min_x, 6), round(min_y, 6)],
|
||
"max": [round(max_x, 6), round(max_y, 6)],
|
||
"width": round(max_x - min_x, 6),
|
||
"height": round(max_y - min_y, 6),
|
||
}
|
||
|
||
|
||
def _point_in_bbox(point: dict[str, float], bbox: dict[str, Any] | None) -> bool:
|
||
if not bbox or point.get("x") is None or point.get("y") is None:
|
||
return False
|
||
min_x, min_y = bbox.get("min", [float("-inf"), float("-inf")])
|
||
max_x, max_y = bbox.get("max", [float("inf"), float("inf")])
|
||
return min_x <= point["x"] <= max_x and min_y <= point["y"] <= max_y
|
||
|
||
|
||
def _load_doc(parsed: dict[str, Any]):
|
||
source = (parsed.get("asset") or {}).get("sourcePath") or parsed.get("sourcePath")
|
||
if not source:
|
||
return None
|
||
try:
|
||
import ezdxf
|
||
|
||
return ezdxf.readfile(Path(source))
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
|
||
def _layout_texts(doc: Any, digest: str) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
|
||
def add_row(entity: Any, space: str) -> None:
|
||
raw = entity.text if entity.dxftype() == "MTEXT" else getattr(entity.dxf, "text", "")
|
||
clean = _clean_text(raw)
|
||
if not clean:
|
||
return
|
||
insert = getattr(entity.dxf, "insert", None)
|
||
if insert is None:
|
||
return
|
||
key = f"{space}:{getattr(entity.dxf, 'handle', '')!s}"
|
||
if key in seen:
|
||
return
|
||
seen.add(key)
|
||
rows.append({
|
||
"id": _stable_id("understandingText", digest, key),
|
||
"space": space,
|
||
"x": round(float(insert[0]), 6),
|
||
"y": round(float(insert[1]), 6),
|
||
"layer": str(getattr(entity.dxf, "layer", "0") or "0"),
|
||
"text": clean,
|
||
})
|
||
|
||
for layout in doc.layouts:
|
||
space = str(layout.name or "Model")
|
||
for entity in layout:
|
||
kind = entity.dxftype()
|
||
if kind in {"TEXT", "MTEXT", "ATTRIB"}:
|
||
add_row(entity, space)
|
||
elif kind == "INSERT":
|
||
for attrib in getattr(entity, "attribs", []):
|
||
add_row(attrib, space)
|
||
if _TABLE_BLOCK_RE.search(str(getattr(entity.dxf, "name", "") or "")):
|
||
try:
|
||
for sub in entity.virtual_entities():
|
||
if sub.dxftype() in {"TEXT", "MTEXT", "ATTRIB"}:
|
||
add_row(sub, space)
|
||
except Exception: # noqa: BLE001, S110
|
||
pass
|
||
return rows
|
||
|
||
|
||
def _insert_bbox(doc: Any, insert: Any) -> dict[str, Any] | None:
|
||
try:
|
||
from ezdxf import bbox
|
||
|
||
box = bbox.extents(list(insert.virtual_entities()), fast=True)
|
||
if box and box.has_data:
|
||
return _bbox_payload(float(box.extmin.x), float(box.extmin.y),
|
||
float(box.extmax.x), float(box.extmax.y))
|
||
block = doc.blocks.get(getattr(insert.dxf, "name", "") or "")
|
||
if block is not None:
|
||
box = bbox.extents(list(block), fast=True)
|
||
if box and box.has_data:
|
||
ix, iy = float(insert.dxf.insert[0]), float(insert.dxf.insert[1])
|
||
return _bbox_payload(ix + float(box.extmin.x), iy + float(box.extmin.y),
|
||
ix + float(box.extmax.x), iy + float(box.extmax.y))
|
||
except Exception: # noqa: BLE001, S110
|
||
pass
|
||
return None
|
||
|
||
|
||
def _extract_title_block(parsed: dict[str, Any], doc: Any) -> dict[str, Any] | None:
|
||
if doc is None:
|
||
return None
|
||
for layout in doc.layouts:
|
||
for entity in layout:
|
||
if entity.dxftype() != "INSERT":
|
||
continue
|
||
attrs = {
|
||
str(getattr(attrib.dxf, "tag", "") or ""): _clean_text(getattr(attrib.dxf, "text", ""))
|
||
for attrib in getattr(entity, "attribs", [])
|
||
}
|
||
name = str(getattr(entity.dxf, "name", "") or "")
|
||
is_title = bool(_TITLE_BLOCK_RE.search(name)) or any(
|
||
key in attrs for key in _TITLE_DRAWING_KEYS
|
||
)
|
||
if not is_title:
|
||
continue
|
||
meaningful = {k: v for k, v in attrs.items() if v and v != "."}
|
||
return {
|
||
"blockName": name,
|
||
"space": str(layout.name or ""),
|
||
"bbox": _insert_bbox(doc, entity),
|
||
"attributes": meaningful,
|
||
"attributeCount": len(meaningful),
|
||
}
|
||
return None
|
||
|
||
|
||
def _extract_frame(parsed: dict[str, Any], doc: Any) -> dict[str, Any] | None:
|
||
if doc is None:
|
||
return None
|
||
for layout in doc.layouts:
|
||
for entity in layout:
|
||
if entity.dxftype() != "INSERT":
|
||
continue
|
||
name = str(getattr(entity.dxf, "name", "") or "")
|
||
if name.startswith("SF_") or not _FRAME_BLOCK_RE.search(name):
|
||
continue
|
||
box = _insert_bbox(doc, entity)
|
||
if box:
|
||
return {
|
||
"blockName": name,
|
||
"space": str(layout.name or ""),
|
||
"bbox": box,
|
||
"confidence": 0.9,
|
||
"evidence": {"source": "layoutBlockGeometry", "blockName": name},
|
||
}
|
||
try:
|
||
from ezdxf import bbox
|
||
|
||
box = bbox.extents(doc.modelspace(), fast=True)
|
||
if box and box.has_data:
|
||
return {
|
||
"blockName": None,
|
||
"space": "Model",
|
||
"bbox": _bbox_payload(float(box.extmin.x), float(box.extmin.y),
|
||
float(box.extmax.x), float(box.extmax.y)),
|
||
"confidence": 0.6,
|
||
"evidence": {"source": "modelspaceExtents"},
|
||
}
|
||
except Exception: # noqa: BLE001, S110
|
||
pass
|
||
return None
|
||
|
||
def _header_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
by_y: dict[float, list[dict[str, Any]]] = {}
|
||
for row in rows:
|
||
for line in _iter_text_lines(row["text"]):
|
||
if _HEADER_TOKEN_RE.fullmatch(line):
|
||
by_y.setdefault(row["y"], []).append({"label": line, "x": row["x"], "y": row["y"]})
|
||
merged: list[dict[str, Any]] = []
|
||
for y in sorted(by_y):
|
||
cells = by_y[y]
|
||
if merged and y - merged[-1]["maxY"] <= 8.0:
|
||
merged[-1]["cells"].extend(cells)
|
||
merged[-1]["maxY"] = y
|
||
else:
|
||
merged.append({"minY": y, "maxY": y, "cells": cells})
|
||
return [row for row in merged if len(row["cells"]) >= 3]
|
||
|
||
|
||
def _column_anchors(region_rows: list[dict[str, Any]]) -> list[float]:
|
||
anchors: list[float] = []
|
||
for row in region_rows:
|
||
for line in _iter_text_lines(row["text"]):
|
||
if _POSITION_HEADER_RE.fullmatch(line) and (not anchors or abs(row["x"] - anchors[-1]) > 1.0):
|
||
anchors.append(row["x"])
|
||
if anchors:
|
||
return sorted(anchors)
|
||
xs = sorted({row["x"] for row in region_rows})
|
||
grouped: list[float] = []
|
||
for x in xs:
|
||
if not grouped or x - grouped[-1] > 15.0:
|
||
grouped.append(x)
|
||
return grouped
|
||
|
||
|
||
def _detect_bom_regions(parsed: dict[str, Any], doc: Any, title: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
if doc is None:
|
||
return []
|
||
digest = str((parsed.get("asset") or {}).get("sha256") or "")
|
||
rows = _layout_texts(doc, digest)
|
||
preferred = [row for row in rows if _TABLE_LAYER_RE.search(row["layer"])]
|
||
if preferred:
|
||
rows = preferred
|
||
else:
|
||
rows = [row for row in rows if row["space"] == "Model"]
|
||
title_box = (title or {}).get("bbox")
|
||
if title_box:
|
||
rows = [row for row in rows if not _point_in_bbox(row, title_box)]
|
||
if not rows:
|
||
return []
|
||
headers = _header_rows(rows)
|
||
regions: list[dict[str, Any]] = []
|
||
for header in headers:
|
||
cells = header["cells"]
|
||
min_x = min(cell["x"] for cell in cells) - 8.0
|
||
max_x = max(cell["x"] for cell in cells) + 8.0
|
||
min_y = min(cell["y"] for cell in cells) - 3.0
|
||
max_y = max(cell["y"] for cell in cells) + 3.0
|
||
in_x = [row for row in rows if min_x <= row["x"] <= max_x]
|
||
data_rows = [row for row in in_x if row["y"] < header["minY"] - 2.0 or row["y"] > header["maxY"] + 2.0]
|
||
if data_rows:
|
||
min_y = min(min_y, min(row["y"] for row in data_rows) - 3.0)
|
||
max_y = max(max_y, max(row["y"] for row in data_rows) + 8.0)
|
||
region_id = _stable_id("bomRegion", digest,
|
||
f"{header['minY']:.2f}:{header['maxY']:.2f}:{min_x:.1f}:{max_x:.1f}")
|
||
layer_names = sorted({row["layer"] for row in in_x if row["layer"]})
|
||
header_labels = sorted({cell["label"].upper() for cell in cells})
|
||
has_quantity = any(_QUANTITY_HEADER_RE.fullmatch(label) for label in header_labels)
|
||
has_sheet = any(_SHEET_HEADER_RE.fullmatch(label) for label in header_labels)
|
||
regions.append({
|
||
"regionId": region_id,
|
||
"kind": "BOM_TABLE",
|
||
"space": next((row["space"] for row in in_x if row["space"]), "Model"),
|
||
"bbox": _bbox_payload(min_x, min_y, max_x, max_y),
|
||
"headerLabels": header_labels,
|
||
"headerCells": cells[:40],
|
||
"columnAnchors": _column_anchors(in_x),
|
||
"hasQuantityColumn": has_quantity,
|
||
"hasSheetColumn": has_sheet,
|
||
"layerNames": layer_names,
|
||
"confidence": round(min(0.95, 0.72 + 0.02 * len(header_labels)), 3),
|
||
"reviewRequired": True,
|
||
"evidence": {
|
||
"source": "layoutTextClustering",
|
||
"headerTokenCount": len(cells),
|
||
"layerNames": layer_names,
|
||
},
|
||
})
|
||
return regions
|
||
|
||
|
||
def _column_bounds(anchors: list[float], index: int, min_x: float, max_x: float) -> tuple[float, float]:
|
||
if not anchors:
|
||
return min_x, max_x
|
||
left = min_x if index == 0 else (anchors[index - 1] + anchors[index]) / 2.0
|
||
right = max_x if index >= len(anchors) - 1 else (anchors[index] + anchors[index + 1]) / 2.0
|
||
return left, right
|
||
|
||
|
||
def _pick_name(texts: list[dict[str, Any]]) -> str | None:
|
||
candidates = []
|
||
for row in texts:
|
||
for line in _iter_text_lines(row["text"]):
|
||
token = line.strip()
|
||
if len(token) < 3 or _HEADER_TOKEN_RE.fullmatch(token) or _ITEM_REF_RE.fullmatch(token):
|
||
continue
|
||
if token.isdigit() or token == ".":
|
||
continue
|
||
candidates.append(token)
|
||
return max(candidates, key=len) if candidates else None
|
||
|
||
|
||
def _part_list_rows(parsed: dict[str, Any], doc: Any, regions: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
digest = str((parsed.get("asset") or {}).get("sha256") or "")
|
||
drawing_no = str((parsed.get("drawing") or {}).get("drawingNumber") or "")
|
||
rows = _layout_texts(doc, digest) if doc is not None else []
|
||
out: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for region in regions:
|
||
min_x, min_y = region["bbox"]["min"]
|
||
max_x, max_y = region["bbox"]["max"]
|
||
in_region = [row for row in rows if min_x <= row["x"] <= max_x and min_y <= row["y"] <= max_y]
|
||
ref_rows = [
|
||
row for row in in_region
|
||
if any(_ITEM_REF_RE.fullmatch(line) for line in _iter_text_lines(row["text"]))
|
||
]
|
||
for ref_row in ref_rows:
|
||
ref_y = ref_row["y"]
|
||
group = [row for row in in_region if abs(row["y"] - ref_y) <= 10.0]
|
||
refs = sorted({line for line in _iter_text_lines(ref_row["text"]) if _ITEM_REF_RE.fullmatch(line)})
|
||
anchors = region.get("columnAnchors") or []
|
||
for ref in refs:
|
||
index = min(range(len(anchors)), key=lambda i: abs(anchors[i] - ref_row["x"])) if anchors else 0
|
||
left, right = _column_bounds(anchors, index, min_x, max_x)
|
||
column_texts = [row for row in group if left <= row["x"] <= right]
|
||
key = f"{ref}:{ref_y:.2f}:{index}"
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
numeric = [
|
||
_to_number(line)
|
||
for row in column_texts
|
||
for line in _iter_text_lines(row["text"])
|
||
if not _ITEM_REF_RE.fullmatch(line) and _to_number(line) is not None
|
||
]
|
||
quantity = None
|
||
sheet = None
|
||
if region.get("hasQuantityColumn") and numeric:
|
||
quantity = numeric[0]
|
||
if region.get("hasSheetColumn") and numeric:
|
||
sheet = str(numeric[0])
|
||
name = _pick_name(column_texts)
|
||
candidate_id = _stable_id("partList", digest, key)
|
||
out.append({
|
||
"candidateId": candidate_id,
|
||
"candidateType": "PART_LIST_ROW",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"parentDrawingNumber": drawing_no,
|
||
"componentReference": ref,
|
||
"componentName": name,
|
||
"quantity": quantity,
|
||
"unit": None,
|
||
"sheetReference": sheet,
|
||
"confidence": round(0.74 if name else 0.6, 3),
|
||
"unknownFields": ["componentMaterialCode", "unit", "materialSpecification"],
|
||
"evidence": {
|
||
"source": "bomRegionText",
|
||
"regionId": region["regionId"],
|
||
"space": region["space"],
|
||
"rowY": round(ref_y, 3),
|
||
"columnGroup": index,
|
||
"texts": [
|
||
{"text": row["text"], "x": row["x"], "y": row["y"], "layer": row["layer"]}
|
||
for row in column_texts[:12]
|
||
],
|
||
},
|
||
})
|
||
if not out:
|
||
for row in parsed.get("itemReferences") or []:
|
||
ref = str(row.get("reference") or "").strip()
|
||
if not ref or ref in seen:
|
||
continue
|
||
seen.add(ref)
|
||
out.append({
|
||
"candidateId": _stable_id("partList", digest, f"callout:{ref}"),
|
||
"candidateType": "PART_LIST_ROW",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"parentDrawingNumber": drawing_no,
|
||
"componentReference": ref,
|
||
"componentName": None,
|
||
"quantity": None,
|
||
"unit": None,
|
||
"sheetReference": None,
|
||
"confidence": round(min(float(row.get("confidence") or 0.0), 0.72), 3),
|
||
"unknownFields": ["componentMaterialCode", "quantity", "unit", "materialSpecification"],
|
||
"evidence": {"source": "inspectItemReference", "entity": row.get("evidence")},
|
||
})
|
||
return out
|
||
|
||
def _assembly_relations(part_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
out: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for row in part_rows:
|
||
parent = str(row.get("parentDrawingNumber") or "").strip()
|
||
ref = str(row.get("componentReference") or "").strip()
|
||
if not parent or not ref or ref == parent:
|
||
continue
|
||
key = f"{parent}:{ref}"
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append({
|
||
"candidateId": _stable_id("assemblyRelation", str((row.get("evidence") or {}).get("regionId") or ""), key),
|
||
"candidateType": "ASSEMBLY_RELATION",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"parentDrawingNumber": parent,
|
||
"parentMaterialCode": parent,
|
||
"componentReference": ref,
|
||
"componentName": row.get("componentName"),
|
||
"quantity": row.get("quantity"),
|
||
"relationshipType": "PART_OF",
|
||
"confidence": round(min(float(row.get("confidence") or 0.0), 0.7), 3),
|
||
"unknownFields": ["componentMaterialCode", "quantity"],
|
||
"evidence": {"source": "assemblyRelationFromPartList", "partListCandidateId": row.get("candidateId")},
|
||
})
|
||
return out
|
||
|
||
|
||
def _material_spec_candidates(title: dict[str, Any] | None, doc: Any,
|
||
digest: str) -> list[dict[str, Any]]:
|
||
out: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
attrs = (title or {}).get("attributes") or {}
|
||
attr_fields = [
|
||
(_TITLE_MATERIAL_KEYS, "materialSpecification"),
|
||
(_TITLE_SURFACE_KEYS, "surfaceSpecification"),
|
||
(_TITLE_MASS_KEYS, "mass"),
|
||
(_TITLE_HALBZEUG_KEYS, "halfFinishedSpecification"),
|
||
]
|
||
for keys, field in attr_fields:
|
||
value = next((str(attrs[k]).strip() for k in keys if attrs.get(k)), None)
|
||
if value and value != "." and field not in seen:
|
||
seen.add(field)
|
||
out.append({
|
||
"candidateId": _stable_id("materialSpec", digest, field),
|
||
"candidateType": "MATERIAL_SPEC",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"field": field,
|
||
"value": value,
|
||
"confidence": 0.82,
|
||
"evidence": {"source": "titleBlockAttribute", "attribute": field},
|
||
})
|
||
rows = _layout_texts(doc, digest) if doc is not None else []
|
||
title_box = (title or {}).get("bbox")
|
||
for row in rows:
|
||
if title_box and _point_in_bbox(row, title_box):
|
||
continue
|
||
for line in _iter_text_lines(row["text"]):
|
||
matches = [*_THREAD_SPEC_RE.finditer(line), *_MATERIAL_WORD_RE.finditer(line)]
|
||
for match in matches:
|
||
field = "threadSpecification" if _THREAD_SPEC_RE.fullmatch(match.group(0)) else "materialSpecificationHint"
|
||
key = f"text:{field}:{match.group(0)}"
|
||
if key in seen or len(out) >= 40:
|
||
continue
|
||
seen.add(key)
|
||
out.append({
|
||
"candidateId": _stable_id("materialSpec", digest, key),
|
||
"candidateType": "MATERIAL_SPEC",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"field": field,
|
||
"value": match.group(0),
|
||
"confidence": 0.55,
|
||
"evidence": {
|
||
"source": "drawingText",
|
||
"text": line,
|
||
"position": [row["x"], row["y"]],
|
||
"layer": row["layer"],
|
||
},
|
||
})
|
||
return out
|
||
|
||
|
||
def _quantity_candidates(part_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
out: list[dict[str, Any]] = []
|
||
for row in part_rows:
|
||
quantity = row.get("quantity")
|
||
if quantity is None:
|
||
continue
|
||
out.append({
|
||
"candidateId": _stable_id("quantity", str((row.get("evidence") or {}).get("regionId") or ""),
|
||
str(row.get("componentReference") or "")),
|
||
"candidateType": "QUANTITY",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"field": "BOM.quantity",
|
||
"value": quantity,
|
||
"componentReference": row.get("componentReference"),
|
||
"confidence": 0.68,
|
||
"evidence": {"source": "bomRegionText", "partListCandidateId": row.get("candidateId")},
|
||
})
|
||
return out
|
||
|
||
|
||
def _drawing_no_relation(inner: str | None, filename_no: str | None) -> tuple[str, float]:
|
||
a = re.sub(r"\s+", "", inner or "").upper()
|
||
b = re.sub(r"\s+", "", filename_no or "").upper()
|
||
if not a or not b:
|
||
return "missing", 0.5
|
||
if a == b:
|
||
return "exact", 0.97
|
||
if len(a) >= 4 and (b.startswith(a) or a.startswith(b)):
|
||
return "partial", 0.86
|
||
return "mismatch", 0.4
|
||
|
||
|
||
def _revision_relation(inner: str | None, filename_rev: str | None) -> tuple[str, float]:
|
||
a = str(inner or "").strip().lower()
|
||
b = str(filename_rev or "").strip().lower()
|
||
if not a:
|
||
return "missing", 0.5
|
||
if not b:
|
||
return "innerOnly", 0.72
|
||
if a == b:
|
||
return "exact", 0.95
|
||
return "mismatch", 0.35
|
||
|
||
|
||
def _field_candidate(field: str, value: Any, confidence: float, evidence: dict[str, Any],
|
||
*, review_required: bool) -> dict[str, Any]:
|
||
return {
|
||
"field": field,
|
||
"value": value,
|
||
"confidence": round(max(0.0, min(float(confidence), 1.0)), 3),
|
||
"reviewRequired": bool(review_required),
|
||
"evidence": evidence,
|
||
}
|
||
|
||
|
||
def _title_value(attrs: dict[str, str], keys: tuple[str, ...]) -> str | None:
|
||
for key in keys:
|
||
value = str(attrs.get(key) or "").strip()
|
||
if value and value != ".":
|
||
return value
|
||
return None
|
||
|
||
def build_drawing_understanding(parsed: dict[str, Any]) -> dict[str, Any]:
|
||
"""Build PENDING_REVIEW drawing-understanding candidates from inspect_dxf output."""
|
||
|
||
asset = parsed.get("asset") or {}
|
||
drawing = parsed.get("drawing") or {}
|
||
digest = str(asset.get("sha256") or "")
|
||
if not digest or len(digest) != 64:
|
||
raise ValueError("parsed drawing is missing a valid sha256")
|
||
filename_no = str(drawing.get("drawingNumber") or "").strip()
|
||
filename_rev = drawing.get("revision")
|
||
doc = _load_doc(parsed)
|
||
title = _extract_title_block(parsed, doc)
|
||
frame = _extract_frame(parsed, doc)
|
||
bom_regions = _detect_bom_regions(parsed, doc, title)
|
||
part_rows = _part_list_rows(parsed, doc, bom_regions)
|
||
relations = _assembly_relations(part_rows)
|
||
spec_candidates = _material_spec_candidates(title, doc, digest)
|
||
quantity_candidates = _quantity_candidates(part_rows)
|
||
|
||
from server.aps_domain.drawing_process import recognize_process_candidates
|
||
process = recognize_process_candidates(parsed, part_rows)
|
||
process_ops = process.get("operations") or []
|
||
process_details = process.get("details") or []
|
||
|
||
fields: list[dict[str, Any]] = []
|
||
title_attrs = (title or {}).get("attributes") or {}
|
||
inner_no = _title_value(title_attrs, _TITLE_DRAWING_KEYS)
|
||
inner_rev = _title_value(title_attrs, _TITLE_REVISION_KEYS)
|
||
no_rel, no_conf = _drawing_no_relation(inner_no, filename_no)
|
||
rev_rel, rev_conf = _revision_relation(inner_rev, filename_rev)
|
||
fields.append(_field_candidate(
|
||
"drawingNumber", filename_no, no_conf,
|
||
{"source": "crossValidation", "filename": filename_no, "titleBlock": inner_no,
|
||
"relation": no_rel},
|
||
review_required=no_rel != "exact",
|
||
))
|
||
revision_value = filename_rev if filename_rev else inner_rev
|
||
fields.append(_field_candidate(
|
||
"revision", revision_value, rev_conf,
|
||
{"source": "crossValidation", "filename": filename_rev, "titleBlock": inner_rev,
|
||
"relation": rev_rel},
|
||
review_required=rev_rel != "exact",
|
||
))
|
||
title_fields = [
|
||
("sheetNo", _title_value(title_attrs, _TITLE_SHEET_KEYS), "titleBlockAttribute"),
|
||
("title", _title_value(title_attrs, _TITLE_NAME_KEYS), "titleBlockAttribute"),
|
||
("scale", _title_value(title_attrs, _TITLE_SCALE_KEYS), "titleBlockAttribute"),
|
||
("format", _title_value(title_attrs, _TITLE_FORMAT_KEYS), "titleBlockAttribute"),
|
||
]
|
||
for field, value, source in title_fields:
|
||
if value:
|
||
fields.append(_field_candidate(
|
||
field, value, 0.85, {"source": source, "blockName": (title or {}).get("blockName")},
|
||
review_required=True,
|
||
))
|
||
for detail in process_details:
|
||
fields.append(_field_candidate(
|
||
detail["field"], detail["value"], float(detail["confidence"]),
|
||
detail["evidence"], review_required=True,
|
||
))
|
||
|
||
warnings = [
|
||
"Candidates must pass engineering review/P2 approval before master-data write.",
|
||
"No BOM quantity, standard time, or production resource is inferred from geometry alone.",
|
||
]
|
||
if no_rel == "partial":
|
||
warnings.append("Title-block drawing number partially matches the filename (sheet/page suffix); review required.")
|
||
if not bom_regions:
|
||
warnings.append("No explicit BOM table detected; part references are treated as callout evidence and remain PENDING_REVIEW.")
|
||
if not quantity_candidates:
|
||
warnings.append("No explicit quantity column detected; quantity candidates are not fabricated.")
|
||
if not process_ops:
|
||
warnings.append("No explicit process/operation keyword detected in drawing text or part-list names; routing stays as engineering-review placeholder.")
|
||
|
||
return {
|
||
"contractVersion": UNDERSTANDING_VERSION,
|
||
"asset": {"id": asset.get("id"), "sha256": digest, "filename": asset.get("filename"),
|
||
"sourcePath": asset.get("sourcePath")},
|
||
"drawing": {"drawingNumber": filename_no, "revision": filename_rev},
|
||
"regions": {
|
||
"titleBlock": title,
|
||
"frame": frame,
|
||
"bomTables": bom_regions,
|
||
},
|
||
"fieldCandidates": fields,
|
||
"partListRows": part_rows,
|
||
"assemblyRelations": relations,
|
||
"materialSpecCandidates": spec_candidates,
|
||
"quantityCandidates": quantity_candidates,
|
||
"processCandidates": process_ops,
|
||
"processDetails": process_details,
|
||
"routingOperations": process_ops,
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"warnings": warnings,
|
||
}
|
||
|
||
def diff_drawing_inspect(older: dict[str, Any], newer: dict[str, Any]) -> dict[str, Any]:
|
||
"""Produce a reviewable diff summary and reschedule-impact hint from two inspect results."""
|
||
|
||
old_asset = older.get("asset") or {}
|
||
new_asset = newer.get("asset") or {}
|
||
old_drawing = older.get("drawing") or {}
|
||
new_drawing = newer.get("drawing") or {}
|
||
old_stats = older.get("entityStatistics") or {}
|
||
new_stats = newer.get("entityStatistics") or {}
|
||
old_refs = {str(row.get("reference") or "") for row in (older.get("itemReferences") or [])}
|
||
new_refs = {str(row.get("reference") or "") for row in (newer.get("itemReferences") or [])}
|
||
|
||
entity_delta: dict[str, dict[str, Any]] = {}
|
||
for key in sorted(set(old_stats) | set(new_stats)):
|
||
old_value = int(old_stats.get(key, 0) or 0)
|
||
new_value = int(new_stats.get(key, 0) or 0)
|
||
if old_value != new_value:
|
||
entity_delta[key] = {"old": old_value, "new": new_value, "delta": new_value - old_value}
|
||
|
||
old_fields = {str(row.get("field") or ""): row.get("value") for row in (older.get("fieldCandidates") or [])}
|
||
new_fields = {str(row.get("field") or ""): row.get("value") for row in (newer.get("fieldCandidates") or [])}
|
||
field_changes = [
|
||
{"field": key, "old": old_fields.get(key), "new": new_fields.get(key)}
|
||
for key in sorted(set(old_fields) | set(new_fields))
|
||
if old_fields.get(key) != new_fields.get(key)
|
||
]
|
||
|
||
old_bbox = old_drawing.get("bbox") or {}
|
||
new_bbox = new_drawing.get("bbox") or {}
|
||
geometry_changed = (
|
||
abs(float(old_bbox.get("width") or 0.0) - float(new_bbox.get("width") or 0.0)) > 0.5
|
||
or abs(float(old_bbox.get("height") or 0.0) - float(new_bbox.get("height") or 0.0)) > 0.5
|
||
or bool(entity_delta)
|
||
)
|
||
part_list_changed = bool(new_refs.symmetric_difference(old_refs))
|
||
revision_changed = str(old_drawing.get("revision") or "") != str(new_drawing.get("revision") or "")
|
||
drawing_no_changed = str(old_drawing.get("drawingNumber") or "") != str(new_drawing.get("drawingNumber") or "")
|
||
sha_changed = str(old_asset.get("sha256") or "") != str(new_asset.get("sha256") or "")
|
||
text_count_changed = len(older.get("texts") or []) != len(newer.get("texts") or [])
|
||
dimension_count_changed = len(older.get("dimensions") or []) != len(newer.get("dimensions") or [])
|
||
|
||
changed = (
|
||
sha_changed or drawing_no_changed or revision_changed or geometry_changed
|
||
or part_list_changed or text_count_changed or dimension_count_changed or bool(field_changes)
|
||
)
|
||
if revision_changed and (geometry_changed or part_list_changed or dimension_count_changed):
|
||
impact_level = "HIGH"
|
||
elif revision_changed or part_list_changed or geometry_changed or dimension_count_changed:
|
||
impact_level = "MEDIUM"
|
||
else:
|
||
impact_level = "LOW"
|
||
reasons = []
|
||
if sha_changed:
|
||
reasons.append("file content hash changed")
|
||
if revision_changed:
|
||
reasons.append(f"revision changed {old_drawing.get('revision')!r} -> {new_drawing.get('revision')!r}")
|
||
if part_list_changed:
|
||
reasons.append("part reference set changed")
|
||
if geometry_changed:
|
||
reasons.append("entity statistics or drawing extents changed")
|
||
if dimension_count_changed:
|
||
reasons.append("dimension count changed")
|
||
|
||
return {
|
||
"contractVersion": DIFF_VERSION,
|
||
"changed": bool(changed),
|
||
"asset": {
|
||
"sameSha": not sha_changed,
|
||
"oldSha": old_asset.get("sha256"),
|
||
"newSha": new_asset.get("sha256"),
|
||
},
|
||
"drawing": {
|
||
"oldDrawingNumber": old_drawing.get("drawingNumber"),
|
||
"newDrawingNumber": new_drawing.get("drawingNumber"),
|
||
"oldRevision": old_drawing.get("revision"),
|
||
"newRevision": new_drawing.get("revision"),
|
||
},
|
||
"summary": {
|
||
"oldTexts": len(older.get("texts") or []),
|
||
"newTexts": len(newer.get("texts") or []),
|
||
"oldDimensions": len(older.get("dimensions") or []),
|
||
"newDimensions": len(newer.get("dimensions") or []),
|
||
"oldBbox": old_bbox,
|
||
"newBbox": new_bbox,
|
||
},
|
||
"entityStatistics": entity_delta,
|
||
"fieldChanges": field_changes,
|
||
"itemReferences": {
|
||
"added": sorted(new_refs - old_refs),
|
||
"removed": sorted(old_refs - new_refs),
|
||
},
|
||
"impact": {
|
||
"level": impact_level,
|
||
"suggestReschedule": bool(changed and impact_level in {"HIGH", "MEDIUM"}),
|
||
"reasons": reasons,
|
||
},
|
||
"warnings": [
|
||
"Version diff is evidence-only; rescheduling requires engineering/P2 review.",
|
||
],
|
||
}
|
||
|
||
|
||
def _order_codes(order: dict[str, Any]) -> list[str]:
|
||
codes = [str(order.get("productCode") or ""), str(order.get("materialCode") or "")]
|
||
for item in order.get("items") or []:
|
||
codes.extend([str(item.get("productCode") or ""), str(item.get("materialCode") or "")])
|
||
return [code for code in codes if code]
|
||
|
||
|
||
def _order_tables(world: dict[str, Any]) -> list[tuple[str, list[dict[str, Any]]]]:
|
||
return [
|
||
("flexOrders", world.get("flexOrders") or []),
|
||
("salesOrders", world.get("salesOrders") or []),
|
||
("productionOrders", world.get("productionOrders") or []),
|
||
("workOrders", world.get("workOrders") or []),
|
||
]
|
||
|
||
|
||
def build_drawing_link_evidence_records(
|
||
world: dict[str, Any],
|
||
*,
|
||
drawing_id: str,
|
||
source_sha256: str,
|
||
selected_candidates: list[dict[str, Any]],
|
||
actor: str,
|
||
confirm_id: str | None = None,
|
||
next_id: Any = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""Build drawingLink evidence records for approved non-material candidates.
|
||
|
||
The workflow keeps material-row creation and its existing material link;
|
||
this helper adds BOM/ROUTING/ORDER evidence without writing master data.
|
||
"""
|
||
|
||
records: list[dict[str, Any]] = []
|
||
now = utc_now()
|
||
seen: set[tuple[str, str, str]] = set()
|
||
|
||
def add(link_type: str, target_type: str, target_id: Any, target_ref: str,
|
||
status: str, master_committed: bool, confidence: float,
|
||
candidate: dict[str, Any], extra: dict[str, Any]) -> None:
|
||
key = (link_type, target_type, target_ref)
|
||
if key in seen:
|
||
return
|
||
seen.add(key)
|
||
records.append({
|
||
"id": next_id("drawingLink") if next_id is not None else None,
|
||
"drawingId": drawing_id,
|
||
"linkType": link_type,
|
||
"targetType": target_type,
|
||
"targetId": target_id,
|
||
"targetRef": target_ref,
|
||
"sourceSha256": source_sha256,
|
||
"revision": candidate.get("revision"),
|
||
"status": status,
|
||
"masterCommitted": master_committed,
|
||
"confidence": round(max(0.0, min(float(confidence), 1.0)), 3),
|
||
"confirmedBy": actor,
|
||
"confirmedAt": now,
|
||
"evidence": {
|
||
"contractVersion": LINK_EVIDENCE_VERSION,
|
||
"confirmId": confirm_id,
|
||
"candidateId": candidate.get("candidateId"),
|
||
"candidateType": candidate.get("candidateType"),
|
||
"sourceSha256": source_sha256,
|
||
"drawingId": drawing_id,
|
||
**extra,
|
||
},
|
||
})
|
||
|
||
for candidate in selected_candidates or []:
|
||
ctype = str(candidate.get("candidateType") or "")
|
||
confidence = float(candidate.get("confidence") or 0.5)
|
||
if ctype == "MATERIAL":
|
||
code = str(candidate.get("code") or "").strip()
|
||
if not code:
|
||
continue
|
||
for table_name, orders in _order_tables(world):
|
||
for order in orders:
|
||
if code not in _order_codes(order):
|
||
continue
|
||
add(
|
||
"ORDER", "order", order.get("id"), str(order.get("orderNo") or ""),
|
||
"CONFIRMED", True, max(0.9, confidence), candidate,
|
||
{"orderTable": table_name, "productCode": code, "orderNo": order.get("orderNo")},
|
||
)
|
||
elif ctype == "BOM_REFERENCE":
|
||
ref = str(candidate.get("componentReference") or "").strip()
|
||
if not ref:
|
||
continue
|
||
add(
|
||
"BOM", "bom_reference", candidate.get("candidateId"), ref,
|
||
"PENDING_REVIEW", False, confidence, candidate,
|
||
{"parentMaterialCode": candidate.get("parentMaterialCode"), "componentReference": ref},
|
||
)
|
||
elif ctype == "ROUTING_OPERATION":
|
||
op = str(candidate.get("operationCode") or "").strip()
|
||
if not op:
|
||
continue
|
||
add(
|
||
"ROUTING", "routing_operation", candidate.get("candidateId"), op,
|
||
"PENDING_REVIEW", False, confidence, candidate,
|
||
{"productCode": candidate.get("productCode"), "operationCode": op},
|
||
)
|
||
return records
|
||
|
||
|
||
__all__ = [
|
||
"build_drawing_link_evidence_records",
|
||
"build_drawing_understanding",
|
||
"diff_drawing_inspect",
|
||
"utc_now",
|
||
]
|