diff --git a/backends/com_backend.py b/backends/com_backend.py index aafe0aa..db4c2c6 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -348,6 +348,54 @@ def _com_bulges(entity, coords, closed: bool, length) -> list[float]: return [_com_bulge(entity, i) for i in range(count)] +def _com_geometry_bbox(entity) -> dict | None: + """WCS extents of a block reference's *drawn* geometry, attributes excluded. + + ``GetBoundingBox`` on an INSERT takes its ATTRIBs with it, so a TAG lettered + above a valve pushes the box past the body — a line ending on the body's + real edge then lies *inside* the box. The definition's members are read in + block space (skipping ``AcDbAttributeDefinition``), each member's box is + carried through the INSERT (origin, scale, rotation) corner by corner, and + the extents of those corners is the answer. None when there is nothing + drawn or ActiveX refuses any step — the caller then omits the key. + """ + try: + try: + doc = entity.Document + except Exception: + doc = _acad_doc() + blk = doc.Blocks.Item(entity.Name) + try: + origin = tuple(blk.Origin) + except Exception: + origin = (0.0, 0.0, 0.0) + ins = entity.InsertionPoint + sx, sy = float(entity.XScaleFactor), float(entity.YScaleFactor) + rot = float(entity.Rotation) + cos_r, sin_r = math.cos(rot), math.sin(rot) + xs: list[float] = [] + ys: list[float] = [] + for i in range(blk.Count): + member = blk.Item(i) + if member.ObjectName == "AcDbAttributeDefinition": + continue + try: + lo, hi = member.GetBoundingBox() + except Exception: + continue + for bx, by in ((lo[0], lo[1]), (hi[0], lo[1]), (lo[0], hi[1]), (hi[0], hi[1])): + lx = (float(bx) - float(origin[0])) * sx + ly = (float(by) - float(origin[1])) * sy + xs.append(float(ins[0]) + lx * cos_r - ly * sin_r) + ys.append(float(ins[1]) + lx * sin_r + ly * cos_r) + if not xs: + return None + return {"min": [min(xs), min(ys)], "max": [max(xs), max(ys)]} + except Exception as exc: + log.debug("geometry bbox failed for block reference: %s", exc) + return None + + # 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 @@ -449,6 +497,9 @@ def _entity_info(entity) -> EntityInfo: props["x_scale"] = entity.XScaleFactor props["y_scale"] = entity.YScaleFactor props["rotation_deg"] = rad2deg(entity.Rotation) + geometry_bbox = _com_geometry_bbox(entity) + if geometry_bbox is not None: + props["geometry_bbox"] = geometry_bbox except Exception as exc: log.debug("Type-specific entity properties extraction failed: %s", exc) diff --git a/backends/ezdxf_backend.py b/backends/ezdxf_backend.py index 7d75bc8..3254222 100644 --- a/backends/ezdxf_backend.py +++ b/backends/ezdxf_backend.py @@ -845,6 +845,24 @@ def _entity_info_dxf(ent) -> EntityInfo: props["x_scale"] = ent.dxf.xscale if ent.dxf.hasattr("xscale") else 1.0 props["y_scale"] = ent.dxf.yscale if ent.dxf.hasattr("yscale") else 1.0 props["rotation_deg"] = ent.dxf.get("rotation", 0.0) + # The drawn symbol alone. ``bounding_box`` below takes the ATTRIBs + # with it, so a TAG above a valve pushes the box past the body and + # a line ending on the body's real edge reads as *inside* it. + # `virtual_entities` already applies the INSERT transform, so this + # is WCS; the key is omitted when the block draws nothing. + try: + from ezdxf import bbox as _bbox + + _gb = _bbox.extents( + e for e in ent.virtual_entities() if e.dxftype() not in ("ATTRIB", "ATTDEF") + ) + if _gb.has_data: + props["geometry_bbox"] = { + "min": [float(_gb.extmin.x), float(_gb.extmin.y)], + "max": [float(_gb.extmax.x), float(_gb.extmax.y)], + } + except Exception as exc: + log.debug("geometry bbox failed for INSERT %s: %s", ent.dxf.name, exc) elif ent_type == "ELLIPSE": props["center"] = _v2(ent.dxf.center) props["major_axis"] = _v2(ent.dxf.major_axis) diff --git a/engineering/pid/graph.py b/engineering/pid/graph.py index e4e23b3..841bc92 100644 --- a/engineering/pid/graph.py +++ b/engineering/pid/graph.py @@ -231,9 +231,12 @@ def _point_segment_distance(p, a, b, bulge: float = 0.0) -> float: 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. + # positive bulge bows to the right of a→b, so its centre is to the left — + # and ``radius - sagitta`` goes negative past the semicircle, which is + # exactly what carries the centre across the chord. ``copysign`` used to + # discard that sign, so a bulge-2 arc was tested as its own reflection. nx, ny = -dy / chord, dx / chord # left of a→b - offset = math.copysign(radius - sagitta, bulge) + offset = (radius - sagitta) * (1.0 if bulge > 0 else -1.0) 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) @@ -253,20 +256,30 @@ def _keyword_kind(block_name: str) -> str | None: def _bbox(info: EntityInfo): - bb = info.properties.get("bounding_box") + """The foreign symbol's box: its drawn geometry when the engine reports it. + + ``bounding_box`` takes the ATTRIBs with it, so a TAG lettered above a + valve pushes the box past the body and a line ending on the body's real + edge on that side lies *inside* the box. Both engines report + ``geometry_bbox`` (attributes excluded) for an INSERT that draws anything; + the attribute-inclusive box is the fallback for a reader that lacks it. + """ + bb = info.properties.get("geometry_bbox") or 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: +def _touches_bbox(p, bbox, tolerance) -> bool: + """Within ``tolerance`` of the box's boundary, or strictly inside it. + + A line drawn into the symbol is a connection just as one that stops on + its edge is: foreign P&IDs overshoot into the body as often as they meet + it, and a box tested on its boundary alone called those ends dangling. + """ 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 + return xmin - tolerance <= x <= xmax + tolerance and ymin - tolerance <= y <= ymax + tolerance async def _collect(backend: AutoCADBackend, scope: str) -> list[tuple[EntityInfo, str]]: @@ -448,8 +461,9 @@ async def build_graph( 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.""" + """``{node, port}`` when this end sits on a port, or on or inside a + foreign block's drawn 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: @@ -463,27 +477,35 @@ async def build_graph( _, (_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} + # Several foreign boxes can hold the same end (a valve drawn inside a + # vessel's outline, a nested detail); the one inserted nearest wins. + node = min( + ( + n + for n in list(nodes.values()) + list(candidates.values()) + if n.source in ("heuristic", "inferred") + and n.bbox + and _touches_bbox((x, y), n.bbox, tolerance) + ), + key=lambda n: math.hypot(x - n.x, y - n.y), + default=None, + ) + if node is not None: + 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: diff --git a/tests/test_com_backend.py b/tests/test_com_backend.py index f17e9d6..7e8088f 100644 --- a/tests/test_com_backend.py +++ b/tests/test_com_backend.py @@ -481,3 +481,126 @@ async def test_a_3dsolid_is_refused_by_name_because_activex_has_no_area_for_it() with pytest.raises(RuntimeError, match="no surface-area member"): await backend.entity_measure("S01") + + +# ── geometry_bbox on a block reference: the drawn body, not its tag ───────── +# +# ``GetBoundingBox`` on an INSERT takes the ATTRIBs with it, so a TAG lettered +# above a valve pushes the box past the body and a line ending on the body's +# real edge reads as *inside* the box. The definition's members are read in +# block space, attribute definitions skipped, and carried through the INSERT. + + +class _Block: + """An AcadBlock: ``Count`` members reachable by ``Item(i)``.""" + + def __init__(self, members, origin=(0.0, 0.0, 0.0)): + self._members = members + self.Count = len(members) + self.Origin = origin + + def Item(self, i): + return self._members[i] + + +def _blockref(name, blocks, insertion=(0.0, 0.0, 0.0), sx=1.0, sy=1.0, rotation=0.0, **extra): + doc = SimpleNamespace(Blocks=SimpleNamespace(Item=lambda n: blocks[n])) + return _Entity( + "AcDbBlockReference", + Name=name, + Document=doc, + InsertionPoint=insertion, + XScaleFactor=sx, + YScaleFactor=sy, + ZScaleFactor=1.0, + Rotation=rotation, + Handle="B1", + Layer="0", + Color=256, + Linetype="ByLayer", + Visible=True, + **extra, + ) + + +def _gate_valve_block(): + body_left = _Entity("AcDbPolyline", GetBoundingBox=lambda: ((-4.0, -2.0, 0.0), (0.0, 2.0, 0.0))) + body_right = _Entity("AcDbPolyline", GetBoundingBox=lambda: ((0.0, -2.0, 0.0), (4.0, 2.0, 0.0))) + tag = _Entity( + "AcDbAttributeDefinition", + GetBoundingBox=lambda: ((0.0, 4.0, 0.0), (10.75, 6.5, 0.0)), + ) + return _Block([body_left, body_right, tag]) + + +def test_com_geometry_bbox_is_the_drawn_body_without_the_attribute(): + from backends.com_backend import _entity_info + + ent = _blockref( + "GATE_VALVE", + {"GATE_VALVE": _gate_valve_block()}, + insertion=(50.0, 50.0, 0.0), + # what ActiveX answers for the INSERT itself: the TAG is in it + GetBoundingBox=lambda: ((46.0, 48.0, 0.0), (60.75, 56.5, 0.0)), + ) + props = _entity_info(ent).properties + assert props["bounding_box"] == {"min": [46.0, 48.0], "max": [60.75, 56.5]} + assert props["geometry_bbox"]["min"] == pytest.approx([46.0, 48.0]) + assert props["geometry_bbox"]["max"] == pytest.approx([54.0, 52.0]) + assert props["block_name"] == "GATE_VALVE" and props["insertion"] == [50.0, 50.0] + + +def test_com_geometry_bbox_follows_the_inserts_scale_rotation_and_block_origin(): + import math + + from backends.com_backend import _com_geometry_bbox + + # scale 2 and a quarter turn: body x∈[-4,4] y∈[-2,2] → x∈[-4,4] y∈[-8,8] + ent = _blockref( + "GATE_VALVE", + {"GATE_VALVE": _gate_valve_block()}, + insertion=(10.0, 20.0, 0.0), + sx=2.0, + sy=2.0, + rotation=math.pi / 2, + ) + box = _com_geometry_bbox(ent) + assert box["min"] == pytest.approx([6.0, 12.0]) and box["max"] == pytest.approx([14.0, 28.0]) + # a mirrored INSERT (YScaleFactor -1) keeps the same extents for a symmetric body + ent = _blockref("GATE_VALVE", {"GATE_VALVE": _gate_valve_block()}, sy=-1.0) + box = _com_geometry_bbox(ent) + assert box["min"] == pytest.approx([-4.0, -2.0]) and box["max"] == pytest.approx([4.0, 2.0]) + # the block's base point is what lands on the insertion point + blk = _Block( + [_Entity("AcDbLine", GetBoundingBox=lambda: ((10.0, 10.0, 0.0), (20.0, 10.0, 0.0)))], + origin=(10.0, 10.0, 0.0), + ) + ent = _blockref("OFFSET", {"OFFSET": blk}, insertion=(100.0, 100.0, 0.0)) + box = _com_geometry_bbox(ent) + assert box["min"] == pytest.approx([100.0, 100.0]) and box["max"] == pytest.approx( + [110.0, 100.0] + ) + + +def test_com_geometry_bbox_is_omitted_when_the_block_draws_nothing_or_activex_refuses(): + from backends.com_backend import _com_geometry_bbox, _entity_info + + only_tag = _Block( + [ + _Entity( + "AcDbAttributeDefinition", GetBoundingBox=lambda: ((0.0, 4.0, 0.0), (5.0, 6.5, 0.0)) + ) + ] + ) + assert _com_geometry_bbox(_blockref("TAG_ONLY", {"TAG_ONLY": only_tag})) is None + props = _entity_info(_blockref("TAG_ONLY", {"TAG_ONLY": only_tag})).properties + assert "geometry_bbox" not in props and props["block_name"] == "TAG_ONLY" + + def _refuse(_name): + raise RuntimeError("Key not found") + + ent = _blockref("MISSING", {}) + ent.Document = SimpleNamespace(Blocks=SimpleNamespace(Item=_refuse)) + assert _com_geometry_bbox(ent) is None + props = _entity_info(ent).properties + assert "geometry_bbox" not in props and props["rotation_deg"] == 0.0 diff --git a/tests/test_pid_graph.py b/tests/test_pid_graph.py index 9db00a7..8ad43fc 100644 --- a/tests/test_pid_graph.py +++ b/tests/test_pid_graph.py @@ -138,6 +138,69 @@ async def test_foreign_blocks_classify_with_reduced_confidence_and_inferred_port assert graph["stats"]["entities_scanned"] >= 3 +async def test_a_foreign_line_ending_on_the_body_behind_the_tag_text_attaches(backend): + """The INSERT's ``bounding_box`` takes the ATTRIB with it, so the TAG + lettered above the valve pushed the box to x~60 and a line ending on the + body's real right edge (54,50) was *inside* the box — and dangled.""" + 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) + blk.add_attdef("TAG", insert=(0, 4), text="", dxfattribs={"height": 2.5}) + valve = (await backend.block_insert("GATE_VALVE", 50, 50, attributes={"TAG": "HV-9"})).handle + props = (await backend.entity_get(valve)).properties + assert props["geometry_bbox"]["min"] == pytest.approx([46.0, 48.0]) + assert props["geometry_bbox"]["max"] == pytest.approx([54.0, 52.0]) + assert props["bounding_box"]["max"][0] > 54.0 # the attribute-inclusive box + line = (await backend.entity_create_line(54, 50, 90, 50, layer="PROCESS-PIPING-MAIN")).handle + graph = await build_graph(backend) + edge = next(e for e in graph["edges"] if e["id"] == line) + assert edge["from"] == {"node": valve, "port": "p1"} + node = next(n for n in graph["nodes"] if n["id"] == valve) + port = node["ports"]["p1"] + assert port["inferred"] is True and (port["x"], port["y"]) == pytest.approx((54.0, 50.0)) + assert (line, "from") not in {(d["edge"], d["end"]) for d in graph["dangling"]} + + +async def test_a_foreign_vessel_takes_lines_on_every_edge_regardless_of_where_its_tag_sits( + backend, +): + blk = backend._doc.blocks.new(name="VESSEL_V") + blk.add_lwpolyline([(-5, -10), (5, -10), (5, 10), (-5, 10)], close=True) + blk.add_attdef("TAG", insert=(0, 13), text="", dxfattribs={"height": 2.5}) + vessel = (await backend.block_insert("VESSEL_V", 100, 100, attributes={"TAG": "V-1"})).handle + props = (await backend.entity_get(vessel)).properties + assert props["geometry_bbox"]["min"] == pytest.approx([95.0, 90.0]) + assert props["geometry_bbox"]["max"] == pytest.approx([105.0, 110.0]) + top = (await backend.entity_create_line(100, 140, 100, 110, layer="PROCESS-PIPING-MAIN")).handle + side = ( + await backend.entity_create_line(140, 100, 105, 100, layer="PROCESS-PIPING-MAIN") + ).handle + graph = await build_graph(backend) + edges = {e["id"]: e for e in graph["edges"]} + assert edges[top]["to"]["node"] == vessel and edges[side]["to"]["node"] == vessel + node = next(n for n in graph["nodes"] if n["id"] == vessel) + assert node["kind"] == "equipment" and node["tag"] == "V-1" + ends = {(round(p["x"], 6), round(p["y"], 6)) for p in node["ports"].values()} + assert ends == {(100.0, 110.0), (105.0, 100.0)} + assert {(d["edge"], d["end"]) for d in graph["dangling"]} == {(top, "from"), (side, "from")} + + +async def test_a_line_drawn_into_a_foreign_body_attaches_to_the_nearest_block(backend): + """An end strictly inside the drawn symbol is a connection, and when two + foreign boxes hold it the one inserted nearest wins.""" + big = backend._doc.blocks.new(name="VESSEL_BIG") + big.add_lwpolyline([(-30, -30), (30, -30), (30, 30), (-30, 30)], close=True) + small = backend._doc.blocks.new(name="GATE_VALVE") + small.add_lwpolyline([(-4, -2), (4, -2), (4, 2), (-4, 2)], close=True) + vessel = (await backend.block_insert("VESSEL_BIG", 0, 0, attributes={"TAG": "V-2"})).handle + valve = (await backend.block_insert("GATE_VALVE", 20, 0, attributes={"TAG": "HV-2"})).handle + line = (await backend.entity_create_line(60, 0, 22, 0, layer="PROCESS-PIPING-MAIN")).handle + graph = await build_graph(backend) + edge = next(e for e in graph["edges"] if e["id"] == line) + assert edge["to"]["node"] == valve + assert next(n for n in graph["nodes"] if n["id"] == vessel)["ports"] == {} + + async def test_include_foreign_false_ignores_untagged_blocks(backend): blk = backend._doc.blocks.new(name="GATE_VALVE") blk.add_line((0, 0), (1, 0)) @@ -258,6 +321,75 @@ def test_point_to_arc_distance_follows_the_bulge(): assert _point_segment_distance((5, 0), a, b, 0.0) == pytest.approx(0.0) # straight +def _arc_distance_sampled(p, a, b, bulge, samples=20000): + """Nearest distance from ``p`` to the bulge arc, measured on ezdxf's own + arc (centre, angles, radius) by dense sampling — an independent answer.""" + from ezdxf.math import Vec2, bulge_to_arc + + centre, start, end, radius = bulge_to_arc(Vec2(a), Vec2(b), bulge) + if end < start: + end += 2.0 * math.pi + return min( + math.dist( + p, + ( + centre.x + radius * math.cos(start + (end - start) * i / samples), + centre.y + radius * math.sin(start + (end - start) * i / samples), + ), + ) + for i in range(samples + 1) + ) + + +def test_point_to_arc_distance_past_a_semicircle_keeps_the_centre_on_the_arcs_side(): + """For |bulge| > 1 the centre crosses the chord (``radius - sagitta`` goes + negative). ``copysign`` threw that sign away, so a bulge-2 arc was tested + as its own reflection: the apex read as far away and empty paper as on it.""" + from ezdxf.math import bulge_center + + from engineering.pid.graph import _point_segment_distance + + a, b = (0.0, 0.0), (10.0, 0.0) + assert tuple(bulge_center(a, b, 2.0)) == pytest.approx((5.0, -3.75)) + assert tuple(bulge_center(a, b, -2.0)) == pytest.approx((5.0, 3.75)) + # centre (5,-3.75), r 6.25 -> apex (5,-10); the mirror image is (5,10) + assert _point_segment_distance((5, -10), a, b, 2.0) == pytest.approx(0.0) + assert _point_segment_distance((5, 10), a, b, 2.0) == pytest.approx( + _arc_distance_sampled((5, 10), a, b, 2.0), abs=1e-6 + ) + assert _point_segment_distance((5, 10), a, b, 2.0) == pytest.approx(math.hypot(5, 10)) + assert _point_segment_distance((5, 10), a, b, -2.0) == pytest.approx(0.0) + assert _point_segment_distance((5, -10), a, b, -2.0) == pytest.approx( + _arc_distance_sampled((5, -10), a, b, -2.0), abs=1e-6 + ) + assert _point_segment_distance((5, -10), a, b, -2.0) == pytest.approx(math.hypot(5, 10)) + + +async def test_a_line_ending_on_the_apex_of_an_arc_past_a_semicircle_is_a_junction(backend): + msp = backend._doc.modelspace() + pipe = msp.add_lwpolyline( + [(-20, 0, 0), (0, 0, 2), (10, 0, 0), (30, 0, 0)], + format="xyb", + dxfattribs={"layer": "PROCESS-PIPING-MAIN"}, + ).dxf.handle + on_arc = ( + await backend.entity_create_line(5, -10, 5, -30, layer="PROCESS-PIPING-SECONDARY") + ).handle + on_paper = ( + await backend.entity_create_line(5, 10, 5, 30, layer="PROCESS-PIPING-SECONDARY") + ).handle + graph = await build_graph(backend) + edges = {e["id"]: e for e in graph["edges"]} + assert edges[on_arc]["from"] and "junction" in edges[on_arc]["from"] + assert edges[on_paper]["from"] is None + assert [j["id"] for j in graph["junctions"]] == [edges[on_arc]["from"]["junction"]] + junction = graph["junctions"][0] + assert (junction["x"], junction["y"]) == pytest.approx((5.0, -10.0)) + assert set(junction["edges"]) == {pipe, on_arc} + dangling = {(d["edge"], d["end"]) for d in graph["dangling"]} + assert (on_paper, "from") in dangling and (on_arc, "from") not in dangling + + async def test_a_mirrored_arc_is_tested_on_the_side_the_drawing_holds(backend): """The bulge is signed in the polyline's own frame, and ``entity_mirror`` leaves that frame a reflection (extrusion -Z). Handing the stored sign