diff --git a/M01_MasterData/M01_MasterData_Store_Flow.py b/M01_MasterData/M01_MasterData_Store_Flow.py index fbbd0b358..2bd82fd28 100644 --- a/M01_MasterData/M01_MasterData_Store_Flow.py +++ b/M01_MasterData/M01_MasterData_Store_Flow.py @@ -22,6 +22,7 @@ from typing import Any from M01_MasterData import M01_MasterData_Store as store from M01_MasterData import M01_MasterData_Store_Make as make from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc +from M02_MasterTemplete.DataFlow import M02_DF_Values as dv FLOW = "흐름" KINDS = ("input", "function", "output") @@ -112,7 +113,18 @@ def check(flow: Any) -> list[str]: # ── 다른 로직 · 계산 ─────────────────────────────────────────────────── def _old(key: str, designs: dict) -> dict: """옛 로직 값 — 옛 셈 · 내보내는 알약 = 노무비 · 재료비 · 경비 · 계(돈 아닌 로직은 결과).""" - got = store.calc(key, {k: v for k, v in designs.items() if v is not None}) + given = {k: v for k, v in designs.items() if v is not None} + whole, mf = dv.SOURCE.get(), store.cm.mf + if whole is not None: # 단가 자리(프로젝트 사본)의 옛 로직 + try: + given = mf.fill_empty(whole.logic(key), {k: _value(v) for k, v in given.items()}) + got = mf.run(whole, key, given) + except (mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e: + return {"outputs": {}, "errors": [str(e) or type(e).__name__]} + if "결과" in got: + return {"outputs": {"결과": got["결과"]}, "errors": []} + return {"outputs": {k: got[k] for k in mf.BUNDLE}, "errors": []} + got = store.calc(key, given) if not got.get("ok"): return {"outputs": {}, "errors": [got.get("reason") or "옛 셈 오류"]} if "result" in got: @@ -122,7 +134,8 @@ def _old(key: str, designs: dict) -> dict: def load(key: str) -> dict: """다른 로직 입력이 부르는 마스터 로직 — 흐름이면 흐름 · 옛 로직이면 옛 셈 함수(`calc`).""" - row = store.logic(key)["logic"] + whole = dv.SOURCE.get() + row = whole.logic(key) if whole is not None else store.logic(key)["logic"] if isinstance(row.get(FLOW), dict): return {**row[FLOW], "key": key} return {"key": key, "calc": lambda designs: _old(key, designs)} @@ -170,9 +183,13 @@ def run(flow: dict, key: str, inputs: dict) -> dict: def _engine(master, ref: str, row: dict, given: dict) -> dict: """옛 엔진 `mf.run` 갈래(조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…)) — `run` 과 같은 꼴. - ponytail: 단가는 마스터에서(`M02_DF_Values`) — 프로젝트 사본 단가(master 인자)는 아직 안 씀. + 단가 · 값 · 표 = 부른 쪽이 넘긴 master(프로젝트 사본이면 그 단가 · `M02_DF_Values.SOURCE`). """ - got = run(row[FLOW], ref, given) + token = dv.SOURCE.set(master) + try: + got = run(row[FLOW], ref, given) + finally: + dv.SOURCE.reset(token) if got["errors"]: raise store.cm.mf.FormulaError(f"「{ref}」 흐름 셈 — {' · '.join(got['errors'])}") cost = got["cost"] diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Values.py b/M02_MasterTemplete/DataFlow/M02_DF_Values.py index cf7ebde93..0cc31950f 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Values.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Values.py @@ -11,6 +11,7 @@ from __future__ import annotations import json import re +from contextvars import ContextVar from typing import Any from M01_MasterData import M01_MasterData_Store as store @@ -21,6 +22,36 @@ RANGE_LOW, RANGE_HIGH = "·아래", "·위" FIELDS = (*cm.mf.COST_ITEMS, "계") +# 단가 자리 — `mf.run` 을 부른 쪽이 넘긴 Master(프로젝트 사본이면 그 단가) · 없으면 마스터 +# ponytail: 단가 · 값 · 표만 이 자리 · 찾기 목록(재료 조건 찾기 · 그룹 찾기 · 기계 짝)은 마스터에서 +SOURCE: ContextVar = ContextVar("df_price_source", default=None) +_DRAFT = "GX999998" # 단가 자리에 잠깐 얹는 견본 로직 키 + + +def _whole(): + return SOURCE.get() or store.loaded()[1] + + +def _calc(inputs: dict, row: dict) -> dict: + """`Store.calc("", inputs, row)` 와 같은 꼴 — 단가 자리가 넘겨졌으면 그 Master 로 셈.""" + whole = SOURCE.get() + if whole is None: + return store.calc("", inputs, row, None) + slot = whole.index.setdefault(_DRAFT[:2], {}) + slot[_DRAFT] = {**store._dec(row), "키": _DRAFT} + try: + given = cm.mf.fill_empty(slot[_DRAFT], {k: store._dec(v) for k, v in inputs.items()}) + got = cm.mf.run(whole, _DRAFT, given) + except (cm.mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e: + return {"ok": False, "reason": str(e) or type(e).__name__} + finally: + slot.pop(_DRAFT, None) + if "결과" in got: + return {"ok": True, "result": got["결과"], "middle": got["중간"]} + sums = {k: got[k] for k in cm.mf.BUNDLE} + return {"ok": True, "lines": got["줄"], "sums": sums, "middle": got["중간"]} + + # ── 입력 값 ──────────────────────────────────────────────────────────── def _row(line: dict, names) -> dict: """단가를 `/calc` 와 같은 셈으로 — 호표 한 줄(수량 1)짜리 저장 안 하는 견본 로직.""" @@ -65,7 +96,7 @@ def _pick(key: Any) -> dict | None: def _material(cond: dict, unit: str) -> dict: """검색어 묶음 재료 단가 — 기존 로직 재료 줄과 같은 길(검색어 · 자동값 차례 · 단위 환산).""" line = {"종류": "재료", "요소": cond, "이름": cond.get("검색어") or "재료", "수량": "1"} - got = store.calc("", {}, _row({**line, "단위": unit, "비목": "재료비"}, ()), None) + got = _calc({}, _row({**line, "단위": unit, "비목": "재료비"}, ())) if not got["ok"]: return got one = got["lines"][0] @@ -85,7 +116,7 @@ def _material(cond: dict, unit: str) -> dict: def _brief(key: str, want: str) -> tuple[dict | None, str | None]: """마스터 줄 하나 + 그룹 확인 — (줄, 까닭) · 유가전력은 재료 그룹 안 MO 키.""" try: - row = store.loaded()[1].get(key) + row = _whole().get(key) except cm.mf.FormulaError as e: return None, str(e) group = store.mk.GROUP_OF.get(store.mk.id_of(key)) @@ -102,7 +133,7 @@ def _price(group: str, key: str, unit: str = "") -> dict: kind, cost = {"인력": ("인력", "노무비"), "재료": ("재료", "재료비")}[group] line = {"종류": kind, "요소": key, "이름": row.get("이름") or key, "수량": "1"} line = {**line, "단위": unit or row.get("단위") or "", "비목": cost} - got = store.calc("", {}, _row(line, ()), None) + got = _calc({}, _row(line, ())) if not got["ok"]: return got one = got["lines"][0] @@ -135,6 +166,12 @@ def _plain_value(group: str, key: str, field: str | None) -> dict: def _table(key: str, at: Any = None) -> dict: """소요량 · 계수 표 → 머리 줄 + 줄(조건 칸 → 값 칸 · 범위가 있는 칸은 「·아래」 「·위」 두 칸 · 그 칸에 수 하나면 아래 · 위 같은 수) · at = 줄 차례(0부터)면 머리 줄 + 그 줄 하나.""" + if SOURCE.get() is not None: # 단가 자리(프로젝트 사본)의 표 + try: + table = SOURCE.get().get(key) + except cm.mf.FormulaError: + table = None + return _table_of(table, key, at) files = store.loaded()[0] table = next( ( @@ -146,6 +183,10 @@ def _table(key: str, at: Any = None) -> dict: ), None, ) + return _table_of(table, key, at) + + +def _table_of(table: dict | None, key: str, at: Any) -> dict: if table is None: return {"ok": False, "reason": f"없는 표 「{key}」"} conds, lines = table.get("조건") or {}, table.get("줄") or [] @@ -236,7 +277,7 @@ def _machine_hourly(key: str, inputs: dict) -> dict: args = "".join(f", {n}='{v}'" for n, v in given.items()) ref = f"로직({logic['키']}, 기계='{code}'{args})" line = {"종류": "로직", "요소": ref, "이름": "기계", "단위": "시간", "수량": "1"} - got = _money(store.calc("", {}, _row(line, ()), None), "계") + got = _money(_calc({}, _row(line, ())), "계") return {**got, "name": row.get("이름"), "unit": "원/시간"} if got["ok"] else got @@ -269,7 +310,7 @@ def one_value(item: dict, local: list[dict] | None = None) -> dict: elif group == "기계" and str(ref or "").startswith("로직("): inputs = item.get("inputs") or {} line = {"종류": "로직", "요소": ref, "이름": "기계", "단위": "시간", "수량": "1"} - got = _money(store.calc("", inputs, _row(line, inputs), None), field) + got = _money(_calc(inputs, _row(line, inputs)), field) elif group == "기계" and field == HOURLY: got = _machine_hourly(str(ref or ""), item.get("inputs") or {}) elif group in ("기계", "유가전력", "환율", "요율"): diff --git a/resources/tester/test_m01_logic_flow_engine.py b/resources/tester/test_m01_logic_flow_engine.py index 124670cfd..58ccea2e3 100644 --- a/resources/tester/test_m01_logic_flow_engine.py +++ b/resources/tester/test_m01_logic_flow_engine.py @@ -119,3 +119,24 @@ def test_표_값_수_식_글은_수로_재료_키_찾기(client: TestClient) -> item = next(r for r in dv._group_rows("재료") if not str(r.get("키")).startswith("MO")) got = dv.slot_price({"mode": "rule", "field": "키", "op": "같음", "scope": {}}, item["키"]) assert got.get("ref") == item["키"], got + + +def test_흐름_셈_단가는_부른_쪽_자리_프로젝트_사본(client: TestClient) -> None: + """옛 길과 같은 입력 → 같은 단가 — 프로젝트 사본(M02_Item_Prices 의 Master)이면 그 사본 노임.""" + flow = _flow() + el = {"ref": "LB000001", "name": "작업반장", "spec": "", "unit": "원/일", "mode": "fixed"} + flow["boxes"].append({"id": "lab", "kind": "input", "name": "노임", "color": "", "unit": "-", "x": 0, "y": 0, + "input": {"mode": "group", "group": "인력", "elements": [el], "variables": []}}) # fmt: skip + flow["links"][2] = { + "id": "l3", + "from": {"box": "lab", "name": "작업반장"}, + "to": {"box": "out", "name": "값"}, + } + _save_flow(client, KEY, flow) + files = store.cm.load(folder=store.FOLDER) + master = store.cm.mf.run(store.cm.mf.Master(files), KEY, {}) + labor = next(r for d in files.values() for r in d.get("줄") or [] if r.get("키") == "LB000001") + assert master["노무비"] == labor["값"] + labor["값"] = Decimal(100000) # 프로젝트 사본 노임 + copy = store.cm.mf.run(store.cm.mf.Master(files), KEY, {}) + assert copy["노무비"] == 100000 and master["노무비"] != 100000