Umut CANandClaude Opus 5
8fcd587b5b
Two reader defects, both about trusting a derived number over the drawing.
Stale junctions: endpoints were resolved before a foreign line's membership
was decided, so a P&ID line ending on a foreign line that was then discarded
(it touched no node) kept `{"junction": "J1"}` while `junctions` was empty
and `stats.dangling` was one short. Node attachment now runs first for every
edge, the foreign lines that attach are admitted, and only then are junctions
and dangling ends resolved over the edges actually in the graph.
Chord length: `length` was a chord walk over the vertices, 12.5% short on a
semicircular crossing jump with no field saying so, and junction detection
tested the chord too, so a line ending on the arc dangled. Both engines now
report per-vertex `bulges` next to `points` (COM asks GetBulge only when
`Length` exceeds the chord walk, since an arc is always longer than its
chord); the reader takes the engine's bulge-aware `length` and tests
endpoints against the arc. The chord sum survives only for an engine that
reports no length, and is then labelled `length_approximate`.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
633 lines
24 KiB
Python
633 lines
24 KiB
Python
"""Read a P&ID back as a graph: nodes (symbols), edges (lines), junctions, dangling ends.
|
|
|
|
Rules the reader keeps (spec §9.4): it never modifies the drawing, it reads
|
|
geometry from the drawing rather than the catalogue, and confidence is the
|
|
minimum of its inputs — never raised by agreement between heuristics.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING
|
|
|
|
from .lines import LAYER_TO_CLASS, PID_LINE_LAYERS
|
|
from .symbols import Port, transform_port
|
|
from .tags import parse_tag
|
|
from .xdata import read_payload
|
|
|
|
if TYPE_CHECKING:
|
|
from backends.base import AutoCADBackend, EntityInfo
|
|
|
|
INSERT_TYPES = {"INSERT", "BLOCKREFERENCE"}
|
|
# Same set ``drawlines`` keeps: ezdxf says LWPOLYLINE, the COM engine derives
|
|
# POLYLINE (AcDbPolyline) / 2DPOLYLINE from the ActiveX ObjectName.
|
|
POLY_TYPES = {"LWPOLYLINE", "POLYLINE", "2DPOLYLINE"}
|
|
LINE_TYPES = {"LINE"}
|
|
TEXT_TYPES = {"TEXT", "MTEXT"}
|
|
ATTRIBUTE_TAGS = ("TAG", "TAGNO", "TAGNUMBER", "EQUIP_TAG", "EQUIPMENT_TAG", "INST_TAG", "INSTTAG")
|
|
# Block-name keywords (spec §9.2 step 2), matched as whole tokens of the name
|
|
# split on non-letters: ``GATE_VALVE`` → VALVE, ``P101PUMP`` → PUMP, but
|
|
# ``COMPANY_LOGO`` is a logo, not a compressor — a substring match would file
|
|
# every title block and logo as equipment at 0.6.
|
|
KEYWORDS = {
|
|
"VALVE": "valve",
|
|
"VLV": "valve",
|
|
"PUMP": "equipment",
|
|
"TANK": "equipment",
|
|
"VESSEL": "equipment",
|
|
"DRUM": "equipment",
|
|
"INST": "instrument",
|
|
"INSTRUMENT": "instrument",
|
|
"BUBBLE": "instrument",
|
|
"HX": "equipment",
|
|
"EXCH": "equipment",
|
|
"EXCHANGER": "equipment",
|
|
"FILTER": "equipment",
|
|
"COMP": "equipment",
|
|
"COMPRESSOR": "equipment",
|
|
}
|
|
_TOKEN_RE = re.compile(r"[A-Z]+")
|
|
FAMILY_KIND = {"instrument": "instrument", "valve": "valve", "connector": "connector"}
|
|
LINE_NUMBER_RE = re.compile(r'^\d+(?:"|MM)?-[A-Z]{1,4}-\d+', re.IGNORECASE)
|
|
CONFIDENCE = {"catalog": 1.0, "xdata": 0.95, "heuristic": 0.6, "inferred": 0.3}
|
|
_PAGE = 1000
|
|
|
|
|
|
@dataclass
|
|
class _Node:
|
|
id: str
|
|
kind: str
|
|
block_name: str
|
|
family: str | None
|
|
symbol: str | None
|
|
x: float
|
|
y: float
|
|
rotation_deg: float
|
|
scale: float
|
|
layer: str
|
|
space: str
|
|
bbox: tuple[float, float, float, float] | None
|
|
source: str
|
|
confidence: float
|
|
attributes: dict
|
|
ports: dict = field(default_factory=dict)
|
|
tag: str | None = None
|
|
tag_source: str | None = None
|
|
|
|
def to_dict(self) -> dict:
|
|
parsed = None
|
|
if self.tag:
|
|
parsed = parse_tag(self.tag, kind="instrument" if self.kind == "instrument" else "auto")
|
|
return {
|
|
"id": self.id,
|
|
"kind": self.kind,
|
|
"block_name": self.block_name,
|
|
"family": self.family,
|
|
"symbol": self.symbol,
|
|
"tag": self.tag,
|
|
"tag_parsed": parsed,
|
|
"tag_source": self.tag_source,
|
|
"x": self.x,
|
|
"y": self.y,
|
|
"rotation_deg": self.rotation_deg,
|
|
"scale": self.scale,
|
|
"layer": self.layer,
|
|
"space": self.space,
|
|
"ports": self.ports,
|
|
"description": self.attributes.get("DESC") or None,
|
|
"source": self.source,
|
|
"confidence": self.confidence,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class _Edge:
|
|
id: str
|
|
vertices: list[tuple[float, float]]
|
|
bulges: list[float] # per vertex, the arc leaving it (DXF convention); 0.0 = straight
|
|
length: float | None # the engine's own measurement (bulge-aware), None when it gave none
|
|
layer: str
|
|
linetype: str
|
|
space: str
|
|
payload: dict | None
|
|
source: str
|
|
line_class: str | None = None
|
|
class_source: str | None = None
|
|
line_number: str | None = None
|
|
number_source: str | None = None
|
|
from_ref: dict | None = None
|
|
to_ref: dict | None = None
|
|
|
|
def to_dict(self, include_geometry: bool) -> dict:
|
|
p = self.payload or {}
|
|
out = {
|
|
"id": self.id,
|
|
"line_class": self.line_class,
|
|
"class_source": self.class_source,
|
|
"line_number": self.line_number,
|
|
"number_source": self.number_source,
|
|
"size": p.get("size"),
|
|
"service": p.get("service"),
|
|
"spec": p.get("spec"),
|
|
"insulation": p.get("insulation"),
|
|
"from": self.from_ref,
|
|
"to": self.to_ref,
|
|
# Both engines measure the real geometry (ezdxf via the shared
|
|
# bulge-aware maths, COM via ActiveX `Length`); a chord walk over
|
|
# the vertices is 12.5% short on a semicircular jump and says
|
|
# nothing. The chord sum is kept only for an engine that reports
|
|
# no length at all, and then the approximation is named.
|
|
"length": round(self.length, 6) if self.length is not None else self.chord_length(),
|
|
"layer": self.layer,
|
|
"space": self.space,
|
|
"source": self.source,
|
|
}
|
|
if self.length is None:
|
|
out["length_approximate"] = True
|
|
if include_geometry:
|
|
out["vertices"] = [[x, y] for x, y in self.vertices]
|
|
out["bulges"] = list(self.bulges)
|
|
return out
|
|
|
|
def chord_length(self) -> float:
|
|
return round(
|
|
sum(math.dist(a, b) for a, b in zip(self.vertices, self.vertices[1:], strict=False)),
|
|
6,
|
|
)
|
|
|
|
def segments(self):
|
|
"""``(a, b, bulge)`` per segment, the bulge belonging to the segment's start."""
|
|
for i, (a, b) in enumerate(zip(self.vertices, self.vertices[1:], strict=False)):
|
|
yield a, b, self.bulges[i] if i < len(self.bulges) else 0.0
|
|
|
|
|
|
class _Grid:
|
|
"""Uniform grid for nearest-within-tolerance queries."""
|
|
|
|
def __init__(self, cell: float):
|
|
self.cell = cell
|
|
self.cells: dict[tuple[int, int], list] = {}
|
|
|
|
def _key(self, x: float, y: float) -> tuple[int, int]:
|
|
return (int(math.floor(x / self.cell)), int(math.floor(y / self.cell)))
|
|
|
|
def add(self, x: float, y: float, item) -> None:
|
|
self.cells.setdefault(self._key(x, y), []).append((x, y, item))
|
|
|
|
def near(self, x: float, y: float, radius: float):
|
|
kx, ky = self._key(x, y)
|
|
reach = int(math.ceil(radius / self.cell))
|
|
for dx in range(-reach, reach + 1):
|
|
for dy in range(-reach, reach + 1):
|
|
for px, py, item in self.cells.get((kx + dx, ky + dy), []):
|
|
d = math.hypot(px - x, py - y)
|
|
if d <= radius:
|
|
yield d, item
|
|
|
|
|
|
def _vertices(info: EntityInfo) -> list[tuple[float, float]]:
|
|
if info.type in LINE_TYPES:
|
|
s, e = info.properties["start"], info.properties["end"]
|
|
return [(float(s[0]), float(s[1])), (float(e[0]), float(e[1]))]
|
|
return [(float(p[0]), float(p[1])) for p in info.properties.get("points") or []]
|
|
|
|
|
|
def _bulges(info: EntityInfo, count: int) -> list[float]:
|
|
"""Per-vertex bulges as the engine reports them; straight when it reports none."""
|
|
raw = info.properties.get("bulges") or []
|
|
out = [float(b or 0.0) for b in raw[:count]]
|
|
return out + [0.0] * (count - len(out))
|
|
|
|
|
|
def _length(info: EntityInfo) -> float | None:
|
|
raw = info.properties.get("length")
|
|
try:
|
|
return float(raw) if raw is not None else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _point_segment_distance(p, a, b, bulge: float = 0.0) -> float:
|
|
"""Distance from ``p`` to the polyline segment a→b, an arc when it carries a bulge.
|
|
|
|
The bulge is the DXF one: tan(sweep/4), positive counter-clockwise from
|
|
``a`` to ``b``. Testing against the chord instead misses every line that
|
|
ends on the curved "jump" foreign P&IDs draw where two lines cross.
|
|
"""
|
|
ax, ay = a
|
|
bx, by = b
|
|
px, py = p
|
|
dx, dy = bx - ax, by - ay
|
|
chord = math.hypot(dx, dy)
|
|
if chord < 1e-12:
|
|
return math.hypot(px - ax, py - ay)
|
|
if abs(bulge) < 1e-12:
|
|
t = max(0.0, min(1.0, ((px - ax) * dx + (py - ay) * dy) / (dx * dx + dy * dy)))
|
|
return math.hypot(px - (ax + t * dx), py - (ay + t * dy))
|
|
theta = 4.0 * math.atan(bulge) # signed sweep
|
|
radius = chord / (2.0 * math.sin(abs(theta) / 2.0))
|
|
sagitta = abs(bulge) * chord / 2.0
|
|
# The centre sits on the chord's normal: opposite the arc while it is less
|
|
# than a semicircle, on the arc's side once it is more (|bulge| > 1). A
|
|
# positive bulge bows to the right of a→b, so its centre is to the left.
|
|
nx, ny = -dy / chord, dx / chord # left of a→b
|
|
offset = math.copysign(radius - sagitta, bulge)
|
|
cx, cy = (ax + bx) / 2.0 + offset * nx, (ay + by) / 2.0 + offset * ny
|
|
start = math.atan2(ay - cy, ax - cx)
|
|
phi = math.atan2(py - cy, px - cx)
|
|
swept = (phi - start) % (2.0 * math.pi) if theta > 0 else (start - phi) % (2.0 * math.pi)
|
|
if swept <= abs(theta):
|
|
return abs(math.hypot(px - cx, py - cy) - radius)
|
|
return min(math.hypot(px - ax, py - ay), math.hypot(px - bx, py - by))
|
|
|
|
|
|
def _keyword_kind(block_name: str) -> str | None:
|
|
"""Kind implied by a foreign block's name, or None when no token is a keyword."""
|
|
for token in _TOKEN_RE.findall(block_name.upper()):
|
|
kind = KEYWORDS.get(token)
|
|
if kind:
|
|
return kind
|
|
return None
|
|
|
|
|
|
def _bbox(info: EntityInfo):
|
|
bb = info.properties.get("bounding_box")
|
|
if not bb:
|
|
return None
|
|
return (float(bb["min"][0]), float(bb["min"][1]), float(bb["max"][0]), float(bb["max"][1]))
|
|
|
|
|
|
def _on_bbox_boundary(p, bbox, tolerance) -> bool:
|
|
x, y = p
|
|
xmin, ymin, xmax, ymax = bbox
|
|
inside_x = xmin - tolerance <= x <= xmax + tolerance
|
|
inside_y = ymin - tolerance <= y <= ymax + tolerance
|
|
near_vertical = min(abs(x - xmin), abs(x - xmax)) <= tolerance and inside_y
|
|
near_horizontal = min(abs(y - ymin), abs(y - ymax)) <= tolerance and inside_x
|
|
return near_vertical or near_horizontal
|
|
|
|
|
|
async def _collect(backend: AutoCADBackend, scope: str) -> list[tuple[EntityInfo, str]]:
|
|
async def page_all(space: str):
|
|
out = []
|
|
offset = 0
|
|
while True:
|
|
page = await backend.entity_list(limit=_PAGE, offset=offset)
|
|
out.extend((info, space) for info in page)
|
|
if len(page) < _PAGE:
|
|
return out
|
|
offset += _PAGE
|
|
|
|
layouts = await backend.layout_list()
|
|
current = layouts.get("current") or "Model"
|
|
if scope != "all":
|
|
return await page_all(current)
|
|
entities = []
|
|
try:
|
|
for name in layouts.get("layouts", [current]):
|
|
await backend.layout_set_current(name)
|
|
entities.extend(await page_all(name))
|
|
finally:
|
|
await backend.layout_set_current(current)
|
|
return entities
|
|
|
|
|
|
async def build_graph(
|
|
backend: AutoCADBackend,
|
|
tolerance: float = 0.5,
|
|
label_search: float = 15.0,
|
|
scope: str = "current_space",
|
|
include_foreign: bool = True,
|
|
include_geometry: bool = False,
|
|
) -> dict:
|
|
if tolerance <= 0:
|
|
raise ValueError("tolerance must be > 0")
|
|
if scope not in ("current_space", "all"):
|
|
raise ValueError("scope must be 'current_space' or 'all'")
|
|
entities = await _collect(backend, scope)
|
|
layer_linetypes = {lyr.name.upper(): lyr.linetype.upper() for lyr in await backend.layer_list()}
|
|
|
|
# 1 — classify INSERTs
|
|
nodes: dict[str, _Node] = {}
|
|
candidates: dict[str, _Node] = {} # unknown blocks kept only if a line touches them
|
|
for info, space in entities:
|
|
if info.type not in INSERT_TYPES:
|
|
continue
|
|
name = str(info.properties.get("block_name", ""))
|
|
if name.startswith("PID_MARKER_"):
|
|
continue
|
|
payload = await read_payload(backend, info.handle)
|
|
try:
|
|
attributes = await backend.block_get_attributes(info.handle)
|
|
except Exception:
|
|
attributes = {}
|
|
ins = info.properties.get("insertion", [0.0, 0.0])
|
|
rotation = float(info.properties.get("rotation_deg", 0.0))
|
|
scale = float(info.properties.get("x_scale", 1.0))
|
|
y_scale = float(info.properties.get("y_scale", scale))
|
|
base = dict(
|
|
id=info.handle,
|
|
block_name=name,
|
|
x=float(ins[0]),
|
|
y=float(ins[1]),
|
|
rotation_deg=rotation,
|
|
scale=scale,
|
|
layer=info.layer,
|
|
space=space,
|
|
bbox=_bbox(info),
|
|
attributes=attributes,
|
|
)
|
|
if payload and payload.get("kind") == "symbol":
|
|
source = "catalog" if name.startswith("PID_") else "xdata"
|
|
family = payload.get("family")
|
|
node = _Node(
|
|
kind=FAMILY_KIND.get(family, "equipment"),
|
|
family=family,
|
|
symbol=payload.get("symbol"),
|
|
source=source,
|
|
confidence=CONFIDENCE[source],
|
|
**base,
|
|
)
|
|
for pname, (lx, ly, ldir, kind, radius) in payload.get("ports", {}).items():
|
|
# Measured from the INSERT as it is: a mirrored symbol carries
|
|
# y_scale = -1 and its ports sit on the mirrored geometry.
|
|
local = Port(pname, float(lx), float(ly), ldir, kind, float(radius))
|
|
try:
|
|
port = transform_port(local, node.x, node.y, rotation, scale, y_scale)
|
|
except ValueError:
|
|
# A stretched INSERT (someone scaled one axis by hand): the
|
|
# reader never refuses a drawing, so it measures along the
|
|
# X factor and says so instead of raising.
|
|
port = transform_port(local, node.x, node.y, rotation, scale)
|
|
port["inferred"] = True
|
|
node.ports[pname] = {**port, "edges": []}
|
|
nodes[node.id] = node
|
|
continue
|
|
if not include_foreign:
|
|
continue
|
|
keyword_kind = _keyword_kind(name)
|
|
tag_key = next((k for k in ATTRIBUTE_TAGS if k in attributes), None)
|
|
if keyword_kind or tag_key:
|
|
node = _Node(
|
|
kind=keyword_kind or "equipment",
|
|
family=None,
|
|
symbol=None,
|
|
source="heuristic",
|
|
confidence=CONFIDENCE["heuristic"],
|
|
**base,
|
|
)
|
|
nodes[node.id] = node
|
|
elif base["bbox"] is not None:
|
|
candidates[info.handle] = _Node(
|
|
kind="unknown_block",
|
|
family=None,
|
|
symbol=None,
|
|
source="inferred",
|
|
confidence=CONFIDENCE["inferred"],
|
|
**base,
|
|
)
|
|
|
|
# 2 — candidate edges
|
|
edges: dict[str, _Edge] = {}
|
|
foreign_edges: dict[str, _Edge] = {}
|
|
for info, space in entities:
|
|
if info.type not in POLY_TYPES | LINE_TYPES:
|
|
continue
|
|
verts = _vertices(info)
|
|
if len(verts) < 2:
|
|
continue
|
|
payload = await read_payload(backend, info.handle) if info.type in POLY_TYPES else None
|
|
layer = info.layer.upper()
|
|
edge = _Edge(
|
|
id=info.handle,
|
|
vertices=verts,
|
|
bulges=_bulges(info, len(verts)),
|
|
length=_length(info),
|
|
layer=info.layer,
|
|
linetype=info.linetype,
|
|
space=space,
|
|
payload=payload if payload and payload.get("kind") == "line" else None,
|
|
source="touch",
|
|
)
|
|
if edge.payload:
|
|
edge.source = "xdata"
|
|
edge.line_class, edge.class_source = edge.payload.get("class"), "xdata"
|
|
edge.line_number = edge.payload.get("number")
|
|
edge.number_source = "xdata" if edge.line_number else None
|
|
edges[edge.id] = edge
|
|
elif layer in PID_LINE_LAYERS:
|
|
edge.source = "layer"
|
|
cls = LAYER_TO_CLASS[layer]
|
|
if cls is None:
|
|
own = (info.linetype or "").upper()
|
|
effective = own if own and own != "BYLAYER" else layer_linetypes.get(layer, "")
|
|
cls = "electric" if "DASH" in effective else "signal_unknown"
|
|
edge.line_class, edge.class_source = cls, "layer"
|
|
edges[edge.id] = edge
|
|
elif include_foreign:
|
|
foreign_edges[edge.id] = edge
|
|
|
|
# 3 — spatial index of ports (edge vertices join once membership is settled)
|
|
grid = _Grid(cell=10.0 * tolerance)
|
|
radial_ports: list[tuple[str, str, float, float, float]] = []
|
|
for node in nodes.values():
|
|
for pname, port in node.ports.items():
|
|
if port.get("radius", 0.0) > 0:
|
|
# a bubble's line starts on its circle, not at its centre
|
|
radial_ports.append((node.id, pname, port["x"], port["y"], port["radius"]))
|
|
else:
|
|
grid.add(port["x"], port["y"], ("port", node.id, pname))
|
|
|
|
# 4 — resolve endpoints
|
|
junctions: list[dict] = []
|
|
dangling: list[dict] = []
|
|
|
|
def _end(edge: _Edge, end: str) -> tuple[float, float]:
|
|
return edge.vertices[0] if end == "from" else edge.vertices[-1]
|
|
|
|
def _attach(edge: _Edge, end: str) -> dict | None:
|
|
"""``{node, port}`` when this end sits on a port or a foreign block's
|
|
bounding box; None otherwise. Only a successful match has side effects."""
|
|
x, y = _end(edge, end)
|
|
for node_id, pname, cx, cy, r in radial_ports:
|
|
if abs(math.hypot(x - cx, y - cy) - r) <= tolerance:
|
|
nodes[node_id].ports[pname]["edges"].append(edge.id)
|
|
return {"node": node_id, "port": pname}
|
|
best = None
|
|
for d, item in grid.near(x, y, tolerance):
|
|
if item[0] == "port" and (best is None or d < best[0]):
|
|
best = (d, item)
|
|
if best:
|
|
_, (_kind, node_id, pname) = best
|
|
nodes[node_id].ports[pname]["edges"].append(edge.id)
|
|
return {"node": node_id, "port": pname}
|
|
for node in list(nodes.values()) + list(candidates.values()):
|
|
if (
|
|
node.source in ("heuristic", "inferred")
|
|
and node.bbox
|
|
and _on_bbox_boundary((x, y), node.bbox, tolerance)
|
|
):
|
|
if node.id in candidates:
|
|
nodes[node.id] = candidates.pop(node.id)
|
|
pname = f"p{len(node.ports) + 1}"
|
|
node.ports[pname] = {
|
|
"name": pname,
|
|
"x": x,
|
|
"y": y,
|
|
"direction_deg": None,
|
|
"kind": "process",
|
|
"radius": 0.0,
|
|
"inferred": True,
|
|
"edges": [edge.id],
|
|
}
|
|
grid.add(x, y, ("port", node.id, pname))
|
|
return {"node": node.id, "port": pname}
|
|
return None
|
|
|
|
def _junction_at(x, y, edge_id, other_id) -> dict:
|
|
for j in junctions:
|
|
if math.hypot(j["x"] - x, j["y"] - y) <= tolerance:
|
|
for e in (edge_id, other_id):
|
|
if e not in j["edges"]:
|
|
j["edges"].append(e)
|
|
return {"junction": j["id"]}
|
|
j = {"id": f"J{len(junctions) + 1}", "x": x, "y": y, "edges": [edge_id, other_id]}
|
|
junctions.append(j)
|
|
return {"junction": j["id"]}
|
|
|
|
def _junction_or_dangling(edge: _Edge, end: str) -> dict | None:
|
|
x, y = _end(edge, end)
|
|
for _d, item in grid.near(x, y, tolerance):
|
|
if item[0] == "vertex" and item[1] != edge.id:
|
|
return _junction_at(x, y, edge.id, item[1])
|
|
for other in edges.values():
|
|
if other.id == edge.id:
|
|
continue
|
|
for a, b, bulge in other.segments():
|
|
if _point_segment_distance((x, y), a, b, bulge) <= tolerance:
|
|
return _junction_at(x, y, edge.id, other.id)
|
|
nearest = None
|
|
for d, item in grid.near(x, y, 10.0 * tolerance):
|
|
if item[0] == "port" and (nearest is None or d < nearest[0]):
|
|
nearest = (d, item)
|
|
dangling.append(
|
|
{
|
|
"edge": edge.id,
|
|
"end": end,
|
|
"x": x,
|
|
"y": y,
|
|
"nearest": (
|
|
{"node": nearest[1][1], "port": nearest[1][2], "distance": round(nearest[0], 6)}
|
|
if nearest
|
|
else None
|
|
),
|
|
}
|
|
)
|
|
return None
|
|
|
|
# 4a — node attachment first, so a foreign line's membership (spec §9.2
|
|
# step 4c: it is an edge only when an end touches a node) is settled
|
|
# before any junction exists. Resolving junctions first let a P&ID line
|
|
# end on a foreign line that was then discarded, leaving a `{"junction"}`
|
|
# reference to nothing and one dangling end short.
|
|
for edge in list(edges.values()):
|
|
edge.from_ref = _attach(edge, "from")
|
|
edge.to_ref = _attach(edge, "to")
|
|
for edge in list(foreign_edges.values()):
|
|
from_ref = _attach(edge, "from")
|
|
to_ref = _attach(edge, "to")
|
|
if from_ref or to_ref:
|
|
edge.from_ref, edge.to_ref = from_ref, to_ref
|
|
edges[edge.id] = edge
|
|
# 4b — junctions and dangling ends over the edges that are in the graph
|
|
for edge in edges.values():
|
|
for v in edge.vertices:
|
|
grid.add(v[0], v[1], ("vertex", edge.id))
|
|
for edge in edges.values():
|
|
if edge.from_ref is None:
|
|
edge.from_ref = _junction_or_dangling(edge, "from")
|
|
if edge.to_ref is None:
|
|
edge.to_ref = _junction_or_dangling(edge, "to")
|
|
|
|
# 5 — tags and line numbers from attributes or nearby text
|
|
texts = [(info, space) for info, space in entities if info.type in TEXT_TYPES]
|
|
|
|
def _text_near(x, y, radius, pattern=None):
|
|
best = None
|
|
for info, _space in texts:
|
|
ins = info.properties.get("insertion")
|
|
content = str(info.properties.get("text", "")).strip()
|
|
if not ins or not content or (pattern and not pattern.match(content)):
|
|
continue
|
|
d = math.hypot(float(ins[0]) - x, float(ins[1]) - y)
|
|
if d <= radius and (best is None or d < best[0]):
|
|
best = (d, content)
|
|
return best[1] if best else None
|
|
|
|
for node in nodes.values():
|
|
attrs = node.attributes
|
|
if node.kind == "instrument" and attrs.get("FUNC"):
|
|
node.tag = f"{attrs['FUNC']}-{attrs.get('LOOP', '')}".rstrip("-")
|
|
node.tag_source = "attribute"
|
|
else:
|
|
key = next((k for k in ATTRIBUTE_TAGS if attrs.get(k)), None)
|
|
if key:
|
|
node.tag, node.tag_source = str(attrs[key]).strip(), "attribute"
|
|
if node.tag is None:
|
|
radius = max((p.get("radius", 0.0) for p in node.ports.values()), default=0.0)
|
|
if node.kind == "instrument" and radius:
|
|
found = _text_near(node.x, node.y, radius)
|
|
else:
|
|
cx = ((node.bbox[0] + node.bbox[2]) / 2.0) if node.bbox else node.x
|
|
top = node.bbox[3] if node.bbox else node.y
|
|
found = _text_near(cx, top, label_search)
|
|
if found:
|
|
node.tag, node.tag_source = found, "text"
|
|
for edge in edges.values():
|
|
if edge.line_number:
|
|
continue
|
|
a, b = max(zip(edge.vertices, edge.vertices[1:], strict=False), key=lambda s: math.dist(*s))
|
|
found = _text_near((a[0] + b[0]) / 2.0, (a[1] + b[1]) / 2.0, label_search, LINE_NUMBER_RE)
|
|
if found:
|
|
edge.line_number, edge.number_source = found, "label"
|
|
|
|
# 6 — off-page links
|
|
links: dict[str, list[str]] = {}
|
|
for node in nodes.values():
|
|
link = node.attributes.get("LINK")
|
|
if node.kind == "connector" and link:
|
|
links.setdefault(str(link), []).append(node.id)
|
|
offpage_links = [{"link": k, "nodes": sorted(v)} for k, v in sorted(links.items())]
|
|
|
|
# 7 — assemble
|
|
node_rows = [n.to_dict() for n in nodes.values()]
|
|
edge_rows = [e.to_dict(include_geometry) for e in edges.values()]
|
|
by_kind: dict[str, int] = {}
|
|
for n in node_rows:
|
|
by_kind[n["kind"]] = by_kind.get(n["kind"], 0) + 1
|
|
by_class: dict[str, int] = {}
|
|
for e in edge_rows:
|
|
key = e["line_class"] or "unknown"
|
|
by_class[key] = by_class.get(key, 0) + 1
|
|
return {
|
|
"nodes": node_rows,
|
|
"edges": edge_rows,
|
|
"junctions": junctions,
|
|
"dangling": dangling,
|
|
"unclassified": [{"handle": c.id, "block_name": c.block_name} for c in candidates.values()],
|
|
"offpage_links": offpage_links,
|
|
"stats": {
|
|
"nodes_by_kind": dict(sorted(by_kind.items())),
|
|
"edges_by_class": dict(sorted(by_class.items())),
|
|
"dangling": len(dangling),
|
|
"unclassified": len(candidates),
|
|
"confidence_min": min((n["confidence"] for n in node_rows), default=1.0),
|
|
"entities_scanned": len(entities),
|
|
},
|
|
"tolerance": tolerance,
|
|
"scope": scope,
|
|
"backend": backend.name,
|
|
}
|