Header, global constraints (confidentiality, read-only by default, nothing invented, inferred units), the five real-world inputs no happy-path test exercises (AutoCAD text codes, Cyrillic look-alike tags, repeated tags, double-drawn pipe, a layout whose INSUNITS lies), file structure, the interface block, execution with the worktree fast-forward rule, and the merge notes. Task sections are drafted next. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
14 KiB
Track H — understanding foreign drawings, takeoffs, diff, topology: implementation plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Read drawings somebody else made — a one-call report, a scale check between a P&ID and its layout, pipe and cable takeoffs whose topology comes from the P&ID and whose lengths come from the layout, a revision diff and topology checks — as six tools on both engines.
Architecture: One bulk read per drawing into frozen EntityRecords (ezdxf
directly; a live document through a Document.Export DXF snapshot that does
not rename it); pure analysis modules under engineering/understand/; a thin
async layer and the server tools. No new contract member.
Tech Stack: Python 3.11, FastMCP 3, ezdxf 1.4.x, pywin32 (COM), openpyxl
(new office extra), pytest + pytest-asyncio, ruff — all through
uv run --frozen.
Spec: docs/superpowers/specs/2026-09-24-v1.6-understand-design.md
Global Constraints
- Run everything through
uv run --frozen …. Neveruv sync, with one exception: Task 1 adds theofficeextra, runsuv lockand thenuv sync --frozen --all-extrasexactly once. Ifuv run --frozenfails withos error 32on.venv/Scripts/autocad-mcp.exe, useuv run --frozen --no-sync …for the rest of the task and say so. A fresh worktree venv lacks extras: run commands there asuv run --frozen --all-extras …. - Commit with
git -c user.name="Umut CAN" -c user.email="umutcan.edizaslan@gmail.com" commit, message ending with these two lines verbatim:Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>andClaude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd. - Known red:
tests/test_release_consistency.py::test_document_numbers_match_the_treedrifts on the README collected-tests figure — deselect exactly that test in a worktree and say so. Task 11 retakes the figure on main. - Confidentiality (spec §14): no client drawing, spreadsheet, file name,
client or site name, or takeoff quantity from the user's archive enters the
repository, a test, a fixture, a benchmark result or a commit message. Every
fixture is synthetic and built by
tests/fixtures/plant_pair.py(Task 2). - Read-only by default: every tool of this track leaves the drawing
untouched, except
drawing_diff(markup=True); the takeoffs write a workbook file, never the drawing. - Nothing is invented: a power value missing from the P&ID stays empty; a
diameter that cannot be read stays
unassigned; a length from an unscaled P&ID is never presented as a physical length; no cable-section table is authored (sections only from a caller'ssection_rules). - Units: analysis runs in drawing units; takeoff lengths are reported in
metres from the inferred unit (spec §5), and a disagreement with
INSUNITSis a warning in the result and on the method sheet. - Before committing:
uv run --frozen ruff check <changed files>anduv run --frozen ruff format <changed files>. - Registration gates (when you register a tool): regenerate
docs/tool-inventory.json, bump the README release-snapshot tool count andtests/test_tool_registry.py's frozen group sizes, add an alias record per tool indiscovery/aliases.py. - Deviations are data: fix a plan test to the verified truth, keep the spec's behaviour, record it; never weaken a test.
Review Focus
The five real-world inputs the spec implies and no happy-path test exercises — each seen in, or one keystroke away from, the field drawings of 2026-09-24. Each line's test lives in the owning task.
- AutoCAD text codes in labels —
%%c51(AutoCAD's Ø),\U+2205,⌀51,ø51and MTEXT format runs ({\fArial|b0;Ø51},\P) must all parse as the diameterØ51. Owner: Task 1 (labels.parse_diameter). - Cyrillic look-alike letters in tags — a Russian-labelled P&ID types
Т4100orМ82with Cyrillic Т/М; it must match the layout's LatinT4100/M82. Owner: Task 1 (labels.parse_tag). - The same tag more than once (two plan copies, a legend, a second sheet)
— the scale check and the takeoffs must not pick one arbitrarily: they use
only tags that are unique within the chosen cluster and report the rest as
ambiguous. Owner: Task 4 (scale.match_tags), reused by Task 6. - Double-drawn pipe — two overlapping collinear segments on a pipe layer
(a line drawn twice, or a line over a polyline) must count once, or every
takeoff doubles silently. Owner: Task 5 (
network.build_network). - A layout whose
INSUNITSlies — inches declared over millimetre geometry: the takeoff must measure in the inferred unit (no ×25.4) and say so. Owner: Task 6 (takeoff.pipe_rows).
File Structure
| File | Responsibility | Task | Group |
|---|---|---|---|
engineering/understand/{__init__,snapshot,vocab,labels}.py |
records, bulk read, live snapshot, vocabulary, label parsers | 1 | main |
config.py, pyproject.toml, uv.lock |
MAX_DXF_BYTES 512 MB; office extra |
1 | main |
tests/fixtures/plant_pair.py |
the synthetic P&ID + layout pair and its truth | 2 | main |
engineering/understand/{units,clusters,describe}.py |
drawing_understand |
3 | U |
engineering/understand/scale.py |
drawing_scale_check |
4 | U |
engineering/understand/network.py |
line networks, runs, services, diameters, attachment | 5 | N |
engineering/understand/{route,takeoff}.py |
Manhattan, rectilinear MST, pipe rows | 6 | N |
engineering/understand/report.py |
the takeoff workbook (XLSX + CSV) | 7 | N |
engineering/understand/takeoff.py (cable) |
cable rows | 8 | N |
engineering/understand/diff.py |
drawing_diff |
9 | D |
engineering/understand/{topology,critique}.py |
drawing_topology_check, topo_* focuses |
10 | D |
server.py SECTION 26 (Understanding & QA, core) |
U opens it; D appends | 3, 4, 9, 10 | U, D |
server.py SECTION 27 (Plant takeoffs, pack plant) |
N owns it | 7, 8 | N |
| packs / lean / aliases / golden / counts | integration | 11 | final |
| benchmarks, smoke, field-test script, docs | evidence | 12 | final |
Interfaces (the contract between tasks)
# engineering/understand/snapshot.py (Task 1)
Pt = tuple[float, float]
@dataclass(frozen=True)
class EntityRecord:
handle: str
type: str # DXF type: LINE, LWPOLYLINE, POLYLINE, ARC, CIRCLE, SPLINE, ELLIPSE,
# TEXT, MTEXT, INSERT, DIMENSION, LEADER, MULTILEADER, HATCH, …
layer: str
space: str # "Model" or the paper layout's name
points: tuple[Pt, ...] = () # LINE ends; polyline vertices; TEXT/MTEXT/INSERT insertion; leader vertices
bulges: tuple[float, ...] = () # per vertex, LWPOLYLINE/POLYLINE
closed: bool = False
radius: float | None = None # CIRCLE / ARC
angles: tuple[float, float] | None = None # ARC start/end, degrees CCW
text: str | None = None # plain text: TEXT, MTEXT (format codes stripped), MULTILEADER, DIMENSION override
height: float | None = None # text height
block: str | None = None # INSERT block name
attribs: tuple[tuple[str, str], ...] = () # INSERT attributes (tag, value)
bbox: tuple[float, float, float, float] | None = None # WCS box (minx, miny, maxx, maxy)
rotation: float = 0.0
scale: tuple[float, float] = (1.0, 1.0)
@dataclass(frozen=True)
class Snapshot:
source: str # file path, or "live:<document name>"
insunits: int
extmin: Pt | None
extmax: Pt | None
layers: dict[str, dict] # name -> {"color": int, "linetype": str}
layouts: tuple[str, ...]
records: tuple[EntityRecord, ...]
def records_from_doc(doc, *, source: str) -> Snapshot # an ezdxf Document
def read_snapshot(path: str) -> Snapshot # refuses > config.settings.max_dxf_bytes, naming MAX_DXF_BYTES
async def take_snapshot(backend, path: str | None = None) -> Snapshot
# path given -> read that file (DXF; a DWG is opened read-only on COM and exported);
# None -> the current document: ezdxf reads its in-memory doc, COM uses
# Document.Export(<tmp>/snap, "DXF", <selection set of all>) and deletes the file afterwards.
def length_of(rec: EntityRecord) -> float # LINE, polylines with bulges, ARC, CIRCLE; 0 for others
# engineering/understand/vocab.py (Task 1)
SERVICES: tuple[str, ...] # product, cip_supply, cip_return, glycol_supply, glycol_return, steam, condensate,
# cold_water, hot_water, ice_water, compressed_air, drain, electrical, unknown
DISCIPLINES: tuple[str, ...] # piping, electrical, architecture, equipment, annotation, structure, unknown
def classify_layer(name: str) -> dict # {"discipline", "service", "confidence", "keyword", "language"}
def supply_return(text: str) -> str | None # "supply" | "return" | None — EN/TR/RU/NL/DE words
def room_label(text: str) -> dict | None # {"number": str | None, "name": str | None, "language": str}
# engineering/understand/labels.py (Task 1)
def plain(text: str) -> str # strips MTEXT codes, decodes %%c/%%d/%%p, \U+XXXX
def parse_diameters(text: str) -> tuple[str, ...] # tokens as drawn, normalised spelling: "Ø51", "Ø25,4", "SMS51", "DN20", "20x27", '1,5"'
def parse_tag(text: str) -> str | None # "T4100", "M82", "M41A", "M47.1", "CP-1", "CP-M82", "JB-2"; Cyrillic look-alikes folded
def normalize_panel(text: str) -> str | None # "CP 1" / "CP1" / "cp-1" -> "CP-1"; "CP M82" -> "CP-M82"
def parse_electrical(text: str) -> dict # {"kw": float | None, "phases": int | None, "voltage": int | None,
# "neutral": bool, "vfd": bool}
def wiring_target(text: str) -> str | None # "Wiring to CP1" / "подключение к шкафу CP-1" -> "CP-1"
# tests/fixtures/plant_pair.py (Task 2)
def build_plant_pair(directory: str | Path) -> dict # writes pid.dxf + layout.dxf, returns the truth dict:
# {"pid": str, "layout": str, "scale_verdict": "schematic", "layout_insunits": 1, "layout_true_unit": "mm",
# "outlier_handle": str, "cluster_count": 2, "layer_services": {layer: service},
# "pipe_runs": [{"service", "diameter", "tags": [..], "layout_length_mm": float, "status": "A"|"B"|"C"}],
# "cable_rows": [{"tag", "panel", "kw": float | None, "length_mm": float, "roundup_m": int}],
# "rooms": [{"number", "name", "polygon": [[x, y], ...]}]}
# engineering/understand/describe.py (Task 3)
def describe(snap: Snapshot) -> dict
# {"units": {...}, "extents": {...}, "clusters": [...], "layers": [...], "equipment_tags": [...],
# "rooms": [...], "languages": {...}, "blocks": {...}, "layouts": [...], "title_block": {...}, "warnings": [...]}
# engineering/understand/scale.py (Task 4)
def match_tags(a: Snapshot, b: Snapshot, *, cluster_a=None, cluster_b=None) -> dict # {"matched": {tag: (pa, pb)}, "ambiguous": [...]}
def scale_check(a: Snapshot, b: Snapshot, *, min_pair_mm=2000.0) -> dict
# {"verdict": "to_scale"|"schematic"|"partly"|"insufficient", "factor", "median", "iqr", "within_10pct",
# "pairs", "matched", "ambiguous", "groups": [...]}
# engineering/understand/network.py (Task 5)
@dataclass(frozen=True)
class Run: id: str; edges: tuple[dict, ...]; service: str; ends: tuple[Pt, ...]; tags: tuple[str, ...]
def build_network(snap: Snapshot, *, layers=None, tol=1.0, label_search=None) -> dict # {"runs": [Run], "unassigned": [...], "stats": {...}}
# engineering/understand/route.py (Task 6)
def manhattan(a: Pt, b: Pt) -> float
def rmst(points: Sequence[Pt]) -> tuple[float, tuple[tuple[int, int], ...]] # rectilinear MST length and edges
# engineering/understand/takeoff.py (Tasks 6, 8)
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
def cable_rows(pid: Snapshot, layout: Snapshot | None, *, allowance=0.20, section_rules=None) -> dict
# engineering/understand/report.py (Task 7)
def write_workbook(result: dict, *, kind: str, lang: str, path: str) -> dict # {"xlsx": path | None, "csv": [paths], "refused": {...} | None}
# engineering/understand/diff.py (Task 9)
def diff_snapshots(old: Snapshot, new: Snapshot, *, tol=0.01, move_tol=None) -> dict
# engineering/understand/topology.py (Task 10)
def topology_findings(snap: Snapshot, *, layers, gap=None, tol=0.01) -> dict # {"dangling": [...], "near_miss": [...], "crossing": [...]}
Execution
| Group | Branch | Tasks | Owns |
|---|---|---|---|
| (main) | main |
1, 2 | snapshot, vocab, labels, config, extra, the fixture |
| U | track-h/understand |
3, 4 | units, clusters, describe, scale; SECTION 26 banner + its first two tools |
| N | track-h/network |
5, 6, 7, 8 | network, route, takeoff, report; SECTION 27 |
| D | track-h/diff |
9, 10 | diff, topology, critique; SECTION 26's other two tools |
| final | main (after merge) |
11, 12 | integration, evidence |
Tasks 1 and 2 land on main first; each group branch is then cut from the
current main (a worktree may open at an older commit: after git checkout -b <branch>, run git merge --ff-only main and confirm git log -1 is main's
HEAD before anything else).
SECTION 26 banner protocol: U opens # SECTION 26: Understanding & QA (1 tool) in Task 3 and grows it to 2 in Task 4; D, on its own branch, opens
the same banner with its own count (1, then 2). The merge keeps one banner at 4
with U's block, then D's.
Merge notes
Order U → N → D. Anticipated conflicts: server.py SECTION 26 (one banner,
blocks U then D) and the RESOURCES seam; engineering/understand/__init__.py
(additive re-exports); discovery/aliases.py (additive); docs/tool-inventory.json
(regenerate, never hand-merge); README release-snapshot tool count (from the
inventory; the collected-tests figure is Task 11's); tests/test_tool_registry.py
group sizes (from the live registry); engineering/critique.py focus
registration (D only). Anything else: git merge --abort, record, continue.