diff --git a/backends/com_backend.py b/backends/com_backend.py index 74476bd..4a5362c 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -326,6 +326,15 @@ def _com_bulge(entity, index: int) -> float: return 0.0 +# 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 +# this code once matched on alone, which left every live LWPOLYLINE without +# ``points`` because the branch never fired. Both are accepted so the vertices +# come through whichever spelling a seat reports. +_COM_LWPOLYLINE_NAMES = ("AcDbPolyline", "AcDbLWPolyline") + + def _entity_info(entity) -> EntityInfo: """Convert a COM entity object to EntityInfo dataclass.""" try: @@ -362,16 +371,17 @@ def _entity_info(entity) -> EntityInfo: # Arc length (N3) — parity with ezdxf so length_range selects ARCs. _sweep = (entity.EndAngle - entity.StartAngle) % (2.0 * math.pi) props["length"] = entity.Radius * _sweep - elif obj_name in ("AcDbLWPolyline", "AcDb2dPolyline"): + elif obj_name in _COM_LWPOLYLINE_NAMES or obj_name == "AcDb2dPolyline": coords = list(entity.Coordinates) pts = [[coords[i], coords[i + 1]] for i in range(0, len(coords), 2)] # ActiveX hands back WCS for every point property on this backend # *except* this one: "LightweightPolyline object: the variant is an - # array of 2D points in OCS." Scoped to AcDbLWPolyline deliberately — - # AcDb2dPolyline's Coordinates may be 3D triples, and changing its - # stride on a doc reading alone, with no way to verify against live - # AutoCAD from here, risks breaking a path that works today. - if obj_name == "AcDbLWPolyline": + # array of 2D points in OCS." Scoped to the lightweight polyline + # deliberately — AcDb2dPolyline's Coordinates may be 3D triples, and + # changing its stride on a doc reading alone, with no way to verify + # against live AutoCAD from here, risks breaking a path that works + # today. + if obj_name in _COM_LWPOLYLINE_NAMES: try: normal = tuple(entity.Normal) if not ocs.is_wcs_frame(normal): diff --git a/engineering/pid/drawlines.py b/engineering/pid/drawlines.py index 5a055c5..5586565 100644 --- a/engineering/pid/drawlines.py +++ b/engineering/pid/drawlines.py @@ -27,6 +27,12 @@ if TYPE_CHECKING: TEXT_HEIGHT = 2.5 LABEL_LAYER = "PROCESS-LINE-TEXT" _INSERT_TYPES = {"INSERT", "BLOCKREFERENCE"} +# ezdxf reports a lightweight polyline as LWPOLYLINE; the COM engine derives the +# type from the ActiveX ObjectName, and a live LWPOLYLINE is ``AcDbPolyline`` +# (measured on AutoCAD 2026 — see backends/com_backend.py), so it arrives as +# POLYLINE, with the heavy kind as 2DPOLYLINE. Same set the graph reader keeps. +_POLY_TYPES = {"LWPOLYLINE", "POLYLINE", "2DPOLYLINE"} +_LINE_TYPES = {"LINE"} _PAGE = 1000 @@ -57,13 +63,21 @@ async def resolve_endpoint(backend: AutoCADBackend, ref: dict) -> dict: raise ValueError(f"{handle} has no port {port_name!r}; ports: {', '.join(ports)}") lx, ly, ldir, kind, radius = ports[port_name] ins = info.properties["insertion"] - world = transform_port( - Port(port_name, float(lx), float(ly), ldir, kind, float(radius)), - float(ins[0]), - float(ins[1]), - float(info.properties.get("rotation_deg", 0.0)), - float(info.properties.get("x_scale", 1.0)), - ) + x_scale = float(info.properties.get("x_scale", 1.0)) + # The full INSERT transform: a mirrored symbol (``entity_mirror`` writes + # ``y_scale = -1``) has its ports on the mirrored geometry, not on the + # unmirrored one; a stretched symbol is refused by ``transform_port``. + try: + world = transform_port( + Port(port_name, float(lx), float(ly), ldir, kind, float(radius)), + float(ins[0]), + float(ins[1]), + float(info.properties.get("rotation_deg", 0.0)), + x_scale, + float(info.properties.get("y_scale", x_scale)), + ) + except ValueError as exc: + raise ValueError(f"{handle}: {exc}") from exc return {**world, "handle": handle, "port": port_name, "block_name": block_name} if "x" in ref and "y" in ref: return { @@ -82,29 +96,34 @@ async def resolve_endpoint(backend: AutoCADBackend, ref: dict) -> dict: def _vertices(info) -> list[tuple[float, float]]: - if info.type == "LINE": + 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", [])] + return [(float(p[0]), float(p[1])) for p in info.properties.get("points") or []] async def existing_pid_lines(backend: AutoCADBackend) -> list[dict]: - """Every LINE/LWPOLYLINE on a P&ID line layer, with its payload when it has one.""" + """Every LINE/polyline on a P&ID line layer, with its payload when it has one. + + Listed without a type filter: the two engines name a lightweight polyline + differently (``_POLY_TYPES``), and a filter spelled for one of them is blind + on the other — every auto-built number would restart at 1 and no crossing + or port reuse would ever be reported there. + """ out = [] - for type_filter in ("LWPOLYLINE", "LINE"): - offset = 0 - while True: - page = await backend.entity_list(type_filter=type_filter, limit=_PAGE, offset=offset) - for info in page: - if info.layer.upper() not in PID_LINE_LAYERS: - continue - payload = ( - await read_payload(backend, info.handle) if info.type == "LWPOLYLINE" else None - ) - out.append({"handle": info.handle, "vertices": _vertices(info), "payload": payload}) - if len(page) < _PAGE: - break - offset += _PAGE + offset = 0 + while True: + page = await backend.entity_list(limit=_PAGE, offset=offset) + for info in page: + if info.type not in _POLY_TYPES | _LINE_TYPES: + continue + if info.layer.upper() not in PID_LINE_LAYERS: + continue + payload = await read_payload(backend, info.handle) if info.type in _POLY_TYPES else None + out.append({"handle": info.handle, "vertices": _vertices(info), "payload": payload}) + if len(page) < _PAGE: + break + offset += _PAGE return out diff --git a/engineering/pid/symbols.py b/engineering/pid/symbols.py index e5e27a5..da1cb2a 100644 --- a/engineering/pid/symbols.py +++ b/engineering/pid/symbols.py @@ -176,21 +176,55 @@ def all_specs() -> list[SymbolSpec]: return specs -def transform_port(port: Port, x: float, y: float, rotation_deg: float, scale: float) -> dict: - """Symbol-local port -> WCS, for an INSERT at (x, y) rotated by ``rotation_deg`` - and uniformly scaled.""" +def transform_port( + port: Port, + x: float, + y: float, + rotation_deg: float, + scale: float, + y_scale: float | None = None, +) -> dict: + """Symbol-local port -> WCS, for an INSERT at (x, y) rotated by ``rotation_deg``. + + ``scale`` is the X factor; ``y_scale`` defaults to it (a uniform INSERT). A + negative factor is a mirror — ``entity_mirror`` on an INSERT writes + ``y_scale = -1`` — and is applied in full, to the point, the direction and + the radius, so the port lands where the mirrored geometry actually is. The + magnitudes must match: a stretched symbol has no circle for a radial port to + sit on, so it is refused instead of resolving to a point off the symbol. + """ + sx = float(scale) + sy = sx if y_scale is None else float(y_scale) + if abs(abs(sx) - abs(sy)) > 1e-9: + raise ValueError( + f"non-uniform INSERT scale (x_scale={sx:g}, y_scale={sy:g}): P&ID symbols " + "are placed with one scale factor (a mirror may flip its sign); re-insert " + "the symbol with pid_symbol_insert" + ) a = math.radians(rotation_deg) - lx, ly = port.x * scale, port.y * scale - wx = x + lx * math.cos(a) - ly * math.sin(a) - wy = y + lx * math.sin(a) + ly * math.cos(a) - direction = None if port.direction_deg is None else (port.direction_deg + rotation_deg) % 360.0 + cos_a, sin_a = math.cos(a), math.sin(a) + lx, ly = port.x * sx, port.y * sy + wx = x + lx * cos_a - ly * sin_a + wy = y + lx * sin_a + ly * cos_a + if port.direction_deg is None: + direction = None + else: + # Reflect first (a flipped X negates the angle about 90°, a flipped Y + # negates it about 0°), then rotate — exact, so the uniform case keeps + # the same numbers it always had. + d = float(port.direction_deg) + if sx < 0: + d = 180.0 - d + if sy < 0: + d = -d + direction = (d + rotation_deg) % 360.0 return { "name": port.name, "x": wx, "y": wy, "direction_deg": direction, "kind": port.kind, - "radius": port.radius * scale, + "radius": port.radius * abs(sx), "inferred": False, } diff --git a/tests/test_pid_line_draw.py b/tests/test_pid_line_draw.py index f3aaf3e..4e82522 100644 --- a/tests/test_pid_line_draw.py +++ b/tests/test_pid_line_draw.py @@ -2,11 +2,20 @@ from __future__ import annotations +from types import SimpleNamespace + import pytest -from engineering.pid.drawlines import draw_line +from backends.com_backend import ComBackend +from engineering.pid.drawlines import draw_line, existing_pid_lines, resolve_endpoint from engineering.pid.insert import place_symbol -from engineering.pid.xdata import read_payload +from engineering.pid.xdata import ( + APP_NAME, + encode_payload, + line_payload, + read_payload, + write_payload, +) pytestmark = pytest.mark.asyncio @@ -138,3 +147,119 @@ async def test_refusals(backend): with pytest.raises(ValueError, match="not a P&ID symbol"): line = await backend.entity_create_line(0, 0, 1, 1) await draw_line(backend, {"handle": line.handle, "port": "in"}, {"x": 0, "y": 0}) + + +async def test_mirrored_symbol_ports_follow_the_mirrored_geometry(backend): + """``entity_mirror`` on an INSERT writes ``y_scale = -1``; the port has to be + on the mirrored pump (discharge stub at (200, 104)->(200, 106)), not at the + unmirrored offset — a line attached at y=94 runs through the pump body.""" + pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-1") + about_vertical = await backend.entity_mirror(pump["handle"], 150, 0, 150, 200) + info = await backend.entity_get(about_vertical.handle) + assert info.properties["y_scale"] == -1.0, "the fixture must exercise a real mirror" + ep = await resolve_endpoint(backend, {"handle": about_vertical.handle, "port": "discharge"}) + assert (ep["x"], ep["y"], ep["direction_deg"]) == (200.0, 106.0, 90.0) + suction = await resolve_endpoint(backend, {"handle": about_vertical.handle, "port": "suction"}) + assert (suction["x"], suction["y"], suction["direction_deg"]) == (204.0, 100.0, 0.0) + result = await draw_line( + backend, {"handle": about_vertical.handle, "port": "discharge"}, {"x": 200, "y": 160} + ) + assert result["vertices"][0] == [200.0, 106.0] + + about_horizontal = await backend.entity_mirror(pump["handle"], 0, 150, 300, 150) + ep = await resolve_endpoint(backend, {"handle": about_horizontal.handle, "port": "discharge"}) + assert (ep["x"], ep["y"], ep["direction_deg"]) == (100.0, 194.0, 270.0) + + +async def test_stretched_symbol_is_refused_not_resolved_off_the_geometry(backend): + pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-1") + payload = await read_payload(backend, pump["handle"]) + stretched = await backend.block_insert(pump["block_name"], 300, 100, 2.0, 1.0, 0.0) + await write_payload(backend, stretched.handle, payload) + with pytest.raises(ValueError, match="non-uniform"): + await resolve_endpoint(backend, {"handle": stretched.handle, "port": "discharge"}) + with pytest.raises(ValueError, match="non-uniform"): + await draw_line( + backend, {"handle": stretched.handle, "port": "discharge"}, {"x": 0, "y": 0} + ) + + +# ── the COM engine names a lightweight polyline ``AcDbPolyline`` ───────────── + + +class _ComEntity: + """An ActiveX entity with only the members it is given.""" + + def __init__(self, object_name, **members): + self.ObjectName = object_name + for name, value in members.items(): + setattr(self, name, value) + + def __getattr__(self, name): + raise AttributeError(name) + + +def _com_pid_polyline(handle, coords, payload): + tags = encode_payload(payload) + codes = [1001] + [1000] * len(tags) + values = [APP_NAME] + tags + return _ComEntity( + "AcDbPolyline", + Handle=handle, + Layer="PROCESS-PIPING-MAIN", + Color=256, + Linetype="ByLayer", + Visible=True, + Closed=False, + Length=1.0, + Coordinates=tuple(coords), + Normal=(0.0, 0.0, 1.0), + Elevation=0.0, + GetBoundingBox=lambda: ((0.0, 0.0, 0.0), (1.0, 1.0, 0.0)), + GetXData=lambda app: (codes, values), + ) + + +def _fake_com(monkeypatch, entities): + backend = ComBackend() + + async def run_inline(func, *args, **kwargs): + return func(*args, **kwargs) + + monkeypatch.setattr(backend, "_run", run_inline) + by_handle = {ent.Handle: ent for ent in entities} + space = SimpleNamespace(Count=len(entities), Item=lambda i: entities[i]) + doc = SimpleNamespace(HandleToObject=lambda handle: by_handle[handle]) + monkeypatch.setattr("backends.com_backend._msp", lambda: space) + monkeypatch.setattr("backends.com_backend._acad_doc", lambda: doc) + return backend + + +async def test_existing_lines_are_seen_on_the_com_engine(monkeypatch): + """A live LWPOLYLINE's ObjectName is ``AcDbPolyline`` (measured on AutoCAD + 2026), which the COM engine reports as type POLYLINE. A reader that filters + on LWPOLYLINE sees nothing there: every number restarts at seq 1, crossings + stay 0 and port reuse is never reported.""" + payload = line_payload( + "process_major", + "100-P-7", + "100", + "P", + None, + None, + 7, + {"handle": "2A", "port": "discharge"}, + None, + ) + poly = _com_pid_polyline("3E", (100.0, 106.0, 100.0, 128.0, 207.0, 128.0), payload) + other_layer = _com_pid_polyline("3F", (0.0, 0.0, 1.0, 1.0), payload) + other_layer.Layer = "GEOMETRY" + backend = _fake_com(monkeypatch, [poly, other_layer]) + + assert (await backend.entity_get("3E")).type == "POLYLINE" + lines = await existing_pid_lines(backend) + + assert [ln["handle"] for ln in lines] == ["3E"] + assert lines[0]["vertices"] == [(100.0, 106.0), (100.0, 128.0), (207.0, 128.0)] + assert lines[0]["payload"]["seq"] == 7 + assert lines[0]["payload"]["from"] == {"handle": "2A", "port": "discharge"} diff --git a/tests/test_pid_symbols.py b/tests/test_pid_symbols.py index 9c76d11..9558ede 100644 --- a/tests/test_pid_symbols.py +++ b/tests/test_pid_symbols.py @@ -194,6 +194,21 @@ def test_transform_port_applies_rotation_and_scale_about_the_insertion(): assert radial["direction_deg"] is None and radial["radius"] == pytest.approx(15.0) +def test_transform_port_applies_a_mirror_and_refuses_a_stretch(): + port = Port("out", 0.0, 6.0, 90.0, "process") + flipped_y = transform_port(port, 100.0, 200.0, 0.0, 1.0, -1.0) + assert (flipped_y["x"], flipped_y["y"], flipped_y["direction_deg"]) == (100.0, 194.0, 270.0) + mirrored = transform_port(port, 200.0, 100.0, 180.0, 1.0, -1.0) # entity_mirror about x=150 + assert (mirrored["x"], mirrored["y"]) == pytest.approx((200.0, 106.0)) + assert mirrored["direction_deg"] == pytest.approx(90.0) + flipped_x = transform_port(Port("in", -4.0, 0.0, 180.0, "process"), 0.0, 0.0, 0.0, -1.0, 1.0) + assert (flipped_x["x"], flipped_x["y"], flipped_x["direction_deg"]) == (4.0, 0.0, 0.0) + radial = transform_port(Port("signal", 0.0, 0.0, None, "signal", 5.0), 0.0, 0.0, 0.0, -2.0) + assert radial["radius"] == pytest.approx(10.0) + with pytest.raises(ValueError, match="non-uniform"): + transform_port(port, 0.0, 0.0, 0.0, 2.0, 1.0) + + def test_list_symbols_reports_variants_and_ports(): rows = {row["symbol"]: row for row in list_symbols()} assert rows["instrument"]["variants"] == {