Files
Pier_v1/Code/교각모델링앱/tools/test_geomread.py
T
ahnyongjooandClaude Sonnet 5.5 964a841de1 chore(rebar): Pier_T_rebar 작업 폴더 초기 구성 및 외부 의존 파일 통합
- 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>
2026-10-28 11:20:18 +09:00

274 lines
20 KiB
Python

"""Test of improvement #34-1 (2026-10-20): the drawing readers (engine/geomread.py) and the web screen "도면에서 읽기". No Rhino. Own data root
검증/geomread_test_data (recreated); real piers are only read. Acceptance: the readers reproduce the finished template pier (P1) and the gate-verified
piers (P3, P4) within 1.5 mm; starting from the early P4 backup (a P1 copy) they come out equal to the verified P4 values; a drawing with another frame is
refused; the screen writes only the rows that were chosen, takes a backup, logs it, and the app check runs again. Skipped parts (data missing) are reported."""
import json
import os
import re
import shutil
import sys
from pathlib import Path
APP = Path(__file__).resolve().parents[1]
D = APP / "data"
TD = APP / "검증" / "geomread_test_data"
if TD.exists():
shutil.rmtree(TD)
TD.mkdir(parents=True)
sys.path.insert(0, str(APP / "tools"))
sys.path.insert(0, str(APP))
sys.path.insert(0, str(APP / "web"))
from testlib import verified_pier # noqa: E402
from engine import anchors, geomread # noqa: E402
ok = True
def check(name, cond, detail=""):
global ok
ok &= bool(cond)
print("PASS" if cond else "FAIL", name, str(detail)[:300], flush=True)
def jl(p):
return json.loads(Path(p).read_text(encoding="utf-8"))
ref = jl(D / "_템플릿" / "운곡교P1형" / "params.json")
anc = jl(D / "_템플릿" / "운곡교P1형" / "anchors.json")
def read(d, P=None, E=None, name=None):
d = Path(d)
texts = jl(d / "drawings" / "texts.json")
audit = jl(d / "drawings" / "audit.json")
return geomread.read(P or jl(d / "params.json"), E or jl(d / "expected.json"), ref, texts, geomread.dxf_of(d, audit), name or re.sub(r"_\d{8}_\d{6}$", "", d.name)), texts
def get(doc, path):
x = doc
for k in path.split("."):
x = x[int(k)] if isinstance(x, list) else x[k]
return x
def geomread_paths(doc, key, pre=None):
"""dotted paths of every leaf under doc[key] (key as first token)"""
out = []
def walk(x, path):
if isinstance(x, dict):
for k, v in x.items():
if k in ("src", "_note") or str(k).endswith("_src"):
continue
walk(v, path + [str(k)])
elif isinstance(x, list):
for i, v in enumerate(x):
walk(v, path + [str(i)])
else:
out.append(".".join(path))
walk(doc, [key])
return out
# ---- 1. finished piers: the readers reproduce them
for n in ("운곡교P1", "운곡교P3", "운곡교P4"):
vd = verified_pier(n) if n != "운곡교P1" else D / n # P1 is the template pier; P3 / P4: the gate-verified copy
if vd is None or not vd.exists():
print("SKIP", n)
continue
(P, E, ch, notes), texts = read(vd)
cov = geomread.frame_ok(texts, anc, anchors)
check(f"{n}: frame coverage {cov:.2f} >= {geomread.FRAME_MIN_COVERAGE}", cov >= geomread.FRAME_MIN_COVERAGE)
# diagnostics that were never updated while P3 / P4 were hand-modelled (gate-approved with them, so the data stays): bar-count notes and bar diameters
stale_diag = ("footing_rebar.counts_check", "G2.dia", "G3.dia", "G4.dia") if n != "운곡교P1" else ()
material = [c for c in ch if not c["inferred"] and c["path"] not in stale_diag]
check(f"{n}: the readers reproduce the verified values (0 material changes, {len(ch) - len(material)} inferred row(s) allowed)", not material,
[(c["file"], c["path"], c["old"], c["new"]) for c in material][:4])
# ---- 2. from the early P4 backup (a P1 copy) the readers reach the verified P4 values
p4 = verified_pier("운곡교P4") or D / "운곡교P4"
bks = sorted((p4 / "backup").glob("before_apply_pier_read_*")) if p4.exists() else []
FIX_P4 = APP / "검증" / "fixtures" / "p4_template_state" # P4 straight from the template (the early P1-copy state); kept because the pier data is recreated
if not bks and FIX_P4.exists():
bks = [FIX_P4]
if bks:
P0, E0 = jl(bks[0] / "params.json"), jl(bks[0] / "expected.json")
(P, E, ch, notes), texts = read(p4, P0, E0)
check(f"P4 from the P1-copy backup: {len(ch)} changes proposed (>= 100)", len(ch) >= 100, len(ch))
cur_p, cur_e = jl(p4 / "params.json"), jl(p4 / "expected.json")
bad = []
for c in ch:
if c["inferred"] or c["path"] in ("display.hidden_ids", "footing_rebar.counts_check", "G2.dia", "G3.dia", "G4.dia"): # a display choice: P4 shows the footing / coping (P3 does too), the reader keeps the template's hiding
continue
cur = get(cur_p if c["file"] == "params" else cur_e, c["path"])
if not geomread.same(cur, c["new"], 0.05 if "rotation" in c["path"] else 1.5) and not (c["path"].endswith(("c14.points", "c14_levels")) and
geomread.same(geomread._points_key(cur), geomread._points_key(c["new"]))):
bad.append((c["file"], c["path"], str(cur)[:30], str(c["new"])[:30]))
check("P4 from the P1-copy backup: every proposed value equals the gate-verified P4 value (inferred rows excluded)", not bad, bad[:4])
groups = {c["group"] for c in ch}
check("the proposal covers coping, footing / piles / column, bearings, footing rebar, column rebar", {"코핑", "코핑 철근", "기초·말뚝·기둥", "받침", "기초 철근", "기둥 철근"} <= groups, groups)
P2, E2 = geomread.apply_selected(P0, E0, ch, [i for i, c in enumerate(ch) if not c["inferred"]])
check("apply_selected leaves the unchosen (inferred) rows as they were", P2["coping_rebar"]["stirrups"]["top_left"] == P0["coping_rebar"]["stirrups"]["top_left"])
else:
print("SKIP P4 early backup")
# ---- 2b. a pier whose answer is known: 운곡교 P2 (the 10-12 pier with gate-verified values now in the recycle bin) -- the readers on the NEW P2 (made from the
# P1 template) reach the same values, incl. C14 points whose x offsets (0 / 900 ...) differ from P1 (13 / 913 ...)
newp2 = D / "운곡교P2"
oldp2 = sorted(t for t in (D / "_휴지통").glob("운곡교p2_*") if t.name.startswith("운곡교p2_") and (t / "params.json").exists()) if (D / "_휴지통").exists() else []
b2 = sorted((newp2 / "backup").glob("before_geomread_*")) if newp2.exists() else []
if oldp2 and b2:
old_p, old_e = jl(oldp2[-1] / "params.json"), jl(oldp2[-1] / "expected.json")
(P, E, ch, notes), texts = read(newp2, jl(b2[0] / "params.json"), jl(b2[0] / "expected.json"))
bad = []
for c in ch:
if c["path"] in ("pier_id", "display.hidden_ids") or c["path"] in ("footing_rebar.counts_check", "G2.dia", "G3.dia", "G4.dia"):
continue
ov = get(old_p if c["file"] == "params" else old_e, c["path"]) if True else None
tol = 0.05 if "rotation" in c["path"] else 1.5
if c["path"] == "G1.seats.row_y": # nominal text value in the old pier (-699 / 600) vs the pad polylines (-700 / 598)
continue
okv = geomread.same(ov, c["new"], tol) or (c["path"].endswith(("c14.points", "c14_levels", "hooks_up")) and geomread.same(geomread._points_key(ov), geomread._points_key(c["new"]), tol))
if not okv:
bad.append((c["file"], c["path"], str(ov)[:30], str(c["new"])[:30]))
check(f"P2 (new, from the P1 template): {len(ch)} readings, every one equals the gate-verified old P2 value (row_y nominal values aside)", not bad, bad[:4])
c14 = [c for c in ch if c["path"] == "coping_rebar.c14.points"]
check("P2: the C14 points are read from the drawn circles (the top rows sit at x 0 / 900 ..., the others at 13 / 913 ... as in P1)", c14 and any(p_[0] == 900 for p_ in c14[0]["new"]) and any(p_[0] == 913 for p_ in c14[0]["new"]))
else:
print("SKIP P2 (new pier with an early backup and the old verified P2 are both needed)")
# ---- 2c. the tables are part of the same screen: from the early P4 backup (a P1 copy) the readings + the table rows together bring the tables' fields to the
# gate-verified P4 values (expected G1..G5 tables, bearing / sole plate / bar-mark fields), incl. the per-bearing sole plate length L
if bks:
P0, E0 = jl(bks[0] / "params.json"), jl(bks[0] / "expected.json")
(P, E, ch, notes), texts = read(p4, P0, E0)
tch = geomread.table_changes(P, E, texts)
check(f"table rows are proposed on top of the readings ({len(tch)}), grouped as one", len(tch) >= 60 and {c["group"] for c in tch} == {geomread.TABLE_GROUP}, len(tch))
allc = ch + tch
P2, E2 = geomread.apply_selected(P0, E0, allc, [i for i, c in enumerate(allc) if not c["inferred"]])
vp, ve = jl(p4 / "params.json"), jl(p4 / "expected.json")
bad = []
for which, a, b in (("params", P2, vp), ("expected", E2, ve)):
for key in (("bearing_seats", "sole_plates", "rebar_marks") if which == "params" else ("G1", "G2", "G3", "G4", "G5")):
for path in geomread_paths(a[key], key):
x, y = get(a, path), get(b, path)
if y != "<missing>" and not geomread.same(x, y, 0.05 if "rotation" in path else 1.5):
bad.append((which, path, str(x)[:25], str(y)[:25]))
allowed = ("G1.seats.rear_first_from_tip", "G2.tie_offset", "G2.dia", "G3.dia", "G4.dia", "G1.seats.row_y", "G1.seats.pitch_chain") # anchor / builder constant / stale diagnostics of the old pier
unexpected = [b for b in bad if not b[1].startswith(allowed)]
check("bearing / sole plate / bar mark fields and the expected tables equal the gate-verified P4 after readings + table rows (known exceptions: anchor value, tie_offset constant, diameters)", not unexpected, unexpected[:6])
check("per-bearing sole plate length: expected G5.L became {no: value} where the table differs", isinstance(E2["G5"].get("L"), dict) and len(set(E2["G5"]["L"].values())) > 1, E2["G5"].get("L"))
# ---- 3. another frame is refused
for n in ("새재1교P1", "옥성교P1"):
if (D / n).exists():
cov = geomread.frame_ok(jl(D / n / "drawings" / "texts.json"), anc, anchors)
check(f"{n}: another drawing frame (coverage {cov:.2f} < 0.70) -> refused", cov < geomread.FRAME_MIN_COVERAGE)
# ---- 4. the web screen
if bks:
for n in ("운곡교P4", "새재1교P1"):
shutil.copytree(D / n, TD / n, ignore=shutil.ignore_patterns("runs", "backup", "dwg", "confirm.json", "autocheck.json"))
shutil.copytree(D / "_템플릿", TD / "_템플릿", ignore=shutil.ignore_patterns("_이전"))
shutil.copy2(bks[0] / "params.json", TD / "운곡교P4" / "params.json")
shutil.copy2(bks[0] / "expected.json", TD / "운곡교P4" / "expected.json")
os.environ["PIERAPP_DATA"] = str(TD)
import app as webapp # noqa: E402
import store # noqa: E402
assert Path(str(store.DATA)) == TD
c = webapp.app.test_client()
h = c.get("/pier/운곡교P4/geomread").get_data(as_text=True)
check("page lists the proposal in groups with a checkbox per row", "코핑 철근" in h and 'name="row"' in h and "기초 철근" in h)
check("the inferred row is not pre-selected", "추론" in h and h.count("checked") < h.count('name="row"'))
check("page warns that params and expected from one reading are no independent evidence", "독립 근거가 아닙니다" in h)
h2 = c.get("/pier/새재1교P1/geomread").get_data(as_text=True)
check("another drawing frame: the page refuses", "읽지 않습니다" in h2 and 'name="row"' not in h2)
p = store.Pier("운곡교P4")
before = (p.load("params"), p.load("expected"))
r = c.post("/pier/운곡교P4/geomread/apply", data={"by": "시험자"}, follow_redirects=True)
check("applying nothing is refused and writes nothing", "반영할 줄을 고르세요" in r.get_data(as_text=True) and (p.load("params"), p.load("expected")) == before)
# rows 0 and the footing rows only
ch = geomread.read(*before, ref, p.texts(), geomread.dxf_of(p.dir, store.rjson(p.drawings / "audit.json", [])), "운곡교P4")[2]
pick = [i for i, x in enumerate(ch) if x["path"] in ("footing.size_x", "footing.size_y") and x["file"] == "params"]
p.confirm("params", "footing", "시험자", "시험", "n")
r = c.post("/pier/운곡교P4/geomread/apply", data={"by": "시험자", "row": [str(i) for i in pick]}, follow_redirects=True)
after = (p.load("params"), p.load("expected"))
check("only the chosen rows were written", after[0]["footing"]["size_x"] == 10350 and after[0]["footing"]["size_y"] == 8300 and after[0]["coping"] == before[0]["coping"] and after[1] == before[1])
bk = list((p.dir / "backup").glob("before_geomread_*"))
check("a backup of params / expected / confirm was taken first", len(bk) == 1 and (bk[0] / "params.json").exists() and jl(bk[0] / "params.json")["footing"]["size_x"] == 9350)
lg = store.rjson(p.dir / "read_log.json", [])
check("the reading is logged with who, how many and which fields", lg and lg[-1]["by"] == "시험자" and lg[-1]["count"] == len(pick) and "params:footing.size_x" in lg[-1]["fields"])
check("the changed bundle lost its confirmation (hash changed) and the app check ran again", p.section_state("params", "footing")["stale"] and p.autocheck()["sections"]["params:footing"])
check("the page now lists fewer differing rows", c.get("/pier/운곡교P4/geomread").get_data(as_text=True).count('name="row"') < h.count('name="row"'))
# the planned FAIL proposal: listed, not pre-selected; applying it records the person's choice in `approved`
st, _P, _E, chs, _n, _m = webapp._geomread(p)
pfi = [i for i, x in enumerate(chs) if x["path"] == "planned_fail"]
check("the planned FAIL proposal is on the page and is not pre-selected", len(pfi) == 1 and chs[pfi[0]]["inferred"] and any(x["approved"] == geomread.PENDING for x in chs[pfi[0]]["new"]))
c.post("/pier/운곡교P4/geomread/apply", data={"by": "시험자", "row": [str(pfi[0])]}, follow_redirects=True)
ap = [x["approved"] for x in p.load("expected")["planned_fail"] if x["gate"] == "G3" and x["item"].startswith(("B11", "B12"))]
check("applying the planned FAIL proposal stamps approved with who / when (no longer 'pending')", len(ap) == 2 and all(a.startswith("사용자 선택 ") and "시험자" in a for a in ap), ap)
# ---- 4. G2.tie_offset = (dia P4 + dia P6) / 2 (cross-tie height = hoop level + hoop radius + tie radius); the 12:11 P4 state (template value 17.5) gets 22.5
for n in ("_템플릿/운곡교P1형", "운곡교P1", "운곡교P2", "운곡교P3", "운곡교P4"):
vd = D / n if n.startswith("_템플릿") or n == "운곡교P1" else verified_pier(n) # a pier may be deleted / half-rebuilt: use its gate-approved copy
if vd is None or not (vd / "expected.json").exists():
print("SKIP tie_offset rule", n)
continue
dd = jl(vd / "expected.json")["G2"]
check(f"tie_offset rule holds for {n}", dd["tie_offset"] == (dd["dia"]["P4"] + dd["dia"]["P6"]) / 2, (dd["tie_offset"], dd["dia"]["P4"], dd["dia"]["P6"]))
bk0 = APP / "검증" / "fixtures" / "p4_template_state" # P4 as created from the template (before any reading), kept because the pier data is recreated
if bk0.exists():
(P, E, ch, notes), texts = read(D / "운곡교P4", jl(bk0 / "params.json"), jl(bk0 / "expected.json"))
got = [c for c in ch if c["path"] == "G2.tie_offset"]
check("reading the template-state P4 proposes tie_offset 17.5 -> 22.5", len(got) == 1 and got[0]["new"] == 22.5, [(c["old"], c["new"]) for c in got])
check("reading the template-state P4 also proposes the three coping top-tier a values", sorted(c["new"] for c in ch if re.fullmatch(r"coping_rebar\.top\.tiers\.\d\.a", c["path"])) == [11856, 11881, 11905])
else:
print("SKIP template-state P4 fixture missing")
# ---- 5. rules that used to be hand decisions (P4 rebuild 2026-10-20): top_left back-calculation, anchor candidate, planned FAIL proposals
tpl = [APP / "검증" / "fixtures" / "p4_template_state"]
if tpl[0].exists() and verified_pier("운곡교P4"):
vp = verified_pier("운곡교P4")
(P, E, ch, notes), texts = read(vp, jl(tpl[0] / "params.json"), jl(tpl[0] / "expected.json"))
ex = geomread.extra_changes(P, E, texts, jl(vp / "anchors.json"), [])
vpar, vexp = jl(vp / "params.json"), jl(vp / "expected.json")
check("template-state P4: top_left is back-calculated from the C9 b range = the verified decision", P["coping_rebar"]["stirrups"]["top_left"] == vpar["coping_rebar"]["stirrups"]["top_left"],
P["coping_rebar"]["stirrups"]["top_left"])
anc = [c for c in ex if c["path"] == "G1.seats.rear_first_from_tip"]
check("template-state P4: the anchor candidate proposes rear_first_from_tip 1232 -> 1198 (pre-selected, not inferred)", len(anc) == 1 and anc[0]["old"] == 1232 and anc[0]["new"] == 1198 and not anc[0]["inferred"], [(c["old"], c["new"]) for c in anc])
pf = [c for c in ex if c["path"] == "planned_fail"]
items = {(x["gate"], x.get("item")): x for c in pf for x in c["new"] if x.get("approved") == geomread.PENDING}
check("template-state P4: planned FAIL proposals B11, B12 (numbers of this pier), C13 convention; never pre-selected; approval left to the person",
set(items) == {("G3", "B11 bar count vs schedule"), ("G3", "B12 bar count vs schedule"), ("G4", "C13 schedule L reproduced by convention (max+min)/2")} and all(c["inferred"] for c in pf)
and "32본" in items[("G3", "B12 bar count vs schedule")]["reason"] and "38본" in items[("G3", "B12 bar count vs schedule")]["reason"], sorted(items))
else:
print("SKIP template-state P4 backup missing")
fx2 = APP / "검증" / "fixtures" / "p2_template_state"
vp2 = verified_pier("운곡교P2")
if fx2.exists() and vp2:
(P, E, ch, notes), texts = read(vp2, jl(fx2 / "params.json"), jl(fx2 / "expected.json"))
ex = geomread.extra_changes(P, E, texts, jl(vp2 / "anchors.json") if (vp2 / "anchors.json").exists() else None, [])
items = {(x["gate"], x.get("item")): x for c in ex if c["path"] == "planned_fail" for x in c["new"] if x.get("approved") == geomread.PENDING}
check("template-state P2: planned FAIL proposals B9 (551 vs 493) and B12 (32 vs 38); no C13 (P2's C13 follows the convention)",
set(items) == {("G3", "B9 bar count vs schedule"), ("G3", "B12 bar count vs schedule")} and "551본" in items[("G3", "B9 bar count vs schedule")]["reason"] and "493본" in items[("G3", "B9 bar count vs schedule")]["reason"], sorted(items))
else:
print("SKIP template-state P2 fixture / verified copy missing")
E_t = {"planned_fail": [{"gate": "G3", "item": "x", "approved": geomread.PENDING}, {"gate": "G3", "item": "y", "approved": "사용자 2026-10-07"}]}
check("stamp_pending changes only the pending entries", geomread.stamp_pending(E_t, "심영표", "2026-10-20") == 1 and E_t["planned_fail"][1]["approved"] == "사용자 2026-10-07" and "심영표" in E_t["planned_fail"][0]["approved"])
for n in ("운곡교P1", "운곡교P2", "운곡교P3", "운곡교P4"):
vd = verified_pier(n) if n != "운곡교P1" else D / n
if vd is None or not vd.exists() or not (vd / "anchors.json").exists():
print("SKIP", n)
continue
(P, E, ch, notes), texts = read(vd)
ex = geomread.extra_changes(P, E, texts, jl(vd / "anchors.json"), [])
anc = [c for c in ex if c["path"] == "G1.seats.rear_first_from_tip"]
pf_new = [x for c in ex if c["path"] == "planned_fail" for x in c["new"] if x.get("approved") == geomread.PENDING]
check(f"{n}: finished pier -> no anchor proposal, no new planned FAIL proposal (no false proposals)", not anc and not pf_new, [(c["old"], c["new"]) for c in anc] + [x.get("item") for x in pf_new])
print("RESULT", "PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)