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

merge: track H group D

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-24 12:32:37 +03:00
co-authored by Claude Opus 5.5
parent d0e99bab7b 1795dc7f2c
commit cfcc8ce6bb
16 changed files with 3237 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engineering/understand/critique.py
+205
View File
@@ -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}
engineering/understand/diff.py
+727
View File
@@ -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
engineering/understand/topology.py
+603
View File
@@ -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