anatoly/autocad-mcp-pro
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

test(understand): the synthetic plant pair - a schematic P&ID and its layout with known answers

Browse Source 
Every track H reader is measured on this pair, never on a client drawing
(spec section 14). tests/fixtures/plant_pair.py writes pid.dxf and layout.dxf
with invented names and returns their truth, computed from the positions it
placed:

- the P&ID is room 1 of the layout stretched 1.3x and room 2 rearranged, so
  21.8 % of the 55 tag pairs fall within 10 % of the median ratio (schematic)
  while room 1 alone is 100 % at 1.3;
- service layers in English and Dutch, Russian supply / return words, one CIP
  return drawn on the supply layer, sizes spelled as drafters do (an MTEXT
  format run, %%c, \U+2205, the U+2300 sign, o-stroke, SMS, DN, 20x27), a
  reducer, inline valves, an arc bend, a segment drawn twice;
- electrical texts in Russian with the power stated for all loads but one;
- a layout declaring inches over millimetre geometry, a stray line 5,000 km
  out, two plan copies 100 m apart, wall outlines, room labels and wiring
  callouts in four spellings.

Run lengths are the rectilinear MST of each run's tags, cable metres are
the Manhattan length rounded up after the 20 % allowance; the generator
refuses to write a pair whose own premises fail.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-24 11:42:09 +03:00
co-authored by Claude Opus 5.5
parent 9ab0ae1773
commit 704a44b190
3 changed files with 911 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/fixtures/__init__.py
Vendored
+1
View File
@@ -0,0 +1 @@
"""Synthetic drawings the readers are measured on (never a client's drawing)."""
tests/fixtures/plant_pair.py
Vendored
+666
View File
@@ -0,0 +1,666 @@
"""The synthetic plant pair: a schematic P&ID and its layout, with known answers.
Every track H reader is measured on this pair and never on a client drawing
(spec §14). It reproduces each field finding of the design spec (§1, §15) with
invented names and invented numbers:
* **Tags in both drawings** - T101..T104, M11, CP-1 in room 1; T201..T203, M12,
CP-2 in room 2. T105 is on the P&ID only, so the run to it is not routable.
* **The P&ID is not to scale** - room 1 is the layout stretched 1.3x about the
origin, room 2 is rearranged outright. The generator computes the scale
statistics of its own placement (pairs at least 2 m apart, the ratio of their
distances, the share within ±10 % of the median) and refuses to write a pair
that would not read as ``schematic``.
* **Service layers in English and Dutch** - ``P_product piping``,
``P_cipsupplyline``, ``P_cipreturnline``, ``P_ijswater`` (ice water, a utility
outside the default takeoff scope), valves and the reducer on ``KLEPPEN``.
* **Russian supply / return words** - ПОДАЧА beside the CIP supply runs,
ВОЗВРАТ beside the returns, and one return (``CR2``) drawn on the *supply*
layer: only the word says which way it runs.
* **Size labels as drafters write them** - ``{\\fArial|b0;Ø51}`` (an MTEXT format
run), ``Ø51``, ``%%c38``, ``\\U+2205 38``, ``⌀38``, ``ø38``, ``SMS51``, ``DN20``,
``DN25``, ``20x27``; each 300 mm from the segment it names and at least 600 mm
from every other run (the generator checks it).
* **Fittings on runs** - a reducer (``VERLOOPSTUK``) between Ø51 and Ø38 on
``PR3``, inline valves (``KLEP_VLINDER``) on ``PR2`` and ``CS2``; the pipe stops
at each fitting's box and continues on its far side.
* **A double-drawn segment** - on ``PR2`` a LINE lies exactly over an LWPOLYLINE
segment; it must count once.
* **Electrical texts in Russian** - power, phases, voltage, N and PE, ЧРП (VFD)
1500 mm right of each load's tag; the heater on T202 states no power.
* **The layout lies about its units** - ``$INSUNITS = 1`` (inches) over
millimetre geometry; one stray LINE 5,000 km away; two copies of the plan
100 m apart, so every tag is on the layout twice and is unique only within a
copy.
* **Rooms** - wall outlines (200 mm walls) and labels ``ROOM 1 PROCESS HALL`` /
``ПОМЕЩЕНИЕ 2 МОЙКА``; ``Wiring to CP1`` style MULTILEADER callouts from each
load to its panel, in four spellings.
Truth values are **computed here from the positions placed here** - Manhattan
distances, the rectilinear minimum spanning tree of each run's tags, the cable
lengths rounded up after the allowance - never typed. A position on the layout
is the insertion point of the tag's TEXT, which is also the centre of the
equipment circle it labels, so "the tag's position" and "the equipment's
position" are one point by construction.
"""
from __future__ import annotations
import itertools
import math
import statistics
from dataclasses import dataclass, field
from pathlib import Path
import ezdxf
from ezdxf import bbox as ezbbox
from ezdxf.math import Vec2
from ezdxf.render.mleader import ConnectionSide
Pt = tuple[float, float]
__all__ = ["ALLOWANCE", "LABEL_OFFSET", "build_plant_pair"]
#: The takeoffs' default allowance; the truth's ROUNDUP metres use it.
ALLOWANCE = 0.20
#: How far a size label or a service word sits from the segment it names.
LABEL_OFFSET = 300.0
#: The stretch of room 1 on the P&ID.
ROOM1_FACTOR = 1.3
#: The second plan copy on the layout is this far to the right of the first.
COPY_OFFSET: Pt = (100_000.0, 0.0)
#: The stray entity on the layout.
OUTLIER_START: Pt = (-5.0e9, 3.0e9)
#: Tank, pump and panel sizes (P&ID radii / layout radii / panel box).
PID_TANK_R, PID_PUMP_R = 600.0, 300.0
LAYOUT_TANK_R, LAYOUT_PUMP_R = 1000.0, 400.0
PANEL_W, PANEL_H = 1000.0, 400.0
TEXT_H = 250.0
#: Tag positions on the layout (first copy), mm. The P&ID positions follow.
LAYOUT_TAGS: dict[str, Pt] = {
"T101": (3000.0, 3000.0),
"T102": (8000.0, 3500.0),
"T103": (13500.0, 3000.0),
"T104": (17000.0, 8500.0),
"M11": (5500.0, 8200.0),
"CP-1": (1200.0, 10800.0),
"T201": (23500.0, 3200.0),
"T202": (28700.0, 3000.0),
"T203": (33000.0, 8800.0),
"M12": (26300.0, 8400.0),
"CP-2": (35000.0, 1300.0),
}
ROOM1_TAGS = ("T101", "T102", "T103", "T104", "M11", "CP-1")
#: Room 2 on the P&ID: rearranged, not scaled.
PID_ROOM2: dict[str, Pt] = {
"T201": (25000.0, 3000.0),
"T202": (29000.0, 14000.0),
"T203": (25500.0, 11000.0),
"M12": (32000.0, 4000.0),
"CP-2": (28000.0, 7500.0),
}
#: On the P&ID only: the run to it cannot be routed on the layout.
PID_ONLY = {"T105": (22100.0, 16500.0)}
ROOMS = (
{"number": "1", "name": "PROCESS HALL", "box": (0.0, 0.0, 20000.0, 12000.0)},
{
"number": "2",
"name": "МОЙКА",
"box": (20200.0, 0.0, 36000.0, 12000.0),
},
)
ROOM_LABELS = (
("ROOM 1 PROCESS HALL", (10000.0, 11000.0)),
(
"ПОМЕЩЕНИЕ 2 МОЙКА",
(28000.0, 11000.0),
),
)
OUTER_WALL = (-200.0, -200.0, 36200.0, 12200.0)
SUPPLY_WORD = "ПОДАЧА"
RETURN_WORD = "ВОЗВРАТ"
#: (tag, the Russian electrical text beside it, stated kW or None, VFD).
LOADS = (
("M11", "Насос P = 5,5 кВт\\P3 ф. 400 В + N + PE", 5.5, False),
("T102", "Мешалка P = 2,2 кВт\\P3 ф. 400 В + N + PE", 2.2, False),
("M12", "Насос CIP P = 7,5 кВт\\P3 ф. 400 В + N + PE\\PЧРП", 7.5, True),
("T202", "Нагреватель\\P3 ф. 400 В + N + PE", None, False),
)
#: (load tag, callout text on the layout, the panel it names).
CALLOUTS = (
("M11", "Wiring to CP1", "CP-1"),
("T102", "Wiring to CP 1", "CP-1"),
("M12", "Wiring to CP-2", "CP-2"),
("T202", "WIRING TO CP2", "CP-2"),
)
ELECTRICAL_DX = 1500.0
CALLOUT_DX, CALLOUT_DY = 1500.0, 1500.0
@dataclass(frozen=True)
class _Run:
"""One pipe run of the P&ID and what the takeoffs must find for it."""
id: str
layer: str
service: str
diameter: str
tags: tuple[str, ...]
status: str
paths: tuple[tuple[Pt, ...], ...]
labels: tuple[tuple[str, str, Pt, str], ...] # (TEXT | MTEXT, as drawn, insert, as read)
arcs: tuple[tuple[Pt, float, float, float], ...] = () # centre, radius, start, end (deg)
fittings: tuple[tuple[str, Pt, float], ...] = () # block, centre, rotation
words: tuple[tuple[str, Pt], ...] = ()
duplicate: tuple[Pt, Pt] | None = None
notes: tuple[str, ...] = field(default_factory=tuple)
RUNS: tuple[_Run, ...] = (
_Run(
"PR1",
"P_product piping",
"product",
"Ø51",
("T101", "T102"),
"A",
paths=(((4500.0, 4200.0), (9800.0, 4200.0)),),
labels=(("MTEXT", "{\\fArial|b0;Ø51}", (7150.0, 4500.0), "Ø51"),),
),
_Run(
"PR2",
"P_product piping",
"product",
"SMS51",
("M11", "T102", "T103"),
"B",
paths=(
((10400.0, 5150.0), (10400.0, 7000.0)),
((7150.0, 7000.0), (13850.0, 7000.0)),
((14150.0, 7000.0), (17250.0, 7000.0)),
((17550.0, 6700.0), (17550.0, 4500.0)),
((7150.0, 7000.0), (7150.0, 10360.0)),
),
labels=(("TEXT", "SMS51", (10700.0, 6075.0), "SMS51"),),
arcs=(((17250.0, 6700.0), 300.0, 0.0, 90.0),),
fittings=(("KLEP_VLINDER", (14000.0, 7000.0), 0.0),),
duplicate=((14150.0, 7000.0), (17250.0, 7000.0)),
notes=("T-junction at (10400, 7000)", "valve continued", "one segment drawn twice"),
),
_Run(
"PR3",
"P_product piping",
"product",
"Ø51",
("T103", "T104"),
"A",
paths=(
((18150.0, 3900.0), (20000.0, 3900.0), (20000.0, 7000.0)),
((20000.0, 7300.0), (20000.0, 11050.0), (21500.0, 11050.0)),
),
labels=(
("TEXT", "Ø51", (19075.0, 4200.0), "Ø51"),
("TEXT", "Ø51", (20300.0, 5450.0), "Ø51"),
("TEXT", "%%c38", (20300.0, 9175.0), "Ø38"),
("TEXT", "%%c38", (20750.0, 11350.0), "Ø38"),
),
fittings=(("VERLOOPSTUK", (20000.0, 7150.0), 90.0),),
notes=("reducer between the two sizes",),
),
_Run(
"PR4",
"P_product piping",
"product",
"DN20",
("T104", "T105"),
"C",
paths=(((22100.0, 11650.0), (22100.0, 15900.0)),),
labels=(("TEXT", "DN20", (22400.0, 13775.0), "DN20"),),
notes=("T105 is not on the layout",),
),
_Run(
"CS1",
"P_cipsupplyline",
"cip_supply",
"Ø38",
("T101", "T201"),
"A",
paths=(((25000.0, 2400.0), (25000.0, 1500.0), (3900.0, 1500.0), (3900.0, 3300.0)),),
labels=(
("MTEXT", "\\U+2205 38", (25300.0, 1950.0), "Ø38"),
("MTEXT", "\\U+2205 38", (14450.0, 1800.0), "Ø38"),
("MTEXT", "\\U+2205 38", (4200.0, 2400.0), "Ø38"),
),
words=((SUPPLY_WORD, (14450.0, 1200.0)),),
),
_Run(
"CS2",
"P_cipsupplyline",
"cip_supply",
"20x27",
("M12", "T201"),
"B",
paths=(
((25600.0, 3000.0), (27850.0, 3000.0)),
((28150.0, 3000.0), (32000.0, 3000.0), (32000.0, 3700.0)),
),
labels=(("TEXT", "20x27", (26725.0, 3300.0), "20x27"),),
fittings=(("KLEP_VLINDER", (28000.0, 3000.0), 0.0),),
words=((SUPPLY_WORD, (30075.0, 2700.0)),),
),
_Run(
"CR1",
"P_cipreturnline",
"cip_return",
"Ø38",
("T101", "T202"),
"A",
paths=(
(
(3300.0, 3900.0),
(2500.0, 3900.0),
(2500.0, 19000.0),
(29000.0, 19000.0),
(29000.0, 14600.0),
),
),
labels=(
("TEXT", "⌀38", (2900.0, 4200.0), "Ø38"),
("TEXT", "⌀38", (2200.0, 11450.0), "Ø38"),
("TEXT", "⌀38", (15750.0, 19300.0), "Ø38"),
("TEXT", "⌀38", (29300.0, 16800.0), "Ø38"),
),
words=((RETURN_WORD, (15750.0, 18700.0)),),
),
_Run(
"CR2",
"P_cipsupplyline",
"cip_return",
"Ø38",
("T202", "T203"),
"A",
paths=(((25500.0, 11600.0), (25500.0, 14000.0), (28400.0, 14000.0)),),
labels=(
("TEXT", "ø38", (25200.0, 12800.0), "Ø38"),
("TEXT", "ø38", (26950.0, 14300.0), "Ø38"),
),
words=((RETURN_WORD, (26950.0, 13700.0)),),
notes=("a return drawn on the supply layer",),
),
_Run(
"UT1",
"P_ijswater",
"ice_water",
"DN25",
("M12", "T203"),
"A",
paths=(((32000.0, 4300.0), (32000.0, 11000.0), (26100.0, 11000.0)),),
labels=(
("TEXT", "DN25", (32300.0, 7650.0), "DN25"),
("TEXT", "DN25", (29050.0, 11300.0), "DN25"),
),
notes=("a utility: outside the default takeoff scope",),
),
)
LAYER_SERVICES = {
"P_product piping": "product",
"P_cipsupplyline": "cip_supply",
"P_cipreturnline": "cip_return",
"P_ijswater": "ice_water",
"E_power": "electrical",
}
def pid_tags() -> dict[str, Pt]:
"""Tag positions on the P&ID: room 1 stretched, room 2 rearranged, T105 alone."""
out = {
tag: (ROOM1_FACTOR * x, ROOM1_FACTOR * y)
for tag, (x, y) in LAYOUT_TAGS.items()
if tag in ROOM1_TAGS
}
out.update(PID_ROOM2)
out.update(PID_ONLY)
return out
def manhattan(a: Pt, b: Pt) -> float:
return abs(a[0] - b[0]) + abs(a[1] - b[1])
def rmst_length(points: list[Pt]) -> float:
"""Rectilinear minimum spanning tree length (Prim, Manhattan weights)."""
if len(points) < 2:
return 0.0
inside = {0}
total = 0.0
while len(inside) < len(points):
best = min(
(manhattan(points[i], points[j]), j)
for i in inside
for j in range(len(points))
if j not in inside
)
total += best[0]
inside.add(best[1])
return total
def scale_statistics(pid: dict[str, Pt], layout: dict[str, Pt], tags) -> dict:
"""Spec §6 over ``tags``: pairs at least 2 m apart on both drawings."""
ratios = []
for a, b in itertools.combinations(sorted(tags), 2):
on_layout = math.dist(layout[a], layout[b])
on_pid = math.dist(pid[a], pid[b])
if on_layout >= 2000.0 and on_pid >= 2000.0:
ratios.append(on_pid / on_layout)
median = statistics.median(ratios)
within = sum(1 for r in ratios if abs(r - median) <= 0.1 * median) / len(ratios)
return {"pairs": len(ratios), "median": median, "within_10pct": within}
def _segments(run: _Run) -> list[tuple[Pt, Pt]]:
out = []
for path in run.paths:
out.extend(zip(path, path[1:], strict=False))
return out
def _segment_distance(p: Pt, a: Pt, b: Pt) -> float:
ax, ay = a
bx, by = b
dx, dy = bx - ax, by - ay
t = max(0.0, min(1.0, ((p[0] - ax) * dx + (p[1] - ay) * dy) / (dx * dx + dy * dy)))
return math.hypot(p[0] - (ax + t * dx), p[1] - (ay + t * dy))
def _arc_length(arc) -> float:
_centre, radius, start, end = arc
return radius * math.radians((end - start) % 360.0)
def _check_labels() -> None:
"""Every label names its own segment at LABEL_OFFSET and no other run within 2x."""
for run in RUNS:
own = _segments(run)
others = [s for other in RUNS if other is not run for s in _segments(other)]
for _kind, _raw, at, _read in run.labels:
mine = min(_segment_distance(at, a, b) for a, b in own)
theirs = min(_segment_distance(at, a, b) for a, b in others)
if abs(mine - LABEL_OFFSET) > 1e-9 or theirs <= 2 * LABEL_OFFSET:
raise AssertionError(f"{run.id} label at {at}: own {mine}, other run {theirs}")
if run.status == "A":
for a, b in own:
near = min(_segment_distance(lab[2], a, b) for lab in run.labels)
if abs(near - LABEL_OFFSET) > 1e-9:
raise AssertionError(f"{run.id}: segment {a}-{b} has no direct label")
def _segment_diameters(run: _Run) -> dict[str, float]:
"""Drawn length per size: a segment takes the label 300 mm from it (direct),
else the run's size (continuity); arcs are continuity."""
lengths: dict[str, float] = {}
for a, b in _segments(run):
size = run.diameter
for _kind, _raw, at, read in run.labels:
if abs(_segment_distance(at, a, b) - LABEL_OFFSET) <= 1e-9:
size = read
break
lengths[size] = lengths.get(size, 0.0) + math.dist(a, b)
for arc in run.arcs:
lengths[run.diameter] = lengths.get(run.diameter, 0.0) + _arc_length(arc)
return lengths
def _blocks(doc) -> None:
tank = doc.blocks.new("TANK_V")
tank.add_circle((0.0, 0.0), PID_TANK_R)
pump = doc.blocks.new("POMP")
pump.add_circle((0.0, 0.0), PID_PUMP_R)
pump.add_lwpolyline([(-150.0, -150.0), (150.0, 0.0), (-150.0, 150.0)], close=True)
panel = doc.blocks.new("SCHAKELKAST")
panel.add_lwpolyline(
[
(-PANEL_W / 2, -PANEL_H / 2),
(PANEL_W / 2, -PANEL_H / 2),
(PANEL_W / 2, PANEL_H / 2),
(-PANEL_W / 2, PANEL_H / 2),
],
close=True,
)
valve = doc.blocks.new("KLEP_VLINDER")
valve.add_lwpolyline(
[(-150.0, -100.0), (150.0, 100.0), (150.0, -100.0), (-150.0, 100.0)], close=True
)
reducer = doc.blocks.new("VERLOOPSTUK")
reducer.add_lwpolyline(
[(-150.0, -100.0), (150.0, -60.0), (150.0, 60.0), (-150.0, 100.0)], close=True
)
def _write_pid(path: Path) -> None:
doc = ezdxf.new("R2018")
doc.header["$INSUNITS"] = 4
for name in (*LAYER_SERVICES, "KLEPPEN", "TEKST", "TAGS", "EQUIPMENT"):
if name not in doc.layers:
doc.layers.add(name)
_blocks(doc)
msp = doc.modelspace()
for tag, at in pid_tags().items():
block = (
"SCHAKELKAST" if tag.startswith("CP") else "POMP" if tag.startswith("M") else "TANK_V"
)
msp.add_blockref(block, at, dxfattribs={"layer": "EQUIPMENT"})
msp.add_text(tag, dxfattribs={"insert": at, "height": TEXT_H, "layer": "TAGS"})
for run in RUNS:
attribs = {"layer": run.layer}
for pts in run.paths:
# a two-point path is a LINE - except the one drawn twice, which is
# an LWPOLYLINE, so its duplicate is a line lying over a polyline
if len(pts) == 2 and tuple(pts) != run.duplicate:
msp.add_line(pts[0], pts[1], dxfattribs=attribs)
else:
msp.add_lwpolyline(pts, dxfattribs=attribs)
for centre, radius, start, end in run.arcs:
msp.add_arc(centre, radius, start, end, dxfattribs=attribs)
if run.duplicate is not None:
msp.add_line(run.duplicate[0], run.duplicate[1], dxfattribs=attribs)
for block, at, rotation in run.fittings:
msp.add_blockref(block, at, dxfattribs={"layer": "KLEPPEN", "rotation": rotation})
for kind, raw, at, _read in run.labels:
if kind == "MTEXT":
msp.add_mtext(
raw, dxfattribs={"insert": at, "char_height": TEXT_H, "layer": "TEKST"}
)
else:
msp.add_text(raw, dxfattribs={"insert": at, "height": TEXT_H, "layer": "TEKST"})
for word, at in run.words:
msp.add_text(word, dxfattribs={"insert": at, "height": TEXT_H, "layer": "TEKST"})
tags = pid_tags()
for tag, text, _kw, _vfd in LOADS:
x, y = tags[tag]
msp.add_mtext(
text,
dxfattribs={
"insert": (x + ELECTRICAL_DX, y),
"char_height": TEXT_H,
"layer": "E_power",
},
)
_set_extents(doc)
doc.saveas(path)
def _set_extents(doc) -> None:
"""Declare the model-space extents the way AutoCAD would: over everything.
ezdxf copies the model-space layout's ``extmin`` / ``extmax`` into
``$EXTMIN`` / ``$EXTMAX`` when it saves (``Drawing.update_extents``), so the
layout's attributes are what is set; a header write alone is overwritten.
"""
msp = doc.modelspace()
box = ezbbox.extents(msp, fast=True)
msp.dxf.extmin = (box.extmin.x, box.extmin.y, 0.0)
msp.dxf.extmax = (box.extmax.x, box.extmax.y, 0.0)
def _rect(msp, box, layer) -> None:
x0, y0, x1, y1 = box
corners = [(x0, y0), (x1, y0), (x1, y1), (x0, y1)]
for a, b in zip(corners, corners[1:] + corners[:1], strict=True):
msp.add_line(a, b, dxfattribs={"layer": layer})
def _write_layout(path: Path) -> str:
doc = ezdxf.new("R2018")
doc.header["$INSUNITS"] = 1 # inches - the geometry is millimetres
for name in ("WALLS", "ROOMS", "EQUIPMENT", "TAGS", "E-PANEL", "E-CALLOUT"):
doc.layers.add(name)
msp = doc.modelspace()
for copy in range(2):
ox, oy = COPY_OFFSET[0] * copy, COPY_OFFSET[1] * copy
def at(p: Pt, ox=ox, oy=oy) -> Pt:
return (p[0] + ox, p[1] + oy)
x0, y0, x1, y1 = OUTER_WALL
_rect(msp, (x0 + ox, y0 + oy, x1 + ox, y1 + oy), "WALLS")
for room in ROOMS:
bx0, by0, bx1, by1 = room["box"]
_rect(msp, (bx0 + ox, by0 + oy, bx1 + ox, by1 + oy), "WALLS")
for text, where in ROOM_LABELS:
msp.add_text(text, dxfattribs={"insert": at(where), "height": TEXT_H, "layer": "ROOMS"})
for tag, where in LAYOUT_TAGS.items():
centre = at(where)
if tag.startswith("CP"):
cx, cy = centre
msp.add_lwpolyline(
[
(cx - PANEL_W / 2, cy - PANEL_H / 2),
(cx + PANEL_W / 2, cy - PANEL_H / 2),
(cx + PANEL_W / 2, cy + PANEL_H / 2),
(cx - PANEL_W / 2, cy + PANEL_H / 2),
],
close=True,
dxfattribs={"layer": "E-PANEL"},
)
else:
radius = LAYOUT_PUMP_R if tag.startswith("M") else LAYOUT_TANK_R
msp.add_circle(centre, radius, dxfattribs={"layer": "EQUIPMENT"})
msp.add_text(tag, dxfattribs={"insert": centre, "height": TEXT_H, "layer": "TAGS"})
for tag, text, _panel in CALLOUTS:
tip = at(LAYOUT_TAGS[tag])
builder = msp.add_multileader_mtext("Standard", dxfattribs={"layer": "E-CALLOUT"})
builder.set_content(text, char_height=TEXT_H)
builder.add_leader_line(ConnectionSide.left, [Vec2(tip)])
builder.build(insert=Vec2(tip[0] + CALLOUT_DX, tip[1] + CALLOUT_DY))
stray = msp.add_line(OUTLIER_START, (OUTLIER_START[0] + 1000.0, OUTLIER_START[1]))
_set_extents(doc)
doc.saveas(path)
return str(stray.dxf.handle)
def build_plant_pair(directory: str | Path) -> dict:
"""Write ``pid.dxf`` and ``layout.dxf`` into ``directory``; return the truth.
Refuses (``AssertionError``) to write a pair that would not test what it
claims: a label nearer another run than its own, a status-A segment with no
direct label, a P&ID that is not schematic overall or not 1.3x in room 1,
or a cable length whose rounding sits on an integer.
"""
_check_labels()
layout_tags = dict(LAYOUT_TAGS)
pid = pid_tags()
overall = scale_statistics(pid, layout_tags, LAYOUT_TAGS)
room1 = scale_statistics(pid, layout_tags, ROOM1_TAGS)
room2 = scale_statistics(pid, layout_tags, [t for t in LAYOUT_TAGS if t not in ROOM1_TAGS])
if overall["within_10pct"] > 0.5 or room1["within_10pct"] != 1.0:
raise AssertionError(f"the pair does not read as schematic: {overall}, {room1}")
out = Path(directory)
out.mkdir(parents=True, exist_ok=True)
pid_path, layout_path = out / "pid.dxf", out / "layout.dxf"
_write_pid(pid_path)
outlier = _write_layout(layout_path)
pipe_runs = []
for run in RUNS:
routable = all(tag in layout_tags for tag in run.tags)
pieces = sum(math.dist(a, b) for a, b in _segments(run)) + sum(
_arc_length(a) for a in run.arcs
)
pipe_runs.append(
{
"id": run.id,
"layer": run.layer,
"service": run.service,
"diameter": run.diameter,
"diameters": _segment_diameters(run),
"tags": sorted(run.tags),
"status": run.status,
"pid_length_mm": pieces,
"layout_length_mm": rmst_length([layout_tags[t] for t in run.tags])
if routable
else None,
}
)
cable_rows = []
for (tag, _text, kw, vfd), (callout_tag, _callout, panel) in zip(LOADS, CALLOUTS, strict=True):
assert tag == callout_tag
length = manhattan(layout_tags[tag], layout_tags[panel])
metres = length / 1000.0 * (1.0 + ALLOWANCE)
if abs(metres - round(metres)) < 1e-6:
raise AssertionError(f"{tag}: {metres} m sits on an integer; move the tag")
cable_rows.append(
{
"tag": tag,
"panel": panel,
"kw": kw,
"phases": 3,
"voltage": 400,
"vfd": vfd,
"length_mm": length,
"roundup_m": math.ceil(metres),
}
)
rooms = []
for room in ROOMS:
x0, y0, x1, y1 = room["box"]
rooms.append(
{
"number": room["number"],
"name": room["name"],
"polygon": [[x0, y0], [x1, y0], [x1, y1], [x0, y1]],
}
)
return {
"pid": str(pid_path),
"layout": str(layout_path),
"scale_verdict": "schematic",
"layout_insunits": 1,
"layout_true_unit": "mm",
"pid_insunits": 4,
"outlier_handle": outlier,
"cluster_count": 2,
"copy_offset": list(COPY_OFFSET),
"layer_services": dict(LAYER_SERVICES),
"pipe_runs": pipe_runs,
"cable_rows": cable_rows,
"rooms": rooms,
"allowance": ALLOWANCE,
"label_offset_mm": LABEL_OFFSET,
"tag_positions": {"layout": layout_tags, "pid": pid},
"scale": {
"overall": overall,
"room_1": room1,
"room_2": room2,
"room_1_factor": ROOM1_FACTOR,
},
}
tests/test_plant_pair.py
+244
View File
@@ -0,0 +1,244 @@
"""The synthetic plant pair opens, holds what it claims, and its truth is right.
Every expected number is worked out by hand in the comment beside it, from the
positions listed in tests/fixtures/plant_pair.py - never read back from the
generator's own arithmetic.
"""
from __future__ import annotations
import math
import ezdxf
import pytest
from tests.fixtures import plant_pair
from tests.fixtures.plant_pair import build_plant_pair
@pytest.fixture(scope="module")
def pair(tmp_path_factory):
return build_plant_pair(tmp_path_factory.mktemp("plant"))
def _count(msp, kind, layer=None):
return sum(1 for e in msp if e.dxftype() == kind and (layer is None or e.dxf.layer == layer))
def _texts(msp):
out = []
for e in msp:
if e.dxftype() == "TEXT":
out.append(e.dxf.text)
elif e.dxftype() == "MTEXT":
out.append(e.text)
return out
def test_both_files_open_with_their_declared_units(pair):
pid = ezdxf.readfile(pair["pid"])
layout = ezdxf.readfile(pair["layout"])
assert pid.header["$INSUNITS"] == 4
assert layout.header["$INSUNITS"] == pair["layout_insunits"] == 1
assert pair["layout_true_unit"] == "mm"
def test_the_pid_holds_the_symbols_the_runs_and_the_labels(pair):
msp = ezdxf.readfile(pair["pid"]).modelspace()
# 8 tanks (T101-T105, T201-T203), 2 pumps, 2 panels, 2 valves, 1 reducer
assert _count(msp, "INSERT") == 15
assert _count(msp, "INSERT", "KLEPPEN") == 3
assert _count(msp, "TEXT", "TAGS") == 12 # the 11 shared tags and T105
assert _count(msp, "ARC", "P_product piping") == 1
texts = _texts(msp)
for spelling in (
"{\\fArial|b0;Ø51}",
"%%c38",
"\\U+2205 38",
"⌀38",
"ø38",
"SMS51",
"DN20",
"DN25",
"20x27",
):
assert spelling in texts, spelling
assert texts.count("ПОДАЧА") == 2 # CS1, CS2
assert texts.count("ВОЗВРАТ") == 2 # CR1, CR2
assert _count(msp, "MTEXT", "E_power") == 4
def test_one_segment_is_drawn_twice_a_line_over_a_polyline(pair):
msp = ezdxf.readfile(pair["pid"]).modelspace()
start, end = (14150.0, 7000.0), (17250.0, 7000.0)
lines = [
e
for e in msp.query("LINE")
if (e.dxf.start.x, e.dxf.start.y) == start and (e.dxf.end.x, e.dxf.end.y) == end
]
polylines = [
e for e in msp.query("LWPOLYLINE") if [tuple(p) for p in e.get_points("xy")] == [start, end]
]
assert len(lines) == 1 and len(polylines) == 1
def test_a_cip_return_is_drawn_on_the_supply_layer(pair):
(cr2,) = [run for run in pair["pipe_runs"] if run["id"] == "CR2"]
assert cr2["layer"] == "P_cipsupplyline" and cr2["service"] == "cip_return"
def test_the_layout_carries_two_copies_one_outlier_and_the_rooms(pair):
doc = ezdxf.readfile(pair["layout"])
msp = doc.modelspace()
# per copy: 7 tank and 2 pump circles, 2 panel boxes, 11 tag texts,
# 2 room labels, 4 callouts, 12 wall lines; plus one stray line
assert _count(msp, "CIRCLE") == 18
assert _count(msp, "LWPOLYLINE", "E-PANEL") == 4
assert _count(msp, "TEXT", "TAGS") == 22
assert _count(msp, "TEXT", "ROOMS") == 4
assert _count(msp, "MULTILEADER") == 8
assert _count(msp, "LINE", "WALLS") == 24
tags = [e.dxf.text for e in msp.query("TEXT") if e.dxf.layer == "TAGS"]
assert all(tags.count(tag) == 2 for tag in set(tags))
assert "T105" not in tags
outlier = doc.entitydb[pair["outlier_handle"]]
assert outlier.dxftype() == "LINE" and outlier.dxf.start.x == -5.0e9
assert doc.header["$EXTMIN"][0] == -5.0e9 # the declared extents include it
assert pair["cluster_count"] == 2 and pair["copy_offset"] == [100000.0, 0.0]
def test_the_wiring_callouts_point_at_their_loads(pair):
msp = ezdxf.readfile(pair["layout"]).modelspace()
first_copy = {}
for e in msp.query("MULTILEADER"):
tip = e.context.leaders[0].lines[0].vertices[0]
if tip.x < 50000.0:
first_copy[e.context.mtext.default_content] = (tip.x, tip.y)
assert first_copy == {
"Wiring to CP1": (5500.0, 8200.0), # M11
"Wiring to CP 1": (8000.0, 3500.0), # T102
"Wiring to CP-2": (26300.0, 8400.0), # M12
"WIRING TO CP2": (28700.0, 3000.0), # T202
}
def test_the_layer_services_truth(pair):
assert pair["layer_services"] == {
"P_product piping": "product",
"P_cipsupplyline": "cip_supply",
"P_cipreturnline": "cip_return",
"P_ijswater": "ice_water",
"E_power": "electrical",
}
def test_the_pid_is_schematic_by_its_own_placement(pair):
scale = pair["scale"]
assert pair["scale_verdict"] == "schematic"
assert scale["overall"]["pairs"] == 55 # 11 shared tags, every pair over 2 m
assert scale["overall"]["within_10pct"] <= 0.5
assert scale["room_1"]["within_10pct"] == 1.0
assert scale["room_1"]["median"] == pytest.approx(1.3, abs=1e-12)
# T101 (3000, 3000) and T102 (8000, 3500) on the layout; x1.3 on the P&ID
pid = pair["tag_positions"]["pid"]
assert pid["T101"] == pytest.approx((3900.0, 3900.0))
assert pid["T102"] == pytest.approx((10400.0, 4550.0))
def test_layout_lengths_are_manhattan_and_the_rectilinear_mst(pair):
runs = {run["id"]: run for run in pair["pipe_runs"]}
# PR1: T101 (3000, 3000) - T102 (8000, 3500): 5000 + 500
assert runs["PR1"]["layout_length_mm"] == 5500.0
# PR2: T102 (8000, 3500), T103 (13500, 3000), M11 (5500, 8200):
# T102-T103 6000, T102-M11 7200, T103-M11 13200 -> MST 6000 + 7200
assert runs["PR2"]["layout_length_mm"] == 13200.0
# CS1: T201 (23500, 3200) - T101 (3000, 3000): 20500 + 200
assert runs["CS1"]["layout_length_mm"] == 20700.0
# PR4 ends on T105, which the layout does not have
assert runs["PR4"]["layout_length_mm"] is None and runs["PR4"]["status"] == "C"
assert [runs[i]["status"] for i in ("PR1", "PR2", "PR3", "CS2")] == ["A", "B", "A", "B"]
def test_schematic_lengths_count_the_double_drawn_segment_once(pair):
runs = {run["id"]: run for run in pair["pipe_runs"]}
# PR2: 1850 + 6700 + 3100 + 2200 + 3360 straight, and a quarter arc of r 300
assert runs["PR2"]["pid_length_mm"] == pytest.approx(17210.0 + 150.0 * math.pi, abs=1e-9)
# PR3 across the reducer: 1850 + 3100 at Ø51, 3750 + 1500 at Ø38
assert runs["PR3"]["diameters"] == {"Ø51": 4950.0, "Ø38": 5250.0}
def test_cable_truth_is_manhattan_then_the_allowance_rounded_up(pair):
rows = {row["tag"]: row for row in pair["cable_rows"]}
# M11 (5500, 8200) to CP-1 (1200, 10800): 4300 + 2600 = 6900 mm;
# 6.9 m x 1.2 = 8.28 -> 9
assert (rows["M11"]["panel"], rows["M11"]["length_mm"], rows["M11"]["roundup_m"]) == (
"CP-1",
6900.0,
9,
)
# M12 (26300, 8400) to CP-2 (35000, 1300): 8700 + 7100 = 15800; 18.96 -> 19
assert (rows["M12"]["length_mm"], rows["M12"]["roundup_m"], rows["M12"]["vfd"]) == (
15800.0,
19,
True,
)
assert rows["T202"]["kw"] is None, "the heater states no power; none is invented"
assert [rows[t]["kw"] for t in ("M11", "T102", "M12")] == [5.5, 2.2, 7.5]
def test_the_rooms_truth(pair):
assert [(r["number"], r["name"]) for r in pair["rooms"]] == [
("1", "PROCESS HALL"),
("2", "МОЙКА"),
]
assert pair["rooms"][1]["polygon"] == [
[20200.0, 0.0],
[36000.0, 0.0],
[36000.0, 12000.0],
[20200.0, 12000.0],
]
def test_the_generator_refuses_a_pair_that_would_read_to_scale(tmp_path, monkeypatch):
stretched = {
tag: (1.3 * x, 1.3 * y)
for tag, (x, y) in plant_pair.LAYOUT_TAGS.items()
if tag not in plant_pair.ROOM1_TAGS
}
monkeypatch.setattr(plant_pair, "PID_ROOM2", stretched)
with pytest.raises(AssertionError, match="schematic"):
build_plant_pair(tmp_path)
assert not (tmp_path / "pid.dxf").exists()
def test_every_label_reads_as_the_truth_says_through_the_task_1_parsers(pair):
"""The fixture's declared readings agree with engineering/understand's parsers."""
from engineering.understand.labels import (
parse_diameters,
parse_electrical,
parse_tag,
wiring_target,
)
from engineering.understand.snapshot import read_snapshot
from engineering.understand.vocab import classify_layer, room_label, supply_return
for run in plant_pair.RUNS:
for _kind, raw, _at, read in run.labels:
assert parse_diameters(raw) == (read,), (run.id, raw)
for word, _at in run.words:
expected = "supply" if run.service.endswith("supply") else "return"
assert supply_return(word) == expected, run.id
for layer, service in pair["layer_services"].items():
assert classify_layer(layer)["service"] == service, layer
pid = read_snapshot(pair["pid"])
tags = [parse_tag(r.text) for r in pid.records if r.type == "TEXT" and r.layer == "TAGS"]
assert sorted(tags) == sorted([*pair["tag_positions"]["layout"], "T105"])
electrical = [r.text for r in pid.records if r.layer == "E_power"]
assert sorted(parse_electrical(t)["kw"] or 0.0 for t in electrical) == [0.0, 2.2, 5.5, 7.5]
layout = read_snapshot(pair["layout"])
rooms = [room_label(r.text) for r in layout.records if r.layer == "ROOMS"]
assert {(r["number"], r["name"]) for r in rooms} == {
(room["number"], room["name"]) for room in pair["rooms"]
}
panels = {wiring_target(r.text) for r in layout.records if r.type == "MULTILEADER"}
assert panels == {"CP-1", "CP-2"}