diff --git a/engineering/pid/drawlines.py b/engineering/pid/drawlines.py index 149922d..cbbecb8 100644 --- a/engineering/pid/drawlines.py +++ b/engineering/pid/drawlines.py @@ -13,6 +13,7 @@ from .lines import ( PID_LINE_LAYERS, aim_points, count_crossings, + flatten_bulges, format_line_number, label_placement, marker_positions, @@ -106,13 +107,57 @@ def _vertices(info) -> list[tuple[float, float]]: return [(float(p[0]), float(p[1])) for p in info.properties.get("points") or []] +def _bulges(info, count: int) -> list[float]: + """Per-vertex bulges as the engine reports them (both engines put a + ``bulges`` list next to ``points``); straight when it reports none.""" + raw = info.properties.get("bulges") or [] + out = [] + for value in raw[:count]: + try: + out.append(float(value or 0.0)) + except (TypeError, ValueError): + out.append(0.0) + return out + [0.0] * (count - len(out)) + + +def payload_port_refs(payload) -> list[tuple[str, str]]: + """``(handle, port)`` for each well-formed ``from``/``to`` in a line payload. + + A foreign or hand-edited payload can carry anything under those keys + (``"garbage"``, ``None``, a number); reading it used to raise + ``TypeError`` out of ``draw_line``. Reader results never raise: a + malformed end is simply not a reference. + """ + refs = [] + if not isinstance(payload, dict): + return refs + for key in ("from", "to"): + ref = payload.get(key) + if not isinstance(ref, dict): + continue + handle, port = ref.get("handle"), ref.get("port") + if isinstance(handle, str) and isinstance(port, str): + refs.append((handle, port)) + return refs + + +def payload_seq(payload) -> int | None: + """The payload's sequence number when it is a genuine int (``True`` is not).""" + if not isinstance(payload, dict): + return None + seq = payload.get("seq") + return seq if type(seq) is int else None + + async def existing_pid_lines(backend: AutoCADBackend) -> list[dict]: """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. + or port reuse would ever be reported there. ``vertices`` are the raw + vertices; ``bulges`` (one per vertex, DXF convention) let the crossing test + follow an arc instead of its chord. """ out = [] offset = 0 @@ -124,7 +169,15 @@ async def existing_pid_lines(backend: AutoCADBackend) -> list[dict]: 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}) + vertices = _vertices(info) + out.append( + { + "handle": info.handle, + "vertices": vertices, + "bulges": _bulges(info, len(vertices)), + "payload": payload, + } + ) if len(page) < _PAGE: break offset += _PAGE @@ -184,17 +237,16 @@ async def draw_line( path = route(s_point, s_dir, e_point, e_dir, stub=stub, mode=route_mode) existing = await existing_pid_lines(backend) - crossings = count_crossings(path, [ln["vertices"] for ln in existing]) + crossings = count_crossings( + path, [flatten_bulges(ln["vertices"], ln["bulges"]) for ln in existing] + ) used: set[tuple[str, str]] = set() max_seq = 0 for ln in existing: - payload = ln["payload"] or {} - for key in ("from", "to"): - ref = payload.get(key) - if ref: - used.add((ref["handle"], ref["port"])) - if isinstance(payload.get("seq"), int): - max_seq = max(max_seq, payload["seq"]) + used.update(payload_port_refs(ln["payload"])) + seq = payload_seq(ln["payload"]) + if seq is not None: + max_seq = max(max_seq, seq) port_reuse = [ {"handle": ep["handle"], "port": ep["port"]} for ep in (start, end) diff --git a/engineering/pid/lines.py b/engineering/pid/lines.py index fed14f3..4f9afda 100644 --- a/engineering/pid/lines.py +++ b/engineering/pid/lines.py @@ -238,6 +238,53 @@ def marker_positions(vertices, min_len: float = 15.0) -> list[tuple[float, float return out +#: Max chord deviation (mm) when a bulged polyline edge is flattened for the +#: crossing test — a tenth of the thinnest ISO 128 lineweight. +FLATTEN_SAGITTA = 0.05 + + +def flatten_bulges(points, bulges, sagitta: float = FLATTEN_SAGITTA) -> list[Point]: + """The polyline as straight segments, every bulged edge replaced by chords + that deviate from the arc by at most ``sagitta``. + + Same arc arithmetic as ``engineering/measure.py`` (sweep ``4·atan(bulge)``, + radius ``chord / 2·sin(|sweep|/2)``) and ``graph._point_segment_distance`` + (centre on the chord's left normal, ``radius - sagitta`` carrying it across + the chord past a semicircle). Checked against ``ezdxf.math.bulge_to_arc`` + + ``ConstructionArc.flattening`` in ``tests/test_pid_lines.py``: identical + vertex count and positions to 1e-12 for bulges 0.2 … 2.0 of either sign. + A straight edge (bulge 0) is passed through; ``bulges`` shorter than the + edge count is padded with zeros. + """ + pts = [(float(x), float(y)) for x, y in points] + if len(pts) < 2: + return pts + out: list[Point] = [] + for i, (a, b) in enumerate(zip(pts, pts[1:], strict=False)): + out.append(a) + bulge = float(bulges[i]) if i < len(bulges) and bulges[i] else 0.0 + chord = _dist(a, b) + if abs(bulge) < 1e-12 or chord < 1e-12: + continue + theta = 4.0 * math.atan(bulge) # signed sweep, radians + radius = chord / (2.0 * math.sin(abs(theta) / 2.0)) + sag = abs(bulge) * chord / 2.0 + nx, ny = -(b[1] - a[1]) / chord, (b[0] - a[0]) / chord # left of a→b + offset = (radius - sag) * (1.0 if bulge > 0 else -1.0) + cx = (a[0] + b[0]) / 2.0 + offset * nx + cy = (a[1] + b[1]) / 2.0 + offset * ny + start = math.atan2(a[1] - cy, a[0] - cx) + if sagitta >= radius: + segments = 1 + else: + segments = max(1, math.ceil(abs(theta) / (2.0 * math.acos(1.0 - sagitta / radius)))) + for k in range(1, segments): + angle = start + theta * k / segments + out.append((cx + radius * math.cos(angle), cy + radius * math.sin(angle))) + out.append(pts[-1]) + return out + + def _orient(a: Point, b: Point, c: Point) -> float: return (b[0] - a[0]) * (c[1] - a[1]) - (b[1] - a[1]) * (c[0] - a[0]) diff --git a/tests/test_pid_line_draw.py b/tests/test_pid_line_draw.py index 6ad95d3..de34e1b 100644 --- a/tests/test_pid_line_draw.py +++ b/tests/test_pid_line_draw.py @@ -424,3 +424,72 @@ async def test_dry_run_plans_the_same_bubble_exit_as_the_real_run(backend): drawn = await run_spec(backend, spec) edge = drawn["graph"]["edges"][0] assert edge["line_number"] is None, "signal lines from a spec are unnumbered too" + + +# ── Track A hardening (track E wave 0, Task 7) ─────────────────────────────── + + +async def test_foreign_line_payloads_are_tolerated_not_raised(backend): + """A hand-edited payload with a non-dict ``from``, a ``None`` ``to`` and a + bool ``seq`` used to escape as ``TypeError: string indices must be + integers``; reader results never raise.""" + poly = await backend.entity_create_polyline([[0, 0], [10, 0]], layer="PROCESS-PIPING-MAIN") + await write_payload( + backend, + poly.handle, + {"v": 1, "kind": "line", "seq": True, "from": "garbage", "to": None, "number": 7}, + ) + other = await backend.entity_create_polyline([[0, 5], [10, 5]], layer="PROCESS-PIPING-MAIN") + await write_payload( + backend, + other.handle, + {"v": 1, "kind": "line", "seq": "9", "from": {"handle": 3, "port": "x"}, "to": 4}, + ) + result = await draw_line( + backend, {"x": 0, "y": 50}, {"x": 50, "y": 50}, size="100", service="P" + ) + assert result["line_number"] == "100-P-1", "neither True nor '9' counts as a sequence number" + assert result["port_reuse"] == [] + + +def test_payload_helpers_accept_only_well_formed_values(): + from engineering.pid.drawlines import payload_port_refs, payload_seq + + good = {"from": {"handle": "2A", "port": "discharge"}, "to": {"handle": "2B", "port": None}} + assert payload_port_refs(good) == [("2A", "discharge")] + assert payload_port_refs("garbage") == [] and payload_port_refs(None) == [] + assert payload_port_refs({"from": "x", "to": 5}) == [] + assert payload_seq({"seq": 7}) == 7 + assert payload_seq({"seq": True}) is None and payload_seq({"seq": "9"}) is None + assert payload_seq({"seq": 7.0}) is None and payload_seq(None) is None + + +async def test_crossings_follow_a_bulged_line_not_its_chord(backend): + """A foreign P&ID line with a semicircular jump (bulge -1 from (0,0) to + (100,0) bows to +Y, apex (50,50)): a vertical utility line at x=30 from + y=20 to y=60 crosses the arc once at y≈45.8 and its chord never.""" + msp = backend._msp() + msp.add_lwpolyline( + [(0, 0, 0, 0, -1.0), (100, 0, 0, 0, 0)], + format="xyseb", + dxfattribs={"layer": "PROCESS-PIPING-MAIN"}, + ) + backend._mark_dirty() + lines = await existing_pid_lines(backend) + assert lines[0]["vertices"] == [(0.0, 0.0), (100.0, 0.0)] + assert lines[0]["bulges"] == [-1.0, 0.0] + result = await draw_line(backend, {"x": 30, "y": 20}, {"x": 30, "y": 60}, line_class="utility") + assert result["crossings"] == 1 + below = await draw_line(backend, {"x": 70, "y": -20}, {"x": 70, "y": -60}, line_class="utility") + assert below["crossings"] == 0, "the chord side of the arc is empty" + + +async def test_com_existing_lines_carry_bulges(monkeypatch): + payload = line_payload("process_major", "100-P-7", "100", "P", None, None, 7, None, None) + poly = _com_pid_polyline("3E", (0.0, 0.0, 100.0, 0.0), payload) + poly.Length = 157.0796 # a semicircle: longer than the 100 mm chord walk + poly.GetBulge = lambda index: -1.0 if index == 0 else 0.0 + backend = _fake_com(monkeypatch, [poly]) + lines = await existing_pid_lines(backend) + assert lines[0]["vertices"] == [(0.0, 0.0), (100.0, 0.0)] + assert lines[0]["bulges"] == [-1.0, 0.0] diff --git a/tests/test_pid_lines.py b/tests/test_pid_lines.py index 9ccebe9..8cc4f75 100644 --- a/tests/test_pid_lines.py +++ b/tests/test_pid_lines.py @@ -3,6 +3,8 @@ from __future__ import annotations +import math + import pytest from engineering.pid.lines import ( @@ -176,3 +178,41 @@ def test_route_refuses_non_axis_port_directions_with_a_value_error(): # Equivalent angles and engine float drift still resolve to the axis. assert route((0, 0), -360.0, (50, 0), 540.0) == [(0.0, 0.0), (50.0, 0.0)] assert route((0, 0), 1e-9, (50, 0), 180.0 - 1e-9) == [(0.0, 0.0), (50.0, 0.0)] + + +# ── Track A hardening (track E wave 0, Task 7): bulge flattening ───────────── + + +@pytest.mark.parametrize("bulge", [1.0, -1.0, 0.5, -0.3, 2.0, 0.2]) +def test_flatten_bulges_matches_ezdxf_flattening(bulge): + """Pinned against ezdxf's own arc flattening: same vertex count, same points.""" + from ezdxf.math import ConstructionArc, Vec2, bulge_to_arc + + from engineering.pid.lines import FLATTEN_SAGITTA, flatten_bulges + + a, b = (0.0, 0.0), (10.0, 0.0) + mine = flatten_bulges([a, b], [bulge]) + centre, start_rad, end_rad, radius = bulge_to_arc(Vec2(a), Vec2(b), bulge) + arc = ConstructionArc( + center=centre, + radius=radius, + start_angle=math.degrees(start_rad), + end_angle=math.degrees(end_rad), + ) + reference = [(p.x, p.y) for p in arc.flattening(FLATTEN_SAGITTA)] + assert len(mine) == len(reference) + assert mine[0] == a and mine[-1] == b + assert all(math.hypot(p[0] - centre.x, p[1] - centre.y) == pytest.approx(radius) for p in mine) + assert mine[len(mine) // 2] == pytest.approx(reference[len(reference) // 2]) + + +def test_flatten_bulges_passes_straight_edges_through_and_pads_short_bulge_lists(): + from engineering.pid.lines import flatten_bulges + + pts = [(0.0, 0.0), (10.0, 0.0), (10.0, 10.0)] + assert flatten_bulges(pts, []) == pts + assert flatten_bulges(pts, [0.0, 0.0, 0.0]) == pts + flat = flatten_bulges(pts, [0.0, 1.0]) + assert flat[0] == (0.0, 0.0) and flat[1] == (10.0, 0.0) and flat[-1] == (10.0, 10.0) + assert len(flat) > 3, "the second edge became a semicircle of chords" + assert flatten_bulges([(1.0, 1.0)], [1.0]) == [(1.0, 1.0)]