"""M02 단가산출 흐름도 견본(35-9) 시험 — `M02_MasterTemplete/DataFlow/_sample/*.json` · `M02_DF_Sample.ts`. 견본 넷(메쌓기 · 지반 가르기 · 흄관 조건 찾기 · 다른 로직 잇기) + 넷째가 부르는 작업량 로직(value) · 모양 — 상자(입력 · 함수 · 출력) · 알약 이름 규칙 · 기본 색 · 선(받는 자리마다 하나 · 출력에서 안 나감 · 순환 없음) · 선 규칙(조건 통과 · 끝수) · 재료 방식(조건 찾기) · 다른 로직 입력의 알약이 있는 자리를 가리킴 · 흐름 — 출력에서 거꾸로 따라가면 닿는 출처(선 · 선 규칙의 조건 알약 · 재료 조건 알약 포함)가 견본 뜻과 같음 · 값(내부 확인) — 함수 상자 시트를 우리 스프레드시트 엔진(Node 번들)으로 차례대로 셈 · 선 규칙 · 끝수를 지남 · 입력 값(노임 · 품셈 표 · 재료 단가)은 마스터에서 읽기만 · 메쌓기 = /calc GF000219 · value = /calc GF000133 · `sampleList` · `sampleLoad`(사본 · key 없음) — Node 로 바로 돌림 · 파일 안 씀 """ from __future__ import annotations import json import math import re import subprocess from decimal import Decimal from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] FOLDER = ROOT / "M02_MasterTemplete" / "DataFlow" SAMPLE_TS = FOLDER / "M02_DF_Sample.ts" SAMPLES = {p.stem: json.loads(p.read_text(encoding="utf-8")) for p in sorted((FOLDER / "_sample").glob("*.json"))} # fmt: skip NAMES = ["masonry", "ground", "hume", "linked", "value"] BOX_COLOR = dict( re.findall( r'(\w+): "(#[0-9a-fA-F]{6})"', re.search( r"BOX_COLOR[^{]*\{([^}]*)\}", (FOLDER / "M02_DF_Model.ts").read_text(encoding="utf-8") ).group(1), ) ) HOOKS = r""" import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { stripTypeScriptTypes } from "node:module"; export async function load(url, ctx, next) { if (!url.endsWith(".ts")) return next(url, ctx); let src = readFileSync(fileURLToPath(url), "utf8"); src = src.replace(/import\.meta\.glob(?:<[\s\S]*?>)?\([^)]*\)/g, "globalThis.__samples"); src = src.replace(/import\s*\{([^}]*)\}\s*from\s*"([^"]+)";/g, (m, names) => { const list = names.split(",").map((x) => x.trim()).filter((x) => x && !x.startsWith("type ")); return list.length ? `const { ${list.join(", ")} } = globalThis.__stub;` : ""; }); return { format: "module", source: stripTypeScriptTypes(src, { mode: "transform" }), shortCircuit: true }; } """ def _node(body: str) -> object: files = {f"./_sample/{k}.json": v for k, v in SAMPLES.items()} script = ( "import { register } from 'node:module';\n" f"register('data:text/javascript,' + encodeURIComponent({json.dumps(HOOKS)}));\n" "globalThis.__stub = { currentLanguageIndex: 0 };\n" f"globalThis.__samples = {json.dumps(files, ensure_ascii=False)};\n" f"const S = await import({json.dumps(SAMPLE_TS.as_uri())});\n" "const out = (x) => process.stdout.write(JSON.stringify(x));\n" + body ) result = subprocess.run( ["node", "--no-warnings", "--input-type=module", "-e", script], capture_output=True, text=True, encoding="utf-8", timeout=120, cwd=ROOT, ) # fmt: skip assert result.returncode == 0, result.stderr[-2000:] return json.loads(result.stdout) # ── 알약 · 선 ───────────────────────────────────────────────────────────── def _ports_out(box: dict) -> list[str]: if box["kind"] == "function": return [box["func"]["output"]["name"]] if box["kind"] == "output": return [] inp = box["input"] if inp["mode"] == "logic": return [p["name"] for p in inp["logic"]["outputs"]] return ( [e["name"] for e in inp.get("elements", [])] + (["표"] if inp.get("table") else []) + [v["name"] for v in inp.get("variables", [])] ) def _ports_in(box: dict) -> list[str]: if box["kind"] == "function": return [i["port"] for i in box["func"]["inputs"]] return ["값"] if box["kind"] == "output" else [] def _deps(lg: dict, box_id: str) -> list[dict]: """그 상자가 기대는 알약 — 들어온 선 · 그 선 규칙의 조건 알약 · 재료 조건 알약""" out = [] for link in lg["links"]: if link["to"]["box"] != box_id: continue out.append(link["from"]) rule = link.get("rule") or {} if rule.get("kind") == "gate": out.append(rule["from"]) out += [c["from"] for c in rule.get("conds", [])] box = next(b for b in lg["boxes"] if b["id"] == box_id) for e in (box.get("input") or {}).get("elements", []): if e.get("rule"): out.append(e["rule"]["from"]) return out def _order(lg: dict) -> list[dict]: """상자 차례(기대는 것이 앞) — 순환이면 실패""" boxes = {b["id"]: b for b in lg["boxes"]} need = {i: {p["box"] for p in _deps(lg, i)} for i in boxes} done: list[str] = [] while len(done) < len(boxes): ready = [i for i in boxes if i not in done and need[i] <= set(done)] assert ready, "선이 돌고 돎" done += ready return [boxes[i] for i in done] def _trace_back(lg: dict, box_id: str) -> set[tuple[str, str]]: """그 상자 값이 온 길의 출처 알약(입력 상자) — 선 · 규칙 조건 · 재료 조건을 거꾸로 끝까지""" boxes = {b["id"]: b for b in lg["boxes"]} seen: set[tuple[str, str]] = set() todo = [box_id] while todo: cur = todo.pop() for p in _deps(lg, cur): if boxes[p["box"]]["kind"] == "input": seen.add((p["box"], p["name"])) todo.append(p["box"]) # 재료 조건처럼 입력 상자도 다른 알약에 기댐 else: todo.append(p["box"]) return seen # ── 모양 ───────────────────────────────────────────────────────────────── @pytest.mark.parametrize("name", NAMES) def test_견본_모양(name: str) -> None: lg = SAMPLES[name] assert lg["key"] and lg["name"] and lg["resultUnit"] and lg["userElements"] == [] boxes = {b["id"]: b for b in lg["boxes"]} assert len(boxes) == len(lg["boxes"]) sheets = [b["func"]["sheet"]["id"] for b in lg["boxes"] if b["kind"] == "function"] assert len(set(sheets)) == len(sheets) for b in lg["boxes"]: assert b["kind"] in ("input", "function", "output") assert b["color"] == BOX_COLOR[b["kind"]] and b["unit"] assert isinstance(b["x"], int) and isinstance(b["y"], int) if b["kind"] == "function": fn = b["func"] assert "식" in fn["sheet"]["칸"][fn["output"]["cell"]] # 출력 하나 = 수식 칸 assert len({i["cell"] for i in fn["inputs"]}) == len(fn["inputs"]) if b["kind"] == "output": assert b["output"]["cost"] in ("노무비", "재료비", "경비", "값") units = {e["unit"] for e in (b.get("input") or {}).get("elements") or []} if len(units) == 1: assert b["unit"] == next(iter(units)), b["id"] for link in lg["links"]: a, z = boxes[link["from"]["box"]], boxes[link["to"]["box"]] assert link["from"]["name"] in _ports_out(a), link assert link["to"]["name"] in _ports_in(z), link assert a["kind"] != "output" and a["x"] < z["x"] for b in lg["boxes"]: for port in _ports_in(b): got = [x for x in lg["links"] if x["to"]["box"] == b["id"] and x["to"]["name"] == port] assert len(got) == 1, (b["id"], port) # 규칙 · 재료 조건이 가리키는 알약은 있는 자리 for b in lg["boxes"]: for p in _deps(lg, b["id"]): assert p["name"] in _ports_out(boxes[p["box"]]), p _order(lg) def test_견본_흐름_규칙_방식() -> None: """견본마다 보여 주려는 흐름 — 조건 통과 선 둘 · 재료 조건 찾기 · 다른 로직 입력 · 출처 사슬""" g = SAMPLES["ground"] gates = [x for x in g["links"] if (x.get("rule") or {}).get("kind") == "gate"] assert len(gates) == 2 and {x["from"]["name"] for x in gates} == {"굴착량"} assert {x["rule"]["op"] for x in gates} == {"=", "≠"} and len( {x["rule"]["value"] for x in gates} ) == 1 assert ("in-ground", "지반") in _trace_back(g, "out-exp") # 조건 알약도 출처 사슬에 h = SAMPLES["hume"] pipe = next(b for b in h["boxes"] if b["id"] == "in-pipe")["input"]["elements"][0] assert pipe["mode"] == "rule" and pipe["slot"] == pipe["name"] and pipe["rule"]["op"] == "포함" assert pipe["scope"]["상세구분"] == "흄관" and pipe["default"] == pipe["ref"] assert _trace_back(h, "out-mat") == { ("in-size", "연장"), ("in-pipe", "흄관"), ("in-size", "관경"), } lk = SAMPLES["linked"] logic = next(b for b in lk["boxes"] if b["id"] == "in-logic")["input"]["logic"] called = SAMPLES[logic["key"]] assert [p["name"] for p in logic["outputs"]] == [ b["name"] for b in called["boxes"] if b["kind"] == "output" ] assert [p["name"] for p in logic["designs"]] == [ v["name"] for b in called["boxes"] if b["kind"] == "input" for v in b["input"]["variables"] ] assert ("in-logic", "시간당 작업량") in _trace_back(lk, "out-exp") m = SAMPLES["masonry"] assert _trace_back(m, "out-pay") >= { ("in-var", "높이"), ("in-poom", "표"), ("in-rate", "표"), ("in-labor", "석공"), } def test_견본_목록_열기() -> None: got = _node( "out({ list: S.sampleList(), one: S.sampleLoad('ground'), none: S.sampleLoad('없음') });" ) assert [x["id"] for x in got["list"]] == NAMES assert all(x["note"] for x in got["list"]) assert got["one"]["key"] is None and got["one"]["links"] == SAMPLES["ground"]["links"] assert got["none"] is None # ── 값(내부 확인) ───────────────────────────────────────────────────────── @pytest.fixture(scope="module") def master(): from M01_MasterData import M01_MasterData_Store as store cm = store.cm return cm, cm.load(folder=store.FOLDER) def _plain(v): return float(v) if isinstance(v, Decimal) else v def _table_rows(table: dict) -> list[list]: """품셈 표 → 블록(머리 줄 + 줄) — 서버 「표」 와 같은 규칙""" conds, lines = table.get("조건") or {}, table.get("줄") or [] names = [*conds, *(table.get("값칸") or {})] split = { n for n in names if conds.get(n) == "범위" or any(isinstance(r.get(n), list) for r in lines) } head = [h for n in names for h in ((f"{n}·아래", f"{n}·위") if n in split else (n,))] rows = [] for r in lines: line: list = [] for n in names: v = r.get(n) line += (list(v) if isinstance(v, list) else [v, v]) if n in split else [v] rows.append(line) return [head, *rows] def _cut(value, rnd): if value is None or not rnd: return value k = 10 ** rnd["places"] way = {"반올림": lambda x: math.floor(x + 0.5), "올림": math.ceil, "버림": math.floor}[ rnd["method"] ] return way(round(value * k, 9)) / k def _gate(rule, value, ports): have = ports[rule["from"]["box"]][rule["from"]["name"]] want = rule["value"] ok = {"=": have == want, "≠": have != want, "≥": have >= want, "≤": have <= want, ">": have > want, "<": have < want} # fmt: skip return value if ok[rule["op"]] else None def _run(lg: dict, master, examples: dict | None = None) -> dict: """상자 차례대로 — 입력(예시값 · 노임 · 표 · 재료 · 다른 로직) → 선 규칙 → 함수 시트(엔진) → 끝수 → 출력""" from common_util.common_util_spreadsheet import recalc_workbooks cm, files = master m = cm.mf.Master(files) tables = {str(t["키"]): t for d in files.values() if d.get("그룹") in cm.mf.TABLE_GROUPS for t in d.get("표") or []} # fmt: skip items = next(d for d in files.values() if d.get("그룹") == "재료" and any("흄관" in str(r.get("이름")) for r in d["줄"]))["줄"] # fmt: skip ports: dict[str, dict] = {} for b in _order(lg): into = {} for x in lg["links"]: if x["to"]["box"] == b["id"]: value, rule = ports[x["from"]["box"]][x["from"]["name"]], x.get("rule") or {} if rule.get("kind") == "gate": value = _gate(rule, value, ports) elif rule.get("kind") == "round": value = _cut(value, rule) into[x["to"]["name"]] = value if b["kind"] == "input": inp = b["input"] got = { v["name"]: (examples or {}).get(v["name"], v["example"]) for v in inp.get("variables", []) } for e in inp.get("elements", []): key = e["ref"] if ( e.get("mode") == "rule" ): # 범위 안 품목 중 규격 / 이름에 조건 알약 값이 든 것 하나 want = str(ports[e["rule"]["from"]["box"]][e["rule"]["from"]["name"]]) sc = e["scope"] hits = [r for r in items if r.get("구분") == sc["구분"] and r.get("상세구분") == sc["상세구분"] and sc["찾는말"] in r.get("이름", "") and want in str(r.get(e["rule"]["field"], ""))] # fmt: skip assert len(hits) == 1, (want, [h["키"] for h in hits]) key = hits[0]["키"] got[e["name"]] = _plain(cm.mf.element(m, key, {})[0]) if inp.get("table"): got["표"] = _table_rows(tables[inp["table"]["key"]]) if inp["mode"] == "logic": got = _run(SAMPLES[inp["logic"]["key"]], master) # 설계 값은 그 로직의 예시값 ports[b["id"]] = got elif b["kind"] == "function": fn = b["func"] sheet = json.loads(json.dumps(fn["sheet"])) for i in fn["inputs"]: value = into[i["port"]] col, row = ord(i["cell"][0]) - 65, int(i["cell"][1:]) for r, line in enumerate(value if isinstance(value, list) else [[value]]): for c, x in enumerate(line): if x is not None: sheet["칸"][f"{chr(65 + col + c)}{row + r}"] = {"값": _plain(x)} book = {"종류": "통합문서", "판": 1, "열": "df", "서식": [{}], "시트": [sheet]} got = recalc_workbooks([book])[0][sheet["id"]][fn["output"]["cell"]] assert "수" in got, (b["id"], got) ports[b["id"]] = { fn["output"]["name"]: _cut(float(got["수"]), fn["output"].get("round")) } else: ports[b["id"]] = {"값": _cut(into["값"], b["output"].get("round"))} return {b["name"]: ports[b["id"]]["값"] for b in lg["boxes"] if b["kind"] == "output"} def test_값_메쌓기_value_는_calc_와_같음(master) -> None: from M01_MasterData import M01_MasterData_Store as store given = {v["name"]: v["example"] for v in SAMPLES["masonry"]["boxes"][0]["input"]["variables"]} want = store.calc("GF000219", {**given, "초과증가율": 80}) assert abs(_run(SAMPLES["masonry"], master)["노무비"] - float(want["sums"]["계"])) < 1 given = {v["name"]: v["example"] for v in SAMPLES["value"]["boxes"][0]["input"]["variables"]} given["qt"] = given.pop("q") assert _run(SAMPLES["value"], master)["시간당 작업량"] == float( store.calc("GF000133", given)["result"] ) def test_값_지반_흄관_다른로직(master) -> None: # 지반 = 토사 → 굴삭기 길만 · 암 → 브레이커 길만(조건 통과 안 된 선은 빈 칸 = 0) assert _run(SAMPLES["ground"], master)["경비"] == 100 / 30 * 60000 / 100 assert _run(SAMPLES["ground"], master, {"지반": "암"})["경비"] == 100 / 8 * 80000 / 100 # 관경 600 → Φ600mm 품목 · 4 본 × 단가 ÷ 10 m · 0.1 원 미만 버림 cm, files = master price = float(cm.mf.element(cm.mf.Master(files), "MT014991", {})[0]) assert _run(SAMPLES["hume"], master)["재료비"] == _cut( 4 * price / 10, {"places": 1, "method": "버림"} ) # 다른 로직(작업량 15.552) → 운반 시간 120 ÷ 15.552 = 7.72(소수 둘째 반올림) → 7.72 × 50,000 ÷ 120 assert _run(SAMPLES["linked"], master)["경비"] == _cut( 7.72 * 50000 / 120, {"places": 1, "method": "버림"} )