diff --git a/engineering/understand/route.py b/engineering/understand/route.py new file mode 100644 index 0000000..99e14be --- /dev/null +++ b/engineering/understand/route.py @@ -0,0 +1,63 @@ +"""Orthogonal lengths on a layout: Manhattan distance and the rectilinear MST. + +A pipe or cable on a plant floor runs along the building grid, so its length +between two points is estimated as |dx| + |dy| - never the diagonal. A run that +joins more than two pieces of equipment is estimated as the minimum spanning +tree of their positions under that distance (Prim's algorithm); a two-ended run +is simply |dx| + |dy|. This is the tree on the given points only - no Steiner +points are added - so it is the length a drafter gets by routing each +connection separately, and it is exact for the points given. + +A route is drawn along x first, then y (`route_legs`): deterministic, and +stated on every method sheet that uses it. +""" + +from __future__ import annotations + +import math +from collections.abc import Sequence + +from engineering.understand.snapshot import Pt + +__all__ = ["manhattan", "rmst", "route_legs"] + + +def manhattan(a: Pt, b: Pt) -> float: + """|dx| + |dy| between two points.""" + return abs(a[0] - b[0]) + abs(a[1] - b[1]) + + +def rmst(points: Sequence[Pt]) -> tuple[float, tuple[tuple[int, int], ...]]: + """Rectilinear minimum spanning tree: (total length, edges as (from, to) indices). + + Prim's algorithm from point 0; each edge is (the tree vertex it grows from, + the new vertex), in the order the tree grows. Ties go to the lower index, + so the answer is deterministic. Fewer than two points: (0.0, ()). + """ + n = len(points) + if n < 2: + return 0.0, () + in_tree = [False] * n + best = [math.inf] * n + parent = [-1] * n + best[0] = 0.0 + total = 0.0 + edges: list[tuple[int, int]] = [] + for _ in range(n): + k = min((i for i in range(n) if not in_tree[i]), key=lambda i: (best[i], i)) + in_tree[k] = True + if parent[k] >= 0: + edges.append((parent[k], k)) + total += best[k] + for j in range(n): + if not in_tree[j]: + d = manhattan(points[k], points[j]) + if d < best[j]: + best[j], parent[j] = d, k + return total, tuple(edges) + + +def route_legs(a: Pt, b: Pt) -> tuple[tuple[Pt, Pt], ...]: + """The orthogonal route from a to b: along x first, then y; zero-length legs dropped.""" + corner = (b[0], a[1]) + return tuple(leg for leg in ((a, corner), (corner, b)) if leg[0] != leg[1]) diff --git a/engineering/understand/takeoff.py b/engineering/understand/takeoff.py new file mode 100644 index 0000000..bf340f0 --- /dev/null +++ b/engineering/understand/takeoff.py @@ -0,0 +1,834 @@ +"""Pipe takeoff rows: topology from the P&ID, lengths from the layout (spec §8). + +A P&ID says which equipment a line joins, what it carries and how big it is; +it does not say how long the pipe is unless the scale check calls it to scale. +So each run found by `engineering.understand.network` is measured one of two +ways, and the rows say which: + +* **layout** - the positions of the tags the run joins, on the layout, joined + by the rectilinear minimum spanning tree (`route.rmst`), plus + `vertical_allowance` metres per tree edge. The P&ID's drawn length is kept + beside it as `schematic_m`, a comparison column, never the answer. +* **pid** - the drawn length on the P&ID: divided by the scale factor when the + scale check calls it `to_scale`, taken in the P&ID's own unit when there is + no layout (every row then carries `scale_verified: false`), and on a + schematic P&ID only with `force=True`. + +This module also carries the takeoff's own stand-ins for three readers another +branch of track H owns - the unit inference, the plan-copy clusters and the +scale check (group U's `units`, `clusters` and `scale`). They follow the +spec's rules (§5, §6) in the narrow form a takeoff needs; `pipe_rows` uses a +`scale` result it is given before its own, so the full scale check replaces +the stand-in wherever a caller passes it. + +Every length leaves this module in metres, converted from the unit the +geometry implies - never blindly from `INSUNITS`. Nothing here touches a +drawing. +""" + +from __future__ import annotations + +import math +import statistics +from collections.abc import Iterable + +from engineering.arch.faces import DEFAULT_TOL, Face, face_containing, planar_faces +from engineering.understand.network import MODEL, TEXT_TYPES, tag_occurrences +from engineering.understand.route import rmst, route_legs +from engineering.understand.snapshot import Pt, Snapshot +from engineering.understand.vocab import SERVICES, classify_layer, room_label + +__all__ = [ + "DEFAULT_SERVICES", + "LENGTH_SOURCES", + "UNIT_METRES", + "apportion", + "check_allowance", + "check_pipe_options", + "layout_tags", + "pipe_rows", + "room_at", + "room_regions", + "scale_stand_in", + "unit_of", +] + +#: The user's own takeoff scope; utilities only on request (spec §8). +DEFAULT_SERVICES = ("product", "cip_supply", "cip_return") +LENGTH_SOURCES = ("auto", "pid", "layout") +UNASSIGNED = "unassigned" + +#: INSUNITS codes this module reads, and metres per unit of each. +INSUNITS_NAMES = {1: "in", 2: "ft", 4: "mm", 5: "cm", 6: "m"} +UNIT_METRES = {"mm": 0.001, "cm": 0.01, "m": 1.0, "in": 0.0254, "ft": 0.3048} +#: This module's heuristic, not a standard: the model-space text of any plant +#: drawing - a P&ID sheet at 1:1 or a plan at 1:200 - stands between 0.5 mm and +#: 1 m tall, and the drawing spans between 0.1 m and 5 km. A declared unit that +#: puts the geometry outside either band is not believed. +TEXT_BAND_M = (0.0005, 1.0) +SPAN_BAND_M = (0.1, 5000.0) +#: Among the units that fit both bands, the one nearest (in log) a 50 m plant +#: with 0.25 m text is chosen. +TYPICAL_SPAN_M = 50.0 +TYPICAL_TEXT_M = 0.25 + +#: Plan copies are told apart when the gap between them exceeds 2 % of the +#: drawing's robust span (5th-95th percentile). +CLUSTER_GAP_SHARE = 0.02 +#: Spec §6: pairs of matched tags at least 2 m apart, a ±10 % band, 80 % / 50 %. +MIN_PAIR_MM = 2000.0 +SCALE_BAND = 0.10 +TO_SCALE_SHARE = 0.80 +SCHEMATIC_SHARE = 0.50 +#: Fewer matched tags than this and the verdict is `insufficient`. +MIN_TAGS = 3 +#: A wall layer is one `classify_layer` files under architecture this surely. +WALL_CONFIDENCE = 0.9 + + +# -- options ----------------------------------------------------------------- + + +def _number(value, name: str) -> float: + try: + number = float(value) + except (TypeError, ValueError): + raise ValueError(f"{name}: expected a number, got {value!r}") from None + if not math.isfinite(number): + raise ValueError(f"{name}: must be finite, got {value!r}") + return number + + +def check_allowance(allowance) -> float: + number = _number(allowance, "allowance") + if not 0.0 <= number <= 1.0: + raise ValueError(f"allowance: must lie between 0 and 1 (0.20 is 20 %), got {number:g}") + return number + + +def check_pipe_options(*, services, allowance, vertical_allowance, length_source) -> dict: + """Refuse bad takeoff options by name, before a drawing is read.""" + chosen = DEFAULT_SERVICES if services is None else tuple(dict.fromkeys(services)) + if not chosen: + raise ValueError("services: an empty list takes nothing off; omit it for the default") + unknown = [s for s in chosen if s not in SERVICES] + if unknown: + raise ValueError( + f"services: {', '.join(map(str, unknown))} unknown; choose from {', '.join(SERVICES)}" + ) + vertical = _number(vertical_allowance, "vertical_allowance") + if vertical < 0.0: + raise ValueError(f"vertical_allowance: must be zero or more metres, got {vertical:g}") + if length_source not in LENGTH_SOURCES: + raise ValueError( + f"length_source: {length_source!r} unknown; choose from {', '.join(LENGTH_SOURCES)}" + ) + return { + "services": chosen, + "allowance": check_allowance(allowance), + "vertical_allowance": vertical, + "length_source": length_source, + } + + +# -- units ------------------------------------------------------------------- + + +def _model_points(snap: Snapshot) -> list[Pt]: + out: list[Pt] = [] + for rec in snap.records: + if rec.space != MODEL: + continue + out.extend(rec.points) + if rec.bbox is not None: + out.extend(((rec.bbox[0], rec.bbox[1]), (rec.bbox[2], rec.bbox[3]))) + return out + + +def _robust_box(points: list[Pt]) -> tuple[float, float, float, float] | None: + """The 5th-95th percentile box of the points: one stray entity does not stretch it.""" + if len(points) < 2: + return None + qx = statistics.quantiles([p[0] for p in points], n=20, method="inclusive") + qy = statistics.quantiles([p[1] for p in points], n=20, method="inclusive") + return (qx[0], qy[0], qx[-1], qy[-1]) + + +def _text_heights(snap: Snapshot) -> list[float]: + return [ + rec.height + for rec in snap.records + if rec.space == MODEL and rec.type in TEXT_TYPES and rec.height and rec.height > 0.0 + ] + + +def unit_of(snap: Snapshot) -> dict: + """The unit a drawing's geometry implies, against the one INSUNITS declares. + + ``{"declared", "inferred", "m_per_unit", "warning", "evidence"}``. The + declared unit (mm when INSUNITS declares none) stands when it puts the + median model-space text height between 0.5 mm and 1 m and the robust span + (5th-95th percentile) between 0.1 m and 5 km - this module's heuristic. + Otherwise the unit of mm / cm / m / in / ft that fits both bands and lies + nearest a 50 m drawing with 0.25 m text is taken. A disagreement is a + warning with the numbers - never a silent correction. + """ + declared = INSUNITS_NAMES.get(snap.insunits) + box = _robust_box(_model_points(snap)) + span = max(box[2] - box[0], box[3] - box[1]) if box else 0.0 + heights = _text_heights(snap) + height = statistics.median(heights) if heights else 0.0 + + def fits(unit: str) -> bool: + m = UNIT_METRES[unit] + return (span <= 0.0 or SPAN_BAND_M[0] <= span * m <= SPAN_BAND_M[1]) and ( + height <= 0.0 or TEXT_BAND_M[0] <= height * m <= TEXT_BAND_M[1] + ) + + def score(unit: str) -> float: + m = UNIT_METRES[unit] + total = abs(math.log(span * m / TYPICAL_SPAN_M)) if span > 0.0 else 0.0 + return total + (abs(math.log(height * m / TYPICAL_TEXT_M)) if height > 0.0 else 0.0) + + base = declared or "mm" + inferred = base + if not fits(base): + candidates = [unit for unit in UNIT_METRES if fits(unit)] + if candidates: + inferred = min(candidates, key=lambda unit: (score(unit), unit)) + if declared is not None: + said = declared + elif snap.insunits: + said = f"code {snap.insunits} (a unit this reader does not convert)" + else: + said = "no unit" + warning = None + if inferred != declared: + warning = ( + f"{snap.source}: INSUNITS declares {said} but the geometry reads " + f"as {inferred} (median text height {height:g}, robust span {span:g} drawing " + f"units); lengths are taken in {inferred}" + ) + return { + "declared": declared, + "inferred": inferred, + "m_per_unit": UNIT_METRES[inferred], + "warning": warning, + "evidence": {"robust_span": span, "median_text_height": height}, + } + + +# -- plan copies and tag positions --------------------------------------------- + + +def _mark_segment(cells: set, a: Pt, b: Pt, gap: float, limit: float) -> None: + length = math.dist(a, b) + if length > limit: + steps = 0 # a stray line across the world marks only its ends + else: + steps = int(length / (gap / 2.0)) + for i in range(steps + 1): + t = i / steps if steps else 0.0 + x, y = a[0] + (b[0] - a[0]) * t, a[1] + (b[1] - a[1]) * t + cells.add((math.floor(x / gap), math.floor(y / gap))) + cells.add((math.floor(b[0] / gap), math.floor(b[1] / gap))) + + +def _clusters(snap: Snapshot) -> tuple[float, list[tuple[float, float, float, float]]]: + """(cell size, cluster boxes): separate bodies of model-space content. + + Every record's geometry marks the cells of a grid whose cell is 2 % of the + robust span; 8-connected cells form bodies; bodies whose boxes overlap are + one cluster (a label in the middle of a room belongs to the plan around it). + """ + points = _model_points(snap) + box = _robust_box(points) + if box is None: + return 1.0, [] + span = max(box[2] - box[0], box[3] - box[1], 1.0) + gap = CLUSTER_GAP_SHARE * span + limit = 10.0 * span + cells: set[tuple[int, int]] = set() + for rec in snap.records: + if rec.space != MODEL: + continue + pts = list(rec.points) + if rec.type in ("LINE", "LWPOLYLINE", "POLYLINE") and len(pts) >= 2: + ring = pts + ([pts[0]] if rec.closed else []) + for a, b in zip(ring, ring[1:], strict=False): + _mark_segment(cells, a, b, gap, limit) + elif rec.bbox is not None: + x0, y0, x1, y1 = rec.bbox + corners = [(x0, y0), (x1, y0), (x1, y1), (x0, y1), (x0, y0)] + for a, b in zip(corners, corners[1:], strict=False): + _mark_segment(cells, a, b, gap, limit) + for p in pts[:1]: + cells.add((math.floor(p[0] / gap), math.floor(p[1] / gap))) + seen: set[tuple[int, int]] = set() + bodies: list[list[float]] = [] + for start in sorted(cells): + if start in seen: + continue + seen.add(start) + stack, body = [start], [start[0], start[1], start[0], start[1]] + while stack: + cx, cy = stack.pop() + body = [min(body[0], cx), min(body[1], cy), max(body[2], cx), max(body[3], cy)] + for dx in (-1, 0, 1): + for dy in (-1, 0, 1): + n = (cx + dx, cy + dy) + if n in cells and n not in seen: + seen.add(n) + stack.append(n) + bodies.append(body) + merged = True + while merged: + merged = False + for i in range(len(bodies)): + for j in range(i + 1, len(bodies)): + a, b = bodies[i], bodies[j] + if a[0] <= b[2] and b[0] <= a[2] and a[1] <= b[3] and b[1] <= a[3]: + bodies[i] = [min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3])] + del bodies[j] + merged = True + break + if merged: + break + boxes = [(b[0] * gap, b[1] * gap, (b[2] + 1) * gap, (b[3] + 1) * gap) for b in sorted(bodies)] + return gap, boxes + + +def layout_tags(snap: Snapshot, wanted: Iterable[str] = ()) -> dict: + """Tag positions on a drawing, using only tags that are unique where they are read. + + When no tag is written twice the whole model space is read. Otherwise the + drawing is split into clusters (plan copies, sheets, a far outlier) and the + cluster holding the most `wanted` tags is chosen (ties: the one furthest + left, then lowest); a tag written more than once inside it is `ambiguous`, + never picked arbitrarily. ``{"positions": {tag: Pt}, "ambiguous": [...], + "cluster": box | None, "clusters": n}``. + """ + occurrences = tag_occurrences(snap) + wanted = set(wanted) + if all(len(places) == 1 for places in occurrences.values()): + return { + "positions": {tag: places[0]["at"] for tag, places in occurrences.items()}, + "ambiguous": [], + "cluster": None, + "clusters": 1, + } + _gap, boxes = _clusters(snap) + + def inside(p: Pt, box) -> bool: + return box[0] <= p[0] <= box[2] and box[1] <= p[1] <= box[3] + + def weight(box) -> tuple: + count = sum( + 1 + for tag, places in occurrences.items() + if tag in wanted and any(inside(place["at"], box) for place in places) + ) + return (-count, box[0], box[1]) + + chosen = min(boxes, key=weight) if boxes else None + positions, ambiguous = {}, [] + for tag, places in sorted(occurrences.items()): + here = [place for place in places if chosen is None or inside(place["at"], chosen)] + if len(here) == 1: + positions[tag] = here[0]["at"] + elif len(here) > 1: + ambiguous.append(tag) + return { + "positions": positions, + "ambiguous": ambiguous, + "cluster": chosen, + "clusters": len(boxes), + } + + +def scale_stand_in( + pid: Snapshot, layout: Snapshot, *, pid_tags=None, layout_found=None, layout_unit=None +) -> dict: + """Spec §6 on the tags unique in both drawings - the takeoff's stand-in for scale_check. + + The result has `scale_check`'s shape (group U, Task 4): for every pair of + matched tags at least 2 m apart on the layout, ``ratio`` is their P&ID + distance over their layout distance, both in millimetres of each drawing's + inferred unit; ``median``, ``q1``, ``q3``, ``iqr`` and ``within_10pct`` (the + share within ±10 % of the median) describe them. Verdict ``to_scale`` at + >= 80 %, ``schematic`` at <= 50 %, ``partly`` between, ``insufficient`` + with fewer than three matched tags or no pair; ``factor`` is the median + only when ``to_scale``; a median that is not positive (every P&ID tag at + one point) is ``insufficient``. There are no per-room ``groups`` here. Tag labels + are not equipment centres: the band and the 2 m floor absorb that. + """ + pid_occ = pid_tags if pid_tags is not None else tag_occurrences(pid) + pid_positions = {tag: places[0]["at"] for tag, places in pid_occ.items() if len(places) == 1} + found = layout_found if layout_found is not None else layout_tags(layout, pid_positions) + unit = layout_unit if layout_unit is not None else unit_of(layout) + mm_layout = unit["m_per_unit"] * 1000.0 + mm_pid = unit_of(pid)["m_per_unit"] * 1000.0 + tags = [tag for tag in sorted(pid_positions) if tag in found["positions"]] + matched = {tag: (pid_positions[tag], found["positions"][tag]) for tag in tags} + ambiguous = sorted( + {tag for tag, places in pid_occ.items() if len(places) > 1} | set(found["ambiguous"]) + ) + pairs = [] + for i, a in enumerate(tags): + for b in tags[i + 1 :]: + d_layout = math.dist(matched[a][1], matched[b][1]) * mm_layout + if d_layout >= MIN_PAIR_MM: + d_pid = math.dist(matched[a][0], matched[b][0]) * mm_pid + pairs.append( + {"tags": [a, b], "a_mm": d_pid, "b_mm": d_layout, "ratio": d_pid / d_layout} + ) + out = { + "verdict": "insufficient", + "factor": None, + "median": None, + "q1": None, + "q3": None, + "iqr": None, + "within_10pct": None, + "pair_count": len(pairs), + "pairs": pairs, + "matched": [ + {"tag": tag, "a": list(matched[tag][0]), "b": list(matched[tag][1])} for tag in tags + ], + "ambiguous": ambiguous, + "groups": [], + "notes": ["takeoff stand-in for scale_check (spec §6): no per-room groups"], + } + ratios = [pair["ratio"] for pair in pairs] + if len(tags) < MIN_TAGS or not ratios: + return out + median = statistics.median(ratios) + if median <= 0.0: + return out # every P&ID tag at one point: no factor to divide by (as scale_check) + q1, q3 = ( + statistics.quantiles(ratios, n=4, method="inclusive")[::2] + if len(ratios) >= 2 + else (ratios[0], ratios[0]) + ) + within = sum(1 for r in ratios if abs(r - median) <= SCALE_BAND * median) / len(ratios) + if within >= TO_SCALE_SHARE: + verdict = "to_scale" + elif within <= SCHEMATIC_SHARE: + verdict = "schematic" + else: + verdict = "partly" + out.update( + verdict=verdict, + factor=median if verdict == "to_scale" else None, + median=median, + q1=q1, + q3=q3, + iqr=q3 - q1, + within_10pct=within, + ) + return out + + +# -- rooms ------------------------------------------------------------------- + + +def room_regions(snap: Snapshot, cluster=None) -> dict: + """Room polygons and room labels of a drawing. + + Labels are texts `room_label` reads; polygons are the faces of the wall + layers (`classify_layer` architecture, confidence >= 0.9, straight + segments) that hold a label, found by track F's planar-face finder. Inside + `cluster` only, when one is given. ``{"faces": [(key, Face)], "labels": + [(key, Pt)]}``; the key is the room number, else its name. + """ + + def keep(p: Pt) -> bool: + return cluster is None or ( + cluster[0] <= p[0] <= cluster[2] and cluster[1] <= p[1] <= cluster[3] + ) + + labels = [] + for rec in snap.records: + if rec.space != MODEL or rec.type not in TEXT_TYPES or not rec.text or not rec.points: + continue + found = room_label(rec.text) + if found and keep(rec.points[0]): + key = found.get("number") or found.get("name") or rec.text + labels.append((str(key), rec.points[0])) + walls = set() + for layer in {rec.layer for rec in snap.records}: # classify each layer once + info = classify_layer(layer) + if info["discipline"] == "architecture" and info["confidence"] >= WALL_CONFIDENCE: + walls.add(layer) + segments = [] + for rec in snap.records: + if rec.space != MODEL or rec.layer not in walls: + continue + if rec.type not in ("LINE", "LWPOLYLINE", "POLYLINE") or len(rec.points) < 2: + continue + pts = list(rec.points) + ([rec.points[0]] if rec.closed else []) + bulges = list(rec.bulges) + [0.0] * len(pts) + for i, (a, b) in enumerate(zip(pts, pts[1:], strict=False)): + if bulges[i] == 0.0 and keep(a) and keep(b): + segments.append((a, b)) + faces = planar_faces(segments, tol=DEFAULT_TOL) if segments else () + rooms: list[tuple[str, Face]] = [] + taken: set[int] = set() + for key, at in sorted(labels): + face = face_containing(faces, at) + if face is not None and id(face) not in taken: + taken.add(id(face)) + rooms.append((key, face)) + return {"faces": rooms, "labels": sorted(labels)} + + +def _rooms_mode(regions: dict) -> str: + if regions["faces"]: + return "polygons" + return "nearest_label" if regions["labels"] else "none" + + +def room_at(regions: dict, p: Pt) -> str: + """The room a point stands in; else the nearest room label; else ''.""" + face = face_containing([face for _key, face in regions["faces"]], p) + if face is not None: + return next(key for key, f in regions["faces"] if f is face) + if regions["labels"]: + return min(regions["labels"], key=lambda item: (math.dist(item[1], p), item[0]))[0] + return "" + + +def apportion(legs, regions: dict) -> dict[str, float]: + """Length of each orthogonal leg inside each room polygon (drawing units). + + A leg is cut at every crossing with a room outline or hole; each piece goes + to the room its midpoint stands in, or to '' outside every room. Nothing is + counted twice: the pieces add up to the leg. + """ + faces = [face for _key, face in regions["faces"]] + key_of = {id(face): key for key, face in regions["faces"]} + out: dict[str, float] = {} + for a, b in legs: + length = math.dist(a, b) + if length == 0.0: + continue + cuts = {0.0, 1.0} + for face in faces: + for ring in (face.loop, *face.holes): + for i in range(len(ring)): + t = _cut(a, b, ring[i], ring[(i + 1) % len(ring)]) + if t is not None: + cuts.add(t) + ts = sorted(cuts) + for t0, t1 in zip(ts, ts[1:], strict=False): + if t1 - t0 <= 0.0: + continue + tm = (t0 + t1) / 2.0 + mid = (a[0] + (b[0] - a[0]) * tm, a[1] + (b[1] - a[1]) * tm) + face = face_containing(faces, mid) + key = key_of[id(face)] if face is not None else "" + out[key] = out.get(key, 0.0) + (t1 - t0) * length + return out + + +def _cut(a: Pt, b: Pt, c: Pt, d: Pt) -> float | None: + """Parameter along a->b where it crosses segment c-d, or None.""" + rx, ry = b[0] - a[0], b[1] - a[1] + sx, sy = d[0] - c[0], d[1] - c[1] + denom = rx * sy - ry * sx + if denom == 0.0: + return None + qx, qy = c[0] - a[0], c[1] - a[1] + t = (qx * sy - qy * sx) / denom + u = (qx * ry - qy * rx) / denom + if 0.0 < t < 1.0 and 0.0 <= u <= 1.0: + return t + return None + + +# -- pipe rows --------------------------------------------------------------- + + +def _choose_source(requested: str, layout, scale: dict, force: bool) -> tuple[str, str]: + verdict = scale.get("verdict") + if requested == "layout": + if layout is None: + raise ValueError("length_source='layout' needs a layout drawing; none was given") + return "layout", "requested" + if requested == "pid": + if layout is not None and verdict == "schematic" and not force: + raise ValueError( + "length_source='pid': the scale check calls this P&ID schematic - " + f"{scale['within_10pct']:.0%} of {scale['pair_count']} tag pairs lie within " + f"±10 % of the median ratio {scale['median']:.6g} (IQR {scale['iqr']:.6g}); " + "a length measured on it is drawn line, not pipe. Pass force=True to measure " + "on it anyway, or use length_source='layout'." + ) + return "pid", "forced" if layout is not None and verdict == "schematic" else "requested" + if layout is None: + return "pid", "no_layout" + if verdict == "to_scale": + return "pid", "auto_to_scale" + return "layout", f"auto_{verdict}" + + +def pipe_rows( + network, + pid: Snapshot, + layout: Snapshot | None, + *, + scale=None, + services=DEFAULT_SERVICES, + allowance=0.20, + vertical_allowance=0.0, + length_source="auto", + force=False, +) -> dict: + """Takeoff rows by room x service x diameter, with the runs, checks and method behind them. + + `network` is `build_network(pid)`'s result. `scale` is a scale-check + result (`scale_check`'s shape); omitted, the stand-in runs when a layout + is given. Refused by name before anything is measured: an unknown service + or length source, an allowance outside 0-1, a negative vertical + allowance, `length_source='layout'` without a layout, and + `length_source='pid'` on a P&ID the scale check calls schematic unless + `force=True` (the refusal quotes the statistics). + + Status per run: **A** every end on a tag placed on the layout and every + piece's diameter read directly; **B** a diameter carried by continuity or + unassigned; **C** not routable - an end on no equipment, fewer than two + tags, a tag missing from (or written twice on) the layout. A C run goes to + `control` with its schematic length only, is counted in the totals and is + flagged. A run holding several diameters shares its length among them in + proportion to their drawn length on the P&ID. + """ + options = check_pipe_options( + services=services, + allowance=allowance, + vertical_allowance=vertical_allowance, + length_source=length_source, + ) + allowance = options["allowance"] + vertical = options["vertical_allowance"] + pid_unit = unit_of(pid) + layout_unit = unit_of(layout) if layout is not None else None + wanted = {tag for run in network["runs"] for tag in run.tags} + found = layout_tags(layout, wanted) if layout is not None else None + if scale is None: + scale = ( + scale_stand_in(pid, layout, layout_found=found, layout_unit=layout_unit) + if layout is not None + else {"verdict": "not_checked", "pair_count": 0, "matched": [], "ambiguous": []} + ) + source, reason = _choose_source(length_source, layout, scale, bool(force)) + scale_verified = source == "layout" or scale.get("verdict") == "to_scale" + positions = found["positions"] if found else {} + ambiguous = set(found["ambiguous"]) if found else set() + regions = ( + room_regions(layout, found["cluster"]) + if found is not None and source == "layout" + else room_regions(pid) + ) + m_pid = pid_unit["m_per_unit"] + m_layout = layout_unit["m_per_unit"] if layout_unit else None + + rows: dict[tuple[str, str, str], dict] = {} + runs_out, control, excluded = [], [], {} + for run in network["runs"]: + if run.service not in options["services"]: + excluded[run.service] = excluded.get(run.service, 0) + 1 + continue + ends = network["attachments"].get(run.id, []) + drawn = sum(edge["length"] for edge in run.edges) + schematic_m = drawn * m_pid + parts: dict[str, dict] = {} + for edge in run.edges: + part = parts.setdefault( + edge["diameter"] or UNASSIGNED, + {"drawn": 0.0, "direct": 0.0, "continuity": 0.0, "unassigned": 0.0}, + ) + part["drawn"] += edge["length"] + part[edge["diameter_source"]] += edge["length"] + why = None + if any(end["tag"] is None for end in ends) or not ends: + why = "end_not_on_equipment" + elif len(run.tags) < 2: + why = "single_tag" + elif source == "layout": + missing = [tag for tag in run.tags if tag not in positions] + if missing: + why = ( + "tag_ambiguous" if any(t in ambiguous for t in missing) else "tag_not_on_layout" + ) + detail = { + "id": run.id, + "service": run.service, + "tags": list(run.tags), + "diameters": {d: p["drawn"] * m_pid for d, p in sorted(parts.items())}, + "schematic_m": schematic_m, + "layout_m": None, + "vertical_m": 0.0, + "length_m": None, + "rooms": {}, + "status": "C", + "reason": why, + } + runs_out.append(detail) + for diameter, part in sorted(parts.items()): + if diameter == UNASSIGNED: + control.append( + { + "run": run.id, + "kind": "unassigned_diameter", + "service": run.service, + "tags": list(run.tags), + "schematic_m": part["drawn"] * m_pid, + "reason": "no_diameter_label", + } + ) + if why is not None: + control.append( + { + "run": run.id, + "kind": "unroutable", + "service": run.service, + "tags": list(run.tags), + "schematic_m": schematic_m, + "reason": why, + } + ) + continue + if source == "layout": + points = [positions[tag] for tag in run.tags] + length_units, tree = rmst(points) + if regions["faces"]: + legs = [leg for i, j in tree for leg in route_legs(points[i], points[j])] + rooms = {key: value * m_layout for key, value in apportion(legs, regions).items()} + else: + centre = ( + sum(p[0] for p in points) / len(points), + sum(p[1] for p in points) / len(points), + ) + rooms = {room_at(regions, centre): length_units * m_layout} + for _i, j in tree: + key = room_at(regions, points[j]) + rooms[key] = rooms.get(key, 0.0) + vertical + detail["layout_m"] = length_units * m_layout + detail["vertical_m"] = vertical * len(tree) + length_m = detail["layout_m"] + detail["vertical_m"] + else: + if layout is not None and scale.get("verdict") == "to_scale": + length_m = schematic_m / scale["factor"] + else: + length_m = schematic_m + centre = ( + sum(end["at"][0] for end in ends) / len(ends), + sum(end["at"][1] for end in ends) / len(ends), + ) + rooms = {room_at(regions, centre): length_m} + detail["length_m"] = length_m + detail["rooms"] = rooms + status = ( + "A" + if all(p["continuity"] == 0.0 and p["unassigned"] == 0.0 for p in parts.values()) + else "B" + ) + detail["status"] = status + for diameter, part in parts.items(): + share = part["drawn"] / drawn if drawn else 0.0 + for room, room_m in rooms.items(): + x = room_m * share + row = rows.setdefault( + (room, run.service, diameter), + { + "room": room, + "service": run.service, + "diameter": diameter, + "direct_m": 0.0, + "continuity_m": 0.0, + "unassigned_m": 0.0, + "net_m": 0.0, + "schematic_m": 0.0, + "status": "A", + "runs": [], + }, + ) + row["net_m"] += x + row["direct_m"] += x * part["direct"] / part["drawn"] + row["continuity_m"] += x * part["continuity"] / part["drawn"] + row["unassigned_m"] += x * part["unassigned"] / part["drawn"] + row["schematic_m"] += schematic_m * share * (room_m / length_m if length_m else 0.0) + if part["continuity"] or part["unassigned"]: + row["status"] = "B" + if run.id not in row["runs"]: + row["runs"].append(run.id) + + order = {service: i for i, service in enumerate(SERVICES)} + table = [] + for key in sorted(rows, key=lambda k: (k[0], order.get(k[1], len(order)), k[2])): + row = rows[key] + row["allowance"] = allowance + row["allowance_m"] = row["net_m"] * allowance + row["total_m"] = row["net_m"] + row["allowance_m"] + table.append(row) + + routed = sum(row["net_m"] for row in table) + unroutable = sum(c["schematic_m"] for c in control if c["kind"] == "unroutable") + net_total = routed + unroutable + summary = { + name: [ + { + "key": key, + "net_m": sum(r["net_m"] for r in table if r[field] == key), + "total_m": sum(r["total_m"] for r in table if r[field] == key), + } + for key in sorted({r[field] for r in table}) + ] + for name, field in ( + ("by_service", "service"), + ("by_diameter", "diameter"), + ("by_room", "room"), + ) + } + warnings = [u["warning"] for u in (pid_unit, layout_unit) if u and u["warning"]] + if not scale_verified: + warnings.append( + "scale_verified: false - lengths are drawn lengths on a P&ID that no scale check " + "has confirmed; they are not physical pipe lengths" + ) + return { + "kind": "pipe", + "rows": table, + "runs": runs_out, + "control": control, + "summary": summary, + "totals": { + "routed_m": routed, + "unroutable_m": unroutable, + "net_m": net_total, + "allowance": allowance, + "allowance_m": net_total * allowance, + "total_m": net_total * (1.0 + allowance), + "flagged_runs": sum(1 for r in runs_out if r["status"] == "C"), + }, + "length_source": source, + "scale": scale, + "scale_verified": scale_verified, + "units": {"pid": pid_unit, "layout": layout_unit}, + "warnings": warnings, + "method": { + "sources": {"pid": pid.source, "layout": layout.source if layout else None}, + "length_source": source, + "requested": length_source, + "reason": reason, + "allowance": allowance, + "vertical_allowance": vertical, + "services": list(options["services"]), + "excluded_services": excluded, + "tol": network["stats"]["tol"], + "label_search": network["stats"]["label_search"], + "overlap_length_m": network["stats"]["overlap_length"] * m_pid, + "route": "rectilinear MST of the tag positions; each connection x first, then y", + "rooms": _rooms_mode(regions), + }, + } diff --git a/tests/test_understand_route.py b/tests/test_understand_route.py new file mode 100644 index 0000000..2aa823d --- /dev/null +++ b/tests/test_understand_route.py @@ -0,0 +1,48 @@ +"""Orthogonal lengths: Manhattan distance, the rectilinear MST, the x-then-y route. + +Every expected value is worked by hand in the comment beside it. +""" + +from __future__ import annotations + +from engineering.understand.route import manhattan, rmst, route_legs + +EPS = 1e-9 + + +def test_manhattan_is_the_sum_of_the_axis_distances(): + assert manhattan((0.0, 0.0), (3.0, -4.0)) == 7.0 # 3 + 4, not the diagonal 5 + + +def test_fewer_than_two_points_have_no_tree(): + assert rmst([]) == (0.0, ()) + assert rmst([(5.0, 5.0)]) == (0.0, ()) + + +def test_a_two_ended_run_is_dx_plus_dy(): + length, edges = rmst([(1000.0, 3000.0), (29000.0, 15000.0)]) + assert abs(length - 40000.0) < EPS # 28000 + 12000 + assert edges == ((0, 1),) + + +def test_a_square_of_four_tags_is_three_sides(): + # Prim from point 0: (0,1) at 10 (tie with 3 goes to the lower index), + # then (1,2) at 10, then (0,3) at 10 -> 30, never the 40 of the full ring + length, edges = rmst([(0.0, 0.0), (10.0, 0.0), (10.0, 10.0), (0.0, 10.0)]) + assert abs(length - 30.0) < EPS + assert edges == ((0, 1), (1, 2), (0, 3)) + + +def test_three_tags_in_a_row_chain_instead_of_a_star(): + length, edges = rmst([(0.0, 0.0), (100.0, 0.0), (200.0, 0.0)]) + assert abs(length - 200.0) < EPS # 100 + 100, not 100 + 200 + assert edges == ((0, 1), (1, 2)) + + +def test_the_route_runs_along_x_first_then_y(): + assert route_legs((0.0, 0.0), (30.0, 40.0)) == ( + ((0.0, 0.0), (30.0, 0.0)), + ((30.0, 0.0), (30.0, 40.0)), + ) + assert route_legs((0.0, 0.0), (0.0, 40.0)) == (((0.0, 0.0), (0.0, 40.0)),) + assert route_legs((1.0, 1.0), (1.0, 1.0)) == () diff --git a/tests/test_understand_takeoff.py b/tests/test_understand_takeoff.py new file mode 100644 index 0000000..f332908 --- /dev/null +++ b/tests/test_understand_takeoff.py @@ -0,0 +1,423 @@ +"""Pipe takeoff rows: layout lengths, statuses, rooms, units and the scale rule. + +The hand-built pair: a P&ID (mm) with three tanks and two runs, and a layout +whose INSUNITS says inches over millimetre geometry - two rooms split by a +wall at x = 20000, the tags at T4100 (2000, 3000), T4101 (30000, 3000) and +T4102 (30000, 15000). Every expected length is worked in the comment beside it. +The last four tests read the synthetic plant pair of Task 2 and compare every +run with the generator's own hand-computed rectilinear MST. +""" + +from __future__ import annotations + +import pytest + +from engineering.understand.network import build_network +from engineering.understand.snapshot import EntityRecord, Snapshot, read_snapshot +from engineering.understand.takeoff import ( + DEFAULT_SERVICES, + layout_tags, + pipe_rows, + scale_stand_in, + unit_of, +) +from engineering.understand.vocab import classify_layer, room_label +from tests.fixtures.plant_pair import build_plant_pair + +EPS = 1e-9 +PIPE = "PRODUCT" +WALLS = "WALLS" + + +def rec(handle, kind, layer, points=(), **extra): + return EntityRecord( + handle=handle, type=kind, layer=layer, space="Model", points=tuple(points), **extra + ) + + +def text(handle, value, at, height, layer="TEXT"): + x, y = at + return rec( + handle, + "TEXT", + layer, + (at,), + text=value, + height=height, + bbox=(x, y, x + height * len(value) * 0.6, y + height), + ) + + +def snap(records, insunits, source): + layers = {r.layer: {"color": 7, "linetype": "Continuous"} for r in records} + return Snapshot( + source=source, + insunits=insunits, + extmin=None, + extmax=None, + layers=layers, + layouts=(), + records=tuple(records), + ) + + +def tank(handle, tag, x0, y0): + box = (x0, y0, x0 + 500.0, y0 + 500.0) + return [ + rec( + handle, + "LWPOLYLINE", + "EQUIPMENT", + ((x0, y0), (x0 + 500.0, y0), (x0 + 500.0, y0 + 500.0), (x0, y0 + 500.0)), + closed=True, + bbox=box, + ), + text(handle + "T", tag, (x0 + 150.0, y0 + 225.0), 50.0), + ] + + +def pid_records(): + return [ + *tank("A", "T4100", 0.0, 0.0), + *tank("B", "T4101", 3000.0, 0.0), + *tank("C", "T4102", 3000.0, 2000.0), + rec("L1", "LINE", PIPE, ((500.0, 250.0), (3000.0, 250.0))), + rec("L2", "LINE", PIPE, ((3250.0, 500.0), (3250.0, 2000.0))), + text("D1", "Ø51", (1500.0, 260.0), 50.0), + ] + + +def layout_records(dx=0.0, extra=()): + def at(x, y): + return (x + dx, y) + + return [ + rec( + f"W{dx}", + "LWPOLYLINE", + WALLS, + (at(0, 0), at(40000, 0), at(40000, 20000), at(0, 20000)), + closed=True, + bbox=(dx, 0.0, dx + 40000.0, 20000.0), + ), + rec(f"X{dx}", "LINE", WALLS, (at(20000, 0), at(20000, 20000))), + text(f"R1{dx}", "ROOM 101", at(5000, 10000), 250.0), + text(f"R2{dx}", "ROOM 102", at(30000, 10000), 250.0), + text(f"T0{dx}", "T4100", at(2000, 3000), 250.0), + text(f"T1{dx}", "T4101", at(30000, 3000), 250.0), + text(f"T2{dx}", "T4102", at(30000, 15000), 250.0), + *extra, + ] + + +def room_key(label): + found = room_label(label) + assert found is not None, f"the vocabulary must read {label!r} as a room label" + return str(found.get("number") or found.get("name")) + + +@pytest.fixture +def pair(): + assert classify_layer(PIPE)["service"] == "product", "PRODUCT must be a product layer" + assert classify_layer(WALLS)["discipline"] == "architecture", "WALLS must be a wall layer" + pid = snap(pid_records(), 4, "pid.dxf") + layout = snap(layout_records(), 1, "layout.dxf") # INSUNITS 1 = inches, geometry in mm + return pid, layout, build_network(pid, layers=[PIPE]) + + +def run_by_tags(result, *tags): + (run,) = [r for r in result["runs"] if r["tags"] == list(tags)] + return run + + +def test_layout_lengths_are_manhattan_between_the_tags(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, length_source="layout") + a = run_by_tags(result, "T4100", "T4101") + b = run_by_tags(result, "T4101", "T4102") + assert abs(a["layout_m"] - 28.0) < EPS # |30000 - 2000| + 0 mm + assert abs(b["layout_m"] - 12.0) < EPS # 0 + |15000 - 3000| mm + assert (a["status"], b["status"]) == ("A", "B") # b has no diameter label + assert result["length_source"] == "layout" + + +def test_review_focus_5_inches_declared_over_millimetres(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, length_source="layout") + units = result["units"]["layout"] + assert (units["declared"], units["inferred"], units["m_per_unit"]) == ("in", "mm", 0.001) + a = run_by_tags(result, "T4100", "T4101") + assert abs(a["layout_m"] - 28.0) < EPS # not 28000 x 0.0254 = 711.2 + assert any("INSUNITS declares in" in w and "as mm" in w for w in result["warnings"]) + assert unit_of(pid)["warning"] is None # the P&ID's mm is what its geometry says + + +def test_rooms_share_the_route_x_first(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, length_source="layout") + a = run_by_tags(result, "T4100", "T4101") + # (2000, 3000) -> (30000, 3000): 18000 mm left of the wall at x = 20000, 10000 right + assert a["rooms"] == pytest.approx( + {room_key("ROOM 101"): 18.0, room_key("ROOM 102"): 10.0}, abs=EPS + ) + rows = {(r["room"], r["diameter"]): r for r in result["rows"]} + assert rows[(room_key("ROOM 101"), "Ø51")]["net_m"] == pytest.approx(18.0, abs=EPS) + assert rows[(room_key("ROOM 101"), "Ø51")]["direct_m"] == pytest.approx(18.0, abs=EPS) + assert rows[(room_key("ROOM 102"), "unassigned")]["unassigned_m"] == pytest.approx( + 12.0, abs=EPS + ) + + +def test_allowance_is_its_own_column_and_totals_add_up(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, length_source="layout", allowance=0.25) + for row in result["rows"]: + assert row["allowance"] == 0.25 + assert row["allowance_m"] == pytest.approx(row["net_m"] * 0.25, abs=EPS) + assert row["total_m"] == pytest.approx(row["net_m"] * 1.25, abs=EPS) + totals = result["totals"] + assert totals["net_m"] == pytest.approx(40.0, abs=EPS) # 28 + 12 + assert totals["total_m"] == pytest.approx(50.0, abs=EPS) + assert totals["flagged_runs"] == 0 + + +def test_vertical_allowance_is_added_per_connection(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, length_source="layout", vertical_allowance=0.5) + a = run_by_tags(result, "T4100", "T4101") + assert a["vertical_m"] == 0.5 # one tree edge + assert a["length_m"] == pytest.approx(28.5, abs=EPS) + + +def test_the_schematic_length_is_kept_beside_the_answer(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, length_source="layout") + a = run_by_tags(result, "T4100", "T4101") + assert a["schematic_m"] == pytest.approx(2.5, abs=EPS) # the 2500 mm line on the P&ID + + +def test_a_tag_missing_from_the_layout_makes_the_run_c(pair): + pid, _layout, network = pair + layout = snap([r for r in layout_records() if r.text != "T4102"], 1, "layout.dxf") + result = pipe_rows(network, pid, layout, length_source="layout") + b = run_by_tags(result, "T4101", "T4102") + assert (b["status"], b["reason"]) == ("C", "tag_not_on_layout") + (row,) = [c for c in result["control"] if c["kind"] == "unroutable"] + assert row["schematic_m"] == pytest.approx(1.5, abs=EPS) # the 1500 mm line, drawn length only + assert result["totals"]["unroutable_m"] == pytest.approx(1.5, abs=EPS) + assert result["totals"]["flagged_runs"] == 1 + + +def test_a_tag_written_twice_in_the_plan_is_ambiguous_not_picked(pair): + pid, _layout, network = pair + extra = [text("dup", "T4101", (35000.0, 18000.0), 250.0)] + layout = snap(layout_records(extra=extra), 1, "layout.dxf") + result = pipe_rows(network, pid, layout, length_source="layout") + assert {r["reason"] for r in result["runs"]} == {"tag_ambiguous"} + + +def test_of_two_plan_copies_one_is_chosen_and_nothing_is_ambiguous(pair): + pid, _layout, network = pair + layout = snap(layout_records() + layout_records(dx=100000.0), 1, "layout.dxf") + found = layout_tags(layout, ["T4100", "T4101", "T4102"]) + assert found["clusters"] == 2 + assert found["cluster"][2] < 100000.0 # the left copy: ties go to the leftmost + assert found["ambiguous"] == [] + result = pipe_rows(network, pid, layout, length_source="layout") + assert run_by_tags(result, "T4100", "T4101")["layout_m"] == pytest.approx(28.0, abs=EPS) + + +def test_a_run_with_a_free_end_is_c(pair): + pid, layout, _network = pair + records = pid_records() + [rec("L3", "LINE", PIPE, ((0.0, 500.0), (0.0, 1500.0)))] + drawing = snap(records, 4, "pid.dxf") + result = pipe_rows( + build_network(drawing, layers=[PIPE]), drawing, layout, length_source="layout" + ) + reasons = {r["reason"] for r in result["runs"] if r["status"] == "C"} + assert reasons == {"end_not_on_equipment"} + + +def test_other_services_are_left_out_unless_asked(pair): + pid, layout, network = pair + assert DEFAULT_SERVICES == ("product", "cip_supply", "cip_return") + result = pipe_rows(network, pid, layout, length_source="layout", services=["steam"]) + assert result["rows"] == [] + assert result["method"]["excluded_services"] == {"product": 2} + + +# -- the scale rule ---------------------------------------------------------- + +SCHEMATIC = { + "verdict": "schematic", + "within_10pct": 0.25, + "pair_count": 12, + "median": 1.3, + "iqr": 0.8, + "factor": None, + "matched": [], + "ambiguous": [], +} + + +def test_auto_measures_a_schematic_pid_on_the_layout(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, scale=SCHEMATIC) + assert result["length_source"] == "layout" + assert result["method"]["reason"] == "auto_schematic" + assert result["scale_verified"] is True + + +def test_pid_on_a_schematic_pid_is_refused_with_the_statistics(pair): + pid, layout, network = pair + with pytest.raises(ValueError) as excinfo: + pipe_rows(network, pid, layout, scale=SCHEMATIC, length_source="pid") + message = str(excinfo.value) + assert message.startswith("length_source='pid': the scale check calls this P&ID schematic") + assert "25% of 12 tag pairs" in message and "median ratio 1.3" in message + assert "force=True" in message + + +def test_force_measures_on_the_pid_and_says_it_is_unverified(pair): + pid, layout, network = pair + result = pipe_rows(network, pid, layout, scale=SCHEMATIC, length_source="pid", force=True) + assert result["length_source"] == "pid" + assert result["method"]["reason"] == "forced" + assert result["scale_verified"] is False + assert run_by_tags(result, "T4100", "T4101")["length_m"] == pytest.approx(2.5, abs=EPS) + assert any(w.startswith("scale_verified: false") for w in result["warnings"]) + + +def test_a_to_scale_pid_is_measured_through_its_factor(pair): + pid, layout, network = pair + to_scale = dict(SCHEMATIC, verdict="to_scale", within_10pct=0.9, factor=0.1) + result = pipe_rows(network, pid, layout, scale=to_scale) + assert result["length_source"] == "pid" + # 2.5 m drawn on the P&ID / 0.1 P&ID mm per layout mm = 25 m + assert run_by_tags(result, "T4100", "T4101")["length_m"] == pytest.approx(25.0, abs=EPS) + + +def test_without_a_layout_auto_measures_the_pid_unverified(pair): + pid, _layout, network = pair + result = pipe_rows(network, pid, None) + assert (result["length_source"], result["scale_verified"]) == ("pid", False) + assert result["method"]["reason"] == "no_layout" + with pytest.raises(ValueError, match="^length_source='layout' needs a layout drawing"): + pipe_rows(network, pid, None, length_source="layout") + + +def test_the_stand_in_recognises_a_uniform_scale_and_a_rearranged_one(): + places = { + "T4100": (2000.0, 3000.0), + "T4101": (30000.0, 3000.0), + "T4102": (30000.0, 15000.0), + "T4103": (8000.0, 17000.0), + } + layout = snap( + [text(t, t, p, 250.0) for t, p in places.items()] + + [rec("W", "LINE", WALLS, ((0.0, 0.0), (40000.0, 20000.0)))], + 4, + "l", + ) + half = snap([text(t, t, (x * 0.5, y * 0.5), 50.0) for t, (x, y) in places.items()], 4, "p") + uniform = scale_stand_in(half, layout) + assert uniform["verdict"] == "to_scale" + assert uniform["factor"] == pytest.approx(0.5, abs=EPS) # P&ID mm per layout mm + assert uniform["pair_count"] == 6 # four tags, every pair at least 2 m apart + shuffled = { + "T4100": (30000.0, 15000.0), + "T4101": (2000.0, 3000.0), + "T4102": (8000.0, 17000.0), + "T4103": (30000.0, 3000.0), + } + other = snap([text(t, t, p, 50.0) for t, p in shuffled.items()], 4, "p") + assert scale_stand_in(other, layout)["verdict"] == "schematic" + + +def test_the_stand_in_calls_coinciding_pid_tags_insufficient_not_to_scale(): + # every P&ID tag at one point: each ratio is 0, the median 0 - a factor of 0 + # would divide every length by zero, so the verdict is insufficient (the + # same rule as group U's scale_check: a median that is not positive) + places = {"T4100": (2000.0, 3000.0), "T4101": (30000.0, 3000.0), "T4102": (30000.0, 15000.0)} + layout = snap([text(t, t, p, 250.0) for t, p in places.items()], 4, "l") + pid = snap([text(t, t, (100.0, 100.0), 50.0) for t in places], 4, "p") + result = scale_stand_in(pid, layout) + assert result["pair_count"] == 3 + assert (result["verdict"], result["factor"]) == ("insufficient", None) + + +def test_an_insunits_code_this_reader_does_not_convert_is_named_not_called_none(): + drawing = snap(layout_records(), 14, "layout.dxf") # 14 = decimetres + found = unit_of(drawing) + assert (found["declared"], found["inferred"]) == (None, "mm") + assert "INSUNITS declares code 14 (a unit this reader does not convert)" in found["warning"] + assert "no unit" not in found["warning"] + + +@pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"allowance": 1.5}, "allowance: must lie between 0 and 1"), + ({"allowance": -0.1}, "allowance: must lie between 0 and 1"), + ({"vertical_allowance": -1.0}, "vertical_allowance: must be zero or more metres"), + ({"length_source": "tape"}, "length_source: 'tape' unknown"), + ({"services": ["lava"]}, "services: lava unknown"), + ({"services": []}, "services: an empty list takes nothing off"), + ], +) +def test_refusals_name_the_option(pair, kwargs, message): + pid, layout, network = pair + with pytest.raises(ValueError, match="^" + message): + pipe_rows(network, pid, layout, **kwargs) + + +# -- the synthetic plant pair (Task 2) ------------------------------------------- + + +@pytest.fixture(scope="module") +def plant(tmp_path_factory): + truth = build_plant_pair(tmp_path_factory.mktemp("plant")) + pid, layout = read_snapshot(truth["pid"]), read_snapshot(truth["layout"]) + return truth, pid, layout, build_network(pid) + + +def truth_services(truth): + return sorted({run["service"] for run in truth["pipe_runs"]}) + + +def test_every_synthetic_layout_length_is_the_hand_computed_rmst(plant): + truth, pid, layout, network = plant + result = pipe_rows(network, pid, layout, services=truth_services(truth), length_source="layout") + runs = {(tuple(sorted(r["tags"])), r["service"]): r for r in result["runs"]} + for expected in truth["pipe_runs"]: + run = runs[(tuple(sorted(expected["tags"])), expected["service"])] + assert run["status"] == expected["status"], run["id"] + if expected["status"] != "C": + assert run["layout_m"] * 1000.0 == pytest.approx(expected["layout_length_mm"], abs=1e-6) + assert sum(run["rooms"].values()) == pytest.approx(run["length_m"], abs=1e-9) + + +def test_auto_calls_the_synthetic_pid_schematic_and_measures_the_layout(plant): + truth, pid, layout, network = plant + result = pipe_rows(network, pid, layout, services=truth_services(truth)) + assert result["scale"]["verdict"] == truth["scale_verdict"] == "schematic" + assert result["length_source"] == "layout" + with pytest.raises(ValueError, match="the scale check calls this P&ID schematic"): + pipe_rows(network, pid, layout, services=truth_services(truth), length_source="pid") + + +def test_review_focus_5_the_synthetic_layout_is_read_in_millimetres(plant): + truth, pid, layout, network = plant + assert layout.insunits == truth["layout_insunits"] == 1 + result = pipe_rows(network, pid, layout, services=truth_services(truth), length_source="layout") + assert result["units"]["layout"]["inferred"] == truth["layout_true_unit"] == "mm" + assert result["units"]["layout"]["m_per_unit"] == 0.001 + assert any("INSUNITS declares in" in w for w in result["warnings"]) + + +def test_routed_lengths_land_in_the_synthetic_rooms(plant): + truth, pid, layout, network = plant + result = pipe_rows(network, pid, layout, services=truth_services(truth), length_source="layout") + known = {str(room["number"] or room["name"]) for room in truth["rooms"]} | {""} + for run in result["runs"]: + assert set(run["rooms"]) <= known, run["id"]