- Code\rebar_pipeline, review_pipeline, pier-program을 정션에서 실제 폴더로 전환 - Code\rhino_mcp 및 CLAUDE_rhino배경.md 추가, .mcp.json은 PYTHONPATH로 새 위치 사용 - pier-program 내부 .git 제거(이력은 backup\pier-program_20261028.bundle에 보관) - worktree 가이드 반영(CLAUDE.local.md, HANDOFF.md) - D:\Rhino-MCP 경로 참조를 새 위치로 치환 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
227 lines
12 KiB
Python
227 lines
12 KiB
Python
"""Out-of-table dimension finding (stage C of the 입력 묶음 확인·확정 개편, first part, 2026-10-19).
|
|
|
|
Anchors: for a modelled pier (params / expected known) the app records where each numeric field was read from the drawing — every dimension / EL
|
|
text (layer, sheet, position) whose value equals the field value. For a NEW drawing of the same series the anchors are followed: the shift between
|
|
the two drawings is estimated per view from the texts both drawings share (R5), the candidate is the dimension text of the same layer and kind
|
|
nearest to the predicted position (R1 EL texts, scalar dimension texts). Several instances of one dimension (front / side / plan) vote.
|
|
Trust grades (기획서 4.2): A >= 2 instances followed and all agree, small shift spread -> candidate with evidence, may be applied
|
|
B agreement of >= 2 instances, or a single instance -> candidate with a warning
|
|
C instances disagree -> list only, applying needs a reason
|
|
none: nothing found -> "근거 없음", no candidate is invented.
|
|
R2: dimension chains `n@pitch=total` are parsed and checked (n x pitch = total).
|
|
Validation on the real series (tools/test_anchors.py): grade A had no wrong candidate; B and C are measured and shown as such.
|
|
"""
|
|
import re
|
|
import statistics
|
|
from collections import defaultdict
|
|
|
|
from engine import variability
|
|
|
|
VERSION = "2"
|
|
MIN_MM = 1000 # values below one metre are too common as texts to be anchored
|
|
MAX_HITS = 6 # a value found in more places is not a specific dimension
|
|
RADIUS = 1200.0 # search radius around the predicted position (drawing units, mm)
|
|
K_LANDMARKS = 5
|
|
MIN_COVERAGE = 0.70 # share of the anchors' landmark texts found again in the new drawing
|
|
A_SPREAD = 400.0
|
|
TABLE_RULE = "single" # "single": a candidate whose only followed instance is ONE table cell is demoted to C (2026-10-20); "none": table cells never grade; "off": old behaviour.
|
|
# Measured on the verified Ungok P3 / P4 values: grade B wrong 10/65 (off) -> 2/39 (single) vs 3/29 (none); grade A unchanged (0 wrong)
|
|
MARK = re.compile(r"^[A-Z]{1,2}\d{1,2}$")
|
|
NUM = re.compile(r"^-?\d[\d,]*\.?\d*$")
|
|
CHAIN = re.compile(r"^(\d+)\s*@\s*(\d[\d,]*(?:\.\d+)?)\s*=\s*(\d[\d,]*(?:\.\d+)?)\s*(?:\(.*\))?$")
|
|
SKIP = {("expected", "planned_fail"), ("expected", "rebar_palette"), ("expected", "tol_mm")}
|
|
|
|
|
|
def text_val(t):
|
|
"""(value mm, kind) of a dimension / EL text: 'EL. 38.916' -> (38916, EL), '7.800' -> (7800, m), '7800' -> (7800, mm); else None."""
|
|
s = t.strip().replace(" ", "")
|
|
m = re.match(r"^EL\.?(\d+\.\d{3})$", s)
|
|
if m:
|
|
return round(float(m.group(1)) * 1000, 3), "EL"
|
|
if NUM.match(s):
|
|
x = s.replace(",", "")
|
|
try:
|
|
f = float(x)
|
|
except ValueError:
|
|
return None
|
|
if re.match(r"^\d+\.\d{3}$", x):
|
|
return round(f * 1000, 3), "m"
|
|
if re.match(r"^\d+$", x):
|
|
return f, "mm"
|
|
return None
|
|
|
|
|
|
def leaves(params, expected, max_list=40):
|
|
"""(field, value) of every anchorable numeric leaf: bundles of grade V1 / V2 only, no sources / notes, no human decision records."""
|
|
for which, doc in (("params", params), ("expected", expected)):
|
|
for k, v in doc.items():
|
|
m = variability.classify(which, k, v)
|
|
if m["info"] or m["grade"] in ("F", "V3") or (which, k) in SKIP:
|
|
continue
|
|
stack = [((k,), v)]
|
|
while stack:
|
|
path, x = stack.pop()
|
|
if isinstance(x, dict):
|
|
for a, b in x.items():
|
|
if a in ("src", "_note") or a.endswith("_src"):
|
|
continue
|
|
stack.append((path + (str(a),), b))
|
|
elif isinstance(x, list):
|
|
if len(x) <= max_list:
|
|
stack.extend((path + (str(i),), b) for i, b in enumerate(x))
|
|
elif isinstance(x, (int, float)) and not isinstance(x, bool):
|
|
yield f"{which}:" + ".".join(path), x
|
|
|
|
|
|
def _unique_labels(texts):
|
|
out = defaultdict(list)
|
|
for t in texts:
|
|
s = t["text"].strip()
|
|
if len(s) > 1 and not NUM.match(s.replace(" ", "")) and not CHAIN.match(s):
|
|
out[(t["dwg"], s)].append(t)
|
|
return {k: v[0] for k, v in out.items() if len(v) == 1}
|
|
|
|
|
|
def learn(params, expected, texts, sources=()):
|
|
"""Anchors of a modelled pier. `sources` = sha256 of its DWG files (to recognise a pier that follows its own drawing)."""
|
|
by = defaultdict(list)
|
|
for t in texts:
|
|
v = text_val(t["text"])
|
|
if v:
|
|
by[round(v[0])].append({"text": t["text"].strip(), "x": t["x"], "y": t["y"], "layer": t["layer"], "dwg": t["dwg"], "kind": v[1]})
|
|
fields = {}
|
|
owners = defaultdict(set) # same value, same file, different bundle / leaf name -> which field the text belongs to is unknown
|
|
cand = []
|
|
for field, v in leaves(params, expected):
|
|
key = round(abs(v))
|
|
if key >= MIN_MM and 1 <= len(by.get(key, [])) <= MAX_HITS:
|
|
which, path = field.split(":", 1)
|
|
parts = [q for q in path.split(".") if not q.isdigit() and not MARK.match(q)] # mark rows (P3, C1 ...) are instances of one dimension kind
|
|
owners[(which, key)].add((path.split(".")[0], re.sub(r"_(x|y|z|left|right|front|rear|top|bottom)$", "", parts[-1])))
|
|
cand.append((field, v, key))
|
|
for field, v, key in cand:
|
|
which = field.split(":", 1)[0]
|
|
fields[field] = {"v": v, "hits": by[key], "shared": len(owners[(which, key)])}
|
|
lm = [{"text": s, "x": t["x"], "y": t["y"], "dwg": d} for (d, s), t in _unique_labels(texts).items()]
|
|
return {"version": VERSION, "sources": list(sources), "fields": fields, "landmarks": lm}
|
|
|
|
|
|
def _shift(anchor, common, dwg, x, y):
|
|
near = sorted((c for c in common if c[4] == dwg), key=lambda c: (c[0] - x) ** 2 + (c[1] - y) ** 2)[:K_LANDMARKS]
|
|
if not near:
|
|
return None
|
|
dx, dy = statistics.median(c[2] for c in near), statistics.median(c[3] for c in near)
|
|
return dx, dy, max(abs(c[2] - dx) for c in near) + max(abs(c[3] - dy) for c in near)
|
|
|
|
|
|
def follow(anchors, texts, own_sources=()):
|
|
"""Candidates for a new drawing: {field: {value, grade, followed, agree, spread, text, x, y, dwg, layer, all, self_source}}."""
|
|
new = _unique_labels(texts)
|
|
common = []
|
|
for lm in anchors.get("landmarks", []):
|
|
t = new.get((lm["dwg"], lm["text"]))
|
|
if t:
|
|
common.append((lm["x"], lm["y"], t["x"] - lm["x"], t["y"] - lm["y"], lm["dwg"]))
|
|
by_layer = defaultdict(list)
|
|
for t in texts:
|
|
v = text_val(t["text"])
|
|
if v:
|
|
by_layer[(t["dwg"], t["layer"], v[1])].append((v[0], t))
|
|
self_source = bool(set(anchors.get("sources", [])) & set(own_sources))
|
|
coverage = len(common) / len(anchors["landmarks"]) if anchors.get("landmarks") else 0.0
|
|
out = {}
|
|
for field, a in anchors.get("fields", {}).items():
|
|
res = []
|
|
for h in a["hits"]:
|
|
sh = _shift(a, common, h["dwg"], h["x"], h["y"])
|
|
if not sh:
|
|
continue
|
|
px, py = h["x"] + sh[0], h["y"] + sh[1]
|
|
best = None
|
|
for val, t in by_layer.get((h["dwg"], h["layer"], h["kind"]), []):
|
|
d = ((t["x"] - px) ** 2 + (t["y"] - py) ** 2) ** 0.5
|
|
if d <= RADIUS and (best is None or d < best[0]):
|
|
best = (d, val, t)
|
|
if best:
|
|
res.append({"value": round(best[1]), "dist": round(best[0], 1), "spread": round(sh[2], 1), "text": best[2]["text"].strip(),
|
|
"x": best[2]["x"], "y": best[2]["y"], "dwg": best[2]["dwg"], "layer": best[2]["layer"],
|
|
"table_cell": "TABL" in best[2]["layer"]})
|
|
if not res:
|
|
continue
|
|
# improvement #34-4 (2026-10-20): a text that sits in a schedule / table is found again by POSITION, but the rows of a table move when the pier's
|
|
# bar list differs (Ungok P3 / P4: the length cell '3.800' of S1 was followed as the end of a rebar chain, grade B). Table cells are the job of the
|
|
# table candidates (engine/reader, check C2). A candidate whose only followed instance is one table cell is therefore demoted to C.
|
|
free = [r for r in res if not r["table_cell"]]
|
|
if TABLE_RULE == "none":
|
|
table_only = not free
|
|
if free:
|
|
res = free
|
|
elif TABLE_RULE == "single":
|
|
table_only = not free and len(res) == 1
|
|
else:
|
|
table_only = False
|
|
vals = [r["value"] for r in res]
|
|
cons = max(set(vals), key=lambda v: (vals.count(v), -abs(v)))
|
|
agree = vals.count(cons)
|
|
spread = max(r["spread"] for r in res)
|
|
if table_only:
|
|
grade = "C"
|
|
elif len(res) >= 2 and agree == len(res) and spread < A_SPREAD and a.get("shared", 1) == 1:
|
|
grade = "A"
|
|
elif agree >= 2 or len(res) == 1:
|
|
grade = "B"
|
|
else:
|
|
grade = "C"
|
|
if coverage < MIN_COVERAGE and grade != "C":
|
|
grade = "C" # the drawing frame differs from the one the anchors come from
|
|
pos = min((r for r in res if r["value"] == cons), key=lambda r: r["dist"])
|
|
out[field] = {"value": cons if a["v"] >= 0 else -cons, "grade": grade, "followed": len(res), "agree": agree, "spread": spread,
|
|
"text": pos["text"], "x": pos["x"], "y": pos["y"], "dwg": pos["dwg"], "layer": pos["layer"],
|
|
"all": sorted(set(vals)), "source_value": a["v"], "self_source": self_source, "coverage": round(coverage, 2),
|
|
"shared": a.get("shared", 1), "table_only": table_only}
|
|
return out
|
|
|
|
|
|
def table_values(proposal_rows):
|
|
"""{field: value} of the table candidates (engine.reader.proposals rows); list candidates are expanded per index (schedule [L, n])."""
|
|
out = {}
|
|
for r in proposal_rows:
|
|
field = f"{r['file']}:" + ".".join(str(x) for x in r["path"])
|
|
c = r["candidate"]
|
|
if isinstance(c, (list, tuple)):
|
|
for i, v in enumerate(c):
|
|
if isinstance(v, (int, float)) and not isinstance(v, bool):
|
|
out[f"{field}.{i}"] = v
|
|
elif isinstance(c, (int, float)) and not isinstance(c, bool):
|
|
out[field] = c
|
|
return out
|
|
|
|
|
|
def cross_check(cand, table):
|
|
"""Confirm / correct anchor candidates with table values of the same field: equal -> grade A (table_confirmed), different -> grade C with
|
|
`table` = the table value. Fields without a table value keep their grade. Works on the dict returned by follow()."""
|
|
for f, c in cand.items():
|
|
t = table.get(f)
|
|
if t is None:
|
|
continue
|
|
c["table"] = t
|
|
if abs(abs(t) - abs(c["value"])) <= 1.5:
|
|
c["grade"], c["table_confirmed"] = "A", True
|
|
else:
|
|
c["grade"], c["table_confirmed"] = "C", False
|
|
return cand
|
|
|
|
|
|
def chains(texts):
|
|
"""R2: dimension chains `n@pitch=total` with the check n x pitch = total: [{text, dwg, x, y, n, pitch, total, ok}]."""
|
|
def mm(s):
|
|
s = s.replace(",", "")
|
|
return round(float(s) * 1000) if re.fullmatch(r"\d+\.\d{3}", s) else round(float(s))
|
|
out = []
|
|
for t in texts:
|
|
m = CHAIN.match(t["text"].strip())
|
|
if m:
|
|
n, p, tot = int(m.group(1)), mm(m.group(2)), mm(m.group(3))
|
|
out.append({"text": t["text"].strip(), "dwg": t["dwg"], "x": t["x"], "y": t["y"], "n": n, "pitch": p, "total": tot, "ok": abs(n * p - tot) <= 1})
|
|
return out
|