Files
Aislo/M02_MasterTemplete/DataFlow/M02_DF_Compute.py
T
eomsangdonandClaude Opus 5.5 018177408b feat(m02-df): 끊긴 곳 무게 — 쉼(쉬는 갈래) · 문제
- CfBreak.level: 통과 안 됨 · 그 아래로 번진 빈 값 = 쉼 · 빈 받는 자리 · 표에 없는 조건 · 맞는 재료 없음 · 조건 겹침 · 안 맞음 · 오류 = 문제
- 쉬는 갈래에서 온 빈 값은 받는 상자도 쉼(셈 안 함 · 출력 빈 값 · 문제로 안 셈)
- M02_DF_Compute.py breaks 에 같은 뜻 level 한 칸
- 시험 25개

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VBToTCGk5dqDKeXCEP3vhk
2026-10-01 20:38:21 +09:00

559 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""단가산출 흐름도 서버 계산(PLAN 35-9 · 정함 4 · 5 · 6 · 7 · 12) — 저장 다시 셈 · 다른 로직 값.
모양 = sub7 Model `M02_DF_Model.ts`(CfLogic · CfLink.rule · CfElementRef.mode · CfInput.logic ·
CfRound). 화면 계산 `M02_DF_Eval.evalLogic` 과 같은 차례 · 같은 뜻으로 셈(짝 ① — 화면은 TS · 서버는
이 파일 · 함수 시트는 같은 엔진(Node · `common_util_spreadsheet`)):
· 상자 차례 = 선 · 선 규칙 조건 알약 · 재료 조건 찾기 알약에 기댐(Model `depends`) — 서로 안 기대는
함수 상자들은 한 번에 Node 로 셈(시트마다 통합문서 하나)
· 입력 상자 알약 = 요소(재료는 고정 · 선택 · 조건 찾기 — 알약 이름 = `elementPort`) · 표 · 변수 ·
다른 로직 출력 — 단가는 마스터에서 새로(`M02_DF_Values`) · 설계 값 values = {상자: {알약: 값}}
· 선 규칙 = `M02_DF_Rule.applyRule` 과 같은 뜻(값 기준) — 막히면(통과 안 됨 · 표에 없는 조건 ·
여럿 ·
오류) 받는 함수는 셈하지 않고 빈 값 · 끝수 = 엔진 ROUND · ROUNDUP · ROUNDDOWN 과 같음
· 다른 로직 = 저장 로직 · 견본을 받은 설계 값(선 · 검증 칸 · 없으면 그 로직 예시값)으로 셈 ·
서로 부르면 막음
마스터 파일은 읽기만.
"""
from __future__ import annotations
import copy
import re
from decimal import ROUND_DOWN, ROUND_HALF_UP, ROUND_UP, Decimal
from typing import Any, Callable
from common_util import common_util_spreadsheet as spreadsheet
from M01_MasterData import M01_MasterData_Store as store
from M01_MasterData.M01_MasterData_Router import _num
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
_CELL = re.compile(r"^([A-Z]+)(\d+)$")
TABLE_PORT, OUTPUT_PORT = "표", "값"
COSTS = ("노무비", "재료비", "경비")
GATE = {"=": "=", "≠": "<>", "≥": ">=", "≤": "<=", ">": ">", "<": "<"}
ROUND = {"반올림": "ROUND", "올림": "ROUNDUP", "버림": "ROUNDDOWN"}
HELP = "_df_" # 보조 시트 이름 머리(표 블록 · 끝수)
FAR = Decimal("1E+300") # 범위 칸 열린 끝 — 엑셀 비교가 늘 참이 되게
Loader = Callable[[str], dict]
def _col(letters: str) -> int:
n = 0
for ch in letters:
n = n * 26 + ord(ch) - 64
return n
def _letters(n: int) -> str:
out = ""
while n:
n, rest = divmod(n - 1, 26)
out = chr(65 + rest) + out
return out
def _at(cell: str) -> tuple[int, int, str]:
m = _CELL.match(str(cell or "").strip().upper())
if not m:
raise store.StoreError(400, f"칸 주소 모양이 아님 「{cell}」")
return _col(m.group(1)), int(m.group(2)), f"{m.group(1)}{m.group(2)}"
def _style(cells: dict, cell: str) -> dict:
return {"서식": cells[cell]["서식"]} if "서식" in (cells.get(cell) or {}) else {}
def _put(cells: dict, cell: str, value: Any) -> None:
"""칸 하나 — 값 · 수식 `{"식"}` · 빈 값은 값만 뺌(서식은 둠)."""
keep = _style(cells, cell)
if isinstance(value, dict) and "식" in value:
cells[cell] = {"식": value["식"], **keep}
elif value is None or value == "":
cells.pop(cell, None) if not keep else cells.__setitem__(cell, keep)
else:
cells[cell] = {"값": value, **keep}
def _place(cells: dict, cell: str, value: Any) -> None:
"""받는 칸에 값 · 수식 · 표 블록(cell 이 왼위)."""
col, row, at = _at(cell)
if isinstance(value, list):
for dy, line in enumerate(value):
for dx, one in enumerate(line):
_put(cells, f"{_letters(col + dx)}{row + dy}", one)
else:
_put(cells, at, value)
def _plain(v: Any) -> Any:
"""엔진 계산값 한 칸 → 수 · 글 · 참거짓 · 빈 값(오류는 None)."""
if not isinstance(v, dict):
return v
if "수" in v:
return Decimal(str(v["수"]))
return v.get("글", v.get("참"))
def _literal(v: Any) -> str:
"""수식 안 값 글 — 수는 그대로 · 글은 따옴표 · 빈 값은 빈 글."""
if v is None:
return '""'
if isinstance(v, bool):
return "TRUE" if v else "FALSE"
if isinstance(v, (int, float, Decimal)):
return str(v)
return '"' + str(v).replace('"', '""') + '"'
# ── 입력 상자 ──────────────────────────────────────────────────────────
def _element_value(
group: str, el: dict, given: Any, ports: dict, errors: list, box: dict, ctx: dict
) -> Any:
"""요소 알약 값 — 재료는 방식 셋 · 그 밖은 그룹 값 · 설계 값이 이김(선택 자리 = 키 · 수)."""
mode = el.get("mode") or "fixed"
if group == "재료" and mode == "choice":
if isinstance(given, (int, float, Decimal)) and not isinstance(given, bool):
return given
got = dv.slot_price({"mode": "choice", "default": el.get("default")}, pick=given or None)
elif group == "재료" and mode == "rule":
rule = el.get("rule") or {}
src = rule.get("from") or {}
slot = {
"mode": "rule",
"scope": el.get("scope"),
"field": rule.get("field"),
"op": rule.get("op"),
}
got = dv.slot_price(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.slot_price({"mode": "fixed", "ref": el["default"]})
elif given is not None:
return given
else:
got = dv.one_value(
{"group": group, "ref": el.get("ref"), "unit": el.get("unit")}, ctx["local"]
)
if not got["ok"]:
errors.append(f"「{box.get('name')} · {el.get('name')}」 {got.get('reason')}")
return None
return got.get("value")
def _port(el: dict) -> str:
"""요소 알약 이름 — Model `elementPort`(고정 아니면 자리 이름 slot · 없으면 name)."""
slot = str(el.get("slot") or "").strip()
return slot if el.get("mode") not in (None, "fixed") and slot else el.get("name")
def _input_ports(lg: dict, box: dict, values: dict, ports: dict, errors: list, ctx: dict) -> dict:
inp, given, out = box.get("input") or {}, values.get(box["id"]) or {}, {}
mode = inp.get("mode")
if mode == "logic":
return _logic_ports(lg, box, given, ports, errors, ctx)
if mode == "group":
group = inp.get("group")
for el in inp.get("elements") or []:
out[_port(el)] = _element_value(
group, el, given.get(_port(el)), ports, errors, box, ctx
)
if inp.get("table"):
got = dv.one_value({"group": group, "table": inp["table"]}, ctx["local"])
out[TABLE_PORT] = got.get("rows") if got["ok"] else None
if not got["ok"]:
errors.append(f"「{box.get('name')} · 표」 {got.get('reason')}")
for v in inp.get("variables") or []:
out[v.get("name")] = given.get(v.get("name"), v.get("example"))
return out
def _logic_ports(lg: dict, box: dict, given: dict, ports: dict, errors: list, ctx: dict) -> dict:
"""다른 로직 — 받는 알약(설계 값) = 선 · 검증 칸 값 · 없으면 그 로직 예시값 → 그 로직 출력."""
ref = (box.get("input") or {}).get("logic") or {}
key = str(ref.get("key") or "")
if key in ctx["stack"]:
errors.append(
f"「{box.get('name')}」 로직이 서로를 부름 — {' → '.join([*ctx['stack'], key])}"
)
return {}
try:
other = ctx["load"](key)
except store.StoreError as e:
errors.append(f"「{box.get('name')}」 {e.detail}")
return {}
designs = {**given}
for link in lg.get("links") or []:
if link.get("to", {}).get("box") == box["id"]:
src = link.get("from") or {}
designs[link["to"].get("name")] = ports.get(src.get("box"), {}).get(src.get("name"))
stack = [*ctx["stack"], key]
result = run(other, _design_values(other, designs), ctx["load"], ctx["local"], stack)
errors += [f"「{box.get('name')}」 {x}" for x in result["errors"]]
names = {b["id"]: b.get("name") for b in other.get("boxes") or []}
return {names[i]: v for i, v in result["outputs"].items() if i in names}
def _design_values(lg: dict, designs: dict) -> dict:
"""다른 로직의 설계 값 {알약 이름: 값} → 그 로직의 values {상자 id: {알약: 값}}."""
out: dict[str, dict] = {}
for d in design_inputs(lg):
if d["port"] in designs and designs[d["port"]] is not None:
out.setdefault(d["box"], {})[d["port"]] = designs[d["port"]]
return out
# ── 설계 때 정할 것 ────────────────────────────────────────────────────
def design_inputs(lg: dict) -> list[dict]:
"""변수 · 재료 자리(선택) · 다른 로직의 받는 알약(선이 없을 때) — Model `CfDesign` 꼴."""
links = lg.get("links") or []
fed = {
(ln.get("to") or {}).get("box", "") + "\0" + (ln.get("to") or {}).get("name", "")
for ln in links
}
out = []
for box in lg.get("boxes") or []:
inp = box.get("input") or {}
if box.get("kind") != "input":
for v in (box.get("func") or {}).get("variables") or []:
out.append(
{
"box": box["id"],
"port": v.get("name"),
"kind": "변수",
"unit": v.get("unit") or "-",
"variable": v,
}
)
continue
for v in inp.get("variables") or []:
out.append(
{
"box": box["id"],
"port": v.get("name"),
"kind": "변수",
"unit": v.get("unit") or "-",
"variable": v,
}
)
for el in inp.get("elements") or [] if inp.get("group") == "재료" else []:
if el.get("mode") == "choice":
out.append(
{
"box": box["id"],
"port": _port(el),
"kind": "재료 자리",
"unit": el.get("unit") or "-",
"element": el,
}
)
for pill in (
((inp.get("logic") or {}).get("designs") or []) if inp.get("mode") == "logic" else []
):
if box["id"] + "\0" + pill.get("name", "") not in fed:
out.append(
{
"box": box["id"],
"port": pill.get("name"),
"kind": "다른 로직",
"unit": pill.get("unit") or "-",
}
)
return out
# ── 선 규칙(`M02_DF_Rule.applyRule` 와 같은 뜻 · 값 기준) ────────────────────
def _num_of(v: Any) -> Decimal | None:
if isinstance(v, bool) or v is None:
return None
if isinstance(v, (int, float, Decimal)):
return Decimal(str(v))
return None
def _same(a: Any, b: Any) -> bool:
"""엑셀 「=」 — 수끼리는 수로 · 글끼리는 대소문자 무시 · 빈 칸은 0 · 빈 글과 같음."""
na, nb = _num_of(a), _num_of(b)
if a is None or b is None:
other = b if a is None else a
return other is None or other == "" or _num_of(other) == 0
if na is not None and nb is not None:
return na == nb
if isinstance(a, str) and isinstance(b, str):
return a.lower() == b.lower()
return False
def _compare(a: Any, op: str, b: Any) -> bool:
if op in ("=", "≠"):
return _same(a, b) == (op == "=")
na, nb = _num_of(a if a is not None else 0), _num_of(b)
if na is None or nb is None:
sa, sb = str(a or "").lower(), str(b or "").lower()
na, nb = (sa, sb) if na is None and nb is None else (None, None)
if na is None:
return False
return {"≥": na >= nb, "≤": na <= nb, ">": na > nb, "<": na < nb}.get(op, False)
def _rounded(value: Any, r: dict) -> Any:
"""끝수 — 엔진 ROUND(반올림 · 0 에서 먼 쪽) · ROUNDUP · ROUNDDOWN · 자리는 음수도."""
n = _num_of(value)
if n is None:
return value
mode = {"반올림": ROUND_HALF_UP, "올림": ROUND_UP, "버림": ROUND_DOWN}.get(
r.get("method"), ROUND_HALF_UP
)
places = int(r.get("places") or 0)
return (
n.quantize(Decimal(1).scaleb(-places), rounding=mode)
if places >= 0
else ((n.scaleb(places).quantize(Decimal(1), rounding=mode)).scaleb(-places))
)
def _lookup(rule: dict, block: Any, conds: dict) -> tuple[Any, str | None]:
"""표에서 찾기 — 같음 · 범위(아래 < 값 ≤ 위 · 빈 끝 열림 · 아래 = 위면 같음) · 맞은 줄 하나."""
if not isinstance(block, list) or len(block) < 2:
return None, "오류"
head = [str(h if h is not None else "") for h in block[0]]
out = rule.get("col")
if out not in head:
return None, "오류"
tests = []
for c in rule.get("conds") or []:
col, want = str(c.get("col")), conds.get((c.get("from") or {}).get("name"))
if col in head:
i = head.index(col)
tests.append(lambda row, i=i, want=want: _same(row[i], want))
elif col + dv.RANGE_LOW in head and col + dv.RANGE_HIGH in head:
lo, hi, x = (
head.index(col + dv.RANGE_LOW),
head.index(col + dv.RANGE_HIGH),
_num_of(want),
)
def fits(row, lo=lo, hi=hi, x=x) -> bool:
low, high = _num_of(row[lo]), _num_of(row[hi])
if low is not None and low == high: # 숫자 하나(아래 = 위) = 같음(브레인 정함)
return x is not None and x == low
return (
x is not None
and (row[lo] is None or (low is not None and low < x))
and (row[hi] is None or (high is not None and x <= high))
)
tests.append(fits)
else:
return None, "오류"
hits = [row for row in block[1:] if all(t(list(row) + [None] * len(head)) for t in tests)]
if not hits:
return None, "표에 없는 조건"
if len(hits) > 1:
return None, "표에 여럿 맞음"
value = hits[0][head.index(out)]
return (None, "오류") if value is None else (value, None)
def apply_rule(rule: dict, value: Any, conds: dict) -> tuple[Any, str | None]:
"""(값, 까닭) — 까닭 = 통과 안 됨 · 표에 없는 조건 · 표에 여럿 맞음 · 오류 · None."""
kind = rule.get("kind")
if kind == "lookup":
return _lookup(rule, value, conds)
if kind == "gate":
src = (rule.get("from") or {}).get("name")
return (
(value, None)
if _compare(conds.get(src), rule.get("op"), rule.get("value"))
else (None, "통과 안 됨")
)
n = _num_of(value)
if n is None:
return (None, None) if value is None else (None, "오류")
if kind == "round":
return _rounded(n, rule), None
v = Decimal(str(rule.get("value")))
op = rule.get("op")
if op == "÷" and v == 0:
return None, "오류"
return {"×": n * v, "÷": n / v, "+%": n * (1 + v / 100), "-%": n * (1 - v / 100)}.get(
op, n
), None
def _sources(rule: dict | None) -> list[dict]:
if not rule:
return []
if rule.get("kind") == "lookup":
return [c.get("from") or {} for c in rule.get("conds") or []]
return [rule.get("from") or {}] if rule.get("kind") == "gate" else []
# ── 계산(`M02_DF_Eval.evalLogic` 와 같은 차례) ─────────────────────────────
def _depends(lg: dict) -> list[tuple[str, str]]:
"""상자 사이 기댐 — 선 + 선 규칙 조건 알약 + 재료 조건 찾기 알약(Model `depends`)."""
out = []
for ln in lg.get("links") or []:
out.append((ln["from"].get("box"), ln["to"].get("box")))
out += [(p.get("box"), ln["to"].get("box")) for p in _sources(ln.get("rule"))]
for b in lg.get("boxes") or []:
for el in (b.get("input") or {}).get("elements") or []:
if el.get("mode") == "rule" and el.get("rule"):
out.append(((el["rule"].get("from") or {}).get("box"), b["id"]))
return out
def _waves(lg: dict) -> tuple[list[list[dict]], list[dict]]:
"""차례 물결 — 한 물결 안 상자는 서로 안 기댐(함수 시트를 한 번에 셈) · 돌고 도는 것은 따로."""
deps, boxes = _depends(lg), lg.get("boxes") or []
left, waves = {b["id"] for b in boxes}, []
while left:
ready = [
b
for b in boxes
if b["id"] in left and not any(t == b["id"] and f != t and f in left for f, t in deps)
]
if not ready:
break
waves.append(ready)
left -= {b["id"] for b in ready}
return waves, [b for b in boxes if b["id"] in left]
def _sheet_of(box: dict, arrived: dict, given: dict) -> dict:
"""함수 상자 시트 — 받는 칸 · 변수 칸에 값(Eval `runFunc` 와 같음)."""
func = box["func"]
sheet = copy.deepcopy(func.get("sheet") or {})
cells = sheet.setdefault("칸", {})
for p in func.get("inputs") or []:
if str(p.get("cell") or "").strip():
port = p.get("port")
_place(
cells,
p["cell"],
arrived[port] if arrived.get(port) is not None else given.get(port),
)
for v in func.get("variables") or []:
if v.get("cell"):
_place(cells, v["cell"], given.get(v.get("name"), v.get("example")))
return sheet
def run(lg: dict, values: dict, load: Loader, local: list[dict], stack: list | None = None) -> dict:
"""셈 — {ports, outputs, cost, errors, breaks, filled}(filled = 받는 칸에 값만 되돌려 넣음)."""
ctx = {"load": load, "local": local, "stack": stack or [lg.get("key") or ""]}
boxes = {b["id"]: b for b in lg.get("boxes") or []}
links = lg.get("links") or []
name = lambda i: boxes.get(i, {}).get("name") or i # noqa: E731
errors, breaks, ports, arrived, outputs = [], [], {}, {}, {}
blocked: set[str] = set()
cost: dict[str, Any] = {**dict.fromkeys(COSTS, Decimal(0)), "값들": {}}
def through(ln: dict) -> Any:
src = ln.get("from") or {}
v = ports.get(src.get("box"), {}).get(src.get("name"))
rule = ln.get("rule")
if not rule:
return v
if v is None and rule.get("kind") in ("scale", "round"):
return None
conds = {
p.get("name"): ports.get(p.get("box"), {}).get(p.get("name")) for p in _sources(rule)
}
got, why = apply_rule(rule, v, conds)
if why:
blocked.add(ln["to"].get("box"))
text = f"「{name(src.get('box'))}」 → 「{name(ln['to'].get('box'))}」 {why}"
kind = "오류" if why in ("표에 여럿 맞음", "오류") else why
to = ln["to"]
breaks.append(
{
"kind": kind,
# 무게 — 통과 안 됨 = 쉼(조건이 안 맞아 쉬는 갈래) · 나머지 = 문제(Eval CfBreak.level 과 같음)
"level": "쉼" if kind == "통과 안 됨" else "문제",
"box": to.get("box"),
"port": to.get("name"),
"link": ln.get("id"),
"text": text,
}
)
if why != "통과 안 됨":
errors.append(text)
return None if why else got
def arrive(box: dict) -> dict:
got = {
ln["to"].get("name"): through(ln) for ln in links if ln["to"].get("box") == box["id"]
}
arrived[box["id"]] = got
return got
waves, loop = _waves(lg)
errors += [f"「{b.get('name')}」 상자가 서로를 기다림(돌고 도는 선)" for b in loop]
sheets: dict[str, dict] = {}
for wave in waves:
funcs = []
for b in wave:
given = values.get(b["id"]) or {}
if b.get("kind") == "input" and (b.get("input") or {}).get("mode") == "logic":
ports[b["id"]] = _logic_ports(lg, b, {**given, **arrive(b)}, ports, errors, ctx)
elif b.get("kind") == "input":
ports[b["id"]] = _input_ports(lg, b, values, ports, errors, ctx)
elif b.get("kind") == "function" and b.get("func"):
sheet = _sheet_of(b, arrive(b), given)
sheets[b["id"]] = sheet
if b["id"] in blocked:
ports[b["id"]] = {b["func"]["output"].get("name"): None}
else:
funcs.append(b)
elif b.get("kind") == "output":
outputs[b["id"]] = arrive(b).get(OUTPUT_PORT)
if funcs: # 이 물결의 함수 시트를 한 번에(시트마다 통합문서 하나 — Eval 과 같음)
books = [
_num({"종류": "통합문서", "판": 1, "서식": [{}] * 64, "시트": [sheets[f["id"]]]})
for f in funcs
]
try:
calcs = spreadsheet.recalc_workbooks(books)
except spreadsheet.RecalcError as e:
raise store.StoreError(503, str(e)) from e
for f, book, calc in zip(funcs, books, calcs):
sheet = book["시트"][0]
for cell, v in (calc.get(sheet.get("id")) or {}).items():
if "오류" in v:
errors.append(f"「{f.get('name')}」 {cell} 칸 계산 오류 {v['오류']}")
out = f["func"].get("output") or {}
at = _at(out.get("cell") or "A1")[2]
hit = (calc.get(sheet.get("id")) or {}).get(at)
value = (
_plain(hit)
if hit is not None
else ((sheet.get("칸") or {}).get(at) or {}).get("값")
)
if out.get("round") and _num_of(value) is not None:
value = _rounded(value, out["round"])
ports[f["id"]] = {out.get("name"): value}
for b in boxes.values(): # 출력 상자 · 원가(Eval 과 같음)
if b.get("kind") != "output":
continue
value = outputs.get(b["id"])
value = None if isinstance(value, list) else value
if (b.get("output") or {}).get("round") and _num_of(value) is not None:
value = _rounded(value, b["output"]["round"])
outputs[b["id"]] = value
kind = (b.get("output") or {}).get("cost") or "값"
if kind == "값":
cost["값들"][b.get("name")] = value
elif _num_of(value) is not None:
cost[kind] += _num_of(value)
elif value is not None:
errors.append(f"「{b.get('name')}」 출력 값이 수가 아님")
cost["계"] = sum(cost[k] for k in COSTS)
filled = copy.deepcopy(lg) # 받는 칸 · 변수 칸에 값만 되돌려 넣음
for b in filled.get("boxes") or []:
if b["id"] in sheets:
b["func"]["sheet"] = sheets[b["id"]]
return {"ports": ports, "arrived": arrived, "outputs": outputs, "cost": cost, "errors": errors,
"breaks": breaks, "filled": filled} # fmt: skip