feat(understand): the snapshot, the vocabulary and the label parsers; MAX_DXF_BYTES 512 MB; the office extra
Track H's foundation, landed on main before its three groups branch. - engineering/understand/snapshot.py: one bulk read of a drawing into frozen WCS records (OCS translated through backends/ocs.py, text decoded, an INSERT boxed by its block geometry without its attribute texts). A DXF is read with ezdxf after the size check; the headless engine's document is read under its own lock; the live engine exports the active document with Document.Export(<tmp>/snap, "DXF", <set of all>) - never renamed, deleted afterwards - and a DWG is opened read-only, exported and closed. Measured on AutoCAD 2026: ".dxf" is refused with Invalid argument, the file lands as snap.DXF, model space, paper space and unused block definitions all reach it, and the selection set does not narrow a DXF export. - engineering/understand/labels.py: %%c, \U+2205, the U+2300 sign, o-stroke and MTEXT format runs all read as one diameter spelling; sizes are kept as drawn and never equated; Cyrillic look-alike capitals fold into Latin tags; panels, wiring callouts and electrical data (kW, phases, volts, N, VFD) in EN / TR / RU / NL / DE. A power that is not written stays None. - engineering/understand/vocab.py: layer classification with a confidence and the deciding keyword, supply / return words, room labels and block kinds in five languages. A CIP layer without a direction word is not guessed. - MAX_DXF_BYTES defaults to 512 MB (a real layout was 188 MB); the refusal names the size and the variable. New `office` extra (openpyxl), also in `full`; uv.lock regenerated. 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
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
f0c50d772f
commit
9ab0ae1773
@@ -0,0 +1,57 @@
|
||||
"""Understanding drawings somebody else made (v1.6 track H).
|
||||
|
||||
One bulk read per drawing (:mod:`.snapshot`), multilingual vocabulary
|
||||
(:mod:`.vocab`) and label parsers (:mod:`.labels`) are the foundation; the
|
||||
readers built on them - the one-call report, the scale check, the line
|
||||
networks and takeoffs, the diff and the topology checks - add their modules
|
||||
here. Everything except the snapshot's live export is pure: records in,
|
||||
results out, the drawing never modified.
|
||||
"""
|
||||
|
||||
from engineering.understand.labels import (
|
||||
normalize_panel,
|
||||
parse_diameters,
|
||||
parse_electrical,
|
||||
parse_tag,
|
||||
plain,
|
||||
wiring_target,
|
||||
)
|
||||
from engineering.understand.snapshot import (
|
||||
EntityRecord,
|
||||
Pt,
|
||||
Snapshot,
|
||||
length_of,
|
||||
read_snapshot,
|
||||
records_from_doc,
|
||||
take_snapshot,
|
||||
)
|
||||
from engineering.understand.vocab import (
|
||||
DISCIPLINES,
|
||||
SERVICES,
|
||||
classify_layer,
|
||||
equipment_kind,
|
||||
room_label,
|
||||
supply_return,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"DISCIPLINES",
|
||||
"SERVICES",
|
||||
"EntityRecord",
|
||||
"Pt",
|
||||
"Snapshot",
|
||||
"classify_layer",
|
||||
"equipment_kind",
|
||||
"length_of",
|
||||
"normalize_panel",
|
||||
"parse_diameters",
|
||||
"parse_electrical",
|
||||
"parse_tag",
|
||||
"plain",
|
||||
"read_snapshot",
|
||||
"records_from_doc",
|
||||
"room_label",
|
||||
"supply_return",
|
||||
"take_snapshot",
|
||||
"wiring_target",
|
||||
]
|
||||
@@ -0,0 +1,302 @@
|
||||
"""Label parsers for foreign drawings: diameters, equipment tags, panels, electrical data.
|
||||
|
||||
Pure functions over one string each - no ezdxf, no I/O - so every rule here is
|
||||
tested on the spellings real drawings use, and the
|
||||
readers above it (the network, the takeoffs, the scale check) share one
|
||||
reading of a label instead of three.
|
||||
|
||||
What a label *says* is kept exactly as drawn. ``Ø51``, ``SMS51``, ``DN20``,
|
||||
``Ø25,4`` and ``Ø25`` are five different sizes: the parser normalises only the
|
||||
spelling of the diameter sign and the whitespace, never the number, and never
|
||||
equates two sizing systems (an SMS size and a DN size are different series; the
|
||||
drawing has to say they are the same pipe, not this module).
|
||||
|
||||
Two kinds of foreign-drawing noise are decoded before anything is parsed:
|
||||
|
||||
* **AutoCAD text codes.** ``%%c`` is AutoCAD's diameter sign, ``%%d`` the degree
|
||||
sign, ``%%p`` plus-minus; ``\\U+2205`` is a Unicode escape (AutoCAD's help
|
||||
lists U+2205 as its diameter symbol). MTEXT carries format runs
|
||||
(``{\\fArial|b0;Ø51}``, ``\\H2.5;``, ``\\P`` for a new paragraph) that are
|
||||
formatting, not content.
|
||||
* **Cyrillic look-alikes in tags.** A Russian-labelled P&ID typed on a Russian
|
||||
keyboard writes ``Т4100`` with a Cyrillic Т (U+0422); it must match the
|
||||
layout's Latin ``T4100``. Eleven capitals have a Latin twin of identical
|
||||
shape (Т М С Р А В Е К Н О Х) and are folded; a Cyrillic letter without one
|
||||
(П, Щ, Ш, ...) is left alone, so a Russian word never turns into a tag.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
__all__ = [
|
||||
"CYRILLIC_LOOKALIKES",
|
||||
"fold_lookalikes",
|
||||
"normalize_panel",
|
||||
"parse_diameters",
|
||||
"parse_electrical",
|
||||
"parse_tag",
|
||||
"plain",
|
||||
"wiring_target",
|
||||
]
|
||||
|
||||
#: Cyrillic capitals drawn identically to a Latin capital, and that capital.
|
||||
CYRILLIC_LOOKALIKES: dict[str, str] = {
|
||||
"\u0422": "T", # Т
|
||||
"\u041c": "M", # М
|
||||
"\u0421": "C", # С
|
||||
"\u0420": "P", # Р
|
||||
"\u0410": "A", # А
|
||||
"\u0412": "B", # В
|
||||
"\u0415": "E", # Е
|
||||
"\u041a": "K", # К
|
||||
"\u041d": "H", # Н
|
||||
"\u041e": "O", # О
|
||||
"\u0425": "X", # Х
|
||||
}
|
||||
_LOOKALIKE_TABLE = str.maketrans(CYRILLIC_LOOKALIKES)
|
||||
|
||||
#: ``%%`` codes that stand for a character (AutoCAD's control codes).
|
||||
_PERCENT_CHARS = {"c": "\u00d8", "d": "\u00b0", "p": "\u00b1"}
|
||||
#: MTEXT codes that carry an argument up to the next ``;`` and draw nothing.
|
||||
_ARG_CODES = frozenset("fFHWQTACcp")
|
||||
#: MTEXT on/off toggles (underline, overline, strike-through, column break).
|
||||
_TOGGLE_CODES = frozenset("LlOoKkN")
|
||||
|
||||
_UNICODE_ESCAPE = re.compile(r"\\[Uu]\+([0-9A-Fa-f]{4})")
|
||||
_PERCENT_CODE = re.compile(r"%%(\d{3}|[cdpCDP]|[uoUOkK])")
|
||||
|
||||
#: Every spelling of the diameter sign met in drawings: Ø (U+00D8), ø (U+00F8),
|
||||
#: ∅ (U+2205, AutoCAD's listed diameter symbol), ⌀ (U+2300,
|
||||
#: the Unicode DIAMETER SIGN).
|
||||
_DIAMETER_SIGNS = "\u00d8\u00f8\u2205\u2300"
|
||||
_NUMBER = r"\d+(?:[.,]\d+)?"
|
||||
_DIAMETER_RE = re.compile(
|
||||
rf"(?P<sign>[{_DIAMETER_SIGNS}])\s*(?P<sign_n>{_NUMBER})"
|
||||
rf"|(?<![A-Za-z])(?P<sms>SMS)\s*-?\s*(?P<sms_n>{_NUMBER})"
|
||||
rf"|(?<![A-Za-z])(?P<dn>DN)\s*-?\s*(?P<dn_n>\d+)"
|
||||
rf"|(?<![\d.,])(?P<a>{_NUMBER})\s*[xX\u00d7\u0445\u0425]\s*(?P<b>{_NUMBER})(?![.,]?\d)"
|
||||
r"(?!\s*(?:mm2|mm\u00b2|\u043c\u043c2|\u043c\u043c\u00b2))"
|
||||
rf"|(?<![\d.,])(?P<inch>\d+(?:[.,]\d+)?(?:\s*/\s*\d+)?)\s*(?:\"|''|\u201d|\u2033)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
#: A token that may be a tag: starts with a letter, keeps letters, digits, dots
|
||||
#: and hyphens (``CP-M82``, ``M47.1``).
|
||||
_TAG_TOKEN = re.compile(r"[A-Za-z\u0400-\u04ff][A-Za-z0-9\u0400-\u04ff.\-]*")
|
||||
#: An equipment tag: 1-4 letters, an optional second letter group after a
|
||||
#: hyphen (``CP-M82``), an optional hyphen, 1-5 digits, an optional letter
|
||||
#: suffix (``M41A``) and an optional ``.n`` (``M47.1``).
|
||||
_TAG_RE = re.compile(r"[A-Z]{1,4}(?:-[A-Z]{1,4})?-?\d{1,5}[A-Z]?(?:\.\d{1,3})?")
|
||||
#: Prefixes that look like a tag and are not one: nominal sizes (DN20, SMS51),
|
||||
#: a pressure rating (PN16) and an ingress-protection code (IP65).
|
||||
_NOT_TAG_PREFIXES = ("DN", "SMS", "PN", "IP")
|
||||
|
||||
_PANEL_RE = re.compile(
|
||||
r"(?<![A-Z0-9])([A-Z]{1,4})\s*[-_ ]?\s*((?:[A-Z]{1,3})?\d{1,4}[A-Z]?)(?![A-Z0-9])"
|
||||
)
|
||||
|
||||
_KW_RE = re.compile(
|
||||
r"(\d+(?:[.,]\d+)?)\s*(?:kw|\u043a\u0432\u0442)(?![a-z\u0430-\u044f])", re.IGNORECASE
|
||||
)
|
||||
_PHASE_RE = re.compile(
|
||||
r"(?<!\d)([13])\s*(?:\u0444\b\.?|ph\b\.?|phase|faz|~|/\s*n\b)", re.IGNORECASE
|
||||
)
|
||||
_VOLT_RE = re.compile(
|
||||
r"(?<![\d.,])(\d{2,4})\s*(?:v(?:ac|dc)?|\u0432)(?![a-z\u0430-\u044f])", re.IGNORECASE
|
||||
)
|
||||
_NEUTRAL_RE = re.compile(r"(?:\+|/)\s*n(?![a-z])", re.IGNORECASE)
|
||||
#: Words that say a load is driven by a frequency converter, EN/TR/RU/NL/DE.
|
||||
_VFD_WORDS = (
|
||||
"vfd",
|
||||
"vsd",
|
||||
"inverter",
|
||||
"frekans",
|
||||
"\u0447\u0430\u0441\u0442\u043e\u0442\u043d", # частотн(ый преобразователь)
|
||||
"\u0447\u0440\u043f", # ЧРП
|
||||
"frequentieregelaar",
|
||||
"frequenzumrichter",
|
||||
)
|
||||
#: Phrases that introduce the panel a load is wired to, EN/TR/RU/NL/DE.
|
||||
_WIRING_PHRASES = (
|
||||
"wiring to",
|
||||
"wired to",
|
||||
"cable to",
|
||||
"connected to",
|
||||
"\u043f\u043e\u0434\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0435 \u043a", # подключение к
|
||||
"\u043f\u043e\u0434\u043a\u043b\u044e\u0447\u0438\u0442\u044c \u043a", # подключить к
|
||||
"bedrading naar",
|
||||
"verdrahtung zu",
|
||||
"verdrahtung nach",
|
||||
)
|
||||
#: Turkish puts the panel first: ``CP1 panosuna`` ("to the CP1 panel").
|
||||
_WIRING_PHRASES_AFTER = ("panosuna", "panosundan")
|
||||
|
||||
|
||||
def plain(text: str | None) -> str:
|
||||
"""The text a reader sees: codes decoded, MTEXT formatting removed.
|
||||
|
||||
``\\U+XXXX`` becomes its character; ``%%c`` / ``%%d`` / ``%%p`` become Ø / ° /
|
||||
±, ``%%nnn`` the character with that code, ``%%%`` a percent sign, and the
|
||||
``%%u`` / ``%%o`` / ``%%k`` toggles disappear. In MTEXT, ``\\P`` is a line
|
||||
break, ``\\~`` a space, ``\\S a^b;`` / ``\\S a/b;`` the stacked ``a/b``,
|
||||
``\\\\`` ``\\{`` ``\\}`` the literal characters; every argument code
|
||||
(``\\f...;``, ``\\H...;``, ``\\C...;`` ...) and every toggle is dropped, and so
|
||||
are the grouping braces.
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
source = _UNICODE_ESCAPE.sub(lambda m: chr(int(m.group(1), 16)), str(text))
|
||||
source = source.replace("%%%", "\x00")
|
||||
|
||||
def percent(match: re.Match) -> str:
|
||||
code = match.group(1)
|
||||
if code.isdigit():
|
||||
return chr(int(code))
|
||||
return _PERCENT_CHARS.get(code.lower(), "")
|
||||
|
||||
source = _PERCENT_CODE.sub(percent, source).replace("\x00", "%")
|
||||
out: list[str] = []
|
||||
i, n = 0, len(source)
|
||||
while i < n:
|
||||
ch = source[i]
|
||||
if ch == "\\" and i + 1 < n:
|
||||
code = source[i + 1]
|
||||
if code in "\\{}":
|
||||
out.append(code)
|
||||
i += 2
|
||||
elif code == "P":
|
||||
out.append("\n")
|
||||
i += 2
|
||||
elif code == "~":
|
||||
out.append(" ")
|
||||
i += 2
|
||||
elif code == "S":
|
||||
end = source.find(";", i + 2)
|
||||
end = n if end < 0 else end
|
||||
out.append(re.sub(r"[\^#]", "/", source[i + 2 : end]))
|
||||
i = end + 1
|
||||
elif code in _ARG_CODES:
|
||||
end = source.find(";", i + 2)
|
||||
i = n if end < 0 else end + 1
|
||||
elif code in _TOGGLE_CODES:
|
||||
i += 2
|
||||
else:
|
||||
out.append(code)
|
||||
i += 2
|
||||
elif ch in "{}":
|
||||
i += 1
|
||||
else:
|
||||
out.append(ch)
|
||||
i += 1
|
||||
return "".join(out).strip()
|
||||
|
||||
|
||||
def parse_diameters(text: str | None) -> tuple[str, ...]:
|
||||
"""Every size label in ``text``, in order, each once, spelled one way.
|
||||
|
||||
``Ø51`` / ``%%c51`` / ``\\U+2205 51`` / ``⌀51`` / ``ø51`` -> ``"Ø51"``;
|
||||
``Ø25,4`` keeps its comma; ``SMS 51`` -> ``"SMS51"``; ``DN 20`` ->
|
||||
``"DN20"``; ``20 x 27`` / ``20х27`` (Cyrillic х) -> ``"20x27"``; ``1,5"`` ->
|
||||
``'1,5"'``. An ``a x b`` followed by mm² is a cable cross-section, not a
|
||||
pipe, and is skipped.
|
||||
"""
|
||||
found: list[str] = []
|
||||
for match in _DIAMETER_RE.finditer(plain(text)):
|
||||
if match.group("sign"):
|
||||
token = "\u00d8" + match.group("sign_n")
|
||||
elif match.group("sms"):
|
||||
token = "SMS" + match.group("sms_n")
|
||||
elif match.group("dn"):
|
||||
token = "DN" + match.group("dn_n")
|
||||
elif match.group("a"):
|
||||
token = f"{match.group('a')}x{match.group('b')}"
|
||||
else:
|
||||
token = re.sub(r"\s+", "", match.group("inch")) + '"'
|
||||
if token not in found:
|
||||
found.append(token)
|
||||
return tuple(found)
|
||||
|
||||
|
||||
def fold_lookalikes(text: str) -> str:
|
||||
"""Uppercase, with the eleven Cyrillic look-alike capitals made Latin."""
|
||||
return str(text).upper().translate(_LOOKALIKE_TABLE)
|
||||
|
||||
|
||||
def parse_tag(text: str | None) -> str | None:
|
||||
"""The first equipment tag in ``text``, uppercase, look-alikes folded.
|
||||
|
||||
``"Т4100"`` (Cyrillic Т) -> ``"T4100"``; ``"pump M41A"`` -> ``"M41A"``;
|
||||
``"CP-M82"`` stays ``"CP-M82"``. A size label (``DN20``, ``SMS51``), a
|
||||
pressure rating (``PN16``) or an IP code (``IP65``) is not a tag; a token
|
||||
that keeps a Cyrillic letter without a Latin twin is not one either.
|
||||
"""
|
||||
for match in _TAG_TOKEN.finditer(plain(text)):
|
||||
token = fold_lookalikes(match.group(0)).rstrip(".-")
|
||||
if not _TAG_RE.fullmatch(token):
|
||||
continue
|
||||
if re.match(r"[A-Z]+", token).group(0) in _NOT_TAG_PREFIXES:
|
||||
continue
|
||||
return token
|
||||
return None
|
||||
|
||||
|
||||
def normalize_panel(text: str | None) -> str | None:
|
||||
"""A panel name spelled one way: ``CP 1`` / ``CP1`` / ``cp-1`` -> ``CP-1``.
|
||||
|
||||
The letters, one hyphen, then the rest: ``CP M82`` -> ``CP-M82``. Cyrillic
|
||||
look-alikes are folded first (``СР-1`` typed in Cyrillic is ``CP-1``).
|
||||
Returns None when the text holds no panel-shaped token.
|
||||
"""
|
||||
if not text:
|
||||
return None
|
||||
match = _PANEL_RE.search(fold_lookalikes(plain(text)))
|
||||
if match is None:
|
||||
return None
|
||||
return f"{match.group(1)}-{match.group(2)}"
|
||||
|
||||
|
||||
def parse_electrical(text: str | None) -> dict:
|
||||
"""Power, phases, voltage, neutral and drive type stated in ``text``.
|
||||
|
||||
``"P = 18,5 кВт, 3 ф. 400 В + N + PE, ЧРП"`` ->
|
||||
``{"kw": 18.5, "phases": 3, "voltage": 400, "neutral": True, "vfd": True}``.
|
||||
What the text does not state stays ``None`` / ``False``: a power that is not
|
||||
written is never estimated.
|
||||
"""
|
||||
body = plain(text)
|
||||
lowered = body.lower()
|
||||
kw = _KW_RE.search(body)
|
||||
phases = _PHASE_RE.search(body)
|
||||
volts = _VOLT_RE.search(body)
|
||||
return {
|
||||
"kw": float(kw.group(1).replace(",", ".")) if kw else None,
|
||||
"phases": int(phases.group(1)) if phases else None,
|
||||
"voltage": int(volts.group(1)) if volts else None,
|
||||
"neutral": bool(_NEUTRAL_RE.search(body)),
|
||||
"vfd": any(word in lowered for word in _VFD_WORDS),
|
||||
}
|
||||
|
||||
|
||||
def wiring_target(text: str | None) -> str | None:
|
||||
"""The panel a wiring callout points to, normalised.
|
||||
|
||||
``"Wiring to CP1"`` -> ``"CP-1"``; ``"подключение к шкафу CP-1"`` ->
|
||||
``"CP-1"``; ``"CP 2 panosuna"`` -> ``"CP-2"`` (Turkish names the panel
|
||||
first). None when the text has no wiring phrase or no panel beside it.
|
||||
"""
|
||||
body = plain(text)
|
||||
lowered = body.lower()
|
||||
for phrase in _WIRING_PHRASES:
|
||||
at = lowered.find(phrase)
|
||||
if at >= 0:
|
||||
panel = normalize_panel(body[at + len(phrase) :])
|
||||
if panel is not None:
|
||||
return panel
|
||||
for phrase in _WIRING_PHRASES_AFTER:
|
||||
at = lowered.find(phrase)
|
||||
if at > 0:
|
||||
matches = list(_PANEL_RE.finditer(fold_lookalikes(body[:at])))
|
||||
if matches:
|
||||
return f"{matches[-1].group(1)}-{matches[-1].group(2)}"
|
||||
return None
|
||||
@@ -0,0 +1,420 @@
|
||||
"""One bulk read of a drawing into frozen records - the input of every track H reader.
|
||||
|
||||
Reading a 35,000-entity P&ID one COM call at a time takes minutes; reading its
|
||||
DXF takes about a second. So every analysis in ``engineering/understand/`` runs
|
||||
on a :class:`Snapshot`: the drawing read once, into small frozen
|
||||
:class:`EntityRecord` s, and never touched again.
|
||||
|
||||
Where the snapshot comes from:
|
||||
|
||||
* **A DXF file** - :func:`read_snapshot`, ``ezdxf.readfile`` after the size
|
||||
check (``MAX_DXF_BYTES``, 512 MB by default; the refusal names the size and
|
||||
the variable).
|
||||
* **The headless engine's current document** - :func:`records_from_doc` on the
|
||||
in-memory ``ezdxf`` document, inside the backend's own lock.
|
||||
* **The live engine's current document** - ``Document.Export(<tmp>/snap, "DXF",
|
||||
<selection set of all>)``, then :func:`read_snapshot` on the file, which is
|
||||
deleted afterwards. Measured on AutoCAD 2026 (``25.1s (LMS Tech)``) on
|
||||
2026-09-24, in a new scratch document closed afterwards:
|
||||
|
||||
- the extension argument must be ``"DXF"``; ``".dxf"`` fails with *Invalid
|
||||
argument*; the file is written as ``snap.DXF`` (upper-case extension);
|
||||
- the document keeps its name (``Export`` is not ``SaveAs``), and an unsaved
|
||||
document exports;
|
||||
- **model space, every paper-space layout and every block definition reach
|
||||
the snapshot** - a LINE in Layout1 was read back in Layout1, and an unused
|
||||
block definition was in the BLOCKS section;
|
||||
- **the selection set does not narrow a DXF export**: an empty selection set
|
||||
wrote the same model space, paper space and blocks as the set of all. The
|
||||
set of all is passed anyway, because the ActiveX signature requires one.
|
||||
- xrefs were not measured, so nothing is claimed for the live engine;
|
||||
headless, ezdxf reads the host file only. Either way an xref's own
|
||||
entities are not read in 1.6 - the host drawing is - and
|
||||
``drawing_understand``'s docstring says so.
|
||||
|
||||
A DWG path on the live engine is opened read-only, exported the same way and
|
||||
closed without saving.
|
||||
|
||||
Records hold what the readers need and nothing else: WCS points (OCS is
|
||||
translated here with ``backends/ocs.py``, so a mirrored ARC or LWPOLYLINE is
|
||||
where it looks), plain text (``labels.plain``: AutoCAD codes and MTEXT format
|
||||
runs decoded), block names and attributes, a WCS box. Block definitions are not
|
||||
exploded: an INSERT is one record with its block name, attributes and box.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import math
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import ezdxf
|
||||
from ezdxf import bbox as ezbbox
|
||||
|
||||
import config
|
||||
from backends.capability import UnsupportedCapabilityError
|
||||
from backends.ocs import to_wcs_2d, wcs_arc_angles, wcs_bulge
|
||||
from engineering.measure import polygon_area_perimeter
|
||||
from engineering.understand.labels import plain
|
||||
|
||||
__all__ = [
|
||||
"EntityRecord",
|
||||
"Pt",
|
||||
"Snapshot",
|
||||
"length_of",
|
||||
"read_snapshot",
|
||||
"records_from_doc",
|
||||
"take_snapshot",
|
||||
]
|
||||
|
||||
Pt = tuple[float, float]
|
||||
|
||||
#: ezdxf's placeholder for an extents header nobody computed.
|
||||
_UNSET_EXTENT = 1e19
|
||||
#: The live export's base name inside its temporary folder.
|
||||
_SNAP_BASE = "snap"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EntityRecord:
|
||||
"""One entity of a layout, in WCS, as the readers need it."""
|
||||
|
||||
handle: str
|
||||
type: str
|
||||
layer: str
|
||||
space: str
|
||||
points: tuple[Pt, ...] = ()
|
||||
bulges: tuple[float, ...] = ()
|
||||
closed: bool = False
|
||||
radius: float | None = None
|
||||
angles: tuple[float, float] | None = None
|
||||
text: str | None = None
|
||||
height: float | None = None
|
||||
block: str | None = None
|
||||
attribs: tuple[tuple[str, str], ...] = ()
|
||||
bbox: tuple[float, float, float, float] | None = None
|
||||
rotation: float = 0.0
|
||||
scale: tuple[float, float] = (1.0, 1.0)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Snapshot:
|
||||
"""A whole drawing, read once."""
|
||||
|
||||
source: str
|
||||
insunits: int
|
||||
extmin: Pt | None
|
||||
extmax: Pt | None
|
||||
layers: dict[str, dict]
|
||||
layouts: tuple[str, ...]
|
||||
records: tuple[EntityRecord, ...]
|
||||
|
||||
|
||||
def _xy(point) -> Pt:
|
||||
return (float(point[0]), float(point[1]))
|
||||
|
||||
|
||||
def _box(entity) -> tuple[float, float, float, float] | None:
|
||||
"""The entity's WCS box. An INSERT's box is its block geometry only: the
|
||||
ATTRIB texts are left out, because a TAG lettered beside a tank would push
|
||||
the tank's box past its body (the rule `engineering/pid/graph.py` keeps)."""
|
||||
try:
|
||||
if entity.dxftype() == "INSERT":
|
||||
box = ezbbox.extents(entity.virtual_entities(), fast=True)
|
||||
else:
|
||||
box = ezbbox.extents([entity], fast=True)
|
||||
except Exception: # a proxy or a malformed entity has no box; the record stays
|
||||
return None
|
||||
if not box.has_data:
|
||||
return None
|
||||
return (
|
||||
float(box.extmin.x),
|
||||
float(box.extmin.y),
|
||||
float(box.extmax.x),
|
||||
float(box.extmax.y),
|
||||
)
|
||||
|
||||
|
||||
def _header_point(doc, name: str) -> Pt | None:
|
||||
try:
|
||||
value = doc.header.get(name)
|
||||
except Exception:
|
||||
return None
|
||||
if value is None or abs(float(value[0])) >= _UNSET_EXTENT:
|
||||
return None
|
||||
return _xy(value)
|
||||
|
||||
|
||||
def _extrusion(entity):
|
||||
"""The entity's extrusion, or +Z for a type that has none (MULTILEADER, ...)."""
|
||||
if entity.dxf.is_supported("extrusion"):
|
||||
return entity.dxf.get("extrusion", (0.0, 0.0, 1.0))
|
||||
return (0.0, 0.0, 1.0)
|
||||
|
||||
|
||||
def _polyline(entity) -> tuple[tuple[Pt, ...], tuple[float, ...], bool]:
|
||||
extrusion = _extrusion(entity)
|
||||
if entity.dxftype() == "LWPOLYLINE":
|
||||
elevation = float(entity.dxf.get("elevation", 0.0))
|
||||
rows = [(float(x), float(y), float(b)) for x, y, b in entity.get_points("xyb")]
|
||||
points = tuple(tuple(to_wcs_2d(extrusion, x, y, elevation)) for x, y, _ in rows)
|
||||
bulges = tuple(wcs_bulge(extrusion, b) for _, _, b in rows)
|
||||
return points, bulges, bool(entity.closed)
|
||||
vertices = list(entity.vertices)
|
||||
if entity.is_2d_polyline:
|
||||
points = tuple(
|
||||
tuple(to_wcs_2d(extrusion, v.dxf.location.x, v.dxf.location.y, v.dxf.location.z))
|
||||
for v in vertices
|
||||
)
|
||||
bulges = tuple(wcs_bulge(extrusion, v.dxf.get("bulge", 0.0)) for v in vertices)
|
||||
else: # a 3D polyline is WCS already; its z is dropped
|
||||
points = tuple(_xy(v.dxf.location) for v in vertices)
|
||||
bulges = tuple(0.0 for _ in vertices)
|
||||
return points, bulges, bool(entity.is_closed)
|
||||
|
||||
|
||||
def _record(entity, space: str) -> EntityRecord | None:
|
||||
kind = entity.dxftype()
|
||||
dxf = entity.dxf
|
||||
base = {
|
||||
"handle": str(dxf.handle),
|
||||
"type": kind,
|
||||
"layer": str(dxf.get("layer", "0")),
|
||||
"space": space,
|
||||
"bbox": _box(entity),
|
||||
}
|
||||
extrusion = _extrusion(entity)
|
||||
if kind == "LINE":
|
||||
return EntityRecord(**base, points=(_xy(dxf.start), _xy(dxf.end)))
|
||||
if kind in ("LWPOLYLINE", "POLYLINE"):
|
||||
if kind == "POLYLINE" and (entity.is_poly_face_mesh or entity.is_polygon_mesh):
|
||||
return EntityRecord(**base)
|
||||
points, bulges, closed = _polyline(entity)
|
||||
return EntityRecord(**base, points=points, bulges=bulges, closed=closed)
|
||||
if kind in ("ARC", "CIRCLE"):
|
||||
centre = tuple(to_wcs_2d(extrusion, dxf.center.x, dxf.center.y, dxf.center.z))
|
||||
angles = None
|
||||
if kind == "ARC":
|
||||
angles = wcs_arc_angles(
|
||||
extrusion, dxf.center, dxf.radius, dxf.start_angle, dxf.end_angle
|
||||
)
|
||||
return EntityRecord(**base, points=(centre,), radius=float(dxf.radius), angles=angles)
|
||||
if kind == "TEXT":
|
||||
insert = dxf.insert
|
||||
return EntityRecord(
|
||||
**base,
|
||||
points=(tuple(to_wcs_2d(extrusion, insert.x, insert.y, insert.z)),),
|
||||
text=plain(dxf.get("text", "")),
|
||||
height=float(dxf.get("height", 0.0)),
|
||||
rotation=float(dxf.get("rotation", 0.0)),
|
||||
)
|
||||
if kind == "MTEXT":
|
||||
return EntityRecord(
|
||||
**base,
|
||||
points=(_xy(dxf.insert),),
|
||||
text=plain(entity.text),
|
||||
height=float(dxf.get("char_height", 0.0)),
|
||||
rotation=float(entity.get_rotation()),
|
||||
)
|
||||
if kind == "INSERT":
|
||||
insert = dxf.insert
|
||||
attribs = tuple((str(a.dxf.tag), plain(a.dxf.get("text", ""))) for a in entity.attribs)
|
||||
return EntityRecord(
|
||||
**base,
|
||||
points=(tuple(to_wcs_2d(extrusion, insert.x, insert.y, insert.z)),),
|
||||
block=str(dxf.name),
|
||||
attribs=attribs,
|
||||
rotation=float(dxf.get("rotation", 0.0)),
|
||||
scale=(float(dxf.get("xscale", 1.0)), float(dxf.get("yscale", 1.0))),
|
||||
)
|
||||
if kind == "MULTILEADER":
|
||||
context = entity.context
|
||||
points: list[Pt] = []
|
||||
for leader in context.leaders:
|
||||
for line in leader.lines:
|
||||
points.extend(_xy(v) for v in line.vertices)
|
||||
points.append(_xy(leader.last_leader_point))
|
||||
text = None
|
||||
height = None
|
||||
if context.mtext is not None:
|
||||
text = plain(context.mtext.default_content)
|
||||
points.append(_xy(context.mtext.insert))
|
||||
height = float(context.char_height)
|
||||
return EntityRecord(**base, points=tuple(points), text=text, height=height)
|
||||
if kind == "DIMENSION":
|
||||
override = str(dxf.get("text", "") or "")
|
||||
text = plain(override) if override not in ("", "<>") else None
|
||||
middle = dxf.get("text_midpoint", dxf.defpoint)
|
||||
points = (_xy(dxf.defpoint), tuple(to_wcs_2d(extrusion, middle.x, middle.y, middle.z)))
|
||||
return EntityRecord(**base, points=points, text=text)
|
||||
if kind == "LEADER":
|
||||
return EntityRecord(**base, points=tuple(_xy(v) for v in entity.vertices))
|
||||
if kind == "POINT":
|
||||
return EntityRecord(**base, points=(_xy(dxf.location),))
|
||||
if kind == "SPLINE":
|
||||
return EntityRecord(**base, points=tuple(_xy(p) for p in entity.control_points))
|
||||
if kind == "ELLIPSE":
|
||||
return EntityRecord(**base, points=(_xy(dxf.center),))
|
||||
return EntityRecord(**base)
|
||||
|
||||
|
||||
def records_from_doc(doc, *, source: str) -> Snapshot:
|
||||
"""Every entity of every layout of an ``ezdxf`` document, as records.
|
||||
|
||||
Model space first, then each paper layout in tab order. ``insunits`` is
|
||||
``$INSUNITS`` (0 when absent); ``extmin`` / ``extmax`` are the declared
|
||||
``$EXTMIN`` / ``$EXTMAX``, None when the header holds ezdxf's "never
|
||||
computed" placeholder. Nothing in ``doc`` is modified.
|
||||
"""
|
||||
records: list[EntityRecord] = []
|
||||
layouts = tuple(doc.layouts.names_in_taborder())
|
||||
for name in layouts:
|
||||
for entity in doc.layouts.get(name):
|
||||
record = _record(entity, name)
|
||||
if record is not None:
|
||||
records.append(record)
|
||||
layers = {
|
||||
str(layer.dxf.name): {
|
||||
"color": int(layer.dxf.get("color", 7)),
|
||||
"linetype": str(layer.dxf.get("linetype", "Continuous")),
|
||||
}
|
||||
for layer in doc.layers
|
||||
}
|
||||
return Snapshot(
|
||||
source=source,
|
||||
insunits=int(doc.header.get("$INSUNITS", 0) or 0),
|
||||
extmin=_header_point(doc, "$EXTMIN"),
|
||||
extmax=_header_point(doc, "$EXTMAX"),
|
||||
layers=layers,
|
||||
layouts=layouts,
|
||||
records=tuple(records),
|
||||
)
|
||||
|
||||
|
||||
def _read(path: str, *, source: str) -> Snapshot:
|
||||
try:
|
||||
doc = ezdxf.readfile(path)
|
||||
except (OSError, ezdxf.DXFError) as exc:
|
||||
raise ValueError(f"{path} is not a readable DXF file: {exc}") from exc
|
||||
return records_from_doc(doc, source=source)
|
||||
|
||||
|
||||
def read_snapshot(path: str) -> Snapshot:
|
||||
"""The snapshot of a DXF file on disk, after the size check.
|
||||
|
||||
Refuses a missing file (``FileNotFoundError``) and a file larger than
|
||||
``MAX_DXF_BYTES`` (``ValueError`` naming the size, the limit and the
|
||||
variable; 0 disables the check) before reading a byte of it.
|
||||
"""
|
||||
target = Path(path)
|
||||
if not target.is_file():
|
||||
raise FileNotFoundError(f"{path}: no such file")
|
||||
limit = int(config.settings.max_dxf_bytes)
|
||||
size = target.stat().st_size
|
||||
if limit > 0 and size > limit:
|
||||
raise ValueError(
|
||||
f"{path} is {size:,} bytes ({size / 2**20:.1f} MB), over the MAX_DXF_BYTES "
|
||||
f"limit of {limit:,} bytes ({limit / 2**20:.0f} MB). Raise MAX_DXF_BYTES "
|
||||
"(0 disables the check) to read it."
|
||||
)
|
||||
return _read(str(target), source=str(target))
|
||||
|
||||
|
||||
def _com_export(backend, path: str | None) -> tuple[str, str, str]:
|
||||
"""(snapshot file, its temporary folder, source) - runs on the COM thread."""
|
||||
from backends import com_backend as com
|
||||
|
||||
app = com._acad_app()
|
||||
opened = None
|
||||
if path is None:
|
||||
doc = com._acad_doc()
|
||||
else:
|
||||
opened = app.Documents.Open(str(path), True) # read-only
|
||||
doc = opened
|
||||
folder = tempfile.mkdtemp(prefix="acadmcp_snap_")
|
||||
try:
|
||||
name = str(backend._wait_out_rejected_call(lambda: doc.Name))
|
||||
selection = doc.SelectionSets.Add(f"_SNAP_{uuid.uuid4().hex[:8]}")
|
||||
try:
|
||||
selection.Select(com._AC_SELECTION_SET_ALL)
|
||||
# "DXF", never ".dxf": the dotted spelling is refused with
|
||||
# "Invalid argument" (measured, AutoCAD 2026).
|
||||
doc.Export(os.path.join(folder, _SNAP_BASE), "DXF", selection)
|
||||
finally:
|
||||
try:
|
||||
selection.Delete()
|
||||
except Exception: # a selection set left behind is harmless; the export is not
|
||||
pass
|
||||
except Exception:
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
raise
|
||||
finally:
|
||||
if opened is not None:
|
||||
opened.Close(False)
|
||||
written = sorted(Path(folder).glob(f"{_SNAP_BASE}.*"))
|
||||
if not written:
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
raise RuntimeError(f"AutoCAD reported the export of {name} done but wrote no file")
|
||||
return str(written[0]), folder, f"live:{name}"
|
||||
|
||||
|
||||
async def take_snapshot(backend, path: str | None = None) -> Snapshot:
|
||||
"""The snapshot of ``path``, or of the backend's current document.
|
||||
|
||||
``path`` given: a ``.dxf`` is read directly (on either engine); a ``.dwg``
|
||||
needs the live engine, which opens it read-only, exports it and closes it -
|
||||
headlessly it is refused with the ``dwg`` capability. ``path`` None: the
|
||||
headless engine reads its in-memory document under its own lock; the live
|
||||
engine exports the active document to a temporary DXF (see the module
|
||||
docstring for what was measured) and deletes it afterwards. The snapshot of
|
||||
a live document is not size-checked: it is already in AutoCAD's memory.
|
||||
"""
|
||||
if path is not None:
|
||||
suffix = Path(path).suffix.lower()
|
||||
if suffix == ".dxf":
|
||||
return await asyncio.to_thread(read_snapshot, str(path))
|
||||
if suffix != ".dwg":
|
||||
raise ValueError(f"{path}: a snapshot reads a .dxf or a .dwg, not {suffix or 'this'}")
|
||||
if not Path(path).is_file():
|
||||
raise FileNotFoundError(f"{path}: no such file")
|
||||
if backend.name != "com":
|
||||
raise UnsupportedCapabilityError(
|
||||
"dwg",
|
||||
f"{path} is a DWG, which the headless engine cannot read. Export it to "
|
||||
"DXF (DXFOUT or SAVEAS in AutoCAD) and pass the .dxf, or run the server "
|
||||
"on the live engine (AUTOCAD_MCP_BACKEND=com).",
|
||||
)
|
||||
if backend.name == "com":
|
||||
written, folder, source = await backend._run(_com_export, backend, path)
|
||||
if path is not None:
|
||||
await backend._ensure_document_state()
|
||||
try:
|
||||
return await asyncio.to_thread(_read, written, source=source)
|
||||
finally:
|
||||
shutil.rmtree(folder, ignore_errors=True)
|
||||
source = backend._doc_path or "memory:untitled"
|
||||
return await backend._async(lambda: records_from_doc(backend._doc, source=source))
|
||||
|
||||
|
||||
def length_of(rec: EntityRecord) -> float:
|
||||
"""Drawn length in drawing units: LINE, LWPOLYLINE / POLYLINE with their
|
||||
bulges (``engineering.measure``, the same arithmetic ``analysis_measure_entity``
|
||||
reports), ARC, CIRCLE. Every other type is 0.0.
|
||||
"""
|
||||
if rec.type == "LINE" and len(rec.points) == 2:
|
||||
return math.dist(rec.points[0], rec.points[1])
|
||||
if rec.type in ("LWPOLYLINE", "POLYLINE") and len(rec.points) >= 2:
|
||||
bulges = rec.bulges or tuple(0.0 for _ in rec.points)
|
||||
vertices = [(x, y, b) for (x, y), b in zip(rec.points, bulges, strict=False)]
|
||||
return polygon_area_perimeter(vertices, rec.closed)[1]
|
||||
if rec.type == "ARC" and rec.radius is not None and rec.angles is not None:
|
||||
sweep = (rec.angles[1] - rec.angles[0]) % 360.0
|
||||
return rec.radius * math.radians(sweep)
|
||||
if rec.type == "CIRCLE" and rec.radius is not None:
|
||||
return 2.0 * math.pi * rec.radius
|
||||
return 0.0
|
||||
@@ -0,0 +1,451 @@
|
||||
"""Multilingual keyword tables for reading foreign drawings: EN / TR / RU / NL / DE.
|
||||
|
||||
A foreign drawing says what a layer carries in its name and in the texts on it,
|
||||
in whatever language the office that drew it speaks: a P&ID may name
|
||||
its piping layers in English and Dutch (``P_cipsupplyline``,
|
||||
``KLEPPEN``) and label its lines in Russian. These tables are what lets the
|
||||
reader tell a CIP supply line from a product line without a human.
|
||||
|
||||
They are **vocabulary, not a standard**: words a drafter writes, each with the
|
||||
language it belongs to. A classification made from them carries a confidence
|
||||
(0.9 for a direct keyword, 0.6 for weaker evidence) and the keyword that
|
||||
decided it, so a reader can always see why.
|
||||
|
||||
Matching is done on a *folded* spelling - casefolded, diacritics stripped,
|
||||
Turkish dotless ı made i - so ``DÖNÜŞ``, ``dönüş`` and ``donus`` are one word.
|
||||
Cyrillic keeps its letters (``подача`` stays ``подача``). A keyword of five
|
||||
letters or more matches anywhere in a layer name written without separators
|
||||
(``cipsupplyline`` holds ``supply``); a shorter one must start a word, so
|
||||
``air`` finds ``air_line`` and not ``stairs``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
|
||||
__all__ = [
|
||||
"DISCIPLINES",
|
||||
"EQUIPMENT_KINDS",
|
||||
"LANGUAGES",
|
||||
"SERVICES",
|
||||
"classify_layer",
|
||||
"equipment_kind",
|
||||
"fold",
|
||||
"room_label",
|
||||
"supply_return",
|
||||
]
|
||||
|
||||
LANGUAGES = ("en", "tr", "ru", "nl", "de")
|
||||
|
||||
SERVICES: tuple[str, ...] = (
|
||||
"product",
|
||||
"cip_supply",
|
||||
"cip_return",
|
||||
"glycol_supply",
|
||||
"glycol_return",
|
||||
"steam",
|
||||
"condensate",
|
||||
"cold_water",
|
||||
"hot_water",
|
||||
"ice_water",
|
||||
"compressed_air",
|
||||
"drain",
|
||||
"electrical",
|
||||
"unknown",
|
||||
)
|
||||
|
||||
DISCIPLINES: tuple[str, ...] = (
|
||||
"piping",
|
||||
"electrical",
|
||||
"architecture",
|
||||
"equipment",
|
||||
"annotation",
|
||||
"structure",
|
||||
"unknown",
|
||||
)
|
||||
|
||||
EQUIPMENT_KINDS: tuple[str, ...] = (
|
||||
"tank",
|
||||
"pump",
|
||||
"valve",
|
||||
"reducer",
|
||||
"motor",
|
||||
"agitator",
|
||||
"panel",
|
||||
)
|
||||
|
||||
#: Confidence of a classification decided by one direct keyword.
|
||||
DIRECT = 0.9
|
||||
#: Confidence of a classification from weaker evidence: a CIP or glycol layer
|
||||
#: whose name says no direction, a generic water layer.
|
||||
WEAK = 0.6
|
||||
#: A keyword this long matches anywhere in a separator-free name.
|
||||
_SUBSTRING_MIN = 5
|
||||
|
||||
# (keyword, language) rows. Order matters inside a table: the first row that
|
||||
# matches decides, so a longer or more specific word comes before a word it
|
||||
# contains.
|
||||
|
||||
_SUPPLY_WORDS = (
|
||||
("supply", "en"),
|
||||
("besleme", "tr"),
|
||||
("gidis", "tr"),
|
||||
("подач", "ru"), # подача: supply
|
||||
("подающ", "ru"), # подающий: supplying
|
||||
("aanvoer", "nl"),
|
||||
("vorlauf", "de"),
|
||||
)
|
||||
_RETURN_WORDS = (
|
||||
("return", "en"),
|
||||
("donus", "tr"),
|
||||
("возврат", "ru"), # return
|
||||
("обрат", "ru"), # обратка / обратный: return
|
||||
("retour", "nl"),
|
||||
("rucklauf", "de"),
|
||||
)
|
||||
|
||||
_CIP_WORDS = (
|
||||
("cip", "en"),
|
||||
("мойк", "ru"), # мойка: washing, CIP
|
||||
)
|
||||
_GLYCOL_WORDS = (
|
||||
("glycol", "en"),
|
||||
("glikol", "tr"),
|
||||
("гликол", "ru"), # гликоль: glycol
|
||||
("glykol", "de"),
|
||||
)
|
||||
|
||||
#: Services a layer name can state directly (CIP and glycol are decided with
|
||||
#: their direction words above).
|
||||
_SERVICE_WORDS = (
|
||||
("product", "product", "en"),
|
||||
("product", "urun", "tr"),
|
||||
("product", "продукт", "ru"), # product
|
||||
("product", "produkt", "de"),
|
||||
("steam", "steam", "en"),
|
||||
("steam", "buhar", "tr"),
|
||||
("steam", "пар", "ru"), # steam
|
||||
("steam", "stoom", "nl"),
|
||||
("steam", "dampf", "de"),
|
||||
("condensate", "condensate", "en"),
|
||||
("condensate", "kondensat", "de"),
|
||||
("condensate", "kondens", "tr"),
|
||||
("condensate", "конденсат", "ru"), # condensate
|
||||
("condensate", "condensaat", "nl"),
|
||||
("ice_water", "ice water", "en"),
|
||||
("ice_water", "buzlu su", "tr"),
|
||||
("ice_water", "ледяная вода", "ru"),
|
||||
("ice_water", "ijswater", "nl"),
|
||||
("ice_water", "eiswasser", "de"),
|
||||
("hot_water", "hot water", "en"),
|
||||
("hot_water", "sicak su", "tr"),
|
||||
("hot_water", "горячая вода", "ru"),
|
||||
("hot_water", "warm water", "nl"),
|
||||
("hot_water", "warmwasser", "de"),
|
||||
("cold_water", "cold water", "en"),
|
||||
("cold_water", "soguk su", "tr"),
|
||||
("cold_water", "холодная вода", "ru"),
|
||||
("cold_water", "koud water", "nl"),
|
||||
("cold_water", "kaltwasser", "de"),
|
||||
("compressed_air", "compressed air", "en"),
|
||||
("compressed_air", "basincli hava", "tr"),
|
||||
("compressed_air", "сжатый воздух", "ru"),
|
||||
("compressed_air", "perslucht", "nl"),
|
||||
("compressed_air", "druckluft", "de"),
|
||||
("compressed_air", "air", "en"),
|
||||
("compressed_air", "hava", "tr"),
|
||||
("compressed_air", "воздух", "ru"), # air
|
||||
("drain", "drain", "en"),
|
||||
("drain", "sewer", "en"),
|
||||
("drain", "drenaj", "tr"),
|
||||
("drain", "atik su", "tr"),
|
||||
("drain", "дренаж", "ru"), # drainage
|
||||
("drain", "канализац", "ru"), # канализация: sewer
|
||||
("drain", "riool", "nl"),
|
||||
("drain", "abwasser", "de"),
|
||||
)
|
||||
|
||||
_ELECTRICAL_WORDS = (
|
||||
("electrical", "en"),
|
||||
("electric", "en"),
|
||||
("power", "en"),
|
||||
("cable", "en"),
|
||||
("elektrik", "tr"),
|
||||
("kablo", "tr"),
|
||||
("электр", "ru"), # электрика: electrical
|
||||
("кабел", "ru"), # кабель: cable
|
||||
("elektra", "nl"),
|
||||
("kabel", "nl"),
|
||||
("elektro", "de"),
|
||||
)
|
||||
|
||||
#: Water words that name no temperature: a piping layer of unknown service.
|
||||
_WATER_WORDS = (
|
||||
("water", "en"),
|
||||
("вода", "ru"), # water
|
||||
("wasser", "de"),
|
||||
)
|
||||
|
||||
_DISCIPLINE_WORDS = (
|
||||
("piping", "piping", "en"),
|
||||
("piping", "pipe", "en"),
|
||||
("piping", "valve", "en"),
|
||||
("piping", "boru", "tr"),
|
||||
("piping", "tesisat", "tr"),
|
||||
("piping", "vana", "tr"),
|
||||
("piping", "труб", "ru"), # трубопровод: pipeline
|
||||
("piping", "клапан", "ru"), # valve
|
||||
("piping", "арматур", "ru"), # арматура: valves and fittings
|
||||
("piping", "leiding", "nl"),
|
||||
("piping", "klep", "nl"),
|
||||
("piping", "afsluiter", "nl"),
|
||||
("piping", "rohr", "de"),
|
||||
("piping", "ventil", "de"),
|
||||
("piping", "armatur", "de"),
|
||||
("architecture", "wall", "en"),
|
||||
("architecture", "door", "en"),
|
||||
("architecture", "window", "en"),
|
||||
("architecture", "duvar", "tr"),
|
||||
("architecture", "kapi", "tr"),
|
||||
("architecture", "pencere", "tr"),
|
||||
("architecture", "стен", "ru"), # стена: wall
|
||||
("architecture", "двер", "ru"), # дверь: door
|
||||
("architecture", "окн", "ru"), # окно: window
|
||||
("architecture", "muur", "nl"),
|
||||
("architecture", "muren", "nl"),
|
||||
("architecture", "deur", "nl"),
|
||||
("architecture", "raam", "nl"),
|
||||
("architecture", "wand", "de"),
|
||||
("architecture", "fenster", "de"),
|
||||
("equipment", "equipment", "en"),
|
||||
("equipment", "tank", "en"),
|
||||
("equipment", "pump", "en"),
|
||||
("equipment", "ekipman", "tr"),
|
||||
("equipment", "pompa", "tr"),
|
||||
("equipment", "оборудован", "ru"), # оборудование: equipment
|
||||
("equipment", "насос", "ru"), # pump
|
||||
("equipment", "apparatuur", "nl"),
|
||||
("equipment", "pomp", "nl"),
|
||||
("equipment", "ausrustung", "de"),
|
||||
("equipment", "pumpe", "de"),
|
||||
("annotation", "text", "en"),
|
||||
("annotation", "dim", "en"),
|
||||
("annotation", "anno", "en"),
|
||||
("annotation", "note", "en"),
|
||||
("annotation", "title", "en"),
|
||||
("annotation", "yazi", "tr"),
|
||||
("annotation", "olcu", "tr"),
|
||||
("annotation", "antet", "tr"),
|
||||
("annotation", "текст", "ru"), # text
|
||||
("annotation", "размер", "ru"), # dimension
|
||||
("annotation", "штамп", "ru"), # title block
|
||||
("annotation", "tekst", "nl"),
|
||||
("annotation", "maat", "nl"),
|
||||
("annotation", "bemassung", "de"),
|
||||
("structure", "column", "en"),
|
||||
("structure", "beam", "en"),
|
||||
("structure", "steel", "en"),
|
||||
("structure", "grid", "en"),
|
||||
("structure", "kolon", "tr"),
|
||||
("structure", "kiris", "tr"),
|
||||
("structure", "celik", "tr"),
|
||||
("structure", "aks", "tr"),
|
||||
("structure", "колонн", "ru"), # колонна: column
|
||||
("structure", "балк", "ru"), # балка: beam
|
||||
("structure", "kolom", "nl"),
|
||||
("structure", "balk", "nl"),
|
||||
("structure", "staal", "nl"),
|
||||
("structure", "stutze", "de"),
|
||||
("structure", "trager", "de"),
|
||||
("structure", "stahl", "de"),
|
||||
)
|
||||
|
||||
_ROOM_WORDS = (
|
||||
("room", "en"),
|
||||
("oda", "tr"),
|
||||
("mahal", "tr"),
|
||||
("помещ", "ru"), # помещение: room
|
||||
("комнат", "ru"), # комната: room
|
||||
("ruimte", "nl"),
|
||||
("kamer", "nl"),
|
||||
("raum", "de"),
|
||||
("zimmer", "de"),
|
||||
)
|
||||
|
||||
_EQUIPMENT_KIND_WORDS = (
|
||||
("reducer", "reducer", "en"),
|
||||
("reducer", "reduksiyon", "tr"),
|
||||
("reducer", "переход", "ru"), # reducer
|
||||
("reducer", "verloop", "nl"),
|
||||
("reducer", "reduzier", "de"),
|
||||
("valve", "valve", "en"),
|
||||
("valve", "vlv", "en"),
|
||||
("valve", "vana", "tr"),
|
||||
("valve", "клапан", "ru"), # valve
|
||||
("valve", "кран", "ru"), # valve (cock)
|
||||
("valve", "klep", "nl"),
|
||||
("valve", "afsluiter", "nl"),
|
||||
("valve", "ventil", "de"),
|
||||
("valve", "armatur", "de"),
|
||||
("agitator", "agitator", "en"),
|
||||
("agitator", "mixer", "en"),
|
||||
("agitator", "karistirici", "tr"),
|
||||
("agitator", "мешалк", "ru"), # мешалка: agitator
|
||||
("agitator", "roerwerk", "nl"),
|
||||
("agitator", "ruhrwerk", "de"),
|
||||
("pump", "pump", "en"),
|
||||
("pump", "pompa", "tr"),
|
||||
("pump", "насос", "ru"), # pump
|
||||
("pump", "pomp", "nl"),
|
||||
("pump", "pumpe", "de"),
|
||||
("tank", "tank", "en"),
|
||||
("tank", "vessel", "en"),
|
||||
("tank", "емкост", "ru"), # емкость: vessel
|
||||
("tank", "бак", "ru"), # tank
|
||||
("tank", "vat", "nl"),
|
||||
("tank", "behalter", "de"),
|
||||
("motor", "motor", "en"),
|
||||
("motor", "мотор", "ru"), # motor
|
||||
("motor", "электродвиг", "ru"), # электродвигатель: electric motor
|
||||
("panel", "panel", "en"),
|
||||
("panel", "cabinet", "en"),
|
||||
("panel", "pano", "tr"),
|
||||
("panel", "шкаф", "ru"), # cabinet
|
||||
("panel", "щит", "ru"), # switchboard
|
||||
("panel", "schakelkast", "nl"),
|
||||
("panel", "kast", "nl"),
|
||||
("panel", "schrank", "de"),
|
||||
)
|
||||
|
||||
|
||||
def fold(text: str | None) -> str:
|
||||
"""Casefolded, diacritics stripped, Turkish ı made i; Cyrillic letters kept."""
|
||||
if not text:
|
||||
return ""
|
||||
decomposed = unicodedata.normalize("NFKD", str(text).casefold().replace("ı", "i"))
|
||||
return "".join(ch for ch in decomposed if not unicodedata.combining(ch))
|
||||
|
||||
|
||||
def _words(folded: str) -> list[str]:
|
||||
return [w for w in re.split(r"[^\w]+|_|\d+", folded) if w]
|
||||
|
||||
|
||||
def _compact(folded: str) -> str:
|
||||
return "".join(w for w in _words(folded))
|
||||
|
||||
|
||||
def _matches(keyword: str, folded: str) -> bool:
|
||||
key = fold(keyword)
|
||||
if " " in key or len(key) >= _SUBSTRING_MIN:
|
||||
return key.replace(" ", "") in _compact(folded)
|
||||
return any(word.startswith(key) for word in _words(folded))
|
||||
|
||||
|
||||
def _first(rows, folded: str):
|
||||
for row in rows:
|
||||
if _matches(row[-2], folded):
|
||||
return row
|
||||
return None
|
||||
|
||||
|
||||
def _direction(folded: str) -> tuple[str, str, str] | None:
|
||||
"""("supply" | "return", keyword, language), or None when neither or both."""
|
||||
supply = _first(_SUPPLY_WORDS, folded)
|
||||
ret = _first(_RETURN_WORDS, folded)
|
||||
if (supply is None) == (ret is None):
|
||||
return None
|
||||
if supply is not None:
|
||||
return ("supply", *supply)
|
||||
return ("return", *ret)
|
||||
|
||||
|
||||
def _result(discipline, service, confidence, keyword, language) -> dict:
|
||||
return {
|
||||
"discipline": discipline,
|
||||
"service": service,
|
||||
"confidence": confidence,
|
||||
"keyword": keyword,
|
||||
"language": language,
|
||||
}
|
||||
|
||||
|
||||
def classify_layer(name: str) -> dict:
|
||||
"""What a layer most likely carries, decided by the words in its name.
|
||||
|
||||
``{"discipline", "service", "confidence", "keyword", "language"}``. Order:
|
||||
electrical words; CIP and glycol with a direction word (``P_cipsupplyline``
|
||||
-> ``cip_supply``; without one, the service stays ``unknown`` at 0.6 - the
|
||||
layer alone is not trusted to say which way a CIP line runs); the stated
|
||||
services; a water word with no temperature (piping, ``unknown``, 0.6); the
|
||||
discipline words. Nothing matched: ``unknown`` / ``unknown`` at 0.0.
|
||||
"""
|
||||
folded = fold(name)
|
||||
row = _first(_ELECTRICAL_WORDS, folded)
|
||||
if row is not None:
|
||||
return _result("electrical", "electrical", DIRECT, *row)
|
||||
for rows, stem in ((_CIP_WORDS, "cip"), (_GLYCOL_WORDS, "glycol")):
|
||||
row = _first(rows, folded)
|
||||
if row is None:
|
||||
continue
|
||||
direction = _direction(folded)
|
||||
if direction is None:
|
||||
return _result("piping", "unknown", WEAK, *row)
|
||||
return _result("piping", f"{stem}_{direction[0]}", DIRECT, row[0], row[1])
|
||||
row = _first(_SERVICE_WORDS, folded)
|
||||
if row is not None:
|
||||
return _result("piping", row[0], DIRECT, row[1], row[2])
|
||||
row = _first(_WATER_WORDS, folded)
|
||||
if row is not None:
|
||||
return _result("piping", "unknown", WEAK, *row)
|
||||
row = _first(_DISCIPLINE_WORDS, folded)
|
||||
if row is not None:
|
||||
return _result(row[0], "unknown", DIRECT, row[1], row[2])
|
||||
return _result("unknown", "unknown", 0.0, None, None)
|
||||
|
||||
|
||||
def supply_return(text: str | None) -> str | None:
|
||||
"""``"supply"`` or ``"return"`` when the text says one of them, else None.
|
||||
|
||||
``ПОДАЧА`` / ``BESLEME`` / ``SUPPLY`` / ``AANVOER`` / ``VORLAUF`` are supply;
|
||||
``ВОЗВРАТ`` / ``DÖNÜŞ`` / ``RETURN`` / ``RETOUR`` / ``RÜCKLAUF`` return. A
|
||||
text naming both (``CIP SUPPLY / RETURN``) says neither and is None.
|
||||
"""
|
||||
direction = _direction(fold(text))
|
||||
return None if direction is None else direction[0]
|
||||
|
||||
|
||||
def room_label(text: str | None) -> dict | None:
|
||||
"""A room label read: ``{"number", "name", "language"}``, or None.
|
||||
|
||||
A text is a room label when one of its words is a room word (*room*,
|
||||
*oda*, *mahal*, *помещение*, *комната*, *ruimte*, *kamer*, *Raum*,
|
||||
*Zimmer*). The number is the first word holding a digit; the name is what
|
||||
is left, in the text's own spelling, or None. ``"ROOM 1 PROCESS HALL"`` ->
|
||||
``{"number": "1", "name": "PROCESS HALL", "language": "en"}``.
|
||||
"""
|
||||
if not text:
|
||||
return None
|
||||
words = str(text).split()
|
||||
for index, word in enumerate(words):
|
||||
folded = fold(word)
|
||||
for keyword, language in _ROOM_WORDS:
|
||||
if folded.startswith(fold(keyword)):
|
||||
rest = words[:index] + words[index + 1 :]
|
||||
number = next((w for w in rest if any(ch.isdigit() for ch in w)), None)
|
||||
name = " ".join(w for w in rest if w != number).strip() or None
|
||||
return {"number": number, "name": name, "language": language}
|
||||
return None
|
||||
|
||||
|
||||
def equipment_kind(block_name: str | None) -> dict | None:
|
||||
"""The kind of equipment a block name says it is: ``{"kind", "keyword", "language"}``.
|
||||
|
||||
``KLEP_VLINDER`` -> valve (nl); ``VERLOOPSTUK`` -> reducer (nl); an
|
||||
anonymous ``*U12`` or a name with no keyword -> None.
|
||||
"""
|
||||
folded = fold(block_name)
|
||||
row = _first(_EQUIPMENT_KIND_WORDS, folded)
|
||||
if row is None:
|
||||
return None
|
||||
return {"kind": row[0], "keyword": row[1], "language": row[2]}
|
||||
Reference in New Issue
Block a user