From 466fa56540eebc599f997ab58d993b429465d2d9 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Thu, 1 Oct 2026 23:49:16 +0900 Subject: [PATCH] =?UTF-8?q?feat(M01):=20=ED=9D=90=EB=A6=84=20=EB=A1=9C?= =?UTF-8?q?=EC=A7=81=20=EC=98=9B=20=EC=97=94=EC=A7=84=20=ED=95=9C=20?= =?UTF-8?q?=EA=B3=B3=20=EA=B0=88=EB=9E=98=20=C2=B7=20=EC=9D=B8=EB=A0=A5/?= =?UTF-8?q?=EA=B8=B0=EA=B3=84/=EC=9C=A0=EA=B0=80=20=EC=A1=B0=EA=B1=B4=20?= =?UTF-8?q?=EC=B0=BE=EA=B8=B0=20=C2=B7=20=ED=91=9C=20=EC=88=98=20=EC=8B=9D?= =?UTF-8?q?=20=EA=B8=80=20=C2=B7=20=EC=9E=AC=EB=A3=8C=20=ED=82=A4=20?= =?UTF-8?q?=EC=B0=BE=EA=B8=B0(PLAN=2036-4=20=EB=8D=A7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - master_formula.run 에 FLOW_RUN 자리 — Store_Flow 가 꽂음 · 조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…) 도 흐름 로직이면 흐름 셈(단가는 마스터 · 프로젝트 사본 단가는 아직) - 흐름 → 옛 → 흐름 돌고 도는 부름 막음 · /logic/auto 흐름 로직 = 설계 예시값 - 조건 찾기 요소 인력 · 기계 · 유가전력(group_find · 이름 · 키 · 원문번호 같음/포함 · 기본 키 대체) - 표 값 칸 수 식 글(「(0.05)」 「1/1.30」) = 옛 엔진 파서로 수 · 재료 조건 찾기 field 키(자재 키 · 품번) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL --- M01_MasterData/M01_MasterData_Router.py | 5 +- M01_MasterData/M01_MasterData_Store_Flow.py | 50 +++++++- M02_MasterTemplete/DataFlow/M02_DF_Compute.py | 9 ++ M02_MasterTemplete/DataFlow/M02_DF_Values.py | 80 +++++++++++- .../master_data/scripts/master_formula.py | 7 + .../tester/test_m01_logic_flow_engine.py | 121 ++++++++++++++++++ 6 files changed, 259 insertions(+), 13 deletions(-) create mode 100644 resources/tester/test_m01_logic_flow_engine.py diff --git a/M01_MasterData/M01_MasterData_Router.py b/M01_MasterData/M01_MasterData_Router.py index de50a83e3..3706dd065 100644 --- a/M01_MasterData/M01_MasterData_Router.py +++ b/M01_MasterData/M01_MasterData_Router.py @@ -192,8 +192,9 @@ def get_logic(key: str) -> dict: @router.get("/logic/auto") def get_logic_auto(key: str) -> dict: - """견본 입력 — 마스터 보기용 자동값(프로젝트 기본값 아님).""" - return _call(store.auto, key) + """견본 입력 — 마스터 보기용 자동값(프로젝트 기본값 아님) · 흐름 로직은 설계 예시값.""" + got = _call(flow.auto, key) + return got if got is not None else _call(store.auto, key) @router.get("/elements") diff --git a/M01_MasterData/M01_MasterData_Store_Flow.py b/M01_MasterData/M01_MasterData_Store_Flow.py index eb2d329f7..fbbd0b358 100644 --- a/M01_MasterData/M01_MasterData_Store_Flow.py +++ b/M01_MasterData/M01_MasterData_Store_Flow.py @@ -15,6 +15,7 @@ from __future__ import annotations import importlib +from contextvars import ContextVar from decimal import Decimal, InvalidOperation from typing import Any @@ -26,6 +27,7 @@ FLOW = "흐름" KINDS = ("input", "function", "output") CONVERT = "M02_MasterTemplete.DataFlow.M02_DF_Convert" NOT_READY = "변환 준비 중" +_ACTIVE: ContextVar[tuple] = ContextVar("m01_flow_active", default=()) # 셈 중인 흐름 키 # ── 검사(Model `link` · `ruleWhy` 와 같은 뜻) ────────────────────────────── @@ -152,9 +154,46 @@ def _value(v: Any) -> Any: def run(flow: dict, key: str, inputs: dict) -> dict: """흐름 셈 — 입력 = 「설계 때 정할 것」 이름 → 값.""" - given = {k: _value(v) for k, v in (inputs or {}).items()} - flow = {**flow, "key": key} - return dc.run(flow, dc._design_values(flow, given), load, flow.get("userElements") or [], [key]) + active = _ACTIVE.get() + if key and key in active: # 옛 셈을 건너 다시 돌아온 흐름(흐름 → 옛 로직 → 흐름) + raise store.cm.mf.FormulaError(f"로직이 서로를 부름 — {' → '.join([*active, key])}") + token = _ACTIVE.set((*active, key)) + try: + given = {k: _value(v) for k, v in (inputs or {}).items()} + flow = {**flow, "key": key} + values = dc._design_values(flow, given) + return dc.run(flow, values, load, flow.get("userElements") or [], [key]) + finally: + _ACTIVE.reset(token) + + +def _engine(master, ref: str, row: dict, given: dict) -> dict: + """옛 엔진 `mf.run` 갈래(조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…)) — `run` 과 같은 꼴. + + ponytail: 단가는 마스터에서(`M02_DF_Values`) — 프로젝트 사본 단가(master 인자)는 아직 안 씀. + """ + got = run(row[FLOW], ref, given) + if got["errors"]: + raise store.cm.mf.FormulaError(f"「{ref}」 흐름 셈 — {' · '.join(got['errors'])}") + cost = got["cost"] + kinds = {(b.get("output") or {}).get("cost") for b in row[FLOW].get("boxes") or [] + if b.get("kind") == "output"} # fmt: skip + if kinds and kinds <= {"값"} and len(cost["값들"]) == 1: # 돈 아닌 로직 = 결과 하나 + return {"결과": next(iter(cost["값들"].values())), "중간": {}} + return {"줄": [], **{k: cost[k] for k in dc.COSTS}, "계": cost["계"], "중간": {}} + + +store.cm.mf.FLOW_RUN = _engine # 옛 엔진을 부르는 곳이 모두 흐름 로직을 흐름 셈으로(한 곳 갈래) + + +def auto(key: str) -> dict | None: + """견본 입력 — 흐름 로직이면 「설계 때 정할 것」 예시값 · 아니면 None(옛 자동값).""" + flow = store.logic(key)["logic"].get(FLOW) + if not isinstance(flow, dict): + return None + return { + "값": {d["port"]: (d.get("variable") or {}).get("example") for d in dc.design_inputs(flow)} + } def calc(key: str, inputs: dict, row: dict | None = None) -> dict | None: @@ -168,7 +207,10 @@ def calc(key: str, inputs: dict, row: dict | None = None) -> dict | None: return None if not isinstance(flow, dict): return None - got = run(flow, key, inputs) + try: + got = run(flow, key, inputs) + except store.cm.mf.FormulaError as e: + return {"ok": False, "flow": True, "reason": str(e)} names = {b["id"]: b for b in flow.get("boxes") or []} outputs = [ {"box": i, "name": names[i].get("name"), "cost": (names[i].get("output") or {}).get("cost"), diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py index e3d8cf96b..835319d54 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py @@ -132,6 +132,15 @@ def _element_value( f"「{box.get('name')} · {el.get('name')}」 {got['reason']} — 기본 품목으로 셈" ) got = dv.slot_price({"mode": "fixed", "ref": el["default"]}) + elif group in dv.GROUP_RULE and mode == "rule": # 인력 · 기계 · 유가전력 조건 찾기 + rule = el.get("rule") or {} + src = rule.get("from") or {} + slot = {"field": rule.get("field"), "op": rule.get("op"), "scope": el.get("scope"), + "ref": el.get("ref"), "unit": el.get("unit")} # fmt: skip + got = dv.group_find(group, slot, ports.get(src.get("box"), {}).get(src.get("name"))) + if not got["ok"] and el.get("default"): + errors.append(f"「{box.get('name')} · {el.get('name')}」 {got['reason']} — 기본으로 셈") + got = dv.one_value({"group": group, "ref": el["default"], "unit": el.get("unit")}) elif given is not None: return given else: diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Values.py b/M02_MasterTemplete/DataFlow/M02_DF_Values.py index 1e6fa07c3..cf7ebde93 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Values.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Values.py @@ -158,11 +158,11 @@ def _table(key: str, at: Any = None) -> dict: if not isinstance(at, int) or not 0 <= at < len(lines): return {"ok": False, "reason": f"「{table.get('이름')}」 에 {at} 번째 줄 없음"} lines = [lines[at]] - rows = [] + rows, worth = [], set(table.get("값칸") or {}) for r in lines: line = [] for n in names: - v = r.get(n) + v = _arith(r.get(n)) if n in worth else r.get(n) if n in split: line += list(v) if isinstance(v, list) else [v, v] else: @@ -171,6 +171,16 @@ def _table(key: str, at: Any = None) -> dict: return {"ok": True, "rows": [head, *rows], "name": table.get("이름"), "unit": table.get("기준")} +def _arith(v: Any) -> Any: + """값 칸 수 식 글(「(0.05)」 「1/1.30」) — 옛 셈처럼 수로(사칙연산 · 괄호만 · 옛 엔진 파서).""" + if not isinstance(v, str) or not cm.mf._ARITH.fullmatch(v): + return v + try: + return cm.mf.evaluate(cm.mf.parse(v), {}, None, 0) + except (cm.mf.FormulaError, ArithmeticError): + return v + + def _user(uid: str, local: list[dict]) -> dict: """사용자 요소 값 — local = 다이어그램 안 것 + 공용 목록(부르는 쪽이 모음).""" one = next((r for r in local if r.get("id") == uid), None) @@ -336,7 +346,7 @@ def materials(q: str, sub: str, detail: str, page: int, size: int) -> dict: return {"total": len(hits), "page": page, "size": size, "items": [_brief_row(r) for r in part]} -FIELDS_RULE, OPS_RULE = ("규격", "이름"), ("포함", "같음") # Model CfMaterialRule +FIELDS_RULE, OPS_RULE = ("규격", "이름", "키"), ("포함", "같음") # Model CfMaterialRule def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict: @@ -368,10 +378,13 @@ def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict: want = _squash(value) sub, detail = str(scope.get("구분") or ""), str(scope.get("상세구분") or "") rows = _found(sub, detail, scope.get("찾는말") or "") - if op == "같음": - hits = [r for r in rows if _squash(r.get(field)) == want] - else: - hits = [r for r in rows if want in _squash(r.get(field))] + numbers = _numbers() if field == "키" else {} # 키 = 자재 키 또는 품번(원문번호) + + def pool(r: dict) -> list[str]: + got = [r.get("키"), numbers.get(r.get("키"))] if field == "키" else [r.get(field)] + return [_squash(x) for x in got if x] + + hits = [r for r in rows if any(want == x if op == "같음" else want in x for x in pool(r))] if not hits: where = " · ".join(x for x in (sub, detail) if x) or "전체" return {"ok": False, "reason": f"맞는 재료 없음 — {where} 에서 {field}에 「{value}」 {op}"} @@ -386,6 +399,59 @@ def slot_price(slot: dict, value: Any = None, pick: str | None = None) -> dict: return {**got, "ref": hits[0].get("키"), "name": hits[0].get("이름")} +GROUP_RULE = ( + "인력", + "기계", + "유가전력", +) # 조건 찾기를 재료 밖에서도(옛 LB:{직종} · EQ:{기계} · MO:…) + + +def _group_rows(group: str) -> list[dict]: + """그룹 줄 — 인력 · 기계 = 그 파일 · 유가전력 = 재료 안 MO 키.""" + files = store.loaded()[0] + want = "재료" if group == "유가전력" else group + return [ + r + for d in files.values() + if d.get("그룹") == want + for r in d.get(store.items_key(d)) or [] + if group != "유가전력" or str(r.get("키")).startswith("MO") + ] + + +def group_find(group: str, slot: dict, value: Any) -> dict: + """인력 · 기계 · 유가전력 조건 찾기 — 알약 값이 이름 · 키(원문번호) 와 같음(포함)인 줄 하나. + + slot = {field?: 이름 | 키, op?: 같음 | 포함, scope?: {찾는말?}, ref?: 「키#칸」 = 칸 이어 씀} + · field 가 없으면 키(원문번호) 또는 이름이 같은 줄 · 하나면 그 값(`one_value`). + """ + if value in (None, ""): + return {"ok": False, "reason": "찾을 값이 비어 있음 — 이어진 알약을 확인해 주세요"} + field, op, want = slot.get("field"), slot.get("op") or "같음", _squash(value) + words = str((slot.get("scope") or {}).get("찾는말") or "").split() + + def hit(r: dict) -> bool: + hay = _squash(f"{r.get('이름')}{r.get('규격') or ''}") + if not all(_squash(w) in hay for w in words): + return False + keys = [_squash(r.get("키")), _squash(r.get("원문번호"))] + names = [_squash(r.get("이름"))] + pool = keys if field == "키" else names if field == "이름" else keys + names + return any(want == x if op == "같음" else want in x for x in pool if x) + + hits = [r for r in _group_rows(group) if hit(r)] + if len(hits) != 1: + if not hits: + return {"ok": False, "reason": f"맞는 {group} 없음 — 「{value}」"} + names = ", ".join(f"{r.get('이름')} {r.get('규격') or ''}".strip() for r in hits[:5]) + return {"ok": False, "reason": f"맞는 {group} {len(hits)}개 — 좁혀 주세요({names} …)"} + ref = str(hits[0].get("키")) + suffix = str(slot.get("ref") or "").partition(MACHINE_SEP)[2] + got = one_value({"group": group, "ref": f"{ref}{MACHINE_SEP}{suffix}" if suffix else ref, + "unit": slot.get("unit") or ""}) # fmt: skip + return {**got, "ref": ref, "name": hits[0].get("이름")} + + # ── 단위 · 금액 표준 표 ──────────────────────────────────────────────── STANDARDS = {"unit": "단위표준.json", "money": "금액단위표준.json"} diff --git a/resources/master_data/scripts/master_formula.py b/resources/master_data/scripts/master_formula.py index 83cc99f58..4bb9afb14 100644 --- a/resources/master_data/scripts/master_formula.py +++ b/resources/master_data/scripts/master_formula.py @@ -28,6 +28,11 @@ _OPS = ("<=", ">=", "==", "!=", "<", ">") _ARITH = re.compile(r"[\d.\s+\-*/()]*\d[\d.\s+\-*/()]*") # 분수·식 값 칸(「1/1.30」) +# 흐름도 로직(줄에 「흐름」) 셈 — 서버(`M01_MasterData_Store_Flow`)가 꽂음 · 없으면 옛 셈만 +# 꼴 = (master, 키, 줄, 입력) → `run` 과 같은 꼴(돈 = 줄 · 비목 합 · 계 / 돈 아님 = 결과) +FLOW_RUN = None + + class FormulaError(ValueError): pass @@ -538,6 +543,8 @@ def _blank_line(item: dict, why: str) -> dict: def run(master: Master, ref: str, given: dict, depth: int = 0) -> dict: """로직 한 줄 계산 — 돈 로직은 호표 줄 · 비목 합 · 계, 돈 아닌 로직은 결과 수 하나.""" row = master.logic(ref) + if FLOW_RUN is not None and isinstance(row.get("흐름"), dict): + return FLOW_RUN(master, ref, row, given) env = _inputs(row, given) for step in row.get("중간", []): try: diff --git a/resources/tester/test_m01_logic_flow_engine.py b/resources/tester/test_m01_logic_flow_engine.py new file mode 100644 index 000000000..124670cfd --- /dev/null +++ b/resources/tester/test_m01_logic_flow_engine.py @@ -0,0 +1,121 @@ +# ruff: noqa: E501 +"""흐름도 적용(PLAN 36-4 덧) — 옛 엔진 한 곳 갈래 · 인력 · 기계 · 유가전력 조건 찾기. + +못박는 것(마스터 사본 폴더) + - `mf.run` 을 바로 부르는 곳(조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…))도 흐름 로직이면 흐름 셈 + - `/logic/auto` — 흐름 로직은 「설계 때 정할 것」 예시값 + - 조건 찾기 요소가 인력 · 기계 · 유가전력에서도 이름 · 키(원문번호) 같음으로 찾음 +""" + +from __future__ import annotations + +import shutil +from decimal import Decimal +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from M01_MasterData import M01_MasterData_Router as m01_router +from M01_MasterData import M01_MasterData_Store as store +from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc +from M02_MasterTemplete.DataFlow import M02_DF_Values as dv +from resources.tester.test_m01_logic_flow_save import KEY, _flow, _get, _save_flow + +REAL = store.FOLDER + + +@pytest.fixture +def client(tmp_path: Path, monkeypatch) -> TestClient: + for path in REAL.glob("*.json"): + shutil.copy(path, tmp_path / path.name) + monkeypatch.setattr(store, "FOLDER", tmp_path) + app = FastAPI() + app.include_router(m01_router.router) + return TestClient(app) + + +def test_옛_엔진도_흐름_로직은_흐름_셈(client: TestClient) -> None: + _save_flow(client, KEY, _flow()) + whole = store.cm.mf.Master(store.cm.load(folder=store.FOLDER)) + got = store.cm.mf.run(whole, KEY, {"길이": Decimal(4), "폭": Decimal(5)}) + assert got["줄"] == [] and got["노무비"] == 20000 and got["계"] == 20000 + line = { + "종류": "로직", + "요소": f"로직({KEY}, 길이=4, 폭=5)", + "이름": "흐름 로직", + "단위": "㎡", + "수량": "2", + } + old = store.calc("", {}, dv._row(line, ()), None) # 옛 식이 흐름 로직을 부름 + assert old["ok"] and old["sums"]["노무비"] == 40000, old + auto = _get(client, "/api/m01/logic/auto", key=KEY) + assert auto == {"값": {"길이": 2, "폭": 3}} + + +def _labor_flow(field: str | None, name: str) -> dict: + flow = _flow() + el = {"ref": "", "name": "노임", "spec": "", "unit": "원/인", "mode": "rule", + "rule": {"field": field, "op": "같음", "from": {"box": "who", "name": "직종"}}} # fmt: skip + flow["boxes"] += [ + {"id": "who", "kind": "input", "name": "직종", "color": "", "unit": "-", "x": 0, "y": 0, + "input": {"mode": "variable", "elements": [], "variables": [ + {"name": "직종", "type": "choice", "unit": "-", "example": name, "choices": [name]}]}}, + {"id": "lab", "kind": "input", "name": "노임", "color": "", "unit": "-", "x": 0, "y": 0, + "input": {"mode": "group", "group": "인력", "elements": [el], "variables": []}}, + ] # fmt: skip + flow["boxes"][2]["output"]["cost"] = "값" + flow["links"][2] = { + "id": "l3", + "from": {"box": "lab", "name": "노임"}, + "to": {"box": "out", "name": "값"}, + } + return flow + + +def test_인력_기계_유가전력_조건_찾기(client: TestClient) -> None: + rows = dv._group_rows("인력") + one = next(r for r in rows if sum(x.get("이름") == r.get("이름") for x in rows) == 1) + want = dv.one_value({"group": "인력", "ref": one["키"]}) + for field, value in (("이름", one["이름"]), ("키", one["원문번호"]), (None, one["키"])): + got = dc.run(_labor_flow(field, value), {}, lambda k: {}, [], ["t"]) + assert got["errors"] == [] and got["cost"]["값들"]["노무비"] == want["value"], ( + field, + got["errors"], + ) + miss = dc.run(_labor_flow("이름", "없는직종zz"), {}, lambda k: {}, [], ["t"]) + assert any("맞는 인력 없음" in e for e in miss["errors"]) + machine = dv._group_rows("기계")[0] + got = dv.group_find("기계", {"field": "키"}, machine["원문번호"]) + assert got["ref"] == machine["키"] + oils = dv._group_rows("유가전력") + if oils: + got = dv.group_find("유가전력", {"field": "키"}, oils[0]["키"]) + assert got["ref"] == oils[0]["키"] + + +def test_표_값_수_식_글은_수로_재료_키_찾기(client: TestClient) -> None: + assert dv._arith("(0.05)") == Decimal("0.05") and dv._arith("1/1.30") == Decimal(1) / Decimal( + "1.30" + ) + assert ( + dv._arith("토사") == "토사" and dv._arith("1/0") == "1/0" + ) # 수 식 아니거나 못 세면 그대로 + files = store.loaded()[0] + hit = next( + ((t["키"], n) for d in files.values() if d.get("그룹") in store.cm.mf.TABLE_GROUPS + for t in d.get("표") or [] for r in t.get("줄") or [] for n in t.get("값칸") or {} + if isinstance(r.get(n), str) and store.cm.mf._ARITH.fullmatch(r[n])), + None, + ) # fmt: skip + if hit: # 마스터에 식 글 값이 있으면 표 알약에서 수로 나옴 + got = dv._table(hit[0]) + col = got["rows"][0].index(hit[1]) + assert not any( + isinstance(r[col], str) and store.cm.mf._ARITH.fullmatch(r[col]) + for r in got["rows"][1:] + ) + 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