Aislo/resources/tester/test_m02_df_convert.py

"""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 「키」(입력값으로 요소 고르기 · 정함 ②) · 표 값 식 글을 수로(정함 ③)는 sub4 몫
  — 대조는 같은 뜻 대역(ref 의 {…} 에 알약 값을 끼워 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, re, sys
from decimal import Decimal
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)
        if seen[k].get("rows"):  # 표 값 식 글 → 수(Values._table 대역)
            seen[k]["rows"] = [[arith(x) for x in r] for r in seen[k]["rows"]]
    return seen[k]
def arith(x):
    if isinstance(x, str) and mf._ARITH.fullmatch(x):
        try:
            return mf.evaluate(mf.parse(x), {}, master)
        except (mf.FormulaError, ArithmeticError):
            return x
    return x
dv.one_value = cached
element_value = cp._element_value
HOLE = re.compile(r"\{([^{}]+)\}")
def stand_in(group, el, given, ports, errors, box, ctx):
    rule = el.get("rule") or {}
    if el.get("mode") != "rule" or rule.get("field") != "키":
        return element_value(group, el, given, ports, errors, box, ctx)
    src = rule.get("from") or {}
    v = ports.get(src.get("box"), {}).get(src.get("name"))
    if v is None:
        errors.append(f"「{el.get('name')}」 자리 값 없음")
        return None
    text = format(Decimal(str(v)).normalize(), "f") if isinstance(v, (int, float)) else str(v)
    ref, _, field = HOLE.sub(lambda m: text, el["ref"]).partition("#")
    try:
        ref = ref if ref.lstrip().startswith("{") else str(master.get(ref).get("키"))
    except mf.FormulaError as e:
        errors.append(f"「{el.get('name')}」 {e}")
        return None
    ref += f"#{field}" if field else ""
    got = cached({"group": group, "ref": ref, "unit": el.get("unit")})
    if not got["ok"]:
        errors.append(f"「{el.get('name')}」 {got.get('reason')}")
    return got.get("value") if got["ok"] else None
cp._element_value = stand_in
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:
        design = {"in-design": {k: cv.plain(v) for k, v in given.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))


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()
    for e in r.get("errors") or []:
        tail = e.rsplit("」", 1)[-1].strip()
        if tail:
            return "흐름 셈 알림 — " + tail.split("—")[0].strip()[:40]
    return "값 다름(알림 없음)"


# ── 꼴 시험 ──────────────────────────────────────────────────────────
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} == {"기계"}
    # 레미콘 {a}-{b}-{c} = 글 잇는 함수(&) → 재료 조건 찾기
    remi = got["GC000264"]["flow"]
    assert any("&" in str(b.get("func")) for b in remi["boxes"])
    # 덧줄 줄.'이름' · 비목 합
    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["시험입력"]