feat(M01): 흐름 셈 단가 = mf.run 을 부른 쪽 자리(프로젝트 사본이면 그 단가 · PLAN 36-4 브레인 답)

- M02_DF_Values.SOURCE — 단가 · 값 · 표를 넘겨진 Master 에서(없으면 마스터) · 견본 로직은 그 Master 에 잠깐 얹어 셈
- Store_Flow 옛 엔진 갈래가 master 를 넘김 · 다른 로직(흐름 · 옛)도 같은 자리
- 찾기 목록(재료 조건 찾기 · 그룹 찾기 · 기계 짝)은 마스터 그대로

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
This commit is contained in:
2026-10-01 23:57:25 +09:00
co-authored by Claude Opus 5.5
parent 6c4f852ecc
commit f0ecaa122d
3 changed files with 88 additions and 9 deletions
+21 -4
View File
@@ -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"]
+46 -5
View File
@@ -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 ("기계", "유가전력", "환율", "요율"):
@@ -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