Files
Aislo/M02_MasterTemplete/DataFlow/M02_DF_Compute.py
T

519 lines
22 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`)과 같은 차례 · 같은 엔진 — 서버는 함수 상자 시트들을 한 통합문서로
묶어(시트 id = 상자 id) 우리 스프레드시트 엔진(Node · `common_util_spreadsheet`)으로 한 번에 셈.
· 입력 상자 알약 = 요소(재료는 고정 · 선택 · 조건 찾기) · 표 · 변수 · 다른 로직 출력 —
단가는 마스터에서 새로(`M02_DF_Values`) · 설계 값 values = {상자 id: {알약: 값}} 이 이김
(재료 선택 자리는 고른 품목 키 · 수)
· 선 → 받는 칸: 값 · 표 블록(왼위) · 앞 함수는 시트 사이 참조 · 선 규칙은 받는 칸의 엑셀 수식
(round = ROUND · ROUNDUP · ROUNDDOWN · scale = × ÷ +% −% · gate = IF(조건, 값, "") ·
lookup = 보조 시트에 표 블록 → COUNTIFS 로 하나인지 · SUMIFS 로 꺼냄 ·
범위 칸 「·아래」 이상 · 「·위」 이하)
· 함수 출력 · 출력 상자 끝수 = 보조 시트 ROUND 계열 · 원가 = 출력 상자 비용 종류별 합
· 다른 로직 = 저장 로직 · 견본을 받은 설계 값(없으면 그 로직 예시값)으로 셈 · 서로 부르면 막음
마스터 파일은 읽기만.
"""
from __future__ import annotations
import copy
import re
from decimal import 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 _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[el.get("name")] = _element_value(
group, el, given.get(el.get("name")), 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": el.get("name"),
"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
# ── 선 규칙 → 받는 칸 수식 ─────────────────────────────────────────────
class _Book:
"""한 통합문서 — 함수 시트(id = 상자 id) + 보조 시트(표 블록 · 끝수)."""
def __init__(self) -> None:
self.sheets: list[dict] = []
self.helpers: dict[str, dict] = {}
def helper(self, name: str) -> dict:
if name not in self.helpers:
self.helpers[name] = {"id": name, "이름": name, "칸": {}}
self.sheets.append(self.helpers[name])
return self.helpers[name]
def _ref(sheet: str, cell: str) -> str:
return f"'{sheet}'!{cell}"
def _lookup(book: _Book, link: dict, block: Any, rule: dict, expr: Callable) -> Any:
"""표에서 찾기 — 보조 시트에 표 블록(열린 범위 끝 ±1E300) · 하나면 그 칸 · 아니면 까닭 글."""
if not isinstance(block, list) or not block:
return {"식": '"표 없음"'}
head = [str(h) for h in block[0]]
sheet = book.helper(f"{HELP}{link['id']}")
for y, line in enumerate(block, start=1):
for x, v in enumerate(line, start=1):
name = head[x - 1]
if (
y > 1
and v is None
and (name.endswith(dv.RANGE_LOW) or name.endswith(dv.RANGE_HIGH))
):
v = -FAR if name.endswith(dv.RANGE_LOW) else FAR
_put(sheet["칸"], f"{_letters(x)}{y}", v)
last = len(block)
def rng(name: str) -> str:
c = _letters(head.index(name) + 1)
return f"'{sheet['id']}'!{c}2:{c}{last}"
crit = []
for cond in rule.get("conds") or []:
col, value = str(cond.get("col")), expr(cond.get("from") or {})
if col in head:
crit.append(f"{rng(col)},{value}")
elif col + dv.RANGE_LOW in head and col + dv.RANGE_HIGH in head:
crit.append(
f'{rng(col + dv.RANGE_LOW)},"<="&{value},{rng(col + dv.RANGE_HIGH)},">="&{value}'
)
else:
return {"식": f'"표에 「{col}」 칸 없음"'}
want = str(rule.get("col"))
if want not in head or not crit:
return {"식": f'"표에 「{want}」 칸 없음"'}
joined = ",".join(crit)
return {
"식": f"IF(COUNTIFS({joined})=1,SUMIFS({rng(want)},{joined}),"
f'IF(COUNTIFS({joined})=0,"표에 없는 조건","표에 여럿 맞음"))'
}
def _apply(rule: dict | None, src: str, round_: Callable) -> str:
"""값 수식 src 에 규칙(round · scale · gate) — lookup 은 따로."""
kind = (rule or {}).get("kind")
if kind == "round":
return round_(src, rule)
if kind == "scale":
v = _literal(rule.get("value"))
return {
"×": f"({src})*{v}",
"÷": f"({src})/{v}",
"+%": f"({src})*(1+{v}/100)",
"-%": f"({src})*(1-{v}/100)",
}.get(rule.get("op"), src)
return src
def _round(src: str, r: dict) -> str:
return f"{ROUND.get(r.get('method'), 'ROUND')}({src},{int(r.get('places') or 0)})"
# ── 계산 ───────────────────────────────────────────────────────────────
def _order(lg: dict) -> tuple[list[dict], list[dict]]:
"""함수 상자 차례(Eval `order` 와 같음) — 앞 함수가 먼저 · 돌고 도는 상자는 따로."""
funcs = [b for b in lg.get("boxes") or [] if b.get("kind") == "function" and b.get("func")]
links, done, left = lg.get("links") or [], [], {f["id"] for f in funcs}
moved = True
while moved and left:
moved = False
for f in funcs:
if f["id"] in left and not any(
ln["to"].get("box") == f["id"] and ln["from"].get("box") in left for ln in links
):
done.append(f)
left.discard(f["id"])
moved = True
return done, [f for f in funcs if f["id"] in left]
def _gate(rule: dict, src: str, expr: Callable) -> str:
cond = (
expr(rule.get("from") or {}) + GATE.get(rule.get("op"), "=") + _literal(rule.get("value"))
)
return f'IF({cond},{src},"")'
def _build(lg: dict, funcs: list[dict], ports: dict, values: dict) -> tuple[_Book, list, dict]:
"""(통합문서, 받는 칸 목록 [(상자, 칸, 표인가, 넣은 것)], 출력 자리 {상자: (시트, 칸)})."""
boxes = {b["id"]: b for b in lg.get("boxes") or []}
links = lg.get("links") or []
book, fills, outs = _Book(), [], {}
def expr(port: dict) -> str:
b = boxes.get(port.get("box"))
if b is None:
return '""'
if b.get("kind") == "function":
return _ref(*outs[b["id"]]) if b["id"] in outs else '""'
v = ports.get(b["id"], {}).get(port.get("name"))
return '""' if isinstance(v, list) else _literal(v)
def incoming(link: dict) -> Any:
"""선 하나가 받는 칸에 넣을 것 — 값 · 표 블록 · 수식."""
src, rule = link["from"], link.get("rule") or {}
up = boxes.get(src.get("box")) or {}
func = up.get("kind") == "function"
raw = None if func else ports.get(up.get("id"), {}).get(src.get("name"))
if rule.get("kind") == "lookup":
return _lookup(book, link, raw, rule, expr)
if isinstance(raw, list):
return raw # 표 블록 — 규칙 없이 그대로 펼침
if not rule and not func:
return raw
if rule.get("kind") == "gate":
return {"식": _gate(rule, expr(src), expr)}
return {"식": _apply(rule, expr(src), _round)}
def to(box: str, port: str) -> dict | None:
return next((ln for ln in links if ln.get("to") == {"box": box, "name": port}), None)
for f in funcs:
func = f["func"]
sheet = {**copy.deepcopy(func.get("sheet") or {}), "id": f["id"], "이름": f["id"]}
cells, given = sheet.setdefault("칸", {}), values.get(f["id"]) or {}
for v in func.get("variables") or []:
if v.get("cell"):
_place(cells, v["cell"], given.get(v.get("name"), v.get("example")))
for p in func.get("inputs") or []:
if not str(p.get("cell") or "").strip():
continue
link = to(f["id"], p.get("port"))
value = incoming(link) if link else given.get(p.get("port"))
_place(cells, p["cell"], value)
fills.append((f["id"], _at(p["cell"])[2], isinstance(value, list), value))
book.sheets.append(sheet)
out = func.get("output") or {}
cell = _at(out.get("cell") or "A1")[2]
if out.get("round"): # 함수 출력 끝수 — 보조 시트 칸 · 뒤 상자는 이 칸을 봄
helper = book.helper(f"{HELP}r")
at = f"A{len(helper['칸']) + 1}"
_put(helper["칸"], at, {"식": _round(_ref(f["id"], cell), out["round"])})
outs[f["id"]] = (helper["id"], at)
else:
outs[f["id"]] = (f["id"], cell)
for b in boxes.values(): # 출력 상자 — 받은 값 · 규칙 · 끝수를 보조 시트 칸에
link = to(b["id"], OUTPUT_PORT) if b.get("kind") == "output" else None
if link is None:
continue
value = incoming(link)
src = value["식"] if isinstance(value, dict) else _literal(value)
if (b.get("output") or {}).get("round"):
src = _round(src, b["output"]["round"])
helper = book.helper(f"{HELP}o")
at = f"A{len(helper['칸']) + 1}"
_put(helper["칸"], at, {"식": src})
outs["out:" + b["id"]] = (helper["id"], at)
return book, fills, outs
def _needs_func(boxes: dict) -> bool:
"""재료 조건 찾기가 함수 출력을 쓰나 — 그러면 한 번 더 셈."""
for b in boxes.values():
for el in (b.get("input") or {}).get("elements") or []:
src = (el.get("rule") or {}).get("from") or {}
if el.get("mode") == "rule" and boxes.get(src.get("box"), {}).get("kind") == "function":
return True
return False
def run(lg: dict, values: dict, load: Loader, local: list[dict], stack: list | None = None) -> dict:
"""셈 — {ports, outputs, cost, errors, 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 []}
funcs, loop = _order(lg)
ports: dict[str, dict] = {}
for _ in range(2 if _needs_func(boxes) else 1):
errs = [f"「{f.get('name')}」 함수 상자가 서로를 기다림(돌고 도는 선)" for f in loop]
for b in boxes.values():
if b.get("kind") == "input":
ports[b["id"]] = _input_ports(lg, b, values, ports, errs, ctx)
book, fills, outs = _build(lg, funcs, ports, values)
doc = _num({"종류": "통합문서", "판": 1, "서식": [{}] * 64, "시트": book.sheets})
try:
calc = spreadsheet.recalc_workbooks([doc])[0] if book.sheets else {}
except spreadsheet.RecalcError as e:
raise store.StoreError(503, str(e)) from e
sheets = {s["id"]: s for s in book.sheets}
def read(sheet: str, cell: str) -> Any:
hit = (calc.get(sheet) or {}).get(cell)
if hit is not None:
return _plain(hit)
return ((sheets.get(sheet) or {"칸": {}})["칸"].get(cell) or {}).get("값")
for f in funcs:
ports[f["id"]] = {f["func"]["output"].get("name"): read(*outs[f["id"]])}
errors = errs
for sheet, cells in calc.items():
for cell, v in cells.items():
if "오류" in v:
name = boxes.get(sheet, {}).get("name") or "끝수 · 규칙"
errors.append(f"「{name}」 {cell} 칸 계산 오류 {v['오류']}")
elif v.get("글") in ("표에 없는 조건", "표에 여럿 맞음", "표 없음"):
errors.append(f"표에서 찾기 — {v['글']}")
outputs, cost = {}, {**dict.fromkeys(COSTS, Decimal(0)), "값들": {}}
for b in boxes.values():
if b.get("kind") != "output":
continue
at = outs.get("out:" + b["id"])
value = read(*at) if at else None
value = None if value == "" else value
outputs[b["id"]] = value
kind = (b.get("output") or {}).get("cost") or "값"
if (
kind in COSTS
and isinstance(value, (int, float, Decimal))
and not isinstance(value, bool)
):
cost[kind] += Decimal(str(value))
elif kind == "값":
cost["값들"][b.get("name")] = 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.get("kind") != "function" or not b.get("func"):
continue
cells = b["func"].setdefault("sheet", {}).setdefault("칸", {})
given = values.get(b["id"]) or {}
for v in b["func"].get("variables") or []:
if v.get("cell"):
_place(cells, v["cell"], given.get(v.get("name"), v.get("example")))
for box_id, cell, block, value in fills:
if box_id == b["id"]:
_place(cells, cell, value if block else read(box_id, cell))
return {"ports": ports, "outputs": outputs, "cost": cost, "errors": errors, "filled": filled}