fix(task-13): mirrored-symbol ports and COM-visible existing lines

Two review findings on pid_line_draw, both root-caused:

1. existing_pid_lines() filtered entity_list(type_filter="LWPOLYLINE") and
   tested info.type == "LWPOLYLINE". A live LWPOLYLINE's ActiveX ObjectName
   is AcDbPolyline (the repo's own 2026-08-06 measurement), which the COM
   engine reports as POLYLINE, so on live AutoCAD every auto-built number was
   seq 1, crossings always 0 and port_reuse always []. The reader now lists
   without a type filter and accepts {LWPOLYLINE, POLYLINE, 2DPOLYLINE} plus
   LINE — the same set the Task 14 graph reader keeps. The COM _entity_info
   polyline branch matched only the DXF-flavoured "AcDbLWPolyline", so a live
   LWPOLYLINE never got `points` either; it now accepts both spellings (with
   the OCS normalisation), which is what makes the crossing count real there.

2. resolve_endpoint() applied only x_scale, so a mirrored INSERT
   (entity_mirror writes y_scale = -1) resolved its port to a point off the
   symbol and the line was drawn through the pump body. transform_port takes
   an optional y_scale (default: uniform, signature unchanged) and applies the
   full reflect-then-rotate transform to point, direction and radius; a
   stretched symbol (|x_scale| != |y_scale|) has no circle for a radial port
   and is refused with a clear ValueError instead of a silent wrong number.

Tests: mirror about both axes resolves onto the real geometry, a stretched
copy is refused, and a fake-COM AcDbPolyline probe sees the existing line
with its seq, vertices and from-ref.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-15 23:48:35 +03:00
co-authored by Claude Opus 5
parent e8f1a666f8
commit e924b5db99
5 changed files with 243 additions and 40 deletions
+16 -6
View File
@@ -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):
+43 -24
View File
@@ -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
+42 -8
View File
@@ -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,
}
+127 -2
View File
@@ -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"}
+15
View File
@@ -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"] == {