feat(pid): line classes, orthogonal routing, labels, markers, crossings
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
co-authored by
Claude Opus 5
Claude Opus 5
parent
abc2a82e71
commit
6dd25ff0c2
@@ -0,0 +1 @@
|
||||
"""P&ID expert and reader: symbol catalogue, lines, ISA-5.1 tags, graph."""
|
||||
@@ -0,0 +1,235 @@
|
||||
# engineering/pid/lines.py
|
||||
"""Line classes (ISA-5.1 Table 5.3.1), orthogonal routing, labels, markers, crossings."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LineClass:
|
||||
layer: str
|
||||
linetype: str | None # None = ByLayer
|
||||
marker: str | None # marker symbol placed on long segments
|
||||
arrow: bool # flow arrow at the end by default
|
||||
kind: str # "process" | "signal"
|
||||
|
||||
|
||||
LINE_CLASSES: dict[str, LineClass] = {
|
||||
"process_major": LineClass("PROCESS-PIPING-MAIN", None, None, True, "process"),
|
||||
"process_minor": LineClass("PROCESS-PIPING-SECONDARY", None, None, True, "process"),
|
||||
"utility": LineClass("UTILITY-LINE", None, None, True, "process"),
|
||||
"pneumatic": LineClass(
|
||||
"INSTRUMENT-LINE-SIGNAL", "Continuous", "mark_pneumatic", False, "signal"
|
||||
),
|
||||
"electric": LineClass("INSTRUMENT-LINE-SIGNAL", None, None, False, "signal"),
|
||||
"hydraulic": LineClass(
|
||||
"INSTRUMENT-LINE-SIGNAL", "Continuous", "mark_hydraulic", False, "signal"
|
||||
),
|
||||
"capillary": LineClass(
|
||||
"INSTRUMENT-LINE-SIGNAL", "Continuous", "mark_capillary", False, "signal"
|
||||
),
|
||||
"data": LineClass("INSTRUMENT-LINE-SIGNAL", "Continuous", "mark_data", False, "signal"),
|
||||
}
|
||||
#: Layer → class when a line carries no payload. ``None`` = decide by linetype.
|
||||
LAYER_TO_CLASS: dict[str, str | None] = {
|
||||
"PROCESS-PIPING-MAIN": "process_major",
|
||||
"PROCESS-PIPING-SECONDARY": "process_minor",
|
||||
"UTILITY-LINE": "utility",
|
||||
"ELECTRICAL-LINE": "electric",
|
||||
"INSTRUMENT-LINE-SIGNAL": None,
|
||||
}
|
||||
PID_LINE_LAYERS = frozenset(LAYER_TO_CLASS)
|
||||
AXIS = {0.0: (1.0, 0.0), 90.0: (0.0, 1.0), 180.0: (-1.0, 0.0), 270.0: (0.0, -1.0)}
|
||||
DEFAULT_NUMBER_FORMAT = "{size}-{service}-{seq}-{spec}-{insulation}"
|
||||
_EPS = 1e-9
|
||||
|
||||
Point = tuple[float, float]
|
||||
|
||||
|
||||
def snap_axis(dx: float, dy: float) -> float:
|
||||
if abs(dx) >= abs(dy):
|
||||
return 0.0 if dx >= 0 else 180.0
|
||||
return 90.0 if dy >= 0 else 270.0
|
||||
|
||||
|
||||
def _unit(direction_deg: float) -> Point:
|
||||
return AXIS[float(direction_deg) % 360.0]
|
||||
|
||||
|
||||
def _dist(a: Point, b: Point) -> float:
|
||||
return math.hypot(b[0] - a[0], b[1] - a[1])
|
||||
|
||||
|
||||
def path_length(points) -> float:
|
||||
return sum(_dist(a, b) for a, b in zip(points, points[1:], strict=False))
|
||||
|
||||
|
||||
def simplify(points) -> list[Point]:
|
||||
"""Drop repeated points and merge collinear consecutive segments."""
|
||||
pts: list[Point] = []
|
||||
for p in points:
|
||||
q = (float(p[0]), float(p[1]))
|
||||
if not pts or _dist(pts[-1], q) > _EPS:
|
||||
pts.append(q)
|
||||
out: list[Point] = []
|
||||
for p in pts:
|
||||
if len(out) >= 2:
|
||||
a, b = out[-2], out[-1]
|
||||
cross = (b[0] - a[0]) * (p[1] - b[1]) - (b[1] - a[1]) * (p[0] - b[0])
|
||||
dot = (b[0] - a[0]) * (p[0] - b[0]) + (b[1] - a[1]) * (p[1] - b[1])
|
||||
if abs(cross) < _EPS and dot > 0:
|
||||
out[-1] = p
|
||||
continue
|
||||
out.append(p)
|
||||
return out
|
||||
|
||||
|
||||
def _direction(a: Point, b: Point) -> Point:
|
||||
length = _dist(a, b)
|
||||
return ((b[0] - a[0]) / length, (b[1] - a[1]) / length)
|
||||
|
||||
|
||||
def _same(u: Point, v: Point) -> bool:
|
||||
return abs(u[0] - v[0]) < 1e-6 and abs(u[1] - v[1]) < 1e-6
|
||||
|
||||
|
||||
def _validate(candidate, ds: float | None, de: float | None, stub: float) -> list[Point] | None:
|
||||
pts = simplify(candidate)
|
||||
if len(pts) < 2:
|
||||
return None
|
||||
segs = list(zip(pts, pts[1:], strict=False))
|
||||
straight = len(segs) == 1
|
||||
if ds is not None and not _same(_direction(*segs[0]), _unit(ds)):
|
||||
return None
|
||||
if de is not None:
|
||||
ue = _unit(de)
|
||||
if not _same(_direction(*segs[-1]), (-ue[0], -ue[1])):
|
||||
return None
|
||||
minimum = 0.01 if straight else max(stub, 0.01)
|
||||
if any(_dist(a, b) < minimum - _EPS for a, b in segs):
|
||||
return None
|
||||
for (a1, b1), (a2, b2) in zip(segs, segs[1:], strict=False):
|
||||
u, v = _direction(a1, b1), _direction(a2, b2)
|
||||
if abs(u[0] * v[0] + u[1] * v[1]) > 1e-6:
|
||||
return None
|
||||
return pts
|
||||
|
||||
|
||||
def route(start, start_dir, end, end_dir, stub: float = 5.0, mode="auto") -> list[Point]:
|
||||
"""Orthogonal path from ``start`` (leaving along ``start_dir``) to ``end``
|
||||
(arriving against ``end_dir``). ``None`` directions are radial: the axis
|
||||
towards the other end. Raises ``ValueError`` when no candidate honours the stub."""
|
||||
s: Point = (float(start[0]), float(start[1]))
|
||||
e: Point = (float(end[0]), float(end[1]))
|
||||
if isinstance(mode, (list, tuple)):
|
||||
path = [s] + [(float(p[0]), float(p[1])) for p in mode] + [e]
|
||||
for a, b in zip(path, path[1:], strict=False):
|
||||
if _dist(a, b) < 0.01:
|
||||
raise ValueError("waypoints repeat a point (segment shorter than 0.01 mm)")
|
||||
return path
|
||||
if _dist(s, e) < 0.01:
|
||||
raise ValueError("zero-length line: both ends are the same point")
|
||||
if mode == "direct":
|
||||
return [s, e]
|
||||
if mode != "auto":
|
||||
raise ValueError("route must be 'auto', 'direct' or a list of waypoints")
|
||||
if stub < 0:
|
||||
raise ValueError("stub must be >= 0")
|
||||
ds = snap_axis(e[0] - s[0], e[1] - s[1]) if start_dir is None else float(start_dir) % 360.0
|
||||
de = snap_axis(s[0] - e[0], s[1] - e[1]) if end_dir is None else float(end_dir) % 360.0
|
||||
us, ue = _unit(ds), _unit(de)
|
||||
s1 = (s[0] + us[0] * stub, s[1] + us[1] * stub)
|
||||
e1 = (e[0] + ue[0] * stub, e[1] + ue[1] * stub)
|
||||
xm, ym = (s1[0] + e1[0]) / 2.0, (s1[1] + e1[1]) / 2.0
|
||||
candidates = [
|
||||
[s, e],
|
||||
[s, (e[0], s[1]), e],
|
||||
[s, (s[0], e[1]), e],
|
||||
[s, (xm, s[1]), (xm, e[1]), e],
|
||||
[s, (s[0], ym), (e[0], ym), e],
|
||||
[s, s1, (s1[0], e1[1]), e1, e],
|
||||
[s, s1, (e1[0], s1[1]), e1, e],
|
||||
]
|
||||
valid = [p for p in (_validate(c, ds, de, stub) for c in candidates) if p]
|
||||
if not valid:
|
||||
raise ValueError(
|
||||
f"no orthogonal route from {s} to {e} honours a {stub} mm stub on both ports; "
|
||||
"shorten `stub` or pass waypoints"
|
||||
)
|
||||
valid.sort(key=lambda p: (len(p) - 2, path_length(p)))
|
||||
return valid[0]
|
||||
|
||||
|
||||
def label_placement(vertices, offset: float = 1.5) -> tuple[float, float, float, int]:
|
||||
"""Midpoint of the longest segment, offset to its left, readable rotation."""
|
||||
pts = [(float(x), float(y)) for x, y in vertices]
|
||||
segs = list(zip(pts, pts[1:], strict=False))
|
||||
index = max(range(len(segs)), key=lambda i: _dist(*segs[i]))
|
||||
(x0, y0), (x1, y1) = segs[index]
|
||||
angle = math.degrees(math.atan2(y1 - y0, x1 - x0)) % 360.0
|
||||
if 90.0 < angle <= 270.0:
|
||||
angle = (angle - 180.0) % 360.0
|
||||
ux, uy = math.cos(math.radians(angle)), math.sin(math.radians(angle))
|
||||
mx, my = (x0 + x1) / 2.0, (y0 + y1) / 2.0
|
||||
return (mx - uy * offset, my + ux * offset, round(angle, 6), index)
|
||||
|
||||
|
||||
def marker_positions(vertices, min_len: float = 15.0) -> list[tuple[float, float, float]]:
|
||||
pts = [(float(x), float(y)) for x, y in vertices]
|
||||
out = []
|
||||
for (x0, y0), (x1, y1) in zip(pts, pts[1:], strict=False):
|
||||
if _dist((x0, y0), (x1, y1)) < min_len:
|
||||
continue
|
||||
angle = math.degrees(math.atan2(y1 - y0, x1 - x0)) % 360.0
|
||||
out.append(((x0 + x1) / 2.0, (y0 + y1) / 2.0, round(angle, 6)))
|
||||
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])
|
||||
|
||||
|
||||
def segments_cross(p1, p2, q1, q2) -> bool:
|
||||
"""Proper crossing only: shared endpoints and T-touches are junctions, not crossings."""
|
||||
d1, d2 = _orient(q1, q2, p1), _orient(q1, q2, p2)
|
||||
d3, d4 = _orient(p1, p2, q1), _orient(p1, p2, q2)
|
||||
return d1 * d2 < -_EPS and d3 * d4 < -_EPS
|
||||
|
||||
|
||||
def count_crossings(vertices, others) -> int:
|
||||
pts = [(float(x), float(y)) for x, y in vertices]
|
||||
count = 0
|
||||
for a, b in zip(pts, pts[1:], strict=False):
|
||||
for other in others:
|
||||
opts = [(float(x), float(y)) for x, y in other]
|
||||
for c, d in zip(opts, opts[1:], strict=False):
|
||||
if segments_cross(a, b, c, d):
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
_FIELD_RE = re.compile(r"^\{(\w+)\}$")
|
||||
_KNOWN_FIELDS = {"size", "service", "seq", "spec", "insulation", "area", "unit"}
|
||||
|
||||
|
||||
def format_line_number(fmt: str, **fields) -> str:
|
||||
"""Fill ``fmt`` (``-``-separated tokens) and drop the empty fields with their separators."""
|
||||
kept: list[str] = []
|
||||
for token in fmt.split("-"):
|
||||
match = _FIELD_RE.match(token)
|
||||
if not match:
|
||||
kept.append(token)
|
||||
continue
|
||||
name = match.group(1)
|
||||
if name not in _KNOWN_FIELDS:
|
||||
raise ValueError(
|
||||
f"unknown line-number field {{{name}}}; known: {', '.join(sorted(_KNOWN_FIELDS))}"
|
||||
)
|
||||
value = fields.get(name)
|
||||
if value is None or value == "":
|
||||
continue
|
||||
kept.append(str(value))
|
||||
return "-".join(kept)
|
||||
@@ -0,0 +1,131 @@
|
||||
# tests/test_pid_lines.py
|
||||
"""Routing is geometry, so it is tested without a backend."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from engineering.pid.lines import (
|
||||
DEFAULT_NUMBER_FORMAT,
|
||||
LINE_CLASSES,
|
||||
count_crossings,
|
||||
format_line_number,
|
||||
label_placement,
|
||||
marker_positions,
|
||||
route,
|
||||
snap_axis,
|
||||
)
|
||||
|
||||
|
||||
def test_line_classes_map_to_layers_and_markers():
|
||||
assert set(LINE_CLASSES) == {
|
||||
"process_major",
|
||||
"process_minor",
|
||||
"utility",
|
||||
"pneumatic",
|
||||
"electric",
|
||||
"hydraulic",
|
||||
"capillary",
|
||||
"data",
|
||||
}
|
||||
assert LINE_CLASSES["process_major"].layer == "PROCESS-PIPING-MAIN"
|
||||
assert (
|
||||
LINE_CLASSES["electric"].layer == "INSTRUMENT-LINE-SIGNAL"
|
||||
and LINE_CLASSES["electric"].linetype is None
|
||||
)
|
||||
assert (
|
||||
LINE_CLASSES["pneumatic"].linetype == "Continuous"
|
||||
and LINE_CLASSES["pneumatic"].marker == "mark_pneumatic"
|
||||
)
|
||||
assert LINE_CLASSES["process_major"].arrow is True and LINE_CLASSES["pneumatic"].arrow is False
|
||||
assert LINE_CLASSES["utility"].kind == "process" and LINE_CLASSES["data"].kind == "signal"
|
||||
|
||||
|
||||
def test_snap_axis_picks_the_dominant_component():
|
||||
assert snap_axis(10, 1) == 0.0 and snap_axis(-10, 1) == 180.0
|
||||
assert snap_axis(1, 10) == 90.0 and snap_axis(1, -10) == 270.0
|
||||
|
||||
|
||||
def test_straight_when_aligned_and_facing():
|
||||
assert route((0, 0), 0.0, (50, 0), 180.0) == [(0.0, 0.0), (50.0, 0.0)]
|
||||
|
||||
|
||||
def test_l_route_when_perpendicular():
|
||||
path = route((0, 0), 0.0, (50, 30), 270.0)
|
||||
assert path == [(0.0, 0.0), (50.0, 0.0), (50.0, 30.0)]
|
||||
|
||||
|
||||
def test_z_route_when_parallel_and_offset():
|
||||
path = route((0, 0), 0.0, (60, 20), 180.0, stub=5.0)
|
||||
assert path == [(0.0, 0.0), (30.0, 0.0), (30.0, 20.0), (60.0, 20.0)]
|
||||
|
||||
|
||||
def test_ports_facing_away_on_one_axis_need_waypoints():
|
||||
with pytest.raises(ValueError, match="waypoints"):
|
||||
route((0, 0), 180.0, (40, 0), 0.0, stub=5.0)
|
||||
path = route((0, 0), 180.0, (40, 0), 0.0, mode=[(-5, 0), (-5, -10), (45, -10), (45, 0)])
|
||||
assert len(path) == 6
|
||||
|
||||
|
||||
def test_stub_is_honoured_or_refused():
|
||||
with pytest.raises(ValueError, match="stub"):
|
||||
route((0, 0), 0.0, (3, 8), 180.0, stub=5.0)
|
||||
assert route((0, 0), 0.0, (3, 0), 180.0, stub=5.0) == [(0.0, 0.0), (3.0, 0.0)], (
|
||||
"a straight run is exempt"
|
||||
)
|
||||
|
||||
|
||||
def test_radial_ends_take_the_axis_towards_the_other_end():
|
||||
path = route((0, 0), None, (0, 40), None)
|
||||
assert path == [(0.0, 0.0), (0.0, 40.0)]
|
||||
|
||||
|
||||
def test_direct_and_waypoints():
|
||||
assert route((0, 0), 0.0, (10, 7), 90.0, mode="direct") == [(0.0, 0.0), (10.0, 7.0)]
|
||||
assert route((0, 0), 0.0, (10, 7), 90.0, mode=[(10, 0)]) == [
|
||||
(0.0, 0.0),
|
||||
(10.0, 0.0),
|
||||
(10.0, 7.0),
|
||||
]
|
||||
with pytest.raises(ValueError, match="repeat"):
|
||||
route((0, 0), 0.0, (10, 7), 90.0, mode=[(0, 0)])
|
||||
with pytest.raises(ValueError, match="zero"):
|
||||
route((0, 0), 0.0, (0, 0), 90.0)
|
||||
|
||||
|
||||
def test_label_sits_left_of_the_longest_segment_and_never_upside_down():
|
||||
x, y, rot, index = label_placement([(0, 0), (10, 0), (10, 40), (0, 40)])
|
||||
assert index == 1 and rot == 90.0
|
||||
assert (x, y) == pytest.approx((8.5, 20.0))
|
||||
x, y, rot, index = label_placement([(50, 0), (0, 0)])
|
||||
assert rot == 0.0 and y == pytest.approx(1.5)
|
||||
|
||||
|
||||
def test_markers_only_on_segments_long_enough():
|
||||
marks = marker_positions([(0, 0), (10, 0), (10, 40)], min_len=15.0)
|
||||
assert marks == [(10.0, 20.0, 90.0)]
|
||||
assert marker_positions([(0, 0), (30, 0)]) == [(15.0, 0.0, 0.0)]
|
||||
|
||||
|
||||
def test_crossings_count_proper_intersections_only():
|
||||
others = [[(5, -5), (5, 5)], [(20, 0), (20, 10)], [(0, 3), (8, 3)]]
|
||||
assert count_crossings([(0, 0), (10, 0)], others) == 1
|
||||
assert count_crossings([(0, 0), (20, 0)], [[(20, 0), (20, 10)]]) == 0, (
|
||||
"touching an endpoint is a junction"
|
||||
)
|
||||
|
||||
|
||||
def test_line_number_format_drops_empty_fields():
|
||||
assert DEFAULT_NUMBER_FORMAT == "{size}-{service}-{seq}-{spec}-{insulation}"
|
||||
assert (
|
||||
format_line_number(
|
||||
DEFAULT_NUMBER_FORMAT, size="100", service="P", seq=1001, spec="CS1", insulation="IH"
|
||||
)
|
||||
== "100-P-1001-CS1-IH"
|
||||
)
|
||||
assert (
|
||||
format_line_number(DEFAULT_NUMBER_FORMAT, size="100", service="P", seq=1001) == "100-P-1001"
|
||||
)
|
||||
assert format_line_number("L-{seq}", seq=7) == "L-7"
|
||||
with pytest.raises(ValueError, match="unknown"):
|
||||
format_line_number("{bogus}", seq=1)
|
||||
Reference in New Issue
Block a user