diff --git a/README.md b/README.md index 303ea23..6273618 100644 --- a/README.md +++ b/README.md @@ -27,8 +27,8 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co -> **v1.5 release snapshot:** 243 tools · 8 resources · 5 prompt templates · 4516 collected tests. -> 243 is the **registered** count; a default install advertises 238 over `tools/list`, +> **v1.5 release snapshot:** 245 tools · 8 resources · 5 prompt templates · 4516 collected tests. +> 245 is the **registered** count; a default install advertises 240 over `tools/list`, > because `ENABLE_3D` is unset. `system_about` is the runtime authority. ## Why this exists @@ -122,7 +122,7 @@ Claude Desktop, Cursor, or any stdio MCP host. For HTTP: `autocad-mcp --transpor | Quality loop | `drawing_preflight` → `drawing_plan` → `drawing_critique` → `drawing_refine` → `drawing_finalize` (0–100 score) | | Delivery | `drawing_deliver`: DXF/PDF/PNG + SHA-256 manifest + reopen-parity checks | -243 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (TOOL_PACKS=core,mech keeps the mechanical ones, TOOL_PACKS=core,arch the architectural ones) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (autocad://drawing/info, layers, blocks, entities/stats, entities/{layer_name}, system/status, pid/symbols, standards/isa51) and 5 prompt templates. +245 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (TOOL_PACKS=core,mech keeps the mechanical ones, TOOL_PACKS=core,arch the architectural ones) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (autocad://drawing/info, layers, blocks, entities/stats, entities/{layer_name}, system/status, pid/symbols, standards/isa51) and 5 prompt templates. **Two rules worth knowing.** Every coordinate in and out of a tool is WCS on both engines — the one exception is TEXT `rotation`, which stays in the entity frame because a mirrored TEXT is mirror-imaged and no scalar angle expresses that. And never read vertices back and shoelace them: that loses **28.2%** of the area on a semicircular edge, silently. `analysis_measure_entity(handle)` reads the real geometry and states its own accuracy. diff --git a/discovery/aliases.py b/discovery/aliases.py index 9e00ef7..c536ba4 100644 --- a/discovery/aliases.py +++ b/discovery/aliases.py @@ -2513,6 +2513,31 @@ TOOL_ALIASES: dict[str, ToolAliases] = { "масштаб", ), ), + # ── Understanding & QA (SECTION 26, track H group D) ──────────────────── + "drawing_diff": ToolAliases( + acad=("COMPARE",), + synonyms=( + "compare two drawings", + "compare two revisions", + "what changed between revisions", + "drawing differences", + "cloud the changes", + "revizyonlar arası değişiklik", + "revizyonlar arasi degisiklik", + ), + ), + "drawing_topology_check": ToolAliases( + synonyms=( + "dangling pipe ends", + "pipes that do not connect", + "gaps in the pipe network", + "unconnected cable ends", + "network connectivity check", + "açık uçlar", + "acik uclar", + "kopuk hatlar", + ), + ), } diff --git a/docs/tool-inventory.json b/docs/tool-inventory.json index f7f220a..336ebd1 100644 --- a/docs/tool-inventory.json +++ b/docs/tool-inventory.json @@ -3,13 +3,13 @@ "schema_version": 1, "package_version": "1.5.1", "totals": { - "tools": 243, + "tools": 245, "resources": 8, "prompts": 5, "groups": 26 }, "groups": { - "analysis": 14, + "analysis": 16, "architecture": 12, "batch": 3, "blocks": 9, @@ -897,6 +897,20 @@ "destructiveHint": true } }, + { + "name": "drawing_diff", + "group": "analysis", + "cost": "mutate", + "summary": "Compare two revisions of a drawing: what moved, changed, was added or removed.", + "tags": [ + "analysis", + "query" + ], + "annotations": { + "title": "Drawing: Diff Two Revisions", + "destructiveHint": false + } + }, { "name": "drawing_export_dwg", "group": "sheet", @@ -1175,6 +1189,20 @@ "destructiveHint": false } }, + { + "name": "drawing_topology_check", + "group": "analysis", + "cost": "read", + "summary": "Find dangling ends, near misses and unjoined crossings in a pipe or cable network.", + "tags": [ + "analysis", + "query" + ], + "annotations": { + "title": "Drawing: Topology Check", + "readOnlyHint": true + } + }, { "name": "drawing_understand", "group": "analysis", diff --git a/engineering/critique.py b/engineering/critique.py index 6c84155..07ef286 100644 --- a/engineering/critique.py +++ b/engineering/critique.py @@ -420,6 +420,10 @@ from .arch.critique import ARCH_DISPATCH # noqa: E402 (same reason: after the _FOCUS_DISPATCH.update(ARCH_DISPATCH) +from .understand.critique import TOPO_DISPATCH # noqa: E402 (same reason: after the base focuses) + +_FOCUS_DISPATCH.update(TOPO_DISPATCH) + async def run_critique( backend: AutoCADBackend, diff --git a/engineering/plan_spec.py b/engineering/plan_spec.py index 812d8e5..c93cdbb 100644 --- a/engineering/plan_spec.py +++ b/engineering/plan_spec.py @@ -36,6 +36,9 @@ CritiqueFocus = Literal[ "arch_room_unlabelled", "arch_wall_gap", "arch_opening_clash", + "topo_dangling_endpoint", + "topo_near_miss", + "topo_interior_crossing", ] SnapType = Literal["end", "mid", "center", "quad", "int", "perp", "near"] @@ -62,6 +65,9 @@ ALL_CRITIQUE_FOCUSES: tuple[CritiqueFocus, ...] = ( "arch_room_unlabelled", "arch_wall_gap", "arch_opening_clash", + "topo_dangling_endpoint", + "topo_near_miss", + "topo_interior_crossing", ) diff --git a/engineering/understand/critique.py b/engineering/understand/critique.py new file mode 100644 index 0000000..10040b5 --- /dev/null +++ b/engineering/understand/critique.py @@ -0,0 +1,205 @@ +"""The three topology critique focuses (spec §11). + +``topo_dangling_endpoint`` (warning), ``topo_near_miss`` (error) and +``topo_interior_crossing`` (info) run ``topology.topology_findings`` - the +engine ``drawing_topology_check`` runs - once per critique run, shared through +``run_critique``'s context the way the P&ID, mechanical and architectural +focuses share theirs. + +Which layers: the network layers ``vocab.classify_layer`` calls piping or +electrical with confidence >= 0.9, **minus every layer a semantic focus +already reads**: the P&ID layer set and line layers (``pid_*``) and the +architectural layer set (``arch_*``). One defect is never scored twice, and a +drawing made entirely by this server - a ``pid_from_spec`` sheet, an +``arch_plan_from_spec`` plan, a mechanical part - gets none of these issues. + +Cost: the layer table is read first; a drawing with no candidate layer (every +drawing this server makes) is never snapshotted, so ``focus=None`` stays cheap. +A snapshot that cannot be taken is reported as one ``info`` issue per focus, +never as a clean result. + +Severities are this module's decision: a near miss is an ``error`` (a network +that looks connected and is not - a takeoff counts two runs), a dangling end a +``warning`` (it may be a real open end), a crossing ``info`` (often drawn on +purpose). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from backends.quarantine import DocumentQuarantineError +from engineering.arch.layers import ARCH_LAYER_ROLES, ARCH_LAYERS +from engineering.layers import PID_LAYERS +from engineering.pid.lines import PID_LINE_LAYERS +from engineering.plan_spec import Issue +from engineering.understand.topology import ( + NETWORK_CONFIDENCE, + NETWORK_DISCIPLINES, + network_layers, + topology_findings, +) +from engineering.understand.vocab import classify_layer + +if TYPE_CHECKING: + from backends.base import AutoCADBackend + +__all__ = [ + "SEMANTIC_LAYERS", + "TOPO_DISPATCH", + "TOPO_FOCUSES", + "build_index", + "focus_layers", + "issues_for", +] + +TOPO_FOCUSES = ("topo_dangling_endpoint", "topo_near_miss", "topo_interior_crossing") +_KEY = "topo_index" + +#: Layers the pid_* and arch_* focuses already read. Layer "0" is nobody's. +SEMANTIC_LAYERS: frozenset[str] = frozenset( + {name for name, *_rest in PID_LAYERS} + | set(PID_LINE_LAYERS) + | {name for name, *_rest in ARCH_LAYERS} + | set(ARCH_LAYER_ROLES.values()) +) - {"0"} + + +def _is_candidate(name: str) -> bool: + if name in SEMANTIC_LAYERS: + return False + meaning = classify_layer(name) + return ( + meaning.get("discipline") in NETWORK_DISCIPLINES + and float(meaning.get("confidence") or 0.0) >= NETWORK_CONFIDENCE + ) + + +def focus_layers(snap) -> list[str]: + """The network layers of ``snap`` no semantic focus covers.""" + return [row["layer"] for row in network_layers(snap) if row["layer"] not in SEMANTIC_LAYERS] + + +async def build_index(backend: AutoCADBackend) -> dict: + """One read of the drawing for the three focuses: + ``{"layers": [...], "findings": dict | None, "error": str | None}``.""" + try: + names = [layer.name for layer in await backend.layer_list()] + except DocumentQuarantineError: + raise + except Exception: + names = [] + if not any(_is_candidate(name) for name in names): + return {"layers": [], "findings": None, "error": None} + from engineering.understand.snapshot import take_snapshot + + try: + snap = await take_snapshot(backend, None) + except DocumentQuarantineError: + raise + except Exception as exc: + return { + "layers": sorted(n for n in names if _is_candidate(n)), + "findings": None, + "error": str(exc) or type(exc).__name__, + } + layers = focus_layers(snap) + if not layers: + return {"layers": [], "findings": None, "error": None} + return {"layers": layers, "findings": topology_findings(snap, layers=layers), "error": None} + + +def _dangling(findings: dict) -> list[Issue]: + return [ + Issue( + "warning", + "topo_dangling_endpoint", + f"{row['handle']} on layer {row['layer']} ends at ({row['at'][0]:.3f}, " + f"{row['at'][1]:.3f}) touching nothing - no line, fitting or equipment is at that " + "end. Connect it, or confirm it is a real open end.", + [row["handle"]], + {"layer": row["layer"], "end": row["end"], "at": row["at"], "space": row["space"]}, + ) + for row in findings["dangling"] + ] + + +def _near_miss(findings: dict) -> list[Issue]: + out = [] + for row in findings["near_miss"]: + first, second = row["handles"] + what = "the end of" if row["kind"] == "end_end" else "the body of" + out.append( + Issue( + "error", + "topo_near_miss", + f"{first} on layer {row['layers'][0]} stops {row['gap']:.4g} short of {what} " + f"{second} without joining it: the network reads as two pieces there. Extend or " + "trim it onto the other line.", + sorted({first, second}), + { + "kind": row["kind"], + "gap": row["gap"], + "at": row["at"], + "to": row["to"], + "layers": row["layers"], + "space": row["space"], + }, + ) + ) + return out + + +def _crossing(findings: dict) -> list[Issue]: + return [ + Issue( + "info", + "topo_interior_crossing", + f"{row['handles'][0]} and {row['handles'][1]} on layer {row['layer']} cross at " + f"({row['at'][0]:.3f}, {row['at'][1]:.3f}) away from their ends with no junction. " + "If they connect, split both at the crossing; if not, draw the crossing with a break.", + sorted(set(row["handles"])), + {"layer": row["layer"], "at": row["at"], "space": row["space"]}, + ) + for row in findings["crossing"] + ] + + +_CHECKS = { + "topo_dangling_endpoint": _dangling, + "topo_near_miss": _near_miss, + "topo_interior_crossing": _crossing, +} + + +def issues_for(focus: str, index: dict) -> list[Issue]: + if index.get("error"): + return [ + Issue( + "info", + focus, + f"Topology not checked on {', '.join(index['layers'])}: the drawing could not be " + f"read ({index['error']}).", + [], + {"layers": list(index["layers"])}, + ) + ] + findings = index.get("findings") + if not findings: + return [] + return _CHECKS[focus](findings) + + +def _make(focus: str): + async def check(backend: AutoCADBackend, shared: dict) -> list[Issue]: + index = shared.get(_KEY) + if index is None: + index = shared[_KEY] = await build_index(backend) + return issues_for(focus, index) + + check.needs_shared = True # type: ignore[attr-defined] + check.__name__ = f"check_{focus}" + return check + + +TOPO_DISPATCH = {focus: _make(focus) for focus in TOPO_FOCUSES} diff --git a/engineering/understand/diff.py b/engineering/understand/diff.py new file mode 100644 index 0000000..71da263 --- /dev/null +++ b/engineering/understand/diff.py @@ -0,0 +1,727 @@ +"""``drawing_diff``: two revisions of one drawing, matched by geometric signature. + +Spec §10. Each revision is read once into ``EntityRecord``s +(``engineering.understand.snapshot``) and the two are compared in memory. +Nothing in this module writes to a drawing; the revision clouds of +``markup=True`` are drawn by the tool through +``engineering.sheet.revision.add_revision``. + +Matching runs in three stages, in this order, and a record leaves the pool the +moment it is matched: + +1. **Exact.** Every record has a signature: its type, layer and space, its + geometry rounded to ``tol`` (points, bulges, closed flag, radius, arc + angles; the bounding box when it has no points), its text and text height, + its block name, rotation, scale and attributes. Equal signatures are the + same entity whatever their handles; where several old records share a + signature the one with the same handle is taken first. A LINE's two ends + are sorted, so a line drawn backwards is the same line. +2. **Handle.** A record left over on both sides with the same handle and the + same type is the same entity, edited. A save that renumbers handles would + make this stage pair strangers, so it runs only while handles look stable: + when there is at least one exact match and at least half of the exact + matches kept their handle. With no exact match there is no evidence either + way - two fresh files hand out handles from the same seed, and a drawing + moved as a whole matches nothing exactly - so the stage does not run. The + report says which way it went (``handles_stable``) and on what evidence + (``handle_evidence``: the exact matches and how many kept their handle). +3. **Position.** Within one (type, layer, space), a leftover new record is + paired with the nearest leftover old record whose anchor (the mean of its + points, else the centre of its box) lies within ``move_tol``; the closest + pairs are taken first. ``move_tol`` defaults to 2 % of the drawing's robust + diagonal - the diagonal of the 1st-99th percentile box of every model-space + anchor of both revisions - so one stray entity kilometres away cannot + inflate it. + +What is left is added (new side) or removed (old side). A matched pair is +compared field by field, each number within half a ``tol`` of its old value +counting as equal - the same test a move uses. A pair with nothing to report +is *unchanged*: the signature snaps every value to the ``tol`` grid, so a +value one ulp either side of a half-grid point (AutoCAD writes 0.145 for the +0.14500000000000002 it was given) lands in another bucket, misses the exact +stage and is paired later, yet nothing was edited. Such a pair counts under +``unchanged`` and gets no row and no cloud. Any other matched pair is +*changed*, and the row says what changed: ``moved_by`` +(dx, dy) when the geometry is the old geometry translated, else ``geometry`` +from -> to; ``text``, ``attribs`` (by tag), ``radius``, ``angles``, +``height``, ``rotation``, ``scale``, ``block``, ``layer``, ``space``, +``bulges``, ``closed`` from -> to. + +The changes are grouped into clusters for the revision clouds: every changed, +added and removed record contributes its box (a changed one its old box and +its new box), boxes within ``move_tol`` of each other join one cluster, and a +cluster's box is padded by a quarter of ``move_tol`` (never less than ten +``tol``). ``count`` is the number of distinct changes in the cluster. + +Block definitions are compared apart (``block_signatures`` / +``block_changes``): the entity records carry INSERTs, not the blocks they +reference, so a redrawn definition changes every drawn copy while every INSERT +record stays identical. + +The 2 % move tolerance, the 1st/99th percentiles, the half-of-the-matches +(and at least one) handle rule, the half-a-``tol`` equality and the cluster +padding are this module's declared defaults, not values from a standard. +""" + +from __future__ import annotations + +import hashlib +import heapq +import math +import os +from collections import defaultdict +from collections.abc import Iterable, Sequence + +import config +from engineering.understand.snapshot import EntityRecord, Pt, Snapshot, records_from_doc + +__all__ = [ + "CATEGORIES", + "DEFAULT_LIMIT", + "DEFAULT_TOL", + "MOVE_FRACTION", + "PAD_FRACTION", + "ROBUST_QUANTILE", + "anchor", + "block_changes", + "block_signatures", + "cap_report", + "default_move_tol", + "diff_snapshots", + "read_revision", + "robust_box", + "signature", +] + +DEFAULT_TOL = 0.01 +#: The default move tolerance, as a share of the robust diagonal. +MOVE_FRACTION = 0.02 +#: The robust box runs from this quantile of the anchors to its complement. +ROBUST_QUANTILE = 0.01 +#: A cluster's box is padded by this share of ``move_tol``. +PAD_FRACTION = 0.25 +#: The position stage keeps at most this many nearest old candidates per new +#: entity, so a dense drawing that moved as a whole (every entity left for this +#: stage, hundreds of neighbours within ``move_tol``) cannot build n x m pairs. +NEAREST_PER_ENTITY = 32 +DEFAULT_LIMIT = 200 +CATEGORIES = ("changed", "added", "removed") + +Box = tuple[float, float, float, float] + + +def _positive(name: str, value) -> float: + try: + number = float(value) + except (TypeError, ValueError): + raise ValueError(f"{name} must be a positive number; got {value!r}") from None + if not math.isfinite(number) or number <= 0: + raise ValueError(f"{name} must be a positive number; got {value!r}") + return number + + +def _q(value, tol: float): + """``value`` on the ``tol`` grid; a non-finite value is kept as its repr.""" + number = float(value) + if not math.isfinite(number): + return repr(number) + return int(round(number / tol)) + + +def _qp(point, tol: float) -> tuple: + return (_q(point[0], tol), _q(point[1], tol)) + + +def signature(rec: EntityRecord, tol: float = DEFAULT_TOL) -> tuple: + """Everything that makes two records the same entity, rounded to ``tol``.""" + points = tuple(_qp(p, tol) for p in rec.points) + if rec.type == "LINE" and len(points) == 2: + points = tuple(sorted(points)) + if points: + geometry = points + elif rec.bbox is not None: + geometry = tuple(_q(v, tol) for v in rec.bbox) + else: + geometry = () + return ( + rec.type, + rec.layer, + rec.space, + geometry, + tuple(_q(b, tol) for b in rec.bulges), + bool(rec.closed), + None if rec.radius is None else _q(rec.radius, tol), + None + if rec.angles is None + else (_q(rec.angles[0] % 360.0, tol), _q(rec.angles[1] % 360.0, tol)), + rec.text, + None if rec.height is None else _q(rec.height, tol), + rec.block, + tuple(sorted(rec.attribs)), + _q(rec.rotation % 360.0, tol), + (_q(rec.scale[0], tol), _q(rec.scale[1], tol)), + ) + + +def anchor(rec: EntityRecord) -> Pt | None: + """The mean of the record's points, else the centre of its box, else None.""" + if rec.points: + n = len(rec.points) + point = (sum(p[0] for p in rec.points) / n, sum(p[1] for p in rec.points) / n) + elif rec.bbox is not None: + x0, y0, x1, y1 = rec.bbox + point = ((x0 + x1) / 2.0, (y0 + y1) / 2.0) + else: + return None + if not (math.isfinite(point[0]) and math.isfinite(point[1])): + return None + return point + + +def _box(rec: EntityRecord) -> Box | None: + if rec.bbox is not None: + return tuple(float(v) for v in rec.bbox) + if rec.points: + xs = [p[0] for p in rec.points] + ys = [p[1] for p in rec.points] + return (min(xs), min(ys), max(xs), max(ys)) + return None + + +def _quantile_pair(values: list[float], q: float) -> tuple[float, float]: + values = sorted(values) + last = len(values) - 1 + low = int(math.floor(q * last)) + return values[low], values[last - low] + + +def robust_box(*snaps: Snapshot) -> Box | None: + """The 1st-99th percentile box of the model-space anchors of ``snaps`` + (every space when model space is empty); None when there is no anchor.""" + records = [r for s in snaps for r in s.records] + anchors = [a for r in records if r.space == "Model" and (a := anchor(r)) is not None] + if not anchors: + anchors = [a for r in records if (a := anchor(r)) is not None] + if not anchors: + return None + x0, x1 = _quantile_pair([a[0] for a in anchors], ROBUST_QUANTILE) + y0, y1 = _quantile_pair([a[1] for a in anchors], ROBUST_QUANTILE) + return (x0, y0, x1, y1) + + +def default_move_tol(old: Snapshot, new: Snapshot, tol: float = DEFAULT_TOL) -> float: + """2 % of the robust diagonal of both revisions, never less than ``tol``.""" + box = robust_box(old, new) + diagonal = 0.0 if box is None else math.hypot(box[2] - box[0], box[3] - box[1]) + return max(MOVE_FRACTION * diagonal, tol) + + +def _nearest_pairs( + olds: Sequence[EntityRecord], + news: Sequence[EntityRecord], + left_old: Iterable[int], + left_new: Iterable[int], + move_tol: float, +) -> list[tuple[int, int]]: + """Stage 3: closest pairs first, within one (type, layer, space).""" + groups: dict[tuple, tuple[list[int], list[int]]] = defaultdict(lambda: ([], [])) + for i in left_old: + groups[(olds[i].type, olds[i].layer, olds[i].space)][0].append(i) + for j in left_new: + groups[(news[j].type, news[j].layer, news[j].space)][1].append(j) + candidates: list[tuple[float, str, str, int, int]] = [] + for old_ids, new_ids in groups.values(): + if not old_ids or not new_ids: + continue + grid: dict[tuple[int, int], list[tuple[int, Pt]]] = defaultdict(list) + for i in old_ids: + a = anchor(olds[i]) + if a is not None: + cell = (math.floor(a[0] / move_tol), math.floor(a[1] / move_tol)) + grid[cell].append((i, a)) + for j in new_ids: + b = anchor(news[j]) + if b is None: + continue + cx, cy = math.floor(b[0] / move_tol), math.floor(b[1] / move_tol) + near: list[tuple[float, str, str, int, int]] = [] + for dx in (-1, 0, 1): + for dy in (-1, 0, 1): + for i, a in grid.get((cx + dx, cy + dy), ()): + distance = math.dist(a, b) + if distance <= move_tol: + near.append((distance, olds[i].handle, news[j].handle, i, j)) + candidates.extend(heapq.nsmallest(NEAREST_PER_ENTITY, near)) + candidates.sort() + used_old: set[int] = set() + used_new: set[int] = set() + pairs: list[tuple[int, int]] = [] + for _distance, _old_handle, _new_handle, i, j in candidates: + if i in used_old or j in used_new: + continue + used_old.add(i) + used_new.add(j) + pairs.append((i, j)) + return pairs + + +def _number_changed(a, b, tol: float, *, modulo: float | None = None) -> bool: + """More than half a ``tol`` apart (around the circle when ``modulo``). + + A difference, never two grid buckets: ``_q`` puts 0.14500000000000002 + and 0.145 in different buckets at ``tol=0.01``. + """ + if a is None or b is None: + return (a is None) != (b is None) + a, b = float(a), float(b) + if not (math.isfinite(a) and math.isfinite(b)): + return repr(a) != repr(b) + difference = abs(a - b) + if modulo is not None: + difference %= modulo + difference = min(difference, modulo - difference) + return difference > tol / 2.0 + + +def _movement(o: EntityRecord, n: EntityRecord, tol: float) -> dict: + """``moved_by`` for a pure translation, ``geometry`` for anything else.""" + if o.points and n.points: + before, after = list(o.points), list(n.points) + elif not o.points and not n.points: + if o.bbox is None or n.bbox is None: + if (o.bbox is None) != (n.bbox is None): + return {"geometry": {"from": o.bbox, "to": n.bbox}} + return {} + before = [(o.bbox[0], o.bbox[1]), (o.bbox[2], o.bbox[3])] + after = [(n.bbox[0], n.bbox[1]), (n.bbox[2], n.bbox[3])] + else: + return { + "geometry": {"from": [list(p) for p in o.points], "to": [list(p) for p in n.points]} + } + a, b = anchor(o), anchor(n) + if a is None or b is None: + return {"geometry": {"from": [list(p) for p in before], "to": [list(p) for p in after]}} + dx, dy = b[0] - a[0], b[1] - a[1] + half = tol / 2.0 + + def translates(start: list, end: list) -> bool: + return len(start) == len(end) and all( + abs(p[0] + dx - q[0]) <= half and abs(p[1] + dy - q[1]) <= half + for p, q in zip(start, end, strict=True) + ) + + translated = translates(before, after) + if not translated and o.type == "LINE" and len(after) == 2: + # the signature ignores a LINE's direction, so its movement must too: + # the same line drawn the other way round and moved is a move + translated = translates(before, after[::-1]) + if not translated: + return {"geometry": {"from": [list(p) for p in before], "to": [list(p) for p in after]}} + if abs(dx) > half or abs(dy) > half: + return {"moved_by": [dx, dy]} + return {} + + +def _attrib_changes(o: EntityRecord, n: EntityRecord) -> list[dict]: + """One row per attribute tag whose values differ. A tag an INSERT carries + twice is compared as the list of its values, so one change cannot hide + behind the other; a tag carried once keeps plain ``from`` / ``to``.""" + + def by_tag(attribs) -> dict[str, list]: + values: dict[str, list] = defaultdict(list) + for tag, value in attribs: + values[tag].append(value) + return values + + before, after = by_tag(o.attribs), by_tag(n.attribs) + rows = [] + for tag in sorted(set(before) | set(after)): + old_values, new_values = before.get(tag, []), after.get(tag, []) + if old_values == new_values: + continue + if len(old_values) <= 1 and len(new_values) <= 1: + rows.append( + { + "tag": tag, + "from": old_values[0] if old_values else None, + "to": new_values[0] if new_values else None, + } + ) + else: + rows.append({"tag": tag, "from": old_values, "to": new_values}) + return rows + + +def _changes(o: EntityRecord, n: EntityRecord, tol: float) -> dict: + out: dict = {} + if o.layer != n.layer: + out["layer"] = {"from": o.layer, "to": n.layer} + if o.space != n.space: + out["space"] = {"from": o.space, "to": n.space} + out.update(_movement(o, n, tol)) + if len(o.bulges) != len(n.bulges) or any( + _number_changed(a, b, tol) for a, b in zip(o.bulges, n.bulges, strict=False) + ): + out["bulges"] = {"from": list(o.bulges), "to": list(n.bulges)} + if bool(o.closed) != bool(n.closed): + out["closed"] = {"from": bool(o.closed), "to": bool(n.closed)} + if _number_changed(o.radius, n.radius, tol): + out["radius"] = {"from": o.radius, "to": n.radius} + if (o.angles is None) != (n.angles is None) or ( + o.angles is not None + and n.angles is not None + and any( + _number_changed(a, b, tol, modulo=360.0) + for a, b in zip(o.angles, n.angles, strict=True) + ) + ): + out["angles"] = { + "from": None if o.angles is None else list(o.angles), + "to": None if n.angles is None else list(n.angles), + } + if o.text != n.text: + out["text"] = {"from": o.text, "to": n.text} + if _number_changed(o.height, n.height, tol): + out["height"] = {"from": o.height, "to": n.height} + if o.block != n.block: + out["block"] = {"from": o.block, "to": n.block} + attribs = _attrib_changes(o, n) + if attribs: + out["attribs"] = attribs + if _number_changed(o.rotation, n.rotation, tol, modulo=360.0): + out["rotation"] = {"from": o.rotation, "to": n.rotation} + if any(_number_changed(a, b, tol) for a, b in zip(o.scale, n.scale, strict=True)): + out["scale"] = {"from": list(o.scale), "to": list(n.scale)} + return out + + +def _row(rec: EntityRecord) -> dict: + at = anchor(rec) + return { + "handle": rec.handle, + "type": rec.type, + "layer": rec.layer, + "space": rec.space, + "at": None if at is None else [at[0], at[1]], + } + + +def _tally(changed: list[dict], added: list[dict], removed: list[dict]) -> tuple[dict, dict]: + by_layer: dict[str, dict[str, int]] = {} + by_type: dict[str, dict[str, int]] = {} + for category, rows in (("changed", changed), ("added", added), ("removed", removed)): + for row in rows: + for table, key in ((by_layer, row["layer"]), (by_type, row["type"])): + counts = table.setdefault(key, {"changed": 0, "added": 0, "removed": 0}) + counts[category] += 1 + return dict(sorted(by_layer.items())), dict(sorted(by_type.items())) + + +def _box_gap(a: Box, b: Box) -> float: + dx = max(0.0, b[0] - a[2], a[0] - b[2]) + dy = max(0.0, b[1] - a[3], a[1] - b[3]) + return math.hypot(dx, dy) + + +def _clusters(items: list[tuple[str, Box, int]], *, link: float, pad: float) -> list[dict]: + """Single-linkage clusters of change boxes, per space, swept along x.""" + parent = list(range(len(items))) + + def find(k: int) -> int: + while parent[k] != k: + parent[k] = parent[parent[k]] + k = parent[k] + return k + + by_space: dict[str, list[int]] = defaultdict(list) + for k, (space, _box_value, _key) in enumerate(items): + by_space[space].append(k) + for members in by_space.values(): + members.sort(key=lambda k: items[k][1][0]) + active: list[int] = [] + for k in members: + box = items[k][1] + active = [m for m in active if items[m][1][2] + link >= box[0]] + for m in active: + if _box_gap(items[m][1], box) <= link: + parent[find(k)] = find(m) + active.append(k) + grouped: dict[int, list[int]] = defaultdict(list) + for k in range(len(items)): + grouped[find(k)].append(k) + clusters = [] + for members in grouped.values(): + boxes = [items[k][1] for k in members] + clusters.append( + { + "space": items[members[0]][0], + "box": [ + min(b[0] for b in boxes) - pad, + min(b[1] for b in boxes) - pad, + max(b[2] for b in boxes) + pad, + max(b[3] for b in boxes) + pad, + ], + "count": len({items[k][2] for k in members}), + } + ) + clusters.sort(key=lambda c: (c["space"], c["box"])) + return clusters + + +def diff_snapshots( + old: Snapshot, new: Snapshot, *, tol: float = DEFAULT_TOL, move_tol: float | None = None +) -> dict: + """Match ``new`` against ``old`` and report every change (spec §10). + + Refuses a non-positive ``tol`` or ``move_tol`` before any work. + """ + tol = _positive("tol", tol) + move_tol = ( + default_move_tol(old, new, tol) if move_tol is None else _positive("move_tol", move_tol) + ) + olds = list(old.records) + news = list(new.records) + + # 1. exact + pools: dict[tuple, list[int]] = defaultdict(list) + for i, rec in enumerate(olds): + pools[signature(rec, tol)].append(i) + used_old: set[int] = set() + left_new: list[int] = [] + exact = kept_handle = 0 + for j, rec in enumerate(news): + pool = pools.get(signature(rec, tol)) + if not pool: + left_new.append(j) + continue + at = next((k for k, i in enumerate(pool) if olds[i].handle == rec.handle), 0) + i = pool.pop(at) + used_old.add(i) + exact += 1 + kept_handle += int(olds[i].handle == rec.handle) + left_old = [i for i in range(len(olds)) if i not in used_old] + # No exact match is no evidence: strangers from two fresh files share + # handles, and so do the entities of a drawing that moved as a whole. + handles_stable = exact > 0 and 2 * kept_handle >= exact + + pairs: list[tuple[int, int, str]] = [] + # 2. handle + if handles_stable: + by_handle = {olds[i].handle: i for i in left_old} + still_new: list[int] = [] + for j in left_new: + i = by_handle.get(news[j].handle) + if i is not None and olds[i].type == news[j].type: + pairs.append((i, j, "handle")) + del by_handle[news[j].handle] + else: + still_new.append(j) + left_new = still_new + taken = {i for i, _j, _how in pairs} + left_old = [i for i in left_old if i not in taken] + + # 3. position + pairs.extend( + (i, j, "position") for i, j in _nearest_pairs(olds, news, left_old, left_new, move_tol) + ) + taken_old = {i for i, _j, _how in pairs} + taken_new = {j for _i, j, _how in pairs} + + changed: list[dict] = [] + items: list[tuple[str, Box, int]] = [] + same = 0 + for i, j, how in sorted(pairs, key=lambda p: (olds[p[0]].handle, news[p[1]].handle)): + o, n = olds[i], news[j] + changes = _changes(o, n, tol) + if not changes: + # the signature grid split an unedited pair (a value 1 ulp across a + # half-grid point): unchanged, no row and no cloud + same += 1 + continue + key = len(changed) + at = anchor(n) + changed.append( + { + "old_handle": o.handle, + "new_handle": n.handle, + "type": n.type, + "layer": n.layer, + "space": n.space, + "matched_by": how, + "at": None if at is None else [at[0], at[1]], + "changes": changes, + } + ) + for rec in (o, n): + box = _box(rec) + if box is not None: + items.append((rec.space, box, key)) + base = len(changed) + removed_recs = sorted((olds[i] for i in left_old if i not in taken_old), key=lambda r: r.handle) + added_recs = sorted((news[j] for j in left_new if j not in taken_new), key=lambda r: r.handle) + for offset, rec in enumerate(removed_recs + added_recs): + box = _box(rec) + if box is not None: + items.append((rec.space, box, base + offset)) + removed = [_row(rec) for rec in removed_recs] + added = [_row(rec) for rec in added_recs] + by_layer, by_type = _tally(changed, added, removed) + pad = max(PAD_FRACTION * move_tol, 10.0 * tol) + return { + "tol": tol, + "move_tol": move_tol, + "handles_stable": handles_stable, + "handle_evidence": {"exact_matches": exact, "kept_handle": kept_handle}, + "summary": { + "old_entities": len(olds), + "new_entities": len(news), + "unchanged": exact + same, + "changed": len(changed), + "added": len(added), + "removed": len(removed), + }, + "by_layer": by_layer, + "by_type": by_type, + "changed": changed, + "added": added, + "removed": removed, + "clusters": _clusters(items, link=move_tol, pad=pad), + } + + +def cap_report(result: dict, limit: int = DEFAULT_LIMIT) -> dict: + """``result`` with each list of ``CATEGORIES`` and the ``clusters`` cut to + ``limit`` rows and ``truncated`` saying how many rows each cut dropped. + + The clusters are capped in the report only: a markup still clouds every one. + """ + if int(limit) < 1: + raise ValueError(f"limit must be at least 1; got {limit!r}") + capped = dict(result) + capped["truncated"] = {} + for category in (*CATEGORIES, "clusters"): + rows = list(result.get(category) or []) + capped[category] = rows[: int(limit)] + capped["truncated"][category] = max(0, len(rows) - int(limit)) + return capped + + +# -- block definitions ----------------------------------------------------------------- + +_BLOCK_ATTRS = ( + "start", + "end", + "center", + "insert", + "radius", + "start_angle", + "end_angle", + "rotation", + "height", + "text", + "tag", + "name", + "xscale", + "yscale", + "major_axis", + "ratio", +) + + +def _q_value(value, tol: float): + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return _q(value, tol) + if isinstance(value, str): + return value + try: + return tuple(_q(v, tol) for v in value) + except TypeError: + return repr(value) + + +def _block_entity_key(entity, tol: float) -> str: + key: list = [entity.dxftype(), entity.dxf.get("layer", "0")] + for attr in _BLOCK_ATTRS: + if entity.dxf.hasattr(attr): + key.append((attr, _q_value(entity.dxf.get(attr), tol))) + kind = entity.dxftype() + if kind == "LWPOLYLINE": + key.append(("points", tuple(_q_value(p, tol) for p in entity.get_points("xyb")))) + key.append(("closed", bool(entity.closed))) + elif kind == "POLYLINE": + key.append(("points", tuple(_q_value(v.dxf.location, tol) for v in entity.vertices))) + elif kind == "MTEXT": + key.append(("mtext", entity.text)) + return repr(tuple(key)) + + +def block_signatures(doc, *, tol: float = DEFAULT_TOL) -> dict[str, str]: + """``{block name: digest}`` for every named block definition of an ezdxf + document: its base point and, per entity, its type, layer and main DXF + attributes rounded to ``tol``. Layout blocks and anonymous blocks + (``*U``, ``*D`` ...) are left out - their names are not stable across saves.""" + tol = _positive("tol", tol) + out: dict[str, str] = {} + for block in doc.blocks: + name = block.name + if block.block_record.is_any_layout or name.startswith("*"): + continue + base = _q_value(block.block.dxf.base_point, tol) + rows = sorted(_block_entity_key(entity, tol) for entity in block) + out[name] = hashlib.sha256(repr((base, rows)).encode("utf-8")).hexdigest() + return dict(sorted(out.items())) + + +def block_changes(old: dict[str, str] | None, new: dict[str, str] | None) -> dict: + """Block definitions added, removed and redrawn between two revisions.""" + if old is None or new is None: + return { + "compared": False, + "reason": "block definitions are compared only when both revisions are DXF files " + "or the current document of the headless engine", + } + return { + "compared": True, + "added": sorted(set(new) - set(old)), + "removed": sorted(set(old) - set(new)), + "changed": sorted(name for name in set(old) & set(new) if old[name] != new[name]), + } + + +def read_revision(path: str, *, tol: float = DEFAULT_TOL) -> tuple[Snapshot, dict[str, str]]: + """One read of a DXF revision: its records and its block signatures. + + Refused before reading: a file that cannot be stat'ed and one over + ``MAX_DXF_BYTES`` (the size and the variable named). A file ezdxf cannot + parse or walk is refused as ``not a readable DXF`` with the reason - + whatever ezdxf raised: a truncated file (an interrupted save, a partial + copy) raises ``StopIteration`` from inside its tag reader, which + ``asyncio.to_thread`` cannot hand back to an awaiting caller, so letting it + through would hang ``drawing_diff`` instead of refusing. + """ + import ezdxf + + tol = _positive("tol", tol) + limit = int(config.settings.max_dxf_bytes) + try: + size = os.path.getsize(path) + except OSError as exc: + raise ValueError(f"{path}: cannot be read ({exc.strerror or exc})") from exc + if limit > 0 and size > limit: + raise ValueError( + f"{path} is {size} bytes, over the MAX_DXF_BYTES limit of {limit}; " + "raise MAX_DXF_BYTES to read it" + ) + try: + doc = ezdxf.readfile(path) + return records_from_doc(doc, source=str(path)), block_signatures(doc, tol=tol) + except MemoryError: + raise + except Exception as exc: # noqa: BLE001 - every parse failure is one refusal + if isinstance(exc, StopIteration): + reason = "the file ends early - an interrupted save or a partial copy?" + else: + reason = str(exc) or type(exc).__name__ + raise ValueError(f"{path}: not a readable DXF ({reason})") from exc diff --git a/engineering/understand/topology.py b/engineering/understand/topology.py new file mode 100644 index 0000000..ba82994 --- /dev/null +++ b/engineering/understand/topology.py @@ -0,0 +1,603 @@ +"""The topology of a line network: dangling ends, near misses, interior crossings. + +Spec §11. Pure: a ``Snapshot`` in, three lists out; nothing touches a drawing. + +What is analysed: the entities on ``layers`` that draw a network - LINE, +LWPOLYLINE, POLYLINE (a bulged segment is a true arc) and ARC. Their *ends* +are the ends of the entity: a polyline's first and last vertex, never an +interior vertex, and a closed polyline has none. Everything else in the same +space is context: every LINE / LWPOLYLINE / POLYLINE / ARC / CIRCLE on another +layer (a tank outline, a wall), every CIRCLE on the analysed layers, and the +box of every INSERT on any layer (a valve, a pump, a reducer). + +Each end is exactly one of, tested in this order: + +* **joined** (no finding) - within ``tol`` of another network end, or of a + network segment other than its own segment and that segment's neighbours. + A polyline of three or more segments may join its own other end (a ring + drawn as an open polyline). A segment lying along the end's own segment and + running into it - a line or an arc drawn twice, a line over a polyline - + is not a join, so a pipe drawn twice dangles exactly like a pipe drawn once; +* **attached** (no finding) - within ``tol`` of a context curve, or inside or + within ``tol`` of an INSERT box: a line ending on equipment or on a fitting + is connected, whatever layer the equipment is on; +* **near miss** - the nearest network end or segment (the candidates of + *joined*) is within ``gap``: ``end_end`` when an end is nearest (an end + wins a tie), else ``end_segment``. One row per pair; +* **dangling** - none of the above. + +An **interior crossing** is two straight network segments on the same layer +that cross at a point more than ``tol`` from every end of both segments: a +junction nobody drew, or a crossing drawn without a break. A crossing through +a polyline vertex reads as a touch and is not reported, arcs are not tested +for crossings, and two different layers crossing is how every plan is drawn (a +cable over a pipe), so it is not a finding. Spaces are analysed apart: model +space and each paper layout never join each other. + +``gap`` defaults to 0.2 % of the diagonal of the analysed network's box, never +less than ten ``tol``. The 0.2 %, the ten-``tol`` floor and the same-layer +rule for crossings are this module's declared defaults, not standards values. + +``EntityRecord.points[0]`` is the centre of an ARC or a CIRCLE (the reader in +``engineering.understand.snapshot``); an ARC or CIRCLE without one is skipped +and counted in ``stats["skipped"]``. Spatial lookups go through a uniform +grid, so a large drawing is read in one pass rather than in pairs of +entities. +""" + +from __future__ import annotations + +import math +from collections import Counter, defaultdict +from collections.abc import Iterable +from dataclasses import dataclass + +from engineering.understand.snapshot import EntityRecord, Pt, Snapshot +from engineering.understand.vocab import classify_layer + +__all__ = [ + "DEFAULT_TOL", + "GAP_FLOOR_TOLS", + "GAP_FRACTION", + "NETWORK_CONFIDENCE", + "NETWORK_DISCIPLINES", + "Piece", + "cap_findings", + "default_gap", + "entity_pieces", + "network_layers", + "topology_findings", +] + +DEFAULT_TOL = 0.01 +#: The default gap, as a share of the analysed network's diagonal. +GAP_FRACTION = 0.002 +#: ... and never less than this many ``tol``. +GAP_FLOOR_TOLS = 10.0 +#: What makes a layer a network layer by default (spec §11). +NETWORK_DISCIPLINES = ("piping", "electrical") +NETWORK_CONFIDENCE = 0.9 +FINDING_KINDS = ("dangling", "near_miss", "crossing") + +_NETWORK_TYPES = frozenset({"LINE", "LWPOLYLINE", "POLYLINE", "ARC"}) +_CONTEXT_TYPES = _NETWORK_TYPES | {"CIRCLE"} +_TWO_PI = 2.0 * math.pi +#: A piece whose box covers more grid cells than this is kept on a side list. +_MAX_CELLS = 4096 + + +@dataclass(frozen=True) +class Piece: + """One segment of an entity: straight, or an arc when ``centre`` is set. + + ``a`` and ``b`` are its ends as drawn. An arc runs counter-clockwise from + the angle ``start`` (radians) through ``sweep`` (radians, > 0). + """ + + owner: str + layer: str + index: int + a: Pt + b: Pt + centre: Pt | None = None + radius: float = 0.0 + start: float = 0.0 + sweep: float = 0.0 + + def box(self) -> tuple[float, float, float, float]: + if self.centre is not None: + cx, cy = self.centre + r = self.radius + return (cx - r, cy - r, cx + r, cy + r) + return ( + min(self.a[0], self.b[0]), + min(self.a[1], self.b[1]), + max(self.a[0], self.b[0]), + max(self.a[1], self.b[1]), + ) + + def nearest(self, p: Pt) -> Pt: + """The point of this piece nearest to ``p``.""" + if self.centre is None: + ax, ay = self.a + dx, dy = self.b[0] - ax, self.b[1] - ay + length2 = dx * dx + dy * dy + if length2 == 0.0: + return self.a + t = ((p[0] - ax) * dx + (p[1] - ay) * dy) / length2 + t = min(1.0, max(0.0, t)) + return (ax + t * dx, ay + t * dy) + cx, cy = self.centre + rel = (math.atan2(p[1] - cy, p[0] - cx) - self.start) % _TWO_PI + distance = math.hypot(p[0] - cx, p[1] - cy) + if rel <= self.sweep + 1e-12 and distance > 0.0: + return ( + cx + self.radius * (p[0] - cx) / distance, + cy + self.radius * (p[1] - cy) / distance, + ) + return self.a if math.dist(p, self.a) <= math.dist(p, self.b) else self.b + + +def _arc_from_bulge(owner: str, layer: str, index: int, a: Pt, b: Pt, bulge: float) -> Piece: + """A polyline segment with a bulge: the arc it stands for. + + The bulge is tan(sweep / 4), positive counter-clockwise (the DXF rule). + """ + dx, dy = b[0] - a[0], b[1] - a[1] + chord = math.hypot(dx, dy) + theta = 4.0 * math.atan(bulge) + if chord == 0.0 or theta == 0.0: + return Piece(owner, layer, index, a, b) + offset = chord / (2.0 * math.tan(theta / 2.0)) + nx, ny = -dy / chord, dx / chord + centre = ((a[0] + b[0]) / 2.0 + nx * offset, (a[1] + b[1]) / 2.0 + ny * offset) + radius = abs(chord / (2.0 * math.sin(theta / 2.0))) + first = a if theta > 0 else b + start = math.atan2(first[1] - centre[1], first[0] - centre[0]) + return Piece(owner, layer, index, a, b, centre, radius, start, abs(theta)) + + +def _arc_piece(rec: EntityRecord, *, full: bool) -> Piece | None: + if not rec.points or rec.radius is None: + return None + centre = (float(rec.points[0][0]), float(rec.points[0][1])) + r = float(rec.radius) + if full or rec.angles is None: + a = b = (centre[0] + r, centre[1]) + return Piece(rec.handle, rec.layer, 0, a, b, centre, r, 0.0, _TWO_PI) + s = math.radians(float(rec.angles[0])) + e = math.radians(float(rec.angles[1])) + sweep = (e - s) % _TWO_PI or _TWO_PI + a = (centre[0] + r * math.cos(s), centre[1] + r * math.sin(s)) + b = (centre[0] + r * math.cos(e), centre[1] + r * math.sin(e)) + return Piece(rec.handle, rec.layer, 0, a, b, centre, r, s, sweep) + + +def entity_pieces(rec: EntityRecord) -> tuple[list[Piece], tuple[Pt, Pt] | None] | None: + """The pieces of one record and its two ends (None when it has none). + + None for a record that draws no curve or lacks what it needs. + """ + kind = rec.type + if kind == "LINE": + if len(rec.points) < 2: + return None + a, b = tuple(rec.points[0]), tuple(rec.points[1]) + return [Piece(rec.handle, rec.layer, 0, a, b)], (a, b) + if kind in ("LWPOLYLINE", "POLYLINE"): + points = [(float(p[0]), float(p[1])) for p in rec.points] + if len(points) < 2: + return None + bulges = list(rec.bulges) + [0.0] * (len(points) - len(rec.bulges)) + count = len(points) if rec.closed else len(points) - 1 + pieces = [ + _arc_from_bulge( + rec.handle, + rec.layer, + i, + points[i], + points[(i + 1) % len(points)], + float(bulges[i]), + ) + for i in range(count) + ] + return pieces, None if rec.closed else (points[0], points[-1]) + if kind == "ARC": + piece = _arc_piece(rec, full=False) + return None if piece is None else ([piece], (piece.a, piece.b)) + if kind == "CIRCLE": + piece = _arc_piece(rec, full=True) + return None if piece is None else ([piece], None) + return None + + +def network_layers(snap: Snapshot) -> list[dict]: + """The layers ``vocab.classify_layer`` calls piping or electrical with + confidence >= 0.9, with the classification as evidence.""" + names = set(snap.layers) | {rec.layer for rec in snap.records} + rows = [] + for name in sorted(names): + meaning = classify_layer(name) + confidence = float(meaning.get("confidence") or 0.0) + if meaning.get("discipline") in NETWORK_DISCIPLINES and confidence >= NETWORK_CONFIDENCE: + rows.append( + { + "layer": name, + "discipline": meaning.get("discipline"), + "service": meaning.get("service"), + "confidence": confidence, + "keyword": meaning.get("keyword"), + } + ) + return rows + + +def default_gap(points: Iterable[Pt], tol: float) -> float: + """0.2 % of the diagonal of the points' box, never less than ten ``tol``.""" + xs: list[float] = [] + ys: list[float] = [] + for x, y in points: + xs.append(x) + ys.append(y) + diagonal = math.hypot(max(xs) - min(xs), max(ys) - min(ys)) if xs else 0.0 + return max(GAP_FRACTION * diagonal, GAP_FLOOR_TOLS * tol) + + +def _positive(name: str, value) -> float: + try: + number = float(value) + except (TypeError, ValueError): + raise ValueError(f"{name} must be a positive number; got {value!r}") from None + if not math.isfinite(number) or number <= 0: + raise ValueError(f"{name} must be a positive number; got {value!r}") + return number + + +def _layer_list(layers) -> set[str]: + if isinstance(layers, str): + layers = [layers] + names = {str(name) for name in (layers or []) if str(name)} + if not names: + raise ValueError("layers: name at least one layer to check") + return names + + +class _Grid: + """A uniform grid of item indices by box; oversized boxes on a side list.""" + + def __init__(self, cell: float) -> None: + self.cell = cell + self.cells: dict[tuple[int, int], list[int]] = defaultdict(list) + self.oversize: list[int] = [] + + def _span(self, box) -> tuple[int, int, int, int]: + c = self.cell + return ( + math.floor(box[0] / c), + math.floor(box[1] / c), + math.floor(box[2] / c), + math.floor(box[3] / c), + ) + + def add(self, item: int, box) -> None: + x0, y0, x1, y1 = self._span(box) + if (x1 - x0 + 1) * (y1 - y0 + 1) > _MAX_CELLS: + self.oversize.append(item) + return + for ix in range(x0, x1 + 1): + for iy in range(y0, y1 + 1): + self.cells[(ix, iy)].append(item) + + def near(self, box) -> list[int]: + x0, y0, x1, y1 = self._span(box) + found: set[int] = set(self.oversize) + if (x1 - x0 + 1) * (y1 - y0 + 1) > _MAX_CELLS: + for items in self.cells.values(): + found.update(items) + return sorted(found) + for ix in range(x0, x1 + 1): + for iy in range(y0, y1 + 1): + found.update(self.cells.get((ix, iy), ())) + return sorted(found) + + +def _around(p: Pt, r: float) -> tuple[float, float, float, float]: + return (p[0] - r, p[1] - r, p[0] + r, p[1] + r) + + +def _box_distance(p: Pt, box) -> float: + dx = max(box[0] - p[0], 0.0, p[0] - box[2]) + dy = max(box[1] - p[1], 0.0, p[1] - box[3]) + return math.hypot(dx, dy) + + +def _cross(ax: float, ay: float, bx: float, by: float) -> float: + return ax * by - ay * bx + + +def _crossing(p: Piece, q: Piece, tol: float) -> Pt | None: + """Where two straight pieces cross strictly inside both, away from their ends.""" + rx, ry = p.b[0] - p.a[0], p.b[1] - p.a[1] + sx, sy = q.b[0] - q.a[0], q.b[1] - q.a[1] + denominator = _cross(rx, ry, sx, sy) + if denominator == 0.0: + return None + qx, qy = q.a[0] - p.a[0], q.a[1] - p.a[1] + t = _cross(qx, qy, sx, sy) / denominator + u = _cross(qx, qy, rx, ry) / denominator + if not (0.0 < t < 1.0 and 0.0 < u < 1.0): + return None + point = (p.a[0] + t * rx, p.a[1] + t * ry) + if min(math.dist(point, e) for e in (p.a, p.b, q.a, q.b)) <= tol: + return None + return point + + +def _overlaps(own: Piece, point: Pt, other: Piece, tol: float) -> bool: + """True when ``other`` lies along ``own`` and runs from ``point`` into it: + a line drawn twice, or a line over a polyline. Such a piece is not a join - + otherwise a pipe drawn twice would join itself and never dangle. A + collinear piece that continues the other way (a butt joint) is a join. + + Arcs follow the same rule: an arc on the same circle (centre and radius + within ``tol``) that covers ``own`` just inside ``point`` - an ARC drawn + twice, an ARC over a bulged polyline segment - is not a join; a straight + piece never lies along an arc.""" + if (own.centre is None) != (other.centre is None): + return False + if own.centre is not None: + if math.dist(own.centre, other.centre) > tol or abs(own.radius - other.radius) > tol: + return False + if own.radius <= 0.0: + return False + cx, cy = own.centre + # ``own`` runs counter-clockwise from ``start`` through ``sweep``; probe + # 2 tol of arc (at most half the sweep) inside it from whichever end + # ``point`` is. + step = min(2.0 * tol / own.radius, own.sweep / 2.0) + first = (cx + own.radius * math.cos(own.start), cy + own.radius * math.sin(own.start)) + last_angle = own.start + own.sweep + last = (cx + own.radius * math.cos(last_angle), cy + own.radius * math.sin(last_angle)) + angle = ( + own.start + step + if math.dist(point, first) <= math.dist(point, last) + else (last_angle - step) + ) + probe = (cx + own.radius * math.cos(angle), cy + own.radius * math.sin(angle)) + return math.dist(probe, other.nearest(probe)) <= tol + far = own.b if math.dist(point, own.a) <= math.dist(point, own.b) else own.a + length = math.dist(point, far) + if length == 0.0: + return False + ux, uy = (far[0] - point[0]) / length, (far[1] - point[1]) / length + along = [] + for p in (other.a, other.b): + dx, dy = p[0] - point[0], p[1] - point[1] + if abs(dy * ux - dx * uy) > tol: + return False + along.append(dx * ux + dy * uy) + return max(along) > tol + + +def _analyse_space( + space: str, + records: list[EntityRecord], + wanted: set[str], + gap: float, + tol: float, + out: dict, + skipped: Counter, +) -> tuple[int, int]: + net: list[Piece] = [] + counts: dict[str, int] = {} + closed: set[str] = set() + # handle, layer, "start" | "end", the point, the index in `net` of the piece it ends + ends: list[tuple[str, str, str, Pt, int]] = [] + context: list[Piece] = [] + boxes: list[tuple[float, float, float, float]] = [] + for rec in records: + if rec.type == "INSERT": + if rec.bbox is not None: + boxes.append(tuple(float(v) for v in rec.bbox)) + continue + if rec.type not in _CONTEXT_TYPES: + continue + built = entity_pieces(rec) + if built is None: + skipped[rec.type] += 1 + continue + pieces, entity_ends = built + if rec.layer in wanted and rec.type in _NETWORK_TYPES: + first = len(net) + net.extend(pieces) + counts[rec.handle] = len(pieces) + if entity_ends is None: + closed.add(rec.handle) + else: + ends.append((rec.handle, rec.layer, "start", entity_ends[0], first)) + ends.append((rec.handle, rec.layer, "end", entity_ends[1], len(net) - 1)) + else: + context.extend(pieces) + if not net: + return 0, 0 + + cell = max(8.0 * gap, 100.0 * tol) + net_grid = _Grid(cell) + for k, piece in enumerate(net): + net_grid.add(k, piece.box()) + end_grid = _Grid(cell) + for k, (_h, _l, _w, point, _piece) in enumerate(ends): + end_grid.add(k, _around(point, 0.0)) + context_grid = _Grid(cell) + for k, piece in enumerate(context): + context_grid.add(k, piece.box()) + box_grid = _Grid(cell) + for k, box in enumerate(boxes): + box_grid.add(k, box) + + seen_pairs: set = set() + for handle, layer, which, point, own_k in sorted(ends, key=lambda e: (e[0], e[2])): + own = net[own_k] + pieces_of_owner = counts[handle] + own_index = 0 if which == "start" else pieces_of_owner - 1 + reach = _around(point, gap) + best_end: tuple[float, int] | None = None + for k in end_grid.near(reach): + other_handle, _ol, other_which, other_point, other_k = ends[k] + if other_handle == handle and (other_which == which or pieces_of_owner < 3): + continue + if _overlaps(own, point, net[other_k], tol): + continue + distance = math.dist(point, other_point) + if best_end is None or distance < best_end[0]: + best_end = (distance, k) + best_piece: tuple[float, int, Pt] | None = None + for k in net_grid.near(reach): + piece = net[k] + if piece.owner == handle and abs(piece.index - own_index) <= 1: + continue + if _overlaps(own, point, piece, tol): + continue + foot = piece.nearest(point) + distance = math.dist(point, foot) + if best_piece is None or distance < best_piece[0]: + best_piece = (distance, k, foot) + nearest = min( + best_end[0] if best_end is not None else math.inf, + best_piece[0] if best_piece is not None else math.inf, + ) + if nearest <= tol: + continue # joined + touch = _around(point, tol) + if any( + math.dist(point, context[k].nearest(point)) <= tol for k in context_grid.near(touch) + ) or any(_box_distance(point, boxes[k]) <= tol for k in box_grid.near(touch)): + continue # attached + if nearest <= gap: + if best_end is not None and (best_piece is None or best_end[0] <= best_piece[0]): + other_handle, other_layer, other_which, other_point, _k = ends[best_end[1]] + key = frozenset({(handle, which), (other_handle, other_which)}) + kind, distance, to = "end_end", best_end[0], other_point + else: + piece = net[best_piece[1]] + other_handle, other_layer = piece.owner, piece.layer + key = (handle, which, other_handle) + kind, distance, to = "end_segment", best_piece[0], best_piece[2] + if key in seen_pairs: + continue + seen_pairs.add(key) + out["near_miss"].append( + { + "handles": [handle, other_handle], + "layers": [layer, other_layer], + "kind": kind, + "gap": distance, + "at": [point[0], point[1]], + "to": [to[0], to[1]], + "space": space, + } + ) + continue + out["dangling"].append( + { + "handle": handle, + "layer": layer, + "end": which, + "at": [point[0], point[1]], + "space": space, + } + ) + + for k, piece in enumerate(net): + if piece.centre is not None: + continue + for m in net_grid.near(piece.box()): + if m <= k: + continue + other = net[m] + if other.centre is not None or other.layer != piece.layer: + continue + if other.owner == piece.owner: + apart = abs(other.index - piece.index) + wraps = piece.owner in closed and apart == counts[piece.owner] - 1 + if apart <= 1 or wraps: + continue + point = _crossing(piece, other, tol) + if point is None: + continue + out["crossing"].append( + { + "handles": [piece.owner, other.owner], + "layer": piece.layer, + "at": [point[0], point[1]], + "space": space, + } + ) + return len(counts), len(ends) + + +def topology_findings( + snap: Snapshot, *, layers, gap: float | None = None, tol: float = DEFAULT_TOL +) -> dict: + """Dangling ends, near misses and interior crossings on ``layers`` (spec §11). + + Refused before any work: an empty ``layers``, a non-positive ``tol`` or + ``gap``, and a ``gap`` that does not exceed ``tol``. + """ + tol = _positive("tol", tol) + wanted = _layer_list(layers) + if gap is not None: + gap = _positive("gap", gap) + if gap <= tol: + raise ValueError( + f"gap ({gap:g}) must exceed tol ({tol:g}): an end within tol is joined, and a " + "near miss is a distance between the two" + ) + by_space: dict[str, list[EntityRecord]] = defaultdict(list) + for rec in snap.records: + by_space[rec.space].append(rec) + if gap is None: + points: list[Pt] = [] + for rec in snap.records: + if rec.layer in wanted and rec.type in _NETWORK_TYPES: + built = entity_pieces(rec) + if built is not None: + for piece in built[0]: + points.extend((piece.a, piece.b)) + gap = default_gap(points, tol) + out: dict = {kind: [] for kind in FINDING_KINDS} + skipped: Counter = Counter() + entities = end_count = 0 + for space in sorted(by_space): + n_entities, n_ends = _analyse_space(space, by_space[space], wanted, gap, tol, out, skipped) + entities += n_entities + end_count += n_ends + out["dangling"].sort(key=lambda r: (r["space"], r["handle"], r["end"])) + out["near_miss"].sort(key=lambda r: (r["space"], r["handles"], r["at"])) + out["crossing"].sort(key=lambda r: (r["space"], r["handles"], r["at"])) + return { + "layers": sorted(wanted), + "gap": gap, + "tol": tol, + **out, + "stats": { + "entities": entities, + "ends": end_count, + "skipped": dict(sorted(skipped.items())), + }, + } + + +def cap_findings(findings: dict, limit: int) -> dict: + """``findings`` with each list cut to ``limit`` rows, plus ``counts`` and + ``truncated`` per kind.""" + if int(limit) < 1: + raise ValueError(f"limit must be at least 1; got {limit!r}") + capped = dict(findings) + capped["counts"] = {kind: len(findings.get(kind) or []) for kind in FINDING_KINDS} + capped["truncated"] = { + kind: max(0, capped["counts"][kind] - int(limit)) for kind in FINDING_KINDS + } + for kind in FINDING_KINDS: + capped[kind] = list(findings.get(kind) or [])[: int(limit)] + return capped diff --git a/server.py b/server.py index 3284d6a..da75c52 100644 --- a/server.py +++ b/server.py @@ -5746,7 +5746,9 @@ async def drawing_finalize( mech_view_misaligned, mech_duplicate_dimension, mech_thread_unrepresented, mech_bom_balloon_mismatch — silent on a sheet with no mechanical part, plus the architectural focuses arch_room_unlabelled, arch_wall_gap, arch_opening_clash — - silent on a sheet with no ACADMCP_ARCH record), + silent on a sheet with no ACADMCP_ARCH record, plus the topology focuses + topo_dangling_endpoint, topo_near_miss, topo_interior_crossing — on network + layers no P&ID or architectural focus reads, silent on every sheet this server draws), then saves to disk, exports a screenshot, and returns the DWG path. Raises ToolError if any validator 'error' finding is present, or if critique reports an @@ -6009,7 +6011,9 @@ async def drawing_critique( "mech_duplicate_dimension, mech_thread_unrepresented, " "mech_bom_balloon_mismatch (silent on a drawing with no mechanical part), and " "the architectural focuses arch_room_unlabelled, arch_wall_gap, " - "arch_opening_clash (silent on a drawing with no ACADMCP_ARCH record). " + "arch_opening_clash (silent on a drawing with no ACADMCP_ARCH record), and " + "the topology focuses topo_dangling_endpoint, topo_near_miss, " + "topo_interior_crossing (network layers no P&ID or architectural focus reads). " "None = run all.", ), ] = None, @@ -10555,7 +10559,7 @@ async def arch_plan_from_spec( # --------------------------------------------------------------------------- -# ── SECTION 26: Understanding & QA (2 tools) ──────────────────────────────── +# ── SECTION 26: Understanding & QA (4 tools) ──────────────────────────────── # --------------------------------------------------------------------------- @@ -10713,6 +10717,317 @@ async def drawing_scale_check( raise ToolError(str(exc)) from exc +async def _diff_revision(backend, path: str | None, *, tol: float, role: str): + """One revision for ``drawing_diff``: its records and, where they can be + read, its block signatures (None otherwise). + + A DXF path is read once, here, for both. The current document goes through + ``take_snapshot``; its block table is read only on the headless engine, + under the backend's own lock. Any other path (a DWG) goes through the + snapshot too, which opens it read-only on the live engine. + """ + import asyncio + + from engineering.understand.diff import block_signatures, read_revision + from engineering.understand.snapshot import take_snapshot + + if path is None: + snap = await take_snapshot(backend, None) + blocks = None + run = getattr(backend, "_async", None) + if backend.name == "ezdxf" and run is not None: + blocks = await run(lambda: block_signatures(backend._doc, tol=tol)) + return snap, blocks + validated = validate_path(path) + if not validated.is_file(): + raise ToolError(f"{role}: {validated} does not exist or is not a file") + if validated.suffix.lower() == ".dxf": + try: + return await asyncio.to_thread(read_revision, str(validated), tol=tol) + except ValueError as exc: + raise ToolError(f"{role}: {exc}") from exc + try: + return await take_snapshot(backend, str(validated)), None + except ValueError as exc: # an unreadable suffix; the dwg refusal is not a ValueError + raise ToolError(f"{role}: {exc}") from exc + + +async def _diff_markup(backend, result: dict, new_snap, *, rev: str, description: str) -> dict: + """Revision clouds around every model-space cluster of change, filed under + ``rev`` through ``engineering.sheet.revision.add_revision`` - the only + write ``drawing_diff`` ever makes.""" + from engineering.sheet.bom import LIST_WIDTH, ROW_HEIGHT + from engineering.sheet.revision import add_revision + from engineering.understand.diff import robust_box + + clusters = [c for c in result["clusters"] if c["space"] == "Model"] + paper = len(result["clusters"]) - len(clusters) + if not clusters: + return { + "created": False, + "clouds": [], + "reason": "no change in model space to cloud", + "paper_space_clusters": paper, + } + boxes = [c["box"] for c in clusters] + shortest = min(min(b[2] - b[0], b[3] - b[1]) for b in boxes) + box = robust_box(new_snap) or ( + min(b[0] for b in boxes), + min(b[1] for b in boxes), + max(b[2] for b in boxes), + max(b[3] for b in boxes), + ) + summary = result["summary"] + text = description or ( + f"drawing_diff: {summary['changed']} changed, {summary['added']} added, " + f"{summary['removed']} removed" + ) + drawn = await add_revision( + backend, + rev=rev, + description=text, + clouds=boxes, + at=(box[2] - LIST_WIDTH, box[3] + 3.0 * ROW_HEIGHT), + cloud_segment=shortest / 4.0, + ) + drawn["paper_space_clusters"] = paper + return drawn + + +@cad_tool( + summary="Compare two revisions of a drawing: what moved, changed, was added or removed.", + cost="mutate", +) +@mcp.tool( + annotations={"title": "Drawing: Diff Two Revisions", "destructiveHint": False}, + tags={"analysis", "query"}, +) +async def drawing_diff( + old_path: Annotated[ + str, + Field( + description="The OLDER revision: a DXF path (a DWG only on the live engine, which " + "exports it read-only)." + ), + ], + new_path: Annotated[ + str | None, + Field( + default=None, + description="The NEWER revision: a DXF path, or omitted for the current document.", + ), + ] = None, + tol: Annotated[ + float, + Field( + default=0.01, + gt=0, + description="Geometry is compared rounded to this many drawing units.", + ), + ] = 0.01, + move_tol: Annotated[ + float | None, + Field( + default=None, + gt=0, + description="An unmatched entity is paired with the nearest one of its type and layer " + "within this distance and reported as changed; default 2 % of the drawing's robust " + "diagonal.", + ), + ] = None, + limit: Annotated[ + int, + Field( + default=200, ge=1, le=5000, description="Rows kept per list (changed, added, removed)." + ), + ] = 200, + markup: Annotated[ + bool, + Field( + default=False, + description="Draw revision clouds around every cluster of change on the current " + "document (the newer revision) and file them under `rev`.", + ), + ] = False, + rev: Annotated[ + str, + Field( + default="", + description="Revision code for markup=True: one to three capitals or digits.", + ), + ] = "", + description: Annotated[ + str, + Field(default="", description="Revision-row text for markup=True; default: the counts."), + ] = "", + ctx: Context = None, +) -> dict: + """Compare two revisions of one drawing by geometric signature (spec §10). + + Entities are matched exactly first (type, layer, geometry rounded to `tol`, + text, block, attributes - handles do not matter), then by handle while + handles look stable, then by type + layer + nearest position within + `move_tol`; what is left is added or removed. The report counts changes by + layer and type, lists each changed entity with what changed (moved by, text + from -> to, attribute from -> to, radius from -> to, ...), compares the + block definitions when both revisions are DXF files or the headless current + document, and groups the changes into clusters. + + Read-only unless `markup=True`: then revision clouds are drawn around the + model-space clusters on the current document, which must be the newer + revision, and filed as one revision row - the only write this tool makes. + + Refused before anything is read: `markup=True` with `new_path` (the clouds go + on the current document), `markup=True` without a valid `rev`, `markup=True` + on an engine without the revcloud capability (the live engine - + `capability: "revcloud"`), a path that does not exist, a DXF over + MAX_DXF_BYTES (the size and the variable named), a file that is not a + readable DXF, and a non-positive `tol` or `move_tol`. + """ + from backends.base import UnsupportedCapabilityError + from engineering.sheet.revision import REV_RE + from engineering.understand.diff import block_changes, cap_report, diff_snapshots + + backend = _backend(ctx) + if markup: + if new_path: + raise ToolError( + "markup draws revision clouds on the current document, which must be the newer " + "revision: open it with drawing_open and call again without new_path" + ) + if not REV_RE.fullmatch(rev.strip().upper()): + raise ToolError( + f"markup needs `rev`, the revision code the clouds are filed under - one to " + f"three capitals or digits (A, B, 01); got {rev!r}" + ) + feature = backend.capabilities().features.get("revcloud") + if feature is None or not feature.supported: + raise UnsupportedCapabilityError( + "revcloud", + f"drawing_diff(markup=True): the {backend.name} backend cannot draw revision " + "clouds (no ActiveX member; REVCLOUD is command-line only). Run the diff " + "without markup, or on the ezdxf engine.", + ) + await ctx.info(f"drawing_diff: {old_path} -> {new_path or 'the current document'}") + old_snap, old_blocks = await _diff_revision(backend, old_path, tol=tol, role="old_path") + new_snap, new_blocks = await _diff_revision(backend, new_path, tol=tol, role="new_path") + try: + result = diff_snapshots(old_snap, new_snap, tol=tol, move_tol=move_tol) + except ValueError as exc: + raise ToolError(str(exc)) from exc + result["old"] = old_snap.source + result["new"] = new_snap.source + result["blocks"] = block_changes(old_blocks, new_blocks) + report = cap_report(result, limit) + if markup: + report["markup"] = await _diff_markup( + backend, result, new_snap, rev=rev.strip().upper(), description=description + ) + return report + + +@cad_tool( + summary="Find dangling ends, near misses and unjoined crossings in a pipe or cable network.", + cost="read", +) +@mcp.tool( + annotations={"title": "Drawing: Topology Check", "readOnlyHint": True}, + tags={"analysis", "query"}, +) +async def drawing_topology_check( + layers: Annotated[ + list[str] | None, + Field( + default=None, + description="Layers to check; default: every layer the vocabulary classifies as " + "piping or electrical with confidence >= 0.9.", + ), + ] = None, + path: Annotated[ + str | None, + Field( + default=None, + description="A DXF to check instead of the current document; it is read, never " + "opened or modified.", + ), + ] = None, + gap: Annotated[ + float | None, + Field( + default=None, + gt=0, + description="Two ends, or an end and a line, closer than this without touching are a " + "near miss; default 0.2 % of the network's diagonal (never under 10 x tol).", + ), + ] = None, + tol: Annotated[ + float, + Field(default=0.01, gt=0, description="Within this distance two ends are joined."), + ] = 0.01, + limit: Annotated[ + int, + Field(default=200, ge=1, le=5000, description="Rows kept per list; counts are complete."), + ] = 200, + ctx: Context = None, +) -> dict: + """Topology of a line network (spec §11): dangling ends, near misses (end to end and + end to line within `gap`) and interior crossings (two lines of one layer crossing + away from their ends). Never modifies the drawing. + + An end that touches another network line, any curve on another layer (a tank + outline) or the box of any block reference (a valve, a pump) is connected. Arcs + and bulged polyline segments are measured as true arcs; arcs are not tested for + crossings. The default layers come with their classification as evidence. + + Refused before any work: a `path` that does not exist or cannot be read, a + `layers` entry that is not a layer of the drawing (the drawing's layers named), + no layer classified as a network layer when `layers` is omitted (name them), + a non-positive `tol` or `gap`, and a `gap` that does not exceed `tol`. + """ + from engineering.understand.snapshot import take_snapshot + from engineering.understand.topology import cap_findings, network_layers, topology_findings + + backend = _backend(ctx) + source = None + if path: + validated = validate_path(path) + if not validated.is_file(): + raise ToolError(f"path: {validated} does not exist or is not a file") + source = str(validated) + try: + snap = await take_snapshot(backend, source) + except (OSError, ValueError) as exc: + raise ToolError(f"path: {exc}") from exc + known = set(snap.layers) | {rec.layer for rec in snap.records} + classification = network_layers(snap) + if layers: + missing = [name for name in layers if name not in known] + if missing: + listed = ", ".join(sorted(known)[:20]) + (" ..." if len(known) > 20 else "") + raise ToolError( + f"layers: {', '.join(repr(m) for m in missing)} is not a layer of this drawing; " + f"its layers are {listed}" + ) + chosen = list(layers) + else: + chosen = [row["layer"] for row in classification] + if not chosen: + raise ToolError( + "no layer of this drawing is classified piping or electrical with confidence " + ">= 0.9 - name the layers to check with `layers` (drawing_understand shows what " + "each layer carries)" + ) + await ctx.info(f"drawing_topology_check: {len(chosen)} layer(s) of {snap.source}") + try: + findings = topology_findings(snap, layers=chosen, gap=gap, tol=tol) + except ValueError as exc: + raise ToolError(str(exc)) from exc + report = cap_findings(findings, limit) + report["source"] = snap.source + report["classification"] = [row for row in classification if row["layer"] in set(chosen)] + return report + + # --------------------------------------------------------------------------- # ── RESOURCES ─────────────────────────────────────────────────────────────── # --------------------------------------------------------------------------- diff --git a/tests/fixtures/revisions.py b/tests/fixtures/revisions.py new file mode 100644 index 0000000..6e6d056 --- /dev/null +++ b/tests/fixtures/revisions.py @@ -0,0 +1,96 @@ +"""Two revisions of the synthetic plant P&ID with known edits, for ``drawing_diff``. + +Built from ``build_plant_pair`` (``tests/fixtures/plant_pair.py``) with +ezdxf. The older revision is the P&ID plus one tagged block reference; the +newer one moves a line by half the move tolerance, rewrites a text, changes +the block's attribute, deletes one line, adds a circle - and then gives every +model-space entity a new handle, the way a save through another program +renumbers them. What was edited is +picked from the older revision's own records (entities whose signature is +unique, so a double-drawn line can never be the one edited) and returned with +the paths: the expected answer is read off the fixture, never typed in. +""" + +from __future__ import annotations + +from collections import Counter +from pathlib import Path + +import ezdxf + +from engineering.understand.diff import diff_snapshots, read_revision, robust_box, signature +from engineering.understand.snapshot import records_from_doc +from tests.fixtures.plant_pair import build_plant_pair + +TAG_BLOCK = "DIFF_TAG" +TAG_LAYER = "DIFF-TAGS" +ADDED_LAYER = "DIFF-ADDED" +NEW_TEXT = "REV-B NOTE" +TAG_BEFORE = "X-1" +TAG_AFTER = "X-2" + + +def renumber_modelspace(doc) -> None: + """Every model-space entity replaced by a copy under a fresh handle.""" + msp = doc.modelspace() + for entity in list(msp): + clone = entity.copy() + msp.add_entity(clone) + msp.delete_entity(entity) + + +def build_revision_pair(directory, *, edits: bool = True) -> dict: + directory = Path(directory) + truth = build_plant_pair(directory / "plant") + base = ezdxf.readfile(truth["pid"]) + box = robust_box(records_from_doc(base, source="base")) + centre = ((box[0] + box[2]) / 2.0, (box[1] + box[3]) / 2.0) + block = base.blocks.new(TAG_BLOCK) + block.add_circle((0.0, 0.0), 5.0) + block.add_attdef("TAG", (0.0, -8.0), dxfattribs={"height": 2.5}) + base.layers.add(TAG_LAYER) + tag = base.modelspace().add_blockref(TAG_BLOCK, centre, dxfattribs={"layer": TAG_LAYER}) + tag.add_auto_attribs({"TAG": TAG_BEFORE}) + old_path = directory / "rev_a.dxf" + base.saveas(old_path) + + old_snap, _blocks = read_revision(str(old_path)) + move_tol = diff_snapshots(old_snap, old_snap)["move_tol"] + counts = Counter(signature(r) for r in old_snap.records) + unique = [r for r in old_snap.records if r.space == "Model" and counts[signature(r)] == 1] + moved = next(r for r in unique if r.type == "LINE") + text = next(r for r in unique if r.type in ("TEXT", "MTEXT")) + insert = next(r for r in old_snap.records if r.block == TAG_BLOCK) + deleted = next( + r + for r in unique + if r.type in ("LINE", "LWPOLYLINE") and (r.type, r.layer) != (moved.type, moved.layer) + ) + + new = ezdxf.readfile(old_path) + msp = new.modelspace() + if edits: + new.entitydb[moved.handle].translate(move_tol / 2.0, 0.0, 0.0) + target = new.entitydb[text.handle] + if target.dxftype() == "MTEXT": + target.text = NEW_TEXT + else: + target.dxf.text = NEW_TEXT + new.entitydb[insert.handle].get_attrib("TAG").dxf.text = TAG_AFTER + msp.delete_entity(new.entitydb[deleted.handle]) + new.layers.add(ADDED_LAYER) + msp.add_circle(centre, move_tol / 10.0, dxfattribs={"layer": ADDED_LAYER}) + renumber_modelspace(new) + new_path = directory / "rev_b.dxf" + new.saveas(new_path) + return { + "old": str(old_path), + "new": str(new_path), + "move_tol": move_tol, + "old_count": len(old_snap.records), + "moved": moved.handle, + "text": text.handle, + "text_before": text.text, + "insert": insert.handle, + "deleted": deleted.handle, + } diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py index df20db3..45cc862 100644 --- a/tests/test_tool_registry.py +++ b/tests/test_tool_registry.py @@ -599,10 +599,20 @@ async def test_tool_group_sizes_are_unchanged(): `drawing_scale_check`, tagged the same, followed it (13 -> 14). This snapshot is branch-local: the track H merge task recomputes it once the understand, network and diff branches have landed. + + Track H group D opened its copy of SECTION 26 (Understanding & QA) with + `drawing_diff` and `drawing_topology_check`, both tagged `analysis` and + `query`; `analysis` outranks `query` in `_GROUP_TAG_PRIORITY`, so both file + under `analysis` (12 -> 14). This snapshot is branch-local: the track H + merge recomputes it once the understand, network and diff branches have + landed. + + Merged, the two SECTION 26 halves file all four readers under `analysis` + (12 -> 16). """ sizes = {label: len(names) for label, names in (await server._tool_groups()).items()} assert sizes == { - "analysis": 14, + "analysis": 16, "architecture": 12, "batch": 3, "blocks": 9, @@ -629,4 +639,4 @@ async def test_tool_group_sizes_are_unchanged(): "validation": 1, "view": 4, } - assert sum(sizes.values()) == 243 + assert sum(sizes.values()) == 245 diff --git a/tests/test_understand_critique.py b/tests/test_understand_critique.py new file mode 100644 index 0000000..9fca83d --- /dev/null +++ b/tests/test_understand_critique.py @@ -0,0 +1,201 @@ +"""The three topo_* focuses: a positive and a negative fixture each, and silence +on every drawing this server makes. + +The network layer is the synthetic plant's product-piping layer, read from the +fixture's truth dict (``tests/fixtures/plant_pair.py``), so the test draws on a +layer the vocabulary classifies the way a foreign drawing's layer would be +classified. +""" + +from __future__ import annotations + +import copy + +import pytest + +from engineering.arch.layers import ARCH_ROLE_LAYER +from engineering.critique import run_critique +from engineering.pid.lines import PID_LINE_LAYERS +from engineering.plan_spec import ALL_CRITIQUE_FOCUSES +from engineering.understand.critique import ( + SEMANTIC_LAYERS, + TOPO_DISPATCH, + TOPO_FOCUSES, + build_index, + focus_layers, + issues_for, +) +from engineering.understand.snapshot import EntityRecord, Snapshot +from engineering.understand.vocab import classify_layer +from tests.fixtures.plant_pair import build_plant_pair + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture(scope="module") +def pipe_layer(tmp_path_factory) -> str: + truth = build_plant_pair(tmp_path_factory.mktemp("plant")) + layer = next( + name for name, service in sorted(truth["layer_services"].items()) if service == "product" + ) + meaning = classify_layer(layer) + assert meaning["discipline"] == "piping" and meaning["confidence"] >= 0.9, meaning + return layer + + +async def _lines(backend, layer, *segments): + handles = [] + for x1, y1, x2, y2 in segments: + entity = await backend.entity_create_line(x1, y1, x2, y2, layer=layer) + handles.append(entity.handle) + return handles + + +async def _issues(backend, focus): + return issues_for(focus, await build_index(backend)) + + +# -- the closed enum and the exclusions ------------------------------------------------ + + +async def test_the_three_focuses_are_the_spec_ones_and_are_in_the_closed_enum(): + assert TOPO_FOCUSES == ("topo_dangling_endpoint", "topo_near_miss", "topo_interior_crossing") + for focus in TOPO_FOCUSES: + assert focus in ALL_CRITIQUE_FOCUSES + assert TOPO_DISPATCH[focus].needs_shared is True + + +async def test_pid_line_layers_and_arch_layers_are_excluded(): + assert set(PID_LINE_LAYERS) <= SEMANTIC_LAYERS + assert ARCH_ROLE_LAYER["wall"] in SEMANTIC_LAYERS + assert "0" not in SEMANTIC_LAYERS + snap = Snapshot( + source="test", + insunits=4, + extmin=None, + extmax=None, + layers={"PROCESS-PIPING-MAIN": {}, "A-WALL-E-N": {}}, + layouts=(), + records=( + EntityRecord("1", "LINE", "PROCESS-PIPING-MAIN", "Model", ((0, 0), (10, 0))), + EntityRecord("2", "LINE", "A-WALL-E-N", "Model", ((0, 5), (10, 5))), + ), + ) + assert focus_layers(snap) == [] + + +# -- topo_dangling_endpoint ------------------------------------------------------------ + + +async def test_dangling_fires_on_a_lone_pipe(backend, pipe_layer): + (handle,) = await _lines(backend, pipe_layer, (0, 0, 1000, 0)) + issues = await _issues(backend, "topo_dangling_endpoint") + assert [(i.severity, i.handles, i.detail["end"]) for i in issues] == [ + ("warning", [handle], "end"), + ("warning", [handle], "start"), + ] + + +async def test_dangling_is_quiet_on_a_closed_loop(backend, pipe_layer): + await _lines( + backend, + pipe_layer, + (0, 0, 1000, 0), + (1000, 0, 1000, 1000), + (1000, 1000, 0, 1000), + (0, 1000, 0, 0), + ) + assert await _issues(backend, "topo_dangling_endpoint") == [] + + +# -- topo_near_miss -------------------------------------------------------------------- + + +async def test_near_miss_fires_on_a_3_unit_gap(backend, pipe_layer): + # box 0..2003 wide: default gap 0.002 x 2003 = 4.006 > 3 + first, second = await _lines(backend, pipe_layer, (0, 0, 1000, 0), (1003, 0, 2003, 0)) + (issue,) = await _issues(backend, "topo_near_miss") + assert issue.severity == "error" + assert issue.handles == sorted([first, second]) + assert issue.detail["kind"] == "end_end" + assert issue.detail["gap"] == pytest.approx(3.0) + assert "stops 3 short of the end of" in issue.message + + +async def test_near_miss_is_quiet_when_the_lines_join(backend, pipe_layer): + await _lines(backend, pipe_layer, (0, 0, 1000, 0), (1000, 0, 2000, 0)) + assert await _issues(backend, "topo_near_miss") == [] + + +# -- topo_interior_crossing ------------------------------------------------------------ + + +async def test_crossing_fires_on_two_pipes_crossing_mid_span(backend, pipe_layer): + await _lines(backend, pipe_layer, (0, 0, 1000, 0), (500, -500, 500, 500)) + (issue,) = await _issues(backend, "topo_interior_crossing") + assert issue.severity == "info" + assert issue.detail["at"] == [500.0, 0.0] + + +async def test_crossing_is_quiet_on_a_t_junction(backend, pipe_layer): + await _lines(backend, pipe_layer, (0, 0, 1000, 0), (500, 0, 500, 500)) + assert await _issues(backend, "topo_interior_crossing") == [] + + +# -- silence on this server's own drawings --------------------------------------------- + + +async def test_a_pid_from_spec_sheet_gets_no_topo_issue(backend): + from engineering.pid.spec import EXAMPLE_SPEC, run_spec + + await run_spec(backend, copy.deepcopy(EXAMPLE_SPEC)) + assert await run_critique(backend, list(TOPO_FOCUSES)) == [] + everything = await run_critique(backend, None) + assert not [i for i in everything if i.focus.startswith("topo_")] + + +async def test_an_arch_plan_from_spec_gets_no_topo_issue(backend): + from engineering.arch.spec import EXAMPLE_SPEC, draw_plan_from_spec + + await draw_plan_from_spec(backend, copy.deepcopy(EXAMPLE_SPEC)) + assert await run_critique(backend, list(TOPO_FOCUSES)) == [] + + +async def test_a_drawing_without_a_network_layer_is_never_snapshotted(backend, monkeypatch): + import engineering.understand.snapshot as snapshot + + async def refuse(*_args, **_kwargs): + raise AssertionError("the snapshot was taken") + + monkeypatch.setattr(snapshot, "take_snapshot", refuse) + await backend.entity_create_line(0, 0, 100, 0, layer="GEOMETRY") + assert await build_index(backend) == {"layers": [], "findings": None, "error": None} + + +async def test_a_snapshot_that_fails_is_reported_not_passed(backend, pipe_layer, monkeypatch): + import engineering.understand.snapshot as snapshot + + async def broken(*_args, **_kwargs): + raise RuntimeError("export failed") + + monkeypatch.setattr(snapshot, "take_snapshot", broken) + await _lines(backend, pipe_layer, (0, 0, 1000, 0)) + (issue,) = await _issues(backend, "topo_near_miss") + assert issue.severity == "info" and "export failed" in issue.message + + +# -- the dispatch ---------------------------------------------------------------------- + + +async def test_the_three_focuses_share_one_index_per_run(backend, pipe_layer): + await _lines(backend, pipe_layer, (0, 0, 1000, 0)) + shared: dict = {} + for focus in TOPO_FOCUSES: + await TOPO_DISPATCH[focus](backend, shared) + assert list(shared) == ["topo_index"] + + +async def test_run_critique_reaches_the_topo_focuses(backend, pipe_layer): + await _lines(backend, pipe_layer, (0, 0, 1000, 0), (1003, 0, 2003, 0)) + (issue,) = await run_critique(backend, ["topo_near_miss"]) + assert issue.focus == "topo_near_miss" and issue.severity == "error" diff --git a/tests/test_understand_diff.py b/tests/test_understand_diff.py new file mode 100644 index 0000000..18b4d1b --- /dev/null +++ b/tests/test_understand_diff.py @@ -0,0 +1,445 @@ +"""drawing_diff's engine: signatures, the three matching stages, the report. + +Pure tests build ``EntityRecord``s by hand, so every coordinate is the +answer. The known-edits tests build two revisions of the synthetic plant P&ID +(``tests/fixtures/revisions.py``) and assert exactly the edits made - a move, +a text change, an attribute change, an add, a delete, under renumbered +handles - and nothing else. +""" + +from __future__ import annotations + +import ezdxf +import pytest + +import config +from engineering.understand.diff import ( + block_changes, + block_signatures, + cap_report, + diff_snapshots, + read_revision, + robust_box, + signature, +) +from engineering.understand.snapshot import EntityRecord, Snapshot +from tests.fixtures.revisions import ( + ADDED_LAYER, + NEW_TEXT, + TAG_AFTER, + TAG_BEFORE, + build_revision_pair, +) + + +def rec(handle, type_, layer="PIPE", points=(), **kw): + space = kw.pop("space", "Model") + return EntityRecord( + handle=handle, type=type_, layer=layer, space=space, points=tuple(points), **kw + ) + + +def snap(*records): + return Snapshot( + source="test", + insunits=4, + extmin=None, + extmax=None, + layers={}, + layouts=(), + records=tuple(records), + ) + + +def line(handle, x1, y1, x2, y2, layer="PIPE"): + return rec(handle, "LINE", layer, ((x1, y1), (x2, y2))) + + +STILL = [line("A", 0, 0, 100, 0), line("B", 0, 50, 100, 50), line("C", 0, 100, 100, 100)] +RENUMBERED = [line("1A", 0, 0, 100, 0), line("1B", 0, 50, 100, 50), line("1C", 0, 100, 100, 100)] + + +# -- signatures ------------------------------------------------------------------------ + + +def test_a_signature_is_rounded_to_tol_and_ignores_a_lines_direction(): + a = line("A", 0.0, 0.0, 10.0, 0.0) + b = line("B", 10.001, 0.004, 0.0, 0.0) + assert signature(a, 0.01) == signature(b, 0.01) + assert signature(a, 0.01) != signature(line("C", 0.0, 0.0, 10.02, 0.0), 0.01) + + +def test_identical_revisions_have_no_change_and_no_cluster(): + result = diff_snapshots(snap(*STILL), snap(*STILL)) + assert result["summary"] == { + "old_entities": 3, + "new_entities": 3, + "unchanged": 3, + "changed": 0, + "added": 0, + "removed": 0, + } + assert result["clusters"] == [] and result["handles_stable"] is True + + +# -- the three stages ------------------------------------------------------------------ + + +def test_a_move_under_renumbered_handles_is_matched_by_position(): + old = snap(*STILL, line("D", 0, 200, 100, 200)) + new = snap(*RENUMBERED, line("1D", 3, 200, 103, 200)) + result = diff_snapshots(old, new, move_tol=10.0) + assert result["handles_stable"] is False + (row,) = result["changed"] + assert (row["old_handle"], row["new_handle"], row["matched_by"]) == ("D", "1D", "position") + assert row["changes"] == {"moved_by": [pytest.approx(3.0), pytest.approx(0.0)]} + assert result["summary"]["unchanged"] == 3 + + +def test_a_text_change_is_reported_from_and_to(): + old = snap(*STILL, rec("T", "TEXT", "NOTES", ((10, 10),), text="OLD", height=2.5)) + new = snap(*STILL, rec("T", "TEXT", "NOTES", ((10, 10),), text="NEW", height=2.5)) + (row,) = diff_snapshots(old, new)["changed"] + assert row["matched_by"] == "handle" + assert row["changes"] == {"text": {"from": "OLD", "to": "NEW"}} + + +def test_an_attribute_change_is_reported_by_tag(): + before = (("TAG", "A-1"), ("SERVICE", "CIP")) + after = (("TAG", "A-2"), ("SERVICE", "CIP")) + old = snap(*STILL, rec("I", "INSERT", "TAGS", ((500, 500),), block="TB", attribs=before)) + new = snap(*STILL, rec("I", "INSERT", "TAGS", ((500, 500),), block="TB", attribs=after)) + (row,) = diff_snapshots(old, new)["changed"] + assert row["changes"] == {"attribs": [{"tag": "TAG", "from": "A-1", "to": "A-2"}]} + + +def test_a_radius_change_is_reported_from_and_to(): + old = snap(*STILL, rec("R", "CIRCLE", "PIPE", ((0, 0),), radius=10.0)) + new = snap(*STILL, rec("R", "CIRCLE", "PIPE", ((0, 0),), radius=12.0)) + (row,) = diff_snapshots(old, new)["changed"] + assert row["changes"] == {"radius": {"from": 10.0, "to": 12.0}} + + +def test_the_handle_stage_catches_a_layer_change_the_position_stage_cannot(): + old = snap(*STILL, line("L", 0, 300, 100, 300, layer="P1")) + new = snap(*STILL, line("L", 0, 300, 100, 300, layer="P2")) + (row,) = diff_snapshots(old, new)["changed"] + assert row["matched_by"] == "handle" + assert row["changes"] == {"layer": {"from": "P1", "to": "P2"}} + + +def test_stable_handles_pair_a_far_move_by_handle(): + old = snap(*STILL, rec("20", "CIRCLE", "PIPE", ((0, 0),), radius=5.0)) + new = snap(*STILL, rec("20", "CIRCLE", "PIPE", ((5000, 5000),), radius=5.0)) + (row,) = diff_snapshots(old, new, move_tol=10.0)["changed"] + assert row["matched_by"] == "handle" + assert row["changes"] == {"moved_by": [pytest.approx(5000.0), pytest.approx(5000.0)]} + + +def test_renumbered_handles_never_pair_strangers_that_share_a_handle(): + old = snap(*STILL, rec("20", "CIRCLE", "PIPE", ((0, 0),), radius=5.0)) + new = snap(*RENUMBERED, rec("20", "CIRCLE", "PIPE", ((5000, 5000),), radius=5.0)) + result = diff_snapshots(old, new, move_tol=10.0) + assert result["handles_stable"] is False + assert result["changed"] == [] + assert [r["handle"] for r in result["removed"]] == ["20"] + assert [r["at"] for r in result["added"]] == [[5000.0, 5000.0]] + + +def test_no_exact_match_is_no_evidence_that_handles_are_stable(): + # two fresh files hand out handles from one seed; every circle moved +1000 in x + # and the new file added them in reverse order, so equal handles are strangers + old = snap( + *(rec(f"{0x20 + i:X}", "CIRCLE", "EQ", ((0, 200 * i),), radius=10.0) for i in range(8)) + ) + new = snap( + *( + rec(f"{0x20 + k:X}", "CIRCLE", "EQ", ((1000, 200 * i),), radius=10.0) + for k, i in enumerate(reversed(range(8))) + ) + ) + result = diff_snapshots(old, new) + assert result["handles_stable"] is False + assert result["handle_evidence"] == {"exact_matches": 0, "kept_handle": 0} + assert result["changed"] == [] + assert (result["summary"]["added"], result["summary"]["removed"]) == (8, 8) + # given a move tolerance that reaches, the position stage finds the true move + moved = diff_snapshots(old, new, move_tol=1001.0) + assert {row["matched_by"] for row in moved["changed"]} == {"position"} + assert {tuple(row["changes"]["moved_by"]) for row in moved["changed"]} == {(1000.0, 0.0)} + + +def test_beyond_move_tol_a_line_is_removed_and_another_added(): + old = snap(*STILL, line("D", 0, 200, 100, 200)) + new = snap(*STILL, line("E", 0, 700, 100, 700)) + result = diff_snapshots(old, new, move_tol=10.0) + assert (result["summary"]["added"], result["summary"]["removed"]) == (1, 1) + assert result["changed"] == [] + + +# -- a value split by the signature grid ---------------------------------------------- + + +def test_a_value_one_ulp_across_a_half_grid_point_is_unchanged(): + # AutoCAD writes 16 significant digits: 0.14 + 0.005 = 0.14500000000000002 comes + # back as 0.145, and at tol=0.01 the two snap to buckets 15 and 14 + assert 0.14 + 0.005 != 0.145 + edge = 0.14 + 0.005 + assert signature(line("L", edge, 300, 50, 300)) != signature(line("L", 0.145, 300, 50, 300)) + old = snap(*STILL, line("L", edge, 300, 50, 300)) + for new in ( + snap(*STILL, line("L", 0.145, 300, 50, 300)), # paired by handle + snap(*RENUMBERED, line("1L", 0.145, 300, 50, 300)), # paired by position + ): + result = diff_snapshots(old, new, move_tol=10.0) + assert result["summary"]["unchanged"] == 4 + assert (result["summary"]["changed"], result["summary"]["added"]) == (0, 0) + assert result["summary"]["removed"] == 0 + assert result["changed"] == [] and result["clusters"] == [] + assert result["by_layer"] == {} and result["by_type"] == {} + + +def test_numbers_one_ulp_across_a_grid_boundary_are_not_changes(): + # at tol=1, 12.500000000000002 and 12.5 snap to 13 and 12 + edge = 12.500000000000002 + old = snap( + *STILL, + rec("R", "CIRCLE", "PIPE", ((edge, 0),), radius=edge), + rec("T", "TEXT", "NOTES", ((10, 10),), text="X", height=edge, rotation=edge), + rec("P", "LWPOLYLINE", "PIPE", ((0, 0), (10, 0)), bulges=(edge, 0.0)), + ) + new = snap( + *STILL, + rec("R", "CIRCLE", "PIPE", ((12.5, 0),), radius=12.5), + rec("T", "TEXT", "NOTES", ((10, 10),), text="X", height=12.5, rotation=12.5), + rec("P", "LWPOLYLINE", "PIPE", ((0, 0), (10, 0)), bulges=(12.5, 0.0)), + ) + result = diff_snapshots(old, new, tol=1.0) + assert result["summary"]["unchanged"] == 6 + assert result["changed"] == [] and result["clusters"] == [] + + +def test_a_real_edit_next_to_a_grid_split_is_still_reported(): + old = snap( + *STILL, + line("L", 0.14 + 0.005, 300, 50, 300), + rec("R", "CIRCLE", "PIPE", ((0, 0),), radius=10.0), + ) + new = snap( + *STILL, line("L", 0.145, 300, 50, 300), rec("R", "CIRCLE", "PIPE", ((0, 0),), radius=10.2) + ) + result = diff_snapshots(old, new) + (row,) = result["changed"] + assert row["old_handle"] == "R" + assert row["changes"] == {"radius": {"from": 10.0, "to": 10.2}} + assert result["summary"]["unchanged"] == 4 + assert [c["count"] for c in result["clusters"]] == [1] + + +# -- the move tolerance ---------------------------------------------------------------- + + +def test_the_default_move_tol_is_2_percent_of_the_robust_diagonal(): + # 100 points on the diagonal of a 2970 x 3960 box, and one stray 10 km away: + # the 1st-99th percentile box is (30, 40)-(2970, 3960), diagonal 4900. + points = [rec(f"P{i}", "POINT", "PTS", ((i * 30.0, i * 40.0),)) for i in range(100)] + stray = rec("FAR", "POINT", "PTS", ((1.0e7, 1.0e7),)) + drawing = snap(*points, stray) + assert robust_box(drawing) == (30.0, 40.0, 2970.0, 3960.0) + assert diff_snapshots(drawing, drawing)["move_tol"] == pytest.approx(98.0) + + +def test_a_non_positive_tolerance_is_refused_by_name(): + with pytest.raises(ValueError, match=r"^tol must be a positive number; got 0"): + diff_snapshots(snap(), snap(), tol=0) + with pytest.raises(ValueError, match=r"^move_tol must be a positive number; got -1"): + diff_snapshots(snap(), snap(), move_tol=-1) + + +# -- the report ------------------------------------------------------------------------ + + +def test_changes_are_counted_by_layer_and_type_and_clustered(): + old = snap( + line("U", 0, -500, 100, -500), + line("D", 0, 200, 100, 200), + rec("T", "TEXT", "NOTES", ((10, 10),), text="OLD", height=2.5), + ) + new = snap( + line("U", 0, -500, 100, -500), + line("D", 3, 200, 103, 200), + rec("T", "TEXT", "NOTES", ((10, 10),), text="NEW", height=2.5), + rec("N1", "CIRCLE", "PIPE", ((5000, 5000),), radius=5.0, bbox=(4995, 4995, 5005, 5005)), + ) + result = diff_snapshots(old, new, move_tol=10.0) + assert result["by_layer"] == { + "NOTES": {"changed": 1, "added": 0, "removed": 0}, + "PIPE": {"changed": 1, "added": 1, "removed": 0}, + } + assert result["by_type"] == { + "CIRCLE": {"changed": 0, "added": 1, "removed": 0}, + "LINE": {"changed": 1, "added": 0, "removed": 0}, + "TEXT": {"changed": 1, "added": 0, "removed": 0}, + } + # pad = max(0.25 x 10, 10 x 0.01) = 2.5 around each cluster + assert result["clusters"] == [ + {"space": "Model", "box": [-2.5, 197.5, 105.5, 202.5], "count": 1}, + {"space": "Model", "box": [7.5, 7.5, 12.5, 12.5], "count": 1}, + {"space": "Model", "box": [4992.5, 4992.5, 5007.5, 5007.5], "count": 1}, + ] + + +def test_cap_report_cuts_every_list_and_says_how_much(): + old = snap(*STILL) + new = snap(*STILL, *(line(f"N{i}", 0, 1000 * (i + 1), 100, 1000 * (i + 1)) for i in range(5))) + capped = cap_report(diff_snapshots(old, new, move_tol=10.0), limit=2) + assert len(capped["added"]) == 2 + # five lines 1000 apart are five clusters: the list is capped like the others + assert len(capped["clusters"]) == 2 + assert capped["truncated"] == {"changed": 0, "added": 3, "removed": 0, "clusters": 3} + assert capped["summary"]["added"] == 5 + + +def test_a_line_drawn_back_to_front_and_moved_is_a_move(): + # the signature ignores a LINE's direction; so does its movement + old = snap(*STILL, line("D", 0, 200, 100, 200)) + new = snap(*STILL, line("D", 150, 250, 50, 250)) # reversed, then moved by (50, 50) + (row,) = diff_snapshots(old, new, move_tol=100.0)["changed"] + assert row["changes"] == {"moved_by": [pytest.approx(50.0), pytest.approx(50.0)]} + + +def test_an_attribute_tag_carried_twice_cannot_hide_a_change(): + # as a dict, both revisions read NOTE = "B" and the edit of the first NOTE vanished + before = (("NOTE", "A"), ("NOTE", "B")) + after = (("NOTE", "X"), ("NOTE", "B")) + old = snap(*STILL, rec("I", "INSERT", "TAGS", ((500, 500),), block="TB", attribs=before)) + new = snap(*STILL, rec("I", "INSERT", "TAGS", ((500, 500),), block="TB", attribs=after)) + (row,) = diff_snapshots(old, new)["changed"] + assert row["changes"] == {"attribs": [{"tag": "NOTE", "from": ["A", "B"], "to": ["X", "B"]}]} + + +def test_a_dense_drawing_moved_as_a_whole_pairs_each_entity_with_its_own_copy(): + # 200 lines 10 apart on one layer, all moved (3, 0) under new handles: no exact + # match, so no handle stage, and with move_tol 500 each line has about a hundred + # candidates. The position stage keeps only the nearest few per entity, and each + # line must still pair with its own moved copy. + old = snap(*(line(f"O{i}", 0, 10 * i, 5, 10 * i) for i in range(200))) + new = snap(*(line(f"N{i}", 3, 10 * i, 8, 10 * i) for i in range(200))) + result = diff_snapshots(old, new, move_tol=500.0) + assert result["handles_stable"] is False + assert (result["summary"]["changed"], result["summary"]["added"]) == (200, 0) + assert result["summary"]["removed"] == 0 + assert all( + row["changes"] == {"moved_by": [pytest.approx(3.0), pytest.approx(0.0)]} + and row["old_handle"][1:] == row["new_handle"][1:] + for row in result["changed"] + ) + + +# -- block definitions ----------------------------------------------------------------- + + +def _blocks(radius: float, extra: bool): + doc = ezdxf.new() + doc.blocks.new("B1").add_circle((0, 0), radius) + doc.blocks.new("B2").add_line((0, 0), (10, 0)) + if extra: + doc.blocks.new("B3").add_line((0, 0), (0, 10)) + doc.blocks.new_anonymous_block().add_line((0, 0), (1, 1)) + return doc + + +def test_block_definitions_are_compared_by_content(): + old = block_signatures(_blocks(5.0, extra=False)) + new = block_signatures(_blocks(6.0, extra=True)) + assert sorted(old) == ["B1", "B2"] # the anonymous *U block is left out + assert block_changes(old, new) == { + "compared": True, + "added": ["B3"], + "removed": [], + "changed": ["B1"], + } + assert block_changes(None, new)["compared"] is False + + +# -- reading a revision ---------------------------------------------------------------- + + +def test_read_revision_refuses_a_file_over_the_size_limit(tmp_path, monkeypatch): + path = tmp_path / "big.dxf" + ezdxf.new().saveas(path) + monkeypatch.setattr(config.settings, "max_dxf_bytes", 10) + with pytest.raises(ValueError, match=r"over the MAX_DXF_BYTES limit of 10"): + read_revision(str(path)) + + +def test_read_revision_refuses_a_file_that_is_not_a_dxf(tmp_path): + path = tmp_path / "notes.dxf" + path.write_text("not a drawing\n", encoding="utf-8") + with pytest.raises(ValueError, match=r"not a readable DXF"): + read_revision(str(path)) + + +def test_read_revision_refuses_a_truncated_dxf(tmp_path): + # an interrupted save: ezdxf's tag reader runs out and raises StopIteration, + # which must surface as the refusal, not escape (it would hang asyncio.to_thread) + path = tmp_path / "half.dxf" + path.write_bytes(b" 0\nSECTION\n 2\nHEADER\n 9\n$ACADVER\n 1\nAC1015\n") + with pytest.raises(ValueError, match=r"half\.dxf: not a readable DXF \(the file ends early"): + read_revision(str(path)) + + +def test_read_revision_refuses_a_non_positive_tol_by_name(tmp_path): + path = tmp_path / "empty.dxf" + ezdxf.new().saveas(path) + with pytest.raises(ValueError, match=r"^tol must be a positive number"): + read_revision(str(path), tol=0) + + +# -- the known edits on the synthetic plant -------------------------------------------- + + +def test_the_known_edits_and_nothing_else(tmp_path): + pair = build_revision_pair(tmp_path) + old, old_blocks = read_revision(pair["old"]) + new, new_blocks = read_revision(pair["new"]) + result = diff_snapshots(old, new, move_tol=pair["move_tol"]) + total = pair["old_count"] + assert result["summary"] == { + "old_entities": total, + "new_entities": total, + "unchanged": total - 4, + "changed": 3, + "added": 1, + "removed": 1, + } + changed = {row["old_handle"]: row for row in result["changed"]} + assert set(changed) == {pair["moved"], pair["text"], pair["insert"]} + assert {row["matched_by"] for row in result["changed"]} == {"position"} + moved = changed[pair["moved"]]["changes"] + assert list(moved) == ["moved_by"] + assert moved["moved_by"][0] == pytest.approx(pair["move_tol"] / 2.0, abs=1e-9) + assert moved["moved_by"][1] == pytest.approx(0.0, abs=1e-9) + assert changed[pair["text"]]["changes"] == { + "text": {"from": pair["text_before"], "to": NEW_TEXT} + } + assert changed[pair["insert"]]["changes"] == { + "attribs": [{"tag": "TAG", "from": TAG_BEFORE, "to": TAG_AFTER}] + } + assert [row["handle"] for row in result["removed"]] == [pair["deleted"]] + assert [(row["type"], row["layer"]) for row in result["added"]] == [("CIRCLE", ADDED_LAYER)] + assert block_changes(old_blocks, new_blocks) == { + "compared": True, + "added": [], + "removed": [], + "changed": [], + } + + +def test_renumbered_handles_alone_are_no_change(tmp_path): + pair = build_revision_pair(tmp_path, edits=False) + old, _ = read_revision(pair["old"]) + new, _ = read_revision(pair["new"]) + summary = diff_snapshots(old, new)["summary"] + assert (summary["changed"], summary["added"], summary["removed"]) == (0, 0, 0) + assert summary["unchanged"] == pair["old_count"] diff --git a/tests/test_understand_diff_tool.py b/tests/test_understand_diff_tool.py new file mode 100644 index 0000000..d6f6e29 --- /dev/null +++ b/tests/test_understand_diff_tool.py @@ -0,0 +1,144 @@ +"""drawing_diff over the MCP wire, on the headless engine, against the synthetic revisions.""" + +from __future__ import annotations + +import pytest +import pytest_asyncio +from fastmcp import Client + +import server +from engineering.sheet.frames import SHEET_LAYER +from tests.fixtures.revisions import build_revision_pair + +pytestmark = pytest.mark.asyncio + + +@pytest_asyncio.fixture +async def client(monkeypatch): + """In-memory client on a fresh headless document - never the operator's AutoCAD.""" + monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf") + async with Client(server.mcp) as connected: + status = (await connected.call_tool("system_status", {})).structured_content or {} + assert status.get("backend") == "ezdxf", "these tests must never touch live AutoCAD" + await connected.call_tool("drawing_new", {}) + yield connected + + +async def _total(client) -> int: + stats = (await client.call_tool("analysis_entity_stats", {})).structured_content + return stats["total_entities"] + + +async def test_the_current_document_against_its_older_revision(client, tmp_path): + pair = build_revision_pair(tmp_path) + await client.call_tool("drawing_open", {"path": pair["new"]}) + before = await _total(client) + result = ( + await client.call_tool( + "drawing_diff", {"old_path": pair["old"], "move_tol": pair["move_tol"]} + ) + ).structured_content + assert (result["summary"]["changed"], result["summary"]["added"]) == (3, 1) + assert result["summary"]["removed"] == 1 + assert {row["old_handle"] for row in result["changed"]} == { + pair["moved"], + pair["text"], + pair["insert"], + } + assert result["blocks"]["compared"] is True and result["blocks"]["changed"] == [] + assert result["truncated"] == {"changed": 0, "added": 0, "removed": 0, "clusters": 0} + assert await _total(client) == before, "a diff without markup writes nothing" + + +async def test_two_paths_and_the_list_cap(client, tmp_path): + pair = build_revision_pair(tmp_path) + result = ( + await client.call_tool( + "drawing_diff", + { + "old_path": pair["old"], + "new_path": pair["new"], + "move_tol": pair["move_tol"], + "limit": 1, + }, + ) + ).structured_content + assert len(result["changed"]) == 1 + assert result["truncated"]["changed"] == 2 + assert result["summary"]["changed"] == 3 + + +async def test_markup_clouds_every_model_space_cluster(client, tmp_path): + pair = build_revision_pair(tmp_path) + await client.call_tool("drawing_open", {"path": pair["new"]}) + result = ( + await client.call_tool( + "drawing_diff", + {"old_path": pair["old"], "move_tol": pair["move_tol"], "markup": True, "rev": "b"}, + ) + ).structured_content + model = [c for c in result["clusters"] if c["space"] == "Model"] + markup = result["markup"] + assert markup["created"] is True and markup["rev"] == "B" + assert len(markup["clouds"]) == len(model) >= 1 + for handle in markup["clouds"]: + cloud = (await client.call_tool("entity_get", {"handle": handle})).structured_content + assert (cloud["type"], cloud["layer"]) == ("LWPOLYLINE", SHEET_LAYER) + + +@pytest.mark.parametrize( + ("arguments", "fragment"), + [ + ( + {"markup": True, "rev": "B", "new_path": "NEW"}, + "markup draws revision clouds on the current", + ), + ({"markup": True}, "markup needs `rev`"), + ({"markup": True, "rev": "TOO-LONG"}, "markup needs `rev`"), + ], +) +async def test_markup_refusals_come_before_any_read(client, tmp_path, arguments, fragment): + pair = build_revision_pair(tmp_path) + if arguments.get("new_path") == "NEW": + arguments = {**arguments, "new_path": pair["new"]} + result = await client.call_tool( + "drawing_diff", {"old_path": pair["old"], **arguments}, raise_on_error=False + ) + assert result.is_error is True + assert fragment in result.content[0].text + + +async def test_a_missing_revision_is_refused_by_parameter(client, tmp_path): + result = await client.call_tool( + "drawing_diff", {"old_path": str(tmp_path / "nowhere.dxf")}, raise_on_error=False + ) + assert result.is_error is True + assert "old_path:" in result.content[0].text and "does not exist" in result.content[0].text + + +async def test_a_revision_that_is_neither_dxf_nor_dwg_is_refused_by_parameter(client, tmp_path): + notes = tmp_path / "notes.txt" + notes.write_text("not a drawing\n", encoding="utf-8") + result = await client.call_tool("drawing_diff", {"old_path": str(notes)}, raise_on_error=False) + assert result.is_error is True + text = result.content[0].text + assert "old_path:" in text and "reads a .dxf or a .dwg" in text + + +async def test_a_truncated_revision_is_refused_not_hung(client, tmp_path): + import asyncio + + half = tmp_path / "half.dxf" + half.write_bytes(b" 0\nSECTION\n 2\nHEADER\n 9\n$ACADVER\n 1\nAC1015\n") + result = await asyncio.wait_for( + client.call_tool("drawing_diff", {"old_path": str(half)}, raise_on_error=False), + timeout=30, + ) + assert result.is_error is True + text = result.content[0].text + assert "old_path:" in text and "not a readable DXF" in text + + +async def test_the_diff_is_advertised_as_non_destructive(client): + tools = {tool.name: tool for tool in await client.list_tools()} + assert tools["drawing_diff"].annotations.destructiveHint is False diff --git a/tests/test_understand_topology.py b/tests/test_understand_topology.py new file mode 100644 index 0000000..0816de2 --- /dev/null +++ b/tests/test_understand_topology.py @@ -0,0 +1,317 @@ +"""The topology engine: ends, joins, near misses, crossings - by hand-computed geometry. + +Every record is built by hand, so every coordinate in an assertion is the +answer. The network layer is called PIPE here; which layers are network layers +by default is ``network_layers``' job and is tested on the synthetic plant. +""" + +from __future__ import annotations + +import pytest + +from engineering.understand.snapshot import EntityRecord, Snapshot +from engineering.understand.topology import ( + Piece, + cap_findings, + entity_pieces, + network_layers, + topology_findings, +) +from engineering.understand.vocab import classify_layer +from tests.fixtures.plant_pair import build_plant_pair + +L = "PIPE" + + +def rec(handle, type_, layer=L, points=(), **kw): + space = kw.pop("space", "Model") + return EntityRecord( + handle=handle, type=type_, layer=layer, space=space, points=tuple(points), **kw + ) + + +def line(handle, x1, y1, x2, y2, layer=L, space="Model"): + return rec(handle, "LINE", layer, ((x1, y1), (x2, y2)), space=space) + + +def snap(*records): + return Snapshot( + source="test", + insunits=4, + extmin=None, + extmax=None, + layers={r.layer: {"color": 7, "linetype": "Continuous"} for r in records}, + layouts=("Layout1",), + records=tuple(records), + ) + + +def run(*records, gap=5.0, layers=(L,)): + return topology_findings(snap(*records), layers=list(layers), gap=gap) + + +def ends(rows): + return {(row["handle"], row["end"]) for row in rows} + + +# -- pieces ---------------------------------------------------------------------------- + + +def test_a_bulge_of_one_is_the_semicircle_below_its_chord(): + # bulge +1 from (0,0) to (200,0): counter-clockwise, centre (100,0), radius 100, + # through (100,-100) + pieces, entity_ends = entity_pieces( + rec("PL", "LWPOLYLINE", L, ((0.0, 0.0), (200.0, 0.0)), bulges=(1.0, 0.0)) + ) + (arc,) = pieces + assert arc.centre == (pytest.approx(100.0), pytest.approx(0.0, abs=1e-9)) + assert arc.radius == pytest.approx(100.0) + foot = arc.nearest((100.0, -150.0)) + assert foot == (pytest.approx(100.0), pytest.approx(-100.0)) + assert entity_ends == ((0.0, 0.0), (200.0, 0.0)) + + +def test_a_closed_polyline_has_no_ends_and_a_closing_piece(): + pieces, entity_ends = entity_pieces( + rec("SQ", "LWPOLYLINE", L, ((0, 0), (10, 0), (10, 10), (0, 10)), closed=True) + ) + assert entity_ends is None and len(pieces) == 4 + assert (pieces[3].a, pieces[3].b) == ((0.0, 10.0), (0.0, 0.0)) + + +def test_a_straight_piece_nearest_point_is_clamped_to_its_ends(): + piece = Piece("A", L, 0, (0.0, 0.0), (10.0, 0.0)) + assert piece.nearest((5.0, 3.0)) == (5.0, 0.0) + assert piece.nearest((-4.0, 3.0)) == (0.0, 0.0) + + +# -- dangling and joined --------------------------------------------------------------- + + +def test_a_lone_line_dangles_at_both_ends(): + result = run(line("A", 0, 0, 1000, 0)) + assert ends(result["dangling"]) == {("A", "start"), ("A", "end")} + assert result["near_miss"] == [] and result["crossing"] == [] + + +def test_a_t_junction_joins_the_stem(): + result = run(line("A", 0, 0, 1000, 0), line("B", 500, 0, 500, 500)) + assert ends(result["dangling"]) == {("A", "start"), ("A", "end"), ("B", "end")} + + +def test_a_closed_square_of_lines_has_no_finding(): + square = [ + line("A", 0, 0, 1000, 0), + line("B", 1000, 0, 1000, 1000), + line("C", 1000, 1000, 0, 1000), + line("D", 0, 1000, 0, 0), + ] + result = run(*square) + assert (result["dangling"], result["near_miss"], result["crossing"]) == ([], [], []) + + +def test_an_end_on_an_arc_and_on_an_arc_body_is_joined(): + # ARC centre (0,0) r 100 from 0 to 90 degrees: ends (100,0) and (0,100) + arc = rec("ARC1", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0)) + on_end = line("L1", 100, 0, 500, 0) + on_body = line("L2", 70.71067811865476, 70.71067811865476, 300, 300) + result = run(arc, on_end, on_body) + assert ends(result["dangling"]) == {("ARC1", "end"), ("L1", "end"), ("L2", "end")} + + +def test_an_end_on_a_bulged_segment_is_joined(): + arc = rec("PL", "LWPOLYLINE", L, ((0.0, 0.0), (200.0, 0.0)), bulges=(1.0, 0.0)) + result = run(arc, line("L", 100, -100, 100, -500)) + assert ends(result["dangling"]) == {("PL", "start"), ("PL", "end"), ("L", "end")} + + +def test_an_end_inside_an_insert_box_is_attached(): + valve = rec("V1", "INSERT", "VALVES", ((1000, 0),), block="VALVE", bbox=(995, -5, 1005, 5)) + result = run(line("A", 0, 0, 995, 0), valve) + assert ends(result["dangling"]) == {("A", "start")} + + +def test_an_end_on_a_curve_of_another_layer_is_attached(): + tank = rec( + "TK", + "LWPOLYLINE", + "EQUIPMENT", + ((1000, -500), (2000, -500), (2000, 500), (1000, 500)), + closed=True, + ) + result = run(line("A", 0, 0, 1000, 0), tank) + assert ends(result["dangling"]) == {("A", "start")} + + +def test_model_space_and_a_layout_never_join(): + result = run(line("A", 0, 0, 100, 0), line("B", 100, 0, 200, 0, space="Layout1")) + assert ends(result["dangling"]) == {("A", "start"), ("A", "end"), ("B", "start"), ("B", "end")} + + +# -- lines drawn twice ----------------------------------------------------------------- + + +def test_a_pipe_drawn_twice_still_dangles_at_all_four_ends(): + result = run(line("A", 0, 0, 1000, 0), line("A2", 0, 0, 1000, 0)) + assert ends(result["dangling"]) == { + ("A", "start"), + ("A", "end"), + ("A2", "start"), + ("A2", "end"), + } + assert result["near_miss"] == [] and result["crossing"] == [] + + +def test_a_line_over_a_polyline_joins_only_where_the_polyline_turns(): + polyline = rec("PL", "LWPOLYLINE", L, ((0, 0), (1000, 0), (1000, 1000))) + result = run(polyline, line("OV", 0, 0, 1000, 0)) + assert ends(result["dangling"]) == {("OV", "start"), ("PL", "end"), ("PL", "start")} + + +def test_an_arc_drawn_twice_still_dangles_at_all_four_ends(): + # the arc form of a pipe drawn twice: a bend drawn twice must not join itself + first = rec("A1", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0)) + second = rec("A2", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0)) + result = run(first, second) + assert ends(result["dangling"]) == { + ("A1", "start"), + ("A1", "end"), + ("A2", "start"), + ("A2", "end"), + } + assert result["near_miss"] == [] + + +def test_two_arcs_of_one_circle_butted_end_to_end_join(): + # 0-90 then 90-180 on the same circle: continuing the other way is a join + first = rec("A1", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0)) + second = rec("A2", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(90.0, 180.0)) + assert ends(run(first, second)["dangling"]) == {("A1", "start"), ("A2", "end")} + + +def test_a_bulge_of_minus_one_is_the_semicircle_above_its_chord(): + ((arc,), _ends) = entity_pieces( + rec("PL", "LWPOLYLINE", L, ((0.0, 0.0), (200.0, 0.0)), bulges=(-1.0, 0.0)) + ) + assert arc.radius == pytest.approx(100.0) + assert arc.nearest((100.0, 150.0)) == (pytest.approx(100.0), pytest.approx(100.0)) + + +def test_a_side_drawn_twice_inside_a_closed_loop_is_harmless(): + square = [ + line("A", 0, 0, 1000, 0), + line("A2", 0, 0, 1000, 0), + line("B", 1000, 0, 1000, 1000), + line("C", 1000, 1000, 0, 1000), + line("D", 0, 1000, 0, 0), + ] + result = run(*square) + assert (result["dangling"], result["near_miss"], result["crossing"]) == ([], [], []) + + +# -- near misses ----------------------------------------------------------------------- + + +def test_two_ends_3_apart_are_one_end_end_near_miss(): + result = run(line("A", 0, 0, 1000, 0), line("B", 1003, 0, 2000, 0)) + (row,) = result["near_miss"] + assert row["handles"] == ["A", "B"] and row["kind"] == "end_end" + assert row["gap"] == pytest.approx(3.0) + assert (row["at"], row["to"]) == ([1000.0, 0.0], [1003.0, 0.0]) + assert ends(result["dangling"]) == {("A", "start"), ("B", "end")} + + +def test_an_end_4_above_a_line_is_an_end_segment_near_miss(): + result = run(line("A", 0, 0, 1000, 0), line("B", 500, 4, 500, 500)) + (row,) = result["near_miss"] + assert row["handles"] == ["B", "A"] and row["kind"] == "end_segment" + assert row["gap"] == pytest.approx(4.0) and row["to"] == [500.0, 0.0] + + +def test_an_end_beyond_the_gap_dangles_instead(): + result = run(line("A", 0, 0, 1000, 0), line("B", 500, 6, 500, 500)) + assert result["near_miss"] == [] + assert ("B", "start") in ends(result["dangling"]) + + +def test_a_polyline_that_nearly_closes_on_itself_is_one_near_miss(): + ring = rec("RING", "LWPOLYLINE", L, ((0, 0), (1000, 0), (1000, 1000), (0, 1000), (0, 3))) + result = run(ring) + (row,) = result["near_miss"] + assert row["handles"] == ["RING", "RING"] and row["kind"] == "end_end" + assert row["gap"] == pytest.approx(3.0) + assert result["dangling"] == [] + + +# -- crossings ------------------------------------------------------------------------- + + +def test_two_lines_of_one_layer_crossing_mid_span_are_a_crossing(): + result = run(line("A", 0, 0, 1000, 0), line("B", 500, -500, 500, 500)) + assert result["crossing"] == [ + {"handles": ["A", "B"], "layer": L, "at": [500.0, 0.0], "space": "Model"} + ] + + +def test_two_layers_crossing_and_a_t_junction_are_not_crossings(): + other = line("B", 500, -500, 500, 500, layer="CABLE") + assert run(line("A", 0, 0, 1000, 0), other, layers=(L, "CABLE"))["crossing"] == [] + assert run(line("A", 0, 0, 1000, 0), line("B", 500, 0, 500, 500))["crossing"] == [] + + +def test_a_bow_tie_polyline_crosses_itself(): + bow = rec("BOW", "LWPOLYLINE", L, ((0, 0), (100, 100), (100, 0), (0, 100))) + (row,) = run(bow)["crossing"] + assert row["handles"] == ["BOW", "BOW"] and row["at"] == [50.0, 50.0] + + +# -- defaults and refusals ------------------------------------------------------------- + + +def test_the_default_gap_is_0_2_percent_of_the_network_diagonal(): + # box 0..3000 x 0..4000: diagonal 5000, gap 10 + result = topology_findings(snap(line("A", 0, 0, 3000, 0), line("B", 0, 0, 0, 4000)), layers=[L]) + assert result["gap"] == pytest.approx(10.0) + assert result["stats"] == {"entities": 2, "ends": 4, "skipped": {}} + + +def test_an_arc_without_a_centre_is_skipped_and_counted(): + result = run(line("A", 0, 0, 10, 0), rec("ARC9", "ARC", L, (), radius=5.0, angles=(0, 90))) + assert result["stats"]["skipped"] == {"ARC": 1} + + +@pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"layers": []}, r"^layers: name at least one layer to check"), + ({"layers": [L], "tol": 0}, r"^tol must be a positive number; got 0"), + ({"layers": [L], "gap": -1}, r"^gap must be a positive number; got -1"), + ({"layers": [L], "gap": 0.005}, r"^gap \(0\.005\) must exceed tol \(0\.01\)"), + ], +) +def test_refusals_come_before_any_work(kwargs, message): + with pytest.raises(ValueError, match=message): + topology_findings(snap(line("A", 0, 0, 1, 0)), **kwargs) + + +def test_cap_findings_counts_everything_and_cuts_the_lists(): + lines = [line(f"L{i}", 0, 1000 * i, 100, 1000 * i) for i in range(3)] + capped = cap_findings(run(*lines), limit=2) + assert capped["counts"] == {"dangling": 6, "near_miss": 0, "crossing": 0} + assert capped["truncated"] == {"dangling": 4, "near_miss": 0, "crossing": 0} + assert len(capped["dangling"]) == 2 + + +# -- the default layers on the synthetic plant ----------------------------------------- + + +def test_the_plant_piping_layers_are_network_layers(tmp_path): + from engineering.understand.snapshot import read_snapshot + + truth = build_plant_pair(tmp_path) + pid = read_snapshot(truth["pid"]) + found = {row["layer"] for row in network_layers(pid)} + for layer, service in truth["layer_services"].items(): + if layer in pid.layers and service in ("product", "cip_supply", "cip_return"): + assert layer in found, (layer, classify_layer(layer)) + assert all(row["confidence"] >= 0.9 for row in network_layers(pid)) diff --git a/tests/test_understand_topology_tool.py b/tests/test_understand_topology_tool.py new file mode 100644 index 0000000..e2f0e00 --- /dev/null +++ b/tests/test_understand_topology_tool.py @@ -0,0 +1,101 @@ +"""drawing_topology_check over the MCP wire, on the headless engine.""" + +from __future__ import annotations + +import ezdxf +import pytest +import pytest_asyncio +from fastmcp import Client + +import server +from tests.fixtures.plant_pair import build_plant_pair + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture(scope="module") +def pipe_layer(tmp_path_factory) -> str: + truth = build_plant_pair(tmp_path_factory.mktemp("plant")) + return next( + name for name, service in sorted(truth["layer_services"].items()) if service == "product" + ) + + +@pytest_asyncio.fixture +async def client(monkeypatch): + """In-memory client on a fresh headless document - never the operator's AutoCAD.""" + monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf") + async with Client(server.mcp) as connected: + status = (await connected.call_tool("system_status", {})).structured_content or {} + assert status.get("backend") == "ezdxf", "these tests must never touch live AutoCAD" + await connected.call_tool("drawing_new", {}) + yield connected + + +async def _line(client, layer, x1, y1, x2, y2): + await client.call_tool( + "entity_create_line", {"x1": x1, "y1": y1, "x2": x2, "y2": y2, "layer": layer} + ) + + +async def _total(client) -> int: + stats = (await client.call_tool("analysis_entity_stats", {})).structured_content + return stats["total_entities"] + + +async def test_the_default_layers_are_the_classified_network_layers(client, pipe_layer): + await _line(client, pipe_layer, 0, 0, 1000, 0) + await _line(client, pipe_layer, 1003, 0, 2003, 0) + before = await _total(client) + result = (await client.call_tool("drawing_topology_check", {})).structured_content + assert result["layers"] == [pipe_layer] + assert [row["layer"] for row in result["classification"]] == [pipe_layer] + assert result["counts"] == {"dangling": 2, "near_miss": 1, "crossing": 0} + assert result["near_miss"][0]["gap"] == pytest.approx(3.0) + assert await _total(client) == before, "the check never writes" + + +async def test_a_dxf_path_is_read_without_opening_it(client, pipe_layer, tmp_path): + doc = ezdxf.new() + doc.layers.add(pipe_layer) + msp = doc.modelspace() + msp.add_line((0, 0), (1000, 0), dxfattribs={"layer": pipe_layer}) + msp.add_line((500, -500), (500, 500), dxfattribs={"layer": pipe_layer}) + path = tmp_path / "crossing.dxf" + doc.saveas(path) + result = ( + await client.call_tool("drawing_topology_check", {"path": str(path), "gap": 5.0}) + ).structured_content + assert result["counts"]["crossing"] == 1 + assert result["crossing"][0]["at"] == [500.0, 0.0] + assert await _total(client) == 0, "the current document is untouched" + + +async def test_an_unknown_layer_is_refused_with_the_drawings_layers(client, pipe_layer): + await _line(client, pipe_layer, 0, 0, 10, 0) + result = await client.call_tool( + "drawing_topology_check", {"layers": ["NOPE"]}, raise_on_error=False + ) + assert result.is_error is True + assert "layers: 'NOPE' is not a layer of this drawing" in result.content[0].text + + +async def test_a_drawing_with_no_network_layer_is_refused_not_passed(client): + await _line(client, "GEOMETRY", 0, 0, 10, 0) + result = await client.call_tool("drawing_topology_check", {}, raise_on_error=False) + assert result.is_error is True + assert "no layer of this drawing is classified piping or electrical" in result.content[0].text + + +async def test_gap_must_exceed_tol(client, pipe_layer): + await _line(client, pipe_layer, 0, 0, 10, 0) + result = await client.call_tool( + "drawing_topology_check", {"gap": 0.005, "tol": 0.01}, raise_on_error=False + ) + assert result.is_error is True + assert "gap (0.005) must exceed tol (0.01)" in result.content[0].text + + +async def test_the_check_is_advertised_read_only(client): + tools = {tool.name: tool for tool in await client.list_tools()} + assert tools["drawing_topology_check"].annotations.readOnlyHint is True