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 「키」(입력값으로 요소 고르기 · 정함 ②) · 표 값 식 글 → 수(정함 ③)는 서버 길로
  (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")
    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」 → 수)