From 8fcd587b5b3a6800914df55a43f1951d235b07f5 Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Wed, 16 Sep 2026 00:30:21 +0300 Subject: [PATCH] fix(task-14): foreign-line membership before junctions, and edges measure the arc the drawing holds 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) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- backends/com_backend.py | 28 +++++++ backends/ezdxf_backend.py | 4 + engineering/pid/graph.py | 155 +++++++++++++++++++++++++++----------- tests/test_pid_graph.py | 115 ++++++++++++++++++++++++++++ 4 files changed, 258 insertions(+), 44 deletions(-) diff --git a/backends/com_backend.py b/backends/com_backend.py index 4a5362c..6af6ec5 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -326,6 +326,28 @@ def _com_bulge(entity, index: int) -> float: return 0.0 +def _com_bulges(entity, coords, closed: bool, length) -> list[float]: + """Every vertex's bulge, read only when ``Length`` says an arc exists. + + ``coords`` is the flat ``Coordinates`` variant (x0, y0, x1, y1, ...). A + circular arc is strictly longer than its chord, so when AutoCAD's own + ``Length`` equals the chord walk the polyline is straight everywhere and + the per-vertex COM round trips are skipped. + """ + count = len(coords) // 2 + pts = [(float(coords[i]), float(coords[i + 1])) for i in range(0, 2 * count, 2)] + chord = sum(math.dist(a, b) for a, b in zip(pts, pts[1:], strict=False)) + if closed and count > 1: + chord += math.dist(pts[-1], pts[0]) + try: + straight = abs(float(length) - chord) <= 1e-9 * max(1.0, chord) + except (TypeError, ValueError): + straight = False + if straight: + return [0.0] * count + return [_com_bulge(entity, i) for i in range(count)] + + # What a lightweight polyline calls itself over ActiveX. The live name is # ``AcDbPolyline`` (measured on AutoCAD 2026, 2026-08-06 — the member profile in # tests/test_com_backend.py); ``AcDbLWPolyline`` is the DXF-flavoured spelling @@ -392,6 +414,12 @@ def _entity_info(entity) -> EntityInfo: props["points"] = pts props["closed"] = bool(entity.Closed) props["length"] = entity.Length + # Per-vertex bulges, parallel to `points` (ezdxf parity). GetBulge + # is one cross-process call per vertex, so it is only paid when + # the polyline can carry an arc: an arc is always longer than its + # chord, so a `Length` equal to the chord walk means every bulge + # is zero and no call is needed. + props["bulges"] = _com_bulges(entity, coords, bool(entity.Closed), props["length"]) elif obj_name == "AcDbText": props["text"] = entity.TextString ins = entity.InsertionPoint diff --git a/backends/ezdxf_backend.py b/backends/ezdxf_backend.py index a39b81a..b9ac24f 100644 --- a/backends/ezdxf_backend.py +++ b/backends/ezdxf_backend.py @@ -813,6 +813,10 @@ def _entity_info_dxf(ent) -> EntityInfo: _verts = [(float(x), float(y), float(b)) for x, y, b in ent.get_points(format="xyb")] _area, _length = measure.polygon_area_perimeter(_verts, bool(ent.closed)) props["length"] = round(_length, 6) + # The bulge of the segment leaving each vertex (DXF convention), so + # a reader that walks `points` can test against the arc the drawing + # holds instead of the chord it does not. Parallel to `points`. + props["bulges"] = [b for _x, _y, b in _verts] if ent.closed: props["area"] = round(_area, 6) elif ent_type == "TEXT": diff --git a/engineering/pid/graph.py b/engineering/pid/graph.py index dce9c09..e4e23b3 100644 --- a/engineering/pid/graph.py +++ b/engineering/pid/graph.py @@ -106,6 +106,8 @@ class _Node: 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 @@ -132,20 +134,34 @@ class _Edge: "insulation": p.get("insulation"), "from": self.from_ref, "to": self.to_ref, - "length": round( - sum( - math.dist(a, b) for a, b in zip(self.vertices, self.vertices[1:], strict=False) - ), - 6, - ), + # 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.""" @@ -178,15 +194,53 @@ def _vertices(info: EntityInfo) -> list[tuple[float, float]]: return [(float(p[0]), float(p[1])) for p in info.properties.get("points") or []] -def _point_segment_distance(p, a, b) -> float: +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 - if dx == 0 and dy == 0: + chord = math.hypot(dx, dy) + if chord < 1e-12: return math.hypot(px - ax, py - ay) - 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)) + 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: @@ -349,6 +403,8 @@ async def build_graph( edge = _Edge( id=info.handle, vertices=verts, + bulges=_bulges(info, len(verts)), + length=_length(info), layer=info.layer, linetype=info.linetype, space=space, @@ -373,7 +429,7 @@ async def build_graph( elif include_foreign: foreign_edges[edge.id] = edge - # 3 — spatial index of ports and vertices + # 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(): @@ -383,28 +439,18 @@ async def build_graph( radial_ports.append((node.id, pname, port["x"], port["y"], port["radius"])) else: grid.add(port["x"], port["y"], ("port", node.id, pname)) - all_edges = {**edges, **foreign_edges} - for edge in all_edges.values(): - for v in edge.vertices: - grid.add(v[0], v[1], ("vertex", edge.id)) # 4 — resolve endpoints junctions: list[dict] = [] dangling: list[dict] = [] - 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 _end(edge: _Edge, end: str) -> tuple[float, float]: + return edge.vertices[0] if end == "from" else edge.vertices[-1] - def _resolve(edge: _Edge, end: str) -> dict | None: - x, y = 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) @@ -438,14 +484,29 @@ async def build_graph( } 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 all_edges.values(): + for other in edges.values(): if other.id == edge.id: continue - for a, b in zip(other.vertices, other.vertices[1:], strict=False): - if _point_segment_distance((x, y), a, b) <= tolerance: + 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): @@ -466,23 +527,29 @@ async def build_graph( ) 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 = _resolve(edge, "from") - edge.to_ref = _resolve(edge, "to") + edge.from_ref = _attach(edge, "from") + edge.to_ref = _attach(edge, "to") for edge in list(foreign_edges.values()): - # a foreign line joins the graph only through a node - probe_from = _resolve(edge, "from") - probe_to = _resolve(edge, "to") - attached = any(ref and "node" in ref for ref in (probe_from, probe_to)) - if attached: - edge.from_ref, edge.to_ref = probe_from, probe_to + 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 - else: - dangling[:] = [d for d in dangling if d["edge"] != edge.id] - for j in junctions: - if edge.id in j["edges"]: - j["edges"].remove(edge.id) - junctions[:] = [j for j in junctions if len(j["edges"]) >= 2] + # 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] diff --git a/tests/test_pid_graph.py b/tests/test_pid_graph.py index 3f7ae68..5a654c0 100644 --- a/tests/test_pid_graph.py +++ b/tests/test_pid_graph.py @@ -2,6 +2,8 @@ from __future__ import annotations +import math + import pytest from engineering.pid.drawlines import draw_line @@ -181,3 +183,116 @@ async def test_geometry_is_optional(backend): assert "vertices" not in lean["edges"][0] full = await build_graph(backend, include_geometry=True) assert len(full["edges"][0]["vertices"]) >= 2 + + +async def test_a_line_ending_on_a_discarded_foreign_line_dangles(backend): + """A foreign line is an edge only when an end touches a node (spec §9.2 4c). + + Junctions used to be resolved before that decision, so a P&ID line ending + on a foreign line that was then dropped kept ``{"junction": "J1"}`` while + ``junctions`` was empty and ``stats.dangling`` was one short — an id + Task 15's line list would dereference into nothing.""" + main = (await backend.entity_create_line(0, 0, 50, 0, layer="PROCESS-PIPING-MAIN")).handle + await backend.entity_create_line(50, -20, 50, 20, layer="0") + graph = await build_graph(backend) + edge = graph["edges"][0] + assert edge["id"] == main and edge["from"] is None and edge["to"] is None + assert graph["junctions"] == [] + assert {(d["edge"], d["end"]) for d in graph["dangling"]} == {(main, "from"), (main, "to")} + assert graph["stats"]["dangling"] == 2 + + +async def test_a_foreign_line_ending_on_another_discarded_foreign_line_dangles(backend): + blk = backend._doc.blocks.new(name="GATE_VALVE") + blk.add_lwpolyline([(-4, -2), (-4, 2), (0, 0)], close=True) + blk.add_lwpolyline([(4, -2), (4, 2), (0, 0)], close=True) + valve = (await backend.block_insert("GATE_VALVE", 50, 50)).handle + kept = (await backend.entity_create_line(46, 50, 10, 50, layer="0")).handle # on the bbox + await backend.entity_create_line(10, 50, 10, 0, layer="0") # touches nothing: not an edge + graph = await build_graph(backend) + assert [e["id"] for e in graph["edges"]] == [kept] + edge = graph["edges"][0] + assert edge["from"] == {"node": valve, "port": "p1"} and edge["to"] is None + assert graph["junctions"] == [] + assert [(d["edge"], d["end"]) for d in graph["dangling"]] == [(kept, "to")] + assert graph["stats"]["dangling"] == 1 + + +async def test_edge_length_is_the_engines_measurement_not_a_chord_walk(backend): + """A curved crossing "jump" is an arc in the drawing; the chord walk over + its vertices was 5.7 units short and said nothing about it. The engine + measures the real geometry, and a line ending *on the arc* is a junction.""" + msp = backend._doc.modelspace() + pipe = msp.add_lwpolyline( + [(-20, 0, 0), (0, 0, 1), (10, 0, 0), (30, 0, 0)], + format="xyb", + dxfattribs={"layer": "PROCESS-PIPING-MAIN"}, + ) + pipe = pipe.dxf.handle + measured = (await backend.entity_get(pipe)).properties["length"] + assert measured == pytest.approx(20 + 5 * math.pi + 20) + # bulge 1 from (0,0) to (10,0) bows to -y: centre (5,0), radius 5 → apex (5,-5) + branch = await backend.entity_create_line(5, -5, 5, -30, layer="PROCESS-PIPING-SECONDARY") + branch = branch.handle + graph = await build_graph(backend) + edges = {e["id"]: e for e in graph["edges"]} + assert edges[pipe]["length"] == pytest.approx(measured) + assert "length_approximate" not in edges[pipe] + assert edges[branch]["from"] and "junction" in edges[branch]["from"] + junction = graph["junctions"][0] + assert (junction["x"], junction["y"]) == pytest.approx((5.0, -5.0)) + assert set(junction["edges"]) == {pipe, branch} + full = await build_graph(backend, include_geometry=True) + assert {e["id"]: e for e in full["edges"]}[pipe]["bulges"] == [0.0, 1.0, 0.0, 0.0] + + +def test_point_to_arc_distance_follows_the_bulge(): + from engineering.pid.graph import _point_segment_distance + + a, b = (0.0, 0.0), (10.0, 0.0) + assert _point_segment_distance((5, -5), a, b, 1.0) == pytest.approx(0.0) # CCW apex + assert _point_segment_distance((5, 5), a, b, -1.0) == pytest.approx(0.0) # CW apex + assert _point_segment_distance((5, 5), a, b, 1.0) == pytest.approx(math.hypot(5, 5)) + assert _point_segment_distance((5, -2.5), a, b, 0.5) == pytest.approx(0.0) # sagitta + assert _point_segment_distance((5, 2.5), a, b, 0.5) == pytest.approx(5.0) + assert _point_segment_distance((5, 0), a, b, 0.0) == pytest.approx(0.0) # straight + + +def test_com_polyline_bulges_are_read_only_when_length_says_an_arc_exists(): + """The COM engine pays one round trip per vertex for GetBulge, so it asks + only when ActiveX ``Length`` exceeds the chord walk — an arc is always + longer than its chord, so equality proves every bulge is zero.""" + from types import SimpleNamespace + + from backends.com_backend import _entity_info + + def fake(coords, length, bulges): + calls = [] + + def get_bulge(i): + calls.append(i) + return bulges[i] + + ent = SimpleNamespace( + ObjectName="AcDbPolyline", + Coordinates=coords, + Normal=(0.0, 0.0, 1.0), + Elevation=0.0, + Handle="2F", + Layer="0", + Color=256, + Linetype="ByLayer", + Visible=True, + Closed=False, + Length=length, + GetBoundingBox=lambda: ((0.0, 0.0, 0.0), (1.0, 1.0, 0.0)), + GetBulge=get_bulge, + ) + return ent, calls + + straight, calls = fake((0.0, 0.0, 10.0, 0.0, 10.0, 10.0), 20.0, [0.0, 0.0, 0.0]) + assert _entity_info(straight).properties["bulges"] == [0.0, 0.0, 0.0] + assert calls == [] + curved, calls = fake((0.0, 0.0, 10.0, 0.0, 10.0, 10.0), 10 + 5 * math.pi, [0.0, 1.0, 0.0]) + assert _entity_info(curved).properties["bulges"] == [0.0, 1.0, 0.0] + assert calls == [0, 1, 2]