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]