feat(understand): drawing_understand - units, extents, clusters, layers, tags, rooms

A drawing somebody else made, read back as one report in which every
claim carries its evidence and a confidence; the drawing is never
modified.

- units.py: $INSUNITS against the unit the geometry implies (text
  heights, wall thickness, door swings, labelled room areas, overall
  size). The header is overruled only when the evidence outweighs it,
  and then with a warning that quotes the numbers. Robust extents grow
  from the quartile box of entity centres; what never joins is an
  outlier, named by handle.
- clusters.py: separate bodies of model-space content by raster,
  8-connectivity and box absorption - two copies of one plan are two
  clusters.
- describe.py: layers with drawn length and the vocabulary's meaning
  (name keyword, else the texts on the layer), equipment tags (one tag
  lettered twice beside its symbol merged, repeats flagged ambiguous),
  room labels in the wall-layer faces of engineering.arch.faces, text
  languages by character class, blocks, layouts, the title block.
- server.py opens SECTION 26 (Understanding & QA) with
  drawing_understand: core, read-only, tagged analysis/query. Alias
  record, inventory, README release snapshot 242, frozen group sizes
  (analysis 13).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-24 11:50:41 +03:00
co-authored by Claude Opus 5.5
parent 704a44b190
commit 2459694880
12 changed files with 1818 additions and 8 deletions
+194
View File
@@ -0,0 +1,194 @@
"""Separate bodies of model-space content: sheets, plan copies, details.
A foreign model space is often several drawings side by side - two copies of
one plan, a detail beside it, a legend. Every reader after this one must know
which body a tag or a room belongs to, or it matches a tag in one copy against
the same tag in the other.
The algorithm, in the order it runs:
1. **Raster.** The box of the records is divided into square cells of
``cell`` drawing units (default :data:`CELL_FRACTION` of its diagonal, so
the grid never exceeds about 50 x 50 cells). Every record marks the cells
its box covers.
2. **Connect.** Occupied cells that touch, corners included, are one
component; a record belongs to the component of its cells. Two bodies stay
apart when an empty cell separates them - a gap of two cells is always
enough.
3. **Absorb.** Components whose boxes intersect are merged, repeatedly: a tag
floating in the middle of a room lies inside the plan's box and belongs to
the plan, not to a cluster of its own.
Ids are ``C1``, ``C2``, ... by entity count, largest first (ties left to right,
then bottom to top). Outliers are the caller's to exclude: one entity thousands of
kilometres away would otherwise set the cell size. Pure; the drawing is never touched.
"""
from __future__ import annotations
import math
from collections.abc import Iterable, Sequence
from engineering.understand.snapshot import EntityRecord
from engineering.understand.units import entity_box
__all__ = ["CELL_FRACTION", "find_clusters"]
#: Cell side as a fraction of the content's diagonal.
CELL_FRACTION = 0.02
#: Labels listed per cluster.
LABELS_PER_CLUSTER = 5
_TEXT_TYPES = frozenset({"TEXT", "MTEXT", "MULTILEADER"})
def _intersect(a, b) -> bool:
return a[0] <= b[2] and b[0] <= a[2] and a[1] <= b[3] and b[1] <= a[3]
def _merge(a, b):
return (min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3]))
def _root(parent: list[int], i: int) -> int:
"""Union-find root with path halving."""
while parent[i] != i:
parent[i] = parent[parent[i]]
i = parent[i]
return i
def _absorb(bodies: list[tuple[tuple, list[int]]]) -> list[tuple[tuple, list[int]]]:
"""Merge bodies whose boxes intersect until none do.
One pass is a sweep over boxes sorted by their left edge (a pair can only
intersect while the next left edge is inside the current box) feeding a
union-find; a merged box can reach a body it did not touch before, so
passes repeat until one merges nothing.
"""
while True:
order = sorted(range(len(bodies)), key=lambda i: bodies[i][0][0])
parent = list(range(len(bodies)))
joined = False
for n, i in enumerate(order):
box = bodies[i][0]
for j in order[n + 1 :]:
if bodies[j][0][0] > box[2]:
break
ri, rj = _root(parent, i), _root(parent, j)
if ri != rj and _intersect(box, bodies[j][0]):
parent[rj] = ri
joined = True
if not joined:
return bodies
merged: dict[int, tuple[tuple, list[int]]] = {}
for i, (box, members) in enumerate(bodies):
r = _root(parent, i)
if r in merged:
merged[r] = (_merge(merged[r][0], box), merged[r][1] + members)
else:
merged[r] = (box, list(members))
bodies = list(merged.values())
def find_clusters(
records: Sequence[EntityRecord],
*,
exclude: Iterable[str] = (),
cell: float | None = None,
) -> tuple[list[dict], dict[str, str]]:
"""``(clusters, membership)``: one row per body of content, and handle -> id.
A row is ``{"id", "box": [x0, y0, x1, y1], "size": [w, h], "entities",
"layers": [the three commonest], "labels": [the tallest texts]}``.
``exclude`` names handles to leave out (the outliers). A record without a
box (no geometry the snapshot could read) belongs to no cluster.
"""
skip = set(exclude)
items = [(rec, box) for rec in records if rec.handle not in skip and (box := entity_box(rec))]
if not items:
return [], {}
x0 = min(b[0] for _r, b in items)
y0 = min(b[1] for _r, b in items)
x1 = max(b[2] for _r, b in items)
y1 = max(b[3] for _r, b in items)
if cell is None:
diag = math.hypot(x1 - x0, y1 - y0)
cell = diag * CELL_FRACTION if diag > 0.0 else 1.0
if not cell > 0.0:
raise ValueError(f"cell: must be greater than zero, got {cell!r}")
def span(lo: float, hi: float, origin: float) -> range:
return range(
int(math.floor((lo - origin) / cell)), int(math.floor((hi - origin) / cell)) + 1
)
owner: dict[tuple[int, int], list[int]] = {}
first_cell: list[tuple[int, int]] = []
for k, (_rec, b) in enumerate(items):
cols, rows = span(b[0], b[2], x0), span(b[1], b[3], y0)
first_cell.append((cols[0], rows[0]))
for i in cols:
for j in rows:
owner.setdefault((i, j), []).append(k)
component: dict[tuple[int, int], int] = {}
count = 0
for start in owner:
if start in component:
continue
component[start] = count
stack = [start]
while stack:
ci, cj = stack.pop()
for di in (-1, 0, 1):
for dj in (-1, 0, 1):
nxt = (ci + di, cj + dj)
if nxt in owner and nxt not in component:
component[nxt] = count
stack.append(nxt)
count += 1
groups: dict[int, list[int]] = {}
for k in range(len(items)):
groups.setdefault(component[first_cell[k]], []).append(k)
bodies = []
for members in groups.values():
box = items[members[0]][1]
for k in members[1:]:
box = _merge(box, items[k][1])
bodies.append((box, members))
bodies = _absorb(bodies)
bodies.sort(key=lambda body: (-len(body[1]), body[0][0], body[0][1]))
clusters: list[dict] = []
membership: dict[str, str] = {}
for n, (box, members) in enumerate(bodies, start=1):
cid = f"C{n}"
layers: dict[str, int] = {}
texts = []
for k in members:
rec = items[k][0]
membership[rec.handle] = cid
layers[rec.layer] = layers.get(rec.layer, 0) + 1
if rec.type in _TEXT_TYPES and rec.text and rec.text.strip():
texts.append((-(rec.height or 0.0), rec.text.strip()))
labels: list[str] = []
for _h, text in sorted(texts):
if text not in labels:
labels.append(text)
if len(labels) == LABELS_PER_CLUSTER:
break
clusters.append(
{
"id": cid,
"box": list(box),
"size": [box[2] - box[0], box[3] - box[1]],
"entities": len(members),
"layers": [
name for name, _c in sorted(layers.items(), key=lambda kv: (-kv[1], kv[0]))[:3]
],
"labels": labels,
}
)
return clusters, membership
+463
View File
@@ -0,0 +1,463 @@
"""drawing_understand: one read of a foreign drawing, every claim with its evidence.
`describe` assembles the report from the pure readers - units and extents
(`units`), bodies of content (`clusters`), and here: layers and what they
carry, equipment tags, rooms, text languages, blocks, layouts and the title
block. Each claim carries its evidence (what was measured or matched) and a
confidence; nothing is corrected silently, and the drawing is never modified
(the snapshot is a frozen copy).
Confidences, in one place:
* a layer name matching the vocabulary: `vocab.classify_layer`'s own (0.9 for
a direct keyword); the texts on an unnamed layer voting instead:
:data:`TEXT_VOTE_CONFIDENCE` when most of them agree, :data:`WEAK_VOTE_CONFIDENCE`
otherwise;
* a tag read from a block attribute :data:`ATTRIBUTE_TAG_CONFIDENCE`, from a
text :data:`TEXT_TAG_CONFIDENCE`;
* a room label that sits in a face of the wall layers
:data:`ROOM_FACE_CONFIDENCE`, a label alone :data:`ROOM_LABEL_CONFIDENCE`;
* a title block whose name says so :data:`TITLE_NAME_CONFIDENCE`, one chosen by
its attribute count alone :data:`TITLE_ATTRIBUTE_CONFIDENCE`;
* character classes are exact: a text either holds a Cyrillic letter or not.
"""
from __future__ import annotations
import math
import re
from collections import Counter
from collections.abc import Sequence
from engineering.arch.faces import Face, face_containing, planar_faces
from engineering.pid.graph import ATTRIBUTE_TAGS, KEYWORDS
from engineering.understand.clusters import find_clusters
from engineering.understand.labels import parse_tag
from engineering.understand.snapshot import EntityRecord, Snapshot, length_of
from engineering.understand.units import infer_units, robust_extents, wall_segments
from engineering.understand.vocab import classify_layer, room_label
__all__ = [
"describe",
"find_rooms",
"prepare",
"tag_occurrences",
"text_language",
]
TEXT_VOTE_CONFIDENCE = 0.6
WEAK_VOTE_CONFIDENCE = 0.4
ATTRIBUTE_TAG_CONFIDENCE = 0.95
TEXT_TAG_CONFIDENCE = 0.8
ROOM_FACE_CONFIDENCE = 0.9
ROOM_LABEL_CONFIDENCE = 0.6
TITLE_NAME_CONFIDENCE = 0.9
TITLE_ATTRIBUTE_CONFIDENCE = 0.6
#: The face finder pairs every wall segment with every other; above this many
#: the rooms are listed without faces and a warning says so.
FACE_SEGMENT_CAP = 3000
#: Endpoints closer than this fraction of the robust diagonal are one corner
#: (1 mm on a 50 m plan drawn in millimetres).
FACE_TOL_FRACTION = 2e-5
#: The same tag lettered twice closer than this fraction of the robust diagonal
#: (an attribute and a text beside one symbol) is one occurrence.
TAG_MERGE_FRACTION = 0.01
#: Rows listed for named blocks, most inserted first.
BLOCK_ROW_CAP = 100
#: Block-name words that mark a title block (EN / NL / DE / TR / RU).
TITLE_WORDS = ("TITLE", "TITEL", "STEMPEL", "SCHRIFTFELD", "ANTET", "ШТАМП")
_TEXT_TYPES = frozenset({"TEXT", "MTEXT", "MULTILEADER"})
_TOKEN_RE = re.compile(r"[A-Z]+")
_CYRILLIC = re.compile(r"[\u0400-\u04FF]") # the Cyrillic block
#: Letters only Turkish (of the five languages) writes; ç, ö and ü are shared
#: with German and French and are not evidence on their own.
_TURKISH = re.compile(r"[ĞğİıŞş]")
_LATIN = re.compile(r"[A-Za-z\u00C0-\u024F]") # ASCII and Latin-1 to Latin Extended-B
def _anchor(rec: EntityRecord) -> tuple[float, float] | None:
if rec.points:
return (float(rec.points[0][0]), float(rec.points[0][1]))
if rec.bbox is not None:
b = rec.bbox
return ((b[0] + b[2]) / 2.0, (b[1] + b[3]) / 2.0)
return None
def prepare(snap: Snapshot) -> dict:
"""The shared first pass: model-space records, extents and outliers, the
records kept, clusters and membership. `describe` and `scale.scale_check`
both start here, so they agree on which entity is an outlier and which
cluster a tag sits in."""
model = [r for r in snap.records if r.space == "Model"]
extents = robust_extents(model)
outliers = {o["handle"] for o in extents["outliers"]}
kept = [r for r in model if r.handle not in outliers]
clusters, membership = find_clusters(kept)
robust = extents["robust"]
diag = math.hypot(robust[2] - robust[0], robust[3] - robust[1]) if robust else 0.0
return {
"model": model,
"extents": extents,
"kept": kept,
"clusters": clusters,
"membership": membership,
"diag": diag,
}
def tag_occurrences(snap: Snapshot, *, prep: dict | None = None) -> list[dict]:
"""Every equipment tag on the drawing, one row per occurrence.
A tag is read with `labels.parse_tag` from a TEXT / MTEXT / MULTILEADER,
or from an INSERT attribute named like a tag (`pid.graph.ATTRIBUTE_TAGS`).
Two readings of one tag in one space closer than
:data:`TAG_MERGE_FRACTION` of the diagonal are one occurrence (the
attribute's position and confidence win). Rows carry ``tag``, ``handle``,
``at``, ``space``, ``layer``, ``cluster`` (model space only), ``source``
and ``confidence``.
"""
prep = prep or prepare(snap)
radius = prep["diag"] * TAG_MERGE_FRACTION
found: list[dict] = []
for rec in snap.records:
at = _anchor(rec)
if at is None:
continue
readings = []
if rec.type in _TEXT_TYPES and rec.text:
readings.append((parse_tag(rec.text), "text", TEXT_TAG_CONFIDENCE))
elif rec.type == "INSERT":
for name, value in rec.attribs:
if name.upper() in ATTRIBUTE_TAGS and value:
readings.append((parse_tag(value), "attribute", ATTRIBUTE_TAG_CONFIDENCE))
for tag, source, confidence in readings:
if not tag:
continue
found.append(
{
"tag": tag,
"handle": rec.handle,
"at": [at[0], at[1]],
"space": rec.space,
"layer": rec.layer,
"cluster": prep["membership"].get(rec.handle) if rec.space == "Model" else None,
"source": source,
"confidence": confidence,
}
)
found.sort(key=lambda o: (-o["confidence"], o["tag"], o["handle"]))
kept: list[dict] = []
for occ in found:
twin = next(
(
k
for k in kept
if k["tag"] == occ["tag"]
and k["space"] == occ["space"]
and math.dist(k["at"], occ["at"]) <= radius
),
None,
)
if twin is None:
kept.append(occ)
kept.sort(key=lambda o: (o["tag"], o["space"], o["at"][0], o["at"][1]))
return kept
def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict) -> dict:
"""Room labels and the faces of the wall layers they sit in.
A label is any text `vocab.room_label` recognises. Wall layers are the
ones `vocab.classify_layer` files under ``architecture``; their straight
segments go through `engineering.arch.faces.planar_faces` (arcs are left
out and counted). ``labelled`` pairs each room row with its `Face` for a
caller that needs the geometry; ``rooms`` is JSON-ready.
"""
labels = []
for rec in records:
if rec.type in _TEXT_TYPES and rec.text:
parsed = room_label(rec.text)
at = _anchor(rec)
if parsed and at is not None:
labels.append((rec, parsed, at))
segments, bulged = wall_segments(records)
faces: tuple[Face, ...] = ()
skipped = None
if len(segments) > FACE_SEGMENT_CAP:
skipped = (
f"{len(segments)} wall segments exceed the {FACE_SEGMENT_CAP}-segment cap of the "
"face finder; rooms are listed by label only"
)
elif segments:
faces = planar_faces(segments, tol=tol)
rooms: list[dict] = []
labelled: list[tuple[dict, Face]] = []
used: set[int] = set()
for rec, parsed, at in labels:
face = face_containing(faces, at) if faces else None
row = {
"number": parsed.get("number"),
"name": parsed.get("name"),
"language": parsed.get("language"),
"text": rec.text,
"handle": rec.handle,
"at": [at[0], at[1]],
"cluster": membership.get(rec.handle),
"face": None,
"source": "label",
"confidence": ROOM_LABEL_CONFIDENCE,
}
if face is not None:
used.add(id(face))
row["face"] = {
"area": face.area,
"polygon": [[p[0], p[1]] for p in face.loop],
"centroid": [face.centroid[0], face.centroid[1]],
}
row["source"] = "label+face"
row["confidence"] = ROOM_FACE_CONFIDENCE
labelled.append((row, face))
rooms.append(row)
rooms.sort(key=lambda r: (r["cluster"] or "", r["number"] or "", r["name"] or "", r["at"]))
return {
"rooms": rooms,
"labelled": labelled,
"faces": len(faces),
"faces_unlabelled": sum(1 for f in faces if id(f) not in used),
"wall_segments": len(segments),
"arcs_skipped": bulged,
"skipped": skipped,
}
def text_language(text: str) -> str:
"""``cyrillic`` | ``turkish`` | ``latin`` | ``none`` by character class, in
that precedence (a Russian label with a Latin tag in it is Cyrillic)."""
if _CYRILLIC.search(text):
return "cyrillic"
if _TURKISH.search(text):
return "turkish"
if _LATIN.search(text):
return "latin"
return "none"
def _layers(snap: Snapshot, unit_mm: float | None) -> list[dict]:
counts: Counter = Counter()
lengths: dict[str, float] = {}
texts: dict[str, list[str]] = {}
for rec in snap.records:
counts[rec.layer] += 1
lengths[rec.layer] = lengths.get(rec.layer, 0.0) + length_of(rec)
if rec.type in _TEXT_TYPES and rec.text:
texts.setdefault(rec.layer, []).append(rec.text)
rows = []
for name in sorted(set(snap.layers) | set(counts)):
props = snap.layers.get(name, {})
c = classify_layer(name)
row = {
"name": name,
"entities": counts.get(name, 0),
"length": lengths.get(name, 0.0),
"length_m": lengths.get(name, 0.0) * unit_mm / 1000.0 if unit_mm else None,
"color": props.get("color"),
"linetype": props.get("linetype"),
"discipline": c.get("discipline", "unknown"),
"service": c.get("service", "unknown"),
"keyword": c.get("keyword"),
"language": c.get("language"),
"confidence": c.get("confidence", 0.0),
"evidence": f"layer name keyword {c.get('keyword')!r}" if c.get("keyword") else None,
}
if row["discipline"] == "unknown" and row["service"] == "unknown":
votes: Counter = Counter()
said = texts.get(name, [])
for text in said:
t = classify_layer(text)
if (
t.get("discipline", "unknown") != "unknown"
or t.get("service", "unknown") != "unknown"
):
votes[(t.get("discipline", "unknown"), t.get("service", "unknown"))] += 1
if votes:
(discipline, service), n = sorted(votes.items(), key=lambda kv: (-kv[1], kv[0]))[0]
row.update(
discipline=discipline,
service=service,
confidence=(
TEXT_VOTE_CONFIDENCE if n * 2 > len(said) else WEAK_VOTE_CONFIDENCE
),
evidence=f"{n} of {len(said)} texts on the layer read as {service} / {discipline}",
)
rows.append(row)
rows.sort(key=lambda r: (-r["entities"], r["name"]))
return rows
def _block_kind(name: str) -> str | None:
for token in _TOKEN_RE.findall(name.upper()):
kind = KEYWORDS.get(token)
if kind:
return kind
return None
def _blocks(snap: Snapshot) -> dict:
inserts = [r for r in snap.records if r.type == "INSERT" and r.block]
named = Counter(r.block for r in inserts if not r.block.startswith("*"))
anonymous = Counter(r.block for r in inserts if r.block.startswith("*"))
rows = [
{
"name": name,
"inserts": n,
"kind": _block_kind(name),
"discipline": classify_layer(name).get("discipline", "unknown"),
}
for name, n in sorted(named.items(), key=lambda kv: (-kv[1], kv[0]))
]
kinds = Counter(row["kind"] for row in rows if row["kind"])
return {
"inserts": len(inserts),
"named": rows[:BLOCK_ROW_CAP],
"named_total": len(rows),
"anonymous": {"names": len(anonymous), "inserts": sum(anonymous.values())},
"kinds": dict(sorted(kinds.items())),
}
def _title_block(snap: Snapshot) -> dict:
best = None
for rec in snap.records:
if rec.type != "INSERT" or len(rec.attribs) < 3:
continue
upper = (rec.block or "").upper()
named = any(word in upper for word in TITLE_WORDS)
key = (named, rec.space != "Model", len(rec.attribs))
if best is None or key > best[0]:
best = (key, rec, named)
if best is None:
return {"found": False, "reason": "no block reference with three or more attributes"}
_key, rec, named = best
return {
"found": True,
"handle": rec.handle,
"block": rec.block,
"space": rec.space,
"attributes": {tag: value for tag, value in rec.attribs},
"confidence": TITLE_NAME_CONFIDENCE if named else TITLE_ATTRIBUTE_CONFIDENCE,
"evidence": (
f"block name {rec.block!r} reads as a title block"
if named
else f"the block reference with the most attributes ({len(rec.attribs)})"
),
}
def _languages(snap: Snapshot, layer_rows: list[dict]) -> dict:
by_class: Counter = Counter()
samples: dict[str, list[str]] = {}
for rec in snap.records:
if rec.type in _TEXT_TYPES and rec.text:
lang = text_language(rec.text)
by_class[lang] += 1
if len(samples.setdefault(lang, [])) < 3:
samples[lang].append(rec.handle)
total = sum(by_class.values())
layer_langs = Counter(row["language"] for row in layer_rows if row.get("language"))
return {
"texts": total,
"by_class": {k: by_class.get(k, 0) for k in ("latin", "turkish", "cyrillic", "none")},
"share": {
k: (by_class.get(k, 0) / total if total else 0.0)
for k in ("latin", "turkish", "cyrillic", "none")
},
"samples": samples,
"layer_name_languages": dict(sorted(layer_langs.items())),
"confidence": 1.0,
}
def describe(snap: Snapshot) -> dict:
"""The whole report; see the module docstring for the confidences."""
prep = prepare(snap)
warnings: list[str] = []
tol = prep["diag"] * FACE_TOL_FRACTION if prep["diag"] > 0 else 1.0
rooms = find_rooms(prep["kept"], tol=tol, membership=prep["membership"])
units = infer_units(
snap,
records=prep["kept"],
room_areas=[row["face"]["area"] for row, _face in rooms["labelled"]],
)
if units["warning"]:
warnings.append(units["warning"])
extents = dict(prep["extents"])
extents["declared"] = (
{"min": list(snap.extmin), "max": list(snap.extmax)}
if snap.extmin is not None and snap.extmax is not None
else None
)
if extents["outliers"]:
handles = ", ".join(o["handle"] for o in extents["outliers"][:5])
stretch = extents["stretch"]
warnings.append(
f"{len(extents['outliers'])} entit{'y' if len(extents['outliers']) == 1 else 'ies'} "
f"far outside the content stretch the extents"
+ (f" {stretch:.3g}x" if stretch else "")
+ f": {handles}"
)
if rooms["skipped"]:
warnings.append(rooms["skipped"])
occurrences = tag_occurrences(snap, prep=prep)
by_tag: dict[str, list[dict]] = {}
for occ in occurrences:
by_tag.setdefault(occ["tag"], []).append(occ)
tags = [
{
"tag": tag,
"count": len(rows),
"ambiguous": len(rows) > 1,
"confidence": max(r["confidence"] for r in rows),
"occurrences": [
{k: r[k] for k in ("handle", "at", "space", "cluster", "layer", "source")}
for r in rows
],
}
for tag, rows in sorted(by_tag.items())
]
repeated = [t["tag"] for t in tags if t["ambiguous"]]
if repeated:
warnings.append(
f"{len(repeated)} tag(s) appear more than once (plan copies, legends, second "
f"sheets): {', '.join(repeated[:5])}"
)
layer_rows = _layers(snap, units["mm_per_unit"])
layout_names = list(snap.layouts)
if "Model" not in layout_names:
layout_names.insert(0, "Model")
space_counts = Counter(r.space for r in snap.records)
return {
"source": snap.source,
"entities": len(snap.records),
"units": units,
"extents": extents,
"clusters": prep["clusters"],
"layers": layer_rows,
"equipment_tags": tags,
"rooms": rooms["rooms"],
"room_faces": {
"faces": rooms["faces"],
"unlabelled": rooms["faces_unlabelled"],
"wall_segments": rooms["wall_segments"],
"arcs_skipped": rooms["arcs_skipped"],
"tolerance": tol,
},
"languages": _languages(snap, layer_rows),
"blocks": _blocks(snap),
"layouts": [{"name": n, "entities": space_counts.get(n, 0)} for n in layout_names],
"title_block": _title_block(snap),
"warnings": warnings,
}
+399
View File
@@ -0,0 +1,399 @@
"""Declared versus inferred drawing units, and extents that one stray entity cannot stretch.
`$INSUNITS` is a claim the drawing makes about itself, and a foreign drawing
can make it wrongly: a layout that declares inches over millimetre geometry
reads 25.4 times too large if the header is believed. This module never corrects
the claim silently. It measures what the geometry implies and reports both,
with the numbers that decided it.
**Evidence.** Each item is one median measured in drawing units and the band a
real drawing puts it in, in metres. A candidate unit is *plausible* for an item
when the measurement, converted by that unit, lands in the band. An item that
every candidate or no candidate satisfies says nothing and is dropped as
inconclusive.
======================== ====================== ====== ====================================
item band weight basis
======================== ====================== ====== ====================================
``room_area`` 1 m² … 5 000 m² 1.0 a labelled room's face (this reader's
heuristic)
``wall_thickness`` 0.05 m … 1.0 m 1.0 nearest parallel wall line (heuristic)
``door_leaf`` 0.5 m … 1.5 m 1.0 radius of a quarter-circle arc, a door
swing (heuristic)
``text_height`` 1.8 mm … 4.0 m 0.5 ISO 3098-0 lettering heights 1.8-20 mm
at ISO 5455 scales 1:1 to 1:200 (the
1:200 ceiling is this reader's choice
for plant drawings)
``overall_size`` 0.01 m … 2 000 m 0.25 the larger side of the robust extents
(heuristic: a plant or a building)
======================== ====================== ====== ====================================
**Decision.** A candidate scores the weights of the items it is plausible for;
the header is evidence too and adds :data:`HEADER_WEIGHT` to the declared unit.
The highest score wins and a tie goes to the declared unit, then to the order of
:data:`PREFERENCE`. So the header is overruled only when the geometry
outweighs it - one weak item (the overall size) never does.
**Extents.** The body of the drawing is grown from its middle: the seed is
the quartile box of the entity centres (Q1..Q3 on each axis), and every
entity whose box lies within :data:`OUTLIER_REACH` body diagonals of the body
joins it, pass after pass, until a pass adds nothing. What never joins is an
outlier - an entity farther from everything else than the drawing is large.
The robust extents are the body's box. This is this reader's heuristic, not
a statistical standard: a per-axis fence on centres (Tukey's) flags the edge
walls of a plan whose centres crowd its middle, and a quantile range is
dragged by the outlier itself in a small drawing. Fewer than four entities
have no outliers.
Pure: records in, dicts out. The drawing is never touched.
"""
from __future__ import annotations
import math
import statistics
from collections.abc import Iterable, Sequence
from ezdxf import units as ezdxf_units
from engineering.understand.snapshot import EntityRecord, Snapshot
from engineering.understand.vocab import classify_layer
__all__ = [
"BANDS",
"CANDIDATES",
"HEADER_WEIGHT",
"MM_PER_UNIT",
"PREFERENCE",
"entity_box",
"infer_units",
"robust_extents",
"wall_segments",
]
#: Millimetres per drawing unit, exact by definition: the SI prefixes, and the
#: inch (25.4 mm) and foot (304.8 mm) of the 1959 international yard and pound
#: agreement. ezdxf's own ``METER_FACTOR`` is rounded (39.37007874 in/m), so it
#: is not used for conversion.
MM_PER_UNIT: dict[str, float] = {"mm": 1.0, "cm": 10.0, "m": 1000.0, "in": 25.4, "ft": 304.8}
CANDIDATES: tuple[str, ...] = ("mm", "cm", "m", "in", "ft")
#: Tie-break after the declared unit.
PREFERENCE: tuple[str, ...] = ("mm", "m", "cm", "in", "ft")
#: What the header's own claim weighs against the geometry.
HEADER_WEIGHT = 0.5
#: (low, high, weight, power): the band in metres (m² for power 2).
BANDS: dict[str, tuple[float, float, float, int]] = {
"room_area": (1.0, 5000.0, 1.0, 2),
"wall_thickness": (0.05, 1.0, 1.0, 1),
"door_leaf": (0.5, 1.5, 1.0, 1),
"text_height": (0.0018, 4.0, 0.5, 1),
"overall_size": (0.01, 2000.0, 0.25, 1),
}
#: How far (in body diagonals) an entity may lie from the body and still join it.
OUTLIER_REACH = 1.0
#: A door swing is a quarter circle; the sweep may be drawn a little off.
DOOR_SWEEP = (85.0, 95.0)
#: Wall-thickness pairing reads at most this many wall segments.
WALL_SEGMENT_CAP = 3000
_TEXT_TYPES = frozenset({"TEXT", "MTEXT"})
_POLY_TYPES = frozenset({"LWPOLYLINE", "POLYLINE"})
def entity_box(rec: EntityRecord) -> tuple[float, float, float, float] | None:
"""The record's WCS box: its own ``bbox``, else its points (a circle or arc
grown by its radius about its centre), else None."""
if rec.bbox is not None:
return tuple(float(v) for v in rec.bbox)
if not rec.points:
return None
xs = [float(p[0]) for p in rec.points]
ys = [float(p[1]) for p in rec.points]
grow = float(rec.radius) if rec.type in ("CIRCLE", "ARC") and rec.radius else 0.0
return (min(xs) - grow, min(ys) - grow, max(xs) + grow, max(ys) + grow)
def _union(boxes: Iterable[tuple[float, float, float, float]]):
boxes = list(boxes)
if not boxes:
return None
return (
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),
)
def _diag(box) -> float:
return math.hypot(box[2] - box[0], box[3] - box[1]) if box else 0.0
def _box_gap(a, b) -> float:
"""Distance between two boxes; 0 when they touch or overlap."""
dx = max(a[0] - b[2], b[0] - a[2], 0.0)
dy = max(a[1] - b[3], b[1] - a[3], 0.0)
return math.hypot(dx, dy)
def _grow_body(boxes: Sequence[tuple[float, float, float, float]]) -> set[int]:
"""Indices of the boxes that never join the body grown from the centre."""
if len(boxes) < 4:
return set()
xs = [(b[0] + b[2]) / 2.0 for b in boxes]
ys = [(b[1] + b[3]) / 2.0 for b in boxes]
qx = statistics.quantiles(xs, n=4, method="inclusive")
qy = statistics.quantiles(ys, n=4, method="inclusive")
body = (qx[0], qy[0], qx[2], qy[2])
floor = statistics.median(_diag(b) for b in boxes)
pending = set(range(len(boxes)))
while pending:
reach = OUTLIER_REACH * max(_diag(body), floor)
joined = [i for i in pending if _box_gap(boxes[i], body) <= reach]
if not joined:
break
for i in joined:
body = _union([body, boxes[i]])
pending.difference_update(joined)
return pending
def _distance_to_box(x: float, y: float, box) -> float:
dx = max(box[0] - x, 0.0, x - box[2])
dy = max(box[1] - y, 0.0, y - box[3])
return math.hypot(dx, dy)
def robust_extents(records: Sequence[EntityRecord]) -> dict:
"""Computed and robust extents of ``records`` and the outliers between them.
``outliers`` are sorted farthest first, each with its handle, type, layer,
centre and its distance from the robust box. ``stretch`` is the computed
diagonal over the robust one (None when the robust box is a point).
"""
items = [(rec, box) for rec in records if (box := entity_box(rec)) is not None]
centres = [((b[0] + b[2]) / 2.0, (b[1] + b[3]) / 2.0) for _rec, b in items]
skip = _grow_body([b for _rec, b in items])
outside = sorted(skip)
computed = _union(b for _rec, b in items)
robust = _union(b for i, (_rec, b) in enumerate(items) if i not in skip)
outliers = []
for i in outside:
rec, _box = items[i]
cx, cy = centres[i]
outliers.append(
{
"handle": rec.handle,
"type": rec.type,
"layer": rec.layer,
"at": [cx, cy],
"distance": _distance_to_box(cx, cy, robust) if robust else None,
"confidence": 0.9,
}
)
outliers.sort(key=lambda o: (-(o["distance"] or 0.0), o["handle"]))
robust_diag = _diag(robust)
return {
"computed": list(computed) if computed else None,
"robust": list(robust) if robust else None,
"outliers": outliers,
"stretch": (_diag(computed) / robust_diag) if robust_diag > 0.0 else None,
"method": (
"body grown from the quartile box of entity centres; an entity farther from it "
f"than {OUTLIER_REACH:g} body diagonal(s) never joins and is an outlier"
),
}
def wall_segments(
records: Sequence[EntityRecord], *, is_wall=None
) -> tuple[list[tuple[tuple[float, float], tuple[float, float]]], int]:
"""Straight segments of LINE and polyline records on wall layers, plus the
number of bulged (arc) polyline segments left out.
``is_wall(layer)`` defaults to `vocab.classify_layer` naming the
``architecture`` discipline.
"""
cache: dict[str, bool] = {}
def wall(layer: str) -> bool:
if layer not in cache:
if is_wall is not None:
cache[layer] = bool(is_wall(layer))
else:
cache[layer] = classify_layer(layer).get("discipline") == "architecture"
return cache[layer]
segments: list[tuple[tuple[float, float], tuple[float, float]]] = []
bulged = 0
for rec in records:
if not wall(rec.layer):
continue
if rec.type == "LINE" and len(rec.points) >= 2:
segments.append((tuple(rec.points[0]), tuple(rec.points[1])))
elif rec.type in _POLY_TYPES and len(rec.points) >= 2:
pts = list(rec.points)
count = len(pts) if rec.closed else len(pts) - 1
for i in range(count):
a, b = pts[i], pts[(i + 1) % len(pts)]
bulge = rec.bulges[i] if i < len(rec.bulges) else 0.0
if abs(bulge) > 1e-12:
bulged += 1
continue
if a != b:
segments.append((tuple(a), tuple(b)))
return segments, bulged
def _wall_thickness(segments) -> tuple[float | None, int]:
"""Median distance from an axis-parallel wall segment to its nearest parallel
neighbour on either side that overlaps it; (None, 0) without a pair.
Both sides, because the inner face of a double-line wall has the room on
one side and its own outer face on the other: looking one way only reads
half the faces as room widths and the median lands on neither."""
horizontal: list[tuple[float, float, float]] = [] # (y, x0, x1)
vertical: list[tuple[float, float, float]] = [] # (x, y0, y1)
for (ax, ay), (bx, by) in segments[:WALL_SEGMENT_CAP]:
length = math.hypot(bx - ax, by - ay)
if length <= 0.0:
continue
if abs(by - ay) <= 1e-9 * length:
horizontal.append((ay, min(ax, bx), max(ax, bx)))
elif abs(bx - ax) <= 1e-9 * length:
vertical.append((ax, min(ay, by), max(ay, by)))
gaps: list[float] = []
for group in (horizontal, vertical):
group.sort()
for i, (c, lo, hi) in enumerate(group):
best = None
for step in (1, -1):
j = i + step
while 0 <= j < len(group):
c2, lo2, hi2 = group[j]
d = abs(c2 - c)
if best is not None and d >= best:
break
if d > 0.0 and lo2 < hi and lo < hi2:
best = d
j += step
if best is not None:
gaps.append(best)
if not gaps:
return None, 0
return statistics.median(gaps), len(gaps)
def _declared(code: int) -> dict:
try:
name = ezdxf_units.decode(int(code))
label = ezdxf_units.unit_name(int(code))
except (IndexError, ValueError, TypeError):
name, label = None, None
return {"code": int(code), "name": name, "label": label or "Unitless"}
def infer_units(
snap: Snapshot,
*,
records: Sequence[EntityRecord] | None = None,
room_areas: Sequence[float] = (),
) -> dict:
"""The unit the geometry implies, against the one ``$INSUNITS`` declares.
``records`` defaults to the snapshot's model-space records; pass the ones
left after the outliers are removed. ``room_areas`` are labelled room faces
in square drawing units. Every item of ``evidence`` carries its median,
its sample count, what it is in metres under each candidate, the
candidates it is plausible for and whether it was informative.
"""
recs = [r for r in snap.records if r.space == "Model"] if records is None else list(records)
measured: dict[str, tuple[float, int]] = {}
heights = [float(r.height) for r in recs if r.type in _TEXT_TYPES and r.height and r.height > 0]
if heights:
measured["text_height"] = (statistics.median(heights), len(heights))
radii = []
for r in recs:
if r.type == "ARC" and r.radius and r.angles:
sweep = (float(r.angles[1]) - float(r.angles[0])) % 360.0
if DOOR_SWEEP[0] <= sweep <= DOOR_SWEEP[1]:
radii.append(float(r.radius))
if radii:
measured["door_leaf"] = (statistics.median(radii), len(radii))
thickness, pairs = _wall_thickness(wall_segments(recs)[0])
if thickness is not None:
measured["wall_thickness"] = (thickness, pairs)
areas = [float(a) for a in room_areas if a and a > 0]
if areas:
measured["room_area"] = (statistics.median(areas), len(areas))
box = robust_extents(recs)["robust"]
if box:
side = max(box[2] - box[0], box[3] - box[1])
if side > 0:
measured["overall_size"] = (side, len(recs))
evidence = []
for kind, (value, count) in measured.items():
low, high, weight, power = BANDS[kind]
physical = {u: value * (MM_PER_UNIT[u] / 1000.0) ** power for u in CANDIDATES}
plausible = [u for u in CANDIDATES if low <= physical[u] <= high]
evidence.append(
{
"kind": kind,
"value": value,
"count": count,
"band": [low, high],
"band_unit": "m2" if power == 2 else "m",
"weight": weight,
"as": physical,
"plausible": plausible,
"informative": 0 < len(plausible) < len(CANDIDATES),
}
)
declared = _declared(snap.insunits)
name = declared["name"]
informative = [e for e in evidence if e["informative"]]
scores = {u: sum(e["weight"] for e in informative if u in e["plausible"]) for u in CANDIDATES}
if name in scores:
scores[name] += HEADER_WEIGHT
if informative:
best = max(scores.values())
order = ([name] if name in scores else []) + [u for u in PREFERENCE if u != name]
inferred = next(u for u in order if scores[u] == best)
total = sum(e["weight"] for e in informative) + (HEADER_WEIGHT if name in scores else 0.0)
confidence = round(0.9 * scores[inferred] / total, 2)
elif name in MM_PER_UNIT:
inferred, confidence = name, 0.4
else:
inferred, confidence = None, 0.0
warning = None
if inferred is not None and inferred != name:
support = [e for e in informative if inferred in e["plausible"]]
detail = "; ".join(
f"{e['kind']} {e['value']:g} = {e['as'][inferred]:.4g} {e['band_unit']} as {inferred}"
+ (f", {e['as'][name]:.4g} {e['band_unit']} as {name}" if name in MM_PER_UNIT else "")
for e in support
)
warning = (
f"INSUNITS declares {declared['label']} ({declared['code']}) but the geometry reads "
f"as {inferred}: {detail}. Lengths are measured in {inferred}; the header is not "
"changed."
)
elif inferred is None:
warning = (
f"INSUNITS declares {declared['label']} ({declared['code']}) and no measurement "
"settles the unit; lengths stay in drawing units."
)
return {
"declared": declared,
"inferred": inferred,
"mm_per_unit": MM_PER_UNIT.get(inferred) if inferred else None,
"agree": inferred is not None and inferred == name,
"confidence": confidence,
"scores": scores,
"evidence": evidence,
"warning": warning,
}