eomsangdonandClaude Opus 5.5
d1df48c45f
- 고르기 첫 값 · 범위 아래 끝 · 표에서 가장 낮은 값 — /logic/auto 와 같은 규칙(GF000004 횡단경사 55) · 글 수 → 수 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VBToTCGk5dqDKeXCEP3vhk
290 lines
13 KiB
Python
290 lines
13 KiB
Python
"""M02 흐름도 — 옛 로직 줄 → 흐름도 변환(M02_DF_Convert) 시험 (PLAN 36-2).
|
|
|
|
· 꼴 시험 — 견본 로직 몇을 옮겨 상자 · 선 · 규칙 · 자동 배치(겹침 없음)를 봄
|
|
· 전수 대조 — 사본 폴더(tmp/master_copy_36 · AISLO_MASTER_DIR)의 로직 1,472 마다
|
|
옛 셈(master_formula.run) = 변환 흐름도 셈(Compute.run) · 입력 = check_master.시험입력(정함 ①)
|
|
· 기본 입력(줄의 시험입력 뺌)도 같이 · 맞는 비율 · 안 맞는 까닭 → tmp/convert_report_36.json
|
|
· 조건 찾기 field 「키」(입력값으로 요소 고르기 · 정함 ②) · 표 값 식 글 → 수(정함 ③)는 서버 길로
|
|
(Compute · Values · sub4) · 값 찾기(dv.one_value)는 같은 물음이면 한 번만 셈
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
COPY = ROOT / "tmp" / "master_copy_36"
|
|
REPORT = ROOT / "tmp" / "convert_report_36.json"
|
|
WORKERS = 6
|
|
|
|
# 일꾼(따로 뜬 파이썬 · 사본 폴더) — 표준 입력 = 키 목록 JSON · 표준 출력 = 줄마다 결과 JSON
|
|
WORKER = r"""
|
|
import json, os, sys
|
|
sys.path.insert(0, os.getcwd())
|
|
from M01_MasterData import M01_MasterData_Store as store
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Compute as cp, M02_DF_Convert as cv
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
|
|
mf, cm = store.cm.mf, store.cm
|
|
master = store.loaded()[1]
|
|
seen, flows = {}, {}
|
|
one_value = dv.one_value
|
|
def cached(item, local=None):
|
|
k = json.dumps(item, ensure_ascii=False, sort_keys=True, default=str)
|
|
if k not in seen:
|
|
seen[k] = one_value(item, local)
|
|
return seen[k]
|
|
dv.one_value = cached
|
|
def load(key):
|
|
if key not in flows:
|
|
flows[key] = cv.convert(master.logic(key))["flow"]
|
|
return flows[key]
|
|
def num(v):
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
def same(a, b):
|
|
a, b = num(a), num(b)
|
|
return a is not None and b is not None and abs(a - b) <= 1e-6 * max(1.0, abs(a))
|
|
def check(row, given):
|
|
try:
|
|
old = mf.run(master, row["키"], given)
|
|
except Exception as e:
|
|
return {"state": "옛 셈 오류", "why": str(e)[:200]}
|
|
got = cv.convert(row)
|
|
flows[row["키"]] = got["flow"]
|
|
try:
|
|
env = mf._inputs(row, given) # 옛 엔진처럼 글로 적힌 수 → 수(화면은 수로 넘김)
|
|
design = {"in-design": {k: cv.plain(v) for k, v in env.items()}}
|
|
new = cp.run(got["flow"], design, load, [])
|
|
except Exception as e:
|
|
return {"state": "흐름 셈 멈춤", "why": repr(e)[:200], "notes": got["notes"]}
|
|
cost = new["cost"]
|
|
if "결과" in old:
|
|
pairs = [("결과", old["결과"], cost["값들"].get("결과"))]
|
|
else:
|
|
pairs = [(k, old[k], cost[k]) for k in ("노무비", "재료비", "경비", "계")]
|
|
bad = [[k, num(a), num(b)] for k, a, b in pairs if not same(a, b)]
|
|
if not bad:
|
|
return {"state": "맞음"}
|
|
return {"state": "다름", "diff": bad, "notes": got["notes"], "errors": new["errors"][:6]}
|
|
for key in json.loads(sys.stdin.read()):
|
|
row = master.logic(key)
|
|
plain = cm.시험입력({k: v for k, v in row.items() if k != "시험입력"})
|
|
test = cm.시험입력(row)
|
|
out = {"key": key, "시험입력": check(row, test)}
|
|
out["기본 입력"] = out["시험입력"] if plain == test else check(row, plain)
|
|
print(json.dumps(out, ensure_ascii=False, default=str), flush=True)
|
|
"""
|
|
|
|
|
|
def _copy() -> None:
|
|
"""사본 폴더 — 없으면 정본 JSON 을 복사(정본은 읽기만)."""
|
|
if not any(COPY.glob("*.json")):
|
|
COPY.mkdir(parents=True, exist_ok=True)
|
|
for f in (ROOT / "resources" / "master_data").glob("*.json"):
|
|
shutil.copy2(f, COPY / f.name)
|
|
|
|
|
|
def _env() -> dict:
|
|
return {**os.environ, "AISLO_MASTER_DIR": str(COPY), "PYTHONIOENCODING": "utf-8"}
|
|
|
|
|
|
def _run(code: str, stdin: str = "") -> str:
|
|
got = subprocess.run(
|
|
[sys.executable, "-c", code],
|
|
input=stdin,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
cwd=ROOT,
|
|
env=_env(),
|
|
timeout=7200,
|
|
)
|
|
assert got.returncode == 0, got.stderr[-2000:]
|
|
return got.stdout
|
|
|
|
|
|
def _keys() -> list[str]:
|
|
code = (
|
|
"import json, os, sys; sys.path.insert(0, os.getcwd())\n"
|
|
"from M01_MasterData import M01_MasterData_Store as store\n"
|
|
"files = store.loaded()[0]\n"
|
|
"print(json.dumps([r['키'] for d in files.values() if d.get('그룹') == '로직'"
|
|
" for r in d.get('줄') or []]))"
|
|
)
|
|
return json.loads(_run(code))
|
|
|
|
|
|
REASONS = [ # (알림에 든 말, 까닭) — 앞이 먼저
|
|
("표에 여럿 맞음", "표 범위 규칙 다름 — 여럿 맞음(이상·이하 · 이상·미만 경계)"),
|
|
("표에 없는 조건", "표 범위 규칙 다름 — 없는 조건(이상·이하 · 이상·미만 경계)"),
|
|
("검색어", "재료 검색어 틀 — 서버 검색어 찾기 대기(sub4)"),
|
|
("같음", "조건 찾기 못 찾음(키 같음)"),
|
|
("맞는 재료 없음", "재료 빈 검색어 — 서버 재료 값 찾기 다름"),
|
|
("비어 있음", "단가 빈 품목 — 서버 처리 다름"),
|
|
("값 없음", "단가 빈 품목 — 서버 처리 다름"),
|
|
]
|
|
|
|
|
|
def _reason(r: dict) -> str:
|
|
if r["state"] != "다름":
|
|
return r["state"]
|
|
moved = [n for n in r.get("notes") or [] if n.startswith("못 옮김")]
|
|
if moved:
|
|
return "못 옮김 — " + moved[0].split("—", 1)[1].strip().split("「")[0].strip()
|
|
text = " | ".join(r.get("errors") or [])
|
|
return next((why for word, why in REASONS if word in text), "값 다름(알림 없음)")
|
|
|
|
|
|
# ── 꼴 시험 ──────────────────────────────────────────────────────────
|
|
SHAPE = r"""
|
|
import json, os, sys
|
|
sys.path.insert(0, os.getcwd())
|
|
from M01_MasterData import M01_MasterData_Store as store
|
|
from M02_MasterTemplete.DataFlow import M02_DF_Convert as cv
|
|
master = store.loaded()[1]
|
|
got = {k: cv.convert(master.logic(k)) for k in sys.argv[1:]}
|
|
print(json.dumps(got, ensure_ascii=False, default=str))
|
|
"""
|
|
|
|
|
|
def _shape(*keys: str) -> dict:
|
|
got = subprocess.run(
|
|
[sys.executable, "-c", SHAPE, *keys],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
cwd=ROOT,
|
|
env=_env(),
|
|
timeout=600,
|
|
)
|
|
assert got.returncode == 0, got.stderr[-2000:]
|
|
return json.loads(got.stdout)
|
|
|
|
|
|
def test_꼴_상자_선_규칙_배치() -> None:
|
|
_copy()
|
|
got = _shape("GC000994", "GC000267", "GF000003", "GC000264", "GF000185")
|
|
for key, one in got.items():
|
|
flow = one["flow"]
|
|
ids = [b["id"] for b in flow["boxes"]]
|
|
assert len(ids) == len(set(ids)), key
|
|
assert all(ln["from"]["box"] in ids and ln["to"]["box"] in ids for ln in flow["links"]), key
|
|
spots = [(b["x"], b["y"]) for b in flow["boxes"]]
|
|
assert len(spots) == len(set(spots)), f"{key} 상자 겹침"
|
|
kinds = {b["kind"] for b in flow["boxes"]}
|
|
assert {"input", "function", "output"} <= kinds, key
|
|
outs = [b for b in flow["boxes"] if b["kind"] == "output"]
|
|
ins = [b for b in flow["boxes"] if b["kind"] == "input"]
|
|
assert max(b["x"] for b in ins) < min(b["x"] for b in outs), (
|
|
f"{key} 입력 왼쪽 · 출력 오른쪽"
|
|
)
|
|
for b in flow["boxes"]:
|
|
if b["kind"] == "function":
|
|
f = b["func"]
|
|
assert (
|
|
f["output"]["cell"] in f["sheet"]["칸"]
|
|
and "식" in f["sheet"]["칸"][f["output"]["cell"]]
|
|
)
|
|
# 돈 아닌 로직 = 출력 「결과」(값) · 반올림 → ROUND
|
|
res = got["GF000003"]["flow"]
|
|
assert [b["output"]["cost"] for b in res["boxes"] if b["kind"] == "output"] == ["값"]
|
|
assert any("ROUND(" in str(b.get("func")) for b in res["boxes"])
|
|
# 표 찾기 = 표 알약 + 선 규칙 lookup · 로직 줄 = 다른 로직 입력(비목 셋)
|
|
one = got["GC000994"]["flow"]
|
|
assert any((ln.get("rule") or {}).get("kind") == "lookup" for ln in one["links"])
|
|
logic = [b for b in one["boxes"] if (b.get("input") or {}).get("mode") == "logic"]
|
|
assert logic and logic[0]["input"]["logic"]["key"].startswith("GC")
|
|
assert {b["output"]["cost"] for b in one["boxes"] if b["kind"] == "output"} <= {
|
|
"노무비",
|
|
"재료비",
|
|
"경비",
|
|
}
|
|
# EQ:{기계} = 기계 요소 조건 찾기(field 키 · from = 입력 알약 「기계」)
|
|
els = [
|
|
e
|
|
for b in got["GC000267"]["flow"]["boxes"]
|
|
for e in (b.get("input") or {}).get("elements", [])
|
|
]
|
|
keyed = [e for e in els if e.get("mode") == "rule"]
|
|
assert keyed and all(e["rule"]["field"] == "키" and "{" in e["ref"] for e in keyed)
|
|
machine = [e for e in keyed if e["ref"].startswith("EQ:{기계}#")]
|
|
assert machine and {e["rule"]["from"]["name"] for e in machine} == {"기계"}
|
|
# 「MO:경유:{지역}」 = 원문번호 글을 잇는 함수(&) → 유가 조건 찾기
|
|
assert any("&" in str(b.get("func")) for b in got["GC000267"]["flow"]["boxes"])
|
|
# 레미콘 {a}-{b}-{c} = 재료 조건 찾기 field 검색어 · 틀 · 자리 셋 · 구분 범위(정함 · 선 없음)
|
|
remi = got["GC000264"]["flow"]
|
|
found = [e for b in remi["boxes"] for e in (b.get("input") or {}).get("elements", [])]
|
|
word = [e for e in found if (e.get("rule") or {}).get("field") == "검색어"]
|
|
assert word and word[0]["rule"]["template"] == "레미콘 {굵은골재최대치수}-{강도}-{슬럼프}"
|
|
assert [x["name"] for x in word[0]["rule"]["slots"]] == ["굵은골재최대치수", "강도", "슬럼프"]
|
|
assert word[0]["scope"]["구분"] == "레미콘"
|
|
groups = {b["id"] for b in remi["boxes"] if (b.get("input") or {}).get("mode") == "group"}
|
|
assert not [ln for ln in remi["links"] if ln["to"]["box"] in groups]
|
|
# 덧줄 줄.'이름' · 비목 합
|
|
assert got["GF000185"]["notes"] == [] or all(
|
|
not n.startswith("못 옮김") for n in got["GF000185"]["notes"]
|
|
)
|
|
|
|
|
|
# ── 전수 대조 ────────────────────────────────────────────────────────
|
|
def test_전수_대조_옛_셈_같음() -> None:
|
|
_copy()
|
|
keys = _keys()
|
|
assert len(keys) >= 1400
|
|
chunks = [keys[i::WORKERS] for i in range(WORKERS)]
|
|
with ThreadPoolExecutor(WORKERS) as pool:
|
|
outs = list(pool.map(lambda c: _run(WORKER, json.dumps(c)), chunks))
|
|
rows = [json.loads(x) for out in outs for x in out.splitlines() if x.startswith("{")]
|
|
assert len(rows) == len(keys)
|
|
report: dict = {"로직": len(rows)}
|
|
for kind in ("시험입력", "기본 입력"):
|
|
states = [r[kind]["state"] for r in rows]
|
|
run = [s for s in states if s != "옛 셈 오류"]
|
|
reasons: dict[str, int] = {}
|
|
for r in rows:
|
|
if r[kind]["state"] not in ("맞음", "옛 셈 오류"):
|
|
why = _reason(r[kind])
|
|
reasons[why] = reasons.get(why, 0) + 1
|
|
report[kind] = {
|
|
"옛 셈 오류(대조 밖)": states.count("옛 셈 오류"),
|
|
"대조": len(run),
|
|
"맞음": run.count("맞음"),
|
|
"맞는 비율": round(run.count("맞음") / max(1, len(run)), 4),
|
|
"안 맞는 까닭": dict(sorted(reasons.items(), key=lambda x: -x[1])),
|
|
}
|
|
report["안 맞는 로직"] = [
|
|
{"키": r["key"], **r["시험입력"]} for r in rows if r["시험입력"]["state"] not in ("맞음",)
|
|
]
|
|
REPORT.write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=1, default=str), encoding="utf-8"
|
|
)
|
|
# 변환이 멈추는 로직은 없어야 함
|
|
assert all(r["시험입력"]["state"] != "흐름 셈 멈춤" for r in rows), [
|
|
r for r in rows if r["시험입력"]["state"] == "흐름 셈 멈춤"
|
|
][:3]
|
|
assert report["시험입력"]["맞는 비율"] >= 0.5, report["시험입력"]
|
|
|
|
|
|
def test_변수_예시값_합계_끝수() -> None:
|
|
_copy()
|
|
got = _shape("GF000007", "GF000226", "GF000004")
|
|
for key, one in got.items():
|
|
flow = one["flow"]
|
|
design = next(b for b in flow["boxes"] if b["id"] == "in-design")
|
|
assert all(v["example"] is not None for v in design["input"]["variables"]), key
|
|
assert flow["totalRound"] is None # 옛 계는 끝수 없음
|
|
box = next(b for b in got["GF000226"]["flow"]["boxes"] if b["id"] == "in-design")
|
|
vars226 = {v["name"]: v["example"] for v in box["input"]["variables"]}
|
|
assert vars226["직경"] == 60 # 줄의 시험입력(글 「60.0」 → 수)
|
|
box = next(b for b in got["GF000004"]["flow"]["boxes"] if b["id"] == "in-design")
|
|
vars4 = {v["name"]: v["example"] for v in box["input"]["variables"]}
|
|
assert vars4["횡단경사"] == 55 # 왼쪽 시험 계산 기본값(master_auto)
|