eomsangdonandClaude Opus 5.5
8193f29ae9
- convert(row, ctx={key, file, master}) — sub4 /api/m01/logic/flow 부르는 꼴 · master 는 ctx 것
- {입력} 요소 찾는 값 = 원문번호(「MO:경유:{지역}」 → "경유:"&지역 글 잇는 함수) · 호표 줄 그룹 = 키 앞 두 글자
- 로직(…).계 = 다른 로직 비목 셋 합
- 표 찾기: 여럿이면 범위 위 끝이 가장 작은 줄(옛 find) · 값 칸이 「·아래」 「·위」 로 갈리면 아래 = 위일 때 그 값(Rule · Compute 같음)
- 표 찾기 실패로 넣은 NA() 칸은 칸 계산 오류로 알리지 않음(Eval · Compute)
- 전수 대조는 서버 길 그대로(sub4 조건 찾기 · 표 식 글) · 입력은 옛 엔진처럼 수로 · 까닭 분류
- 시험입력 1,423/1,472(96.7%) · 기본 입력 1,434/1,472(97.4%)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VBToTCGk5dqDKeXCEP3vhk
402 lines
21 KiB
Python
402 lines
21 KiB
Python
"""M02 단가산출 흐름도 — 옛 로직 줄 → 흐름도(CfLogic) 변환 (PLAN 36-2 · 흐름도_적용_정함 2 · 3).
|
|
|
|
옛 식은 지금 서버 파서(`master_formula.parse` · `store.cm.mf`)로 나무를 읽어 함수 상자 시트로
|
|
· 이름마다 받는 알약 → 칸(B2 · B3 …) · 식은 칸 주소 · 함수 상자 하나 = 결과 하나.
|
|
|
|
입력 → 입력 상자 「설계 값」(변수 · 고르기 = choice · 범위)
|
|
중간 · 호표 수량 × 단가 · 덧줄 · 결과 → 함수 상자 · 비목 합 → 출력 상자(비목 셋 · 결과 = 값)
|
|
호표 단가 · 식 속 요소 키 → 그룹 입력 상자(인력 · 재료 · 기계 · 유가전력 · 환율 · 요율) 요소 알약
|
|
· 키에 {입력} 이 끼면 조건 찾기(field 「키」 · 같음 · from = 입력 알약 · 자리 여럿 = 글 잇는 함수)
|
|
찾기(표, 조건=값).칸 → 표 입력 상자 「표」 알약 + 선 규칙 「표에서 찾기」(lookup)
|
|
로직(키, 넘길 값=…) → 다른 로직 입력 상자(받는 알약 = 넘길 값 · 내보내는 알약 = 비목 · 결과)
|
|
만약 → IF · 그리고 / 또는 → AND / OR · 최소 / 최대 → MIN / MAX
|
|
· 올림 / 버림 / 반올림 → ROUNDUP / ROUNDDOWN / ROUND
|
|
자동 배치 — 입력 왼쪽 → 함수 가운데(기댐 깊이마다 한 칸) → 출력 오른쪽 · 칸마다 위에서 아래로
|
|
|
|
부르는 꼴 — convert(row, ctx={key, file, master}) → {flow, notes} · row = 옛 로직 줄 · 파일 안 씀.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from M01_MasterData import M01_MasterData_Store as store
|
|
|
|
mf = store.cm.mf
|
|
COSTS = ("노무비", "재료비", "경비")
|
|
OUTPUT_PORT = "값"
|
|
TABLE_PORT = "표"
|
|
COLORS = {"input": "#2f80ed", "function": "#7b4fd6", "output": "#27ae60"}
|
|
GROUP_OF = {"LB": "인력", "MT": "재료", "MP": "재료", "MO": "유가전력", "EQ": "기계"}
|
|
GROUP_OF |= {"FX": "환율", "RT": "요율"}
|
|
FUNCS = {"만약": "IF", "최소": "MIN", "최대": "MAX", "올림": "ROUNDUP", "버림": "ROUNDDOWN"}
|
|
FUNCS["반올림"] = "ROUND"
|
|
OPS = {"==": "=", "!=": "<>"}
|
|
JOINS = {"그리고": "AND", "또는": "OR"}
|
|
_HOLE = re.compile(r"\{([^{}]+)\}")
|
|
|
|
|
|
class Unsupported(Exception):
|
|
"""옮기지 못하는 꼴 — 쉬운 말 까닭."""
|
|
|
|
|
|
def plain(v: Any) -> Any:
|
|
"""JSON 값 — Decimal → int · float."""
|
|
if isinstance(v, Decimal):
|
|
return int(v) if v == v.to_integral_value() else float(v)
|
|
if isinstance(v, (list, tuple)):
|
|
return [plain(x) for x in v]
|
|
if isinstance(v, dict):
|
|
return {k: plain(x) for k, x in v.items()}
|
|
return v
|
|
|
|
|
|
def literal(v: Any) -> str:
|
|
if isinstance(v, str):
|
|
return '"' + v.replace('"', '""') + '"'
|
|
if isinstance(v, Decimal):
|
|
return format(v.normalize(), "f")
|
|
return str(plain(v))
|
|
|
|
|
|
class _Flow:
|
|
"""흐름 한 벌 — 상자 · 선 · 못 옮긴 곳."""
|
|
|
|
def __init__(self, row: dict, master: Any):
|
|
self.row = row
|
|
self.master = master
|
|
self.specs = {s["이름"]: s for s in row.get("입력") or []}
|
|
self.boxes: dict[str, dict] = {}
|
|
self.links: list[dict] = []
|
|
self.notes: list[str] = []
|
|
self.named: dict[str, dict] = {} # 중간 · 덧줄 이름 → 내보내는 알약
|
|
self.cache: dict[str, dict] = {} # 같은 요소 · 표 · 로직 · 값 → 알약(하나만)
|
|
|
|
# ── 상자 · 선 ─────────────────────────────────────────────────────
|
|
def box(self, kind: str, name: str, **body: Any) -> dict:
|
|
bid = f"{kind[:2]}-{len(self.boxes) + 1}"
|
|
self.boxes[bid] = {"id": bid, "kind": kind, "name": name, "color": COLORS[kind]}
|
|
self.boxes[bid] |= {"unit": "-", "x": 0, "y": 0, **body}
|
|
return self.boxes[bid]
|
|
|
|
def link(self, src: dict, to: dict, rule: dict | None = None) -> None:
|
|
one = {"id": f"l{len(self.links) + 1}", "from": dict(src), "to": dict(to)}
|
|
self.links.append(one | ({"rule": rule} if rule else {}))
|
|
|
|
def design(self) -> None:
|
|
variables = []
|
|
for s in self.row.get("입력") or []:
|
|
pick = s.get("고르기")
|
|
one = {"name": s["이름"], "type": "choice" if pick else "number"}
|
|
one |= {"unit": str(s.get("단위") or "-"), "example": plain(pick[0]) if pick else None}
|
|
if pick:
|
|
one["choices"] = plain(pick)
|
|
if "범위" in s:
|
|
one["range"] = plain(s["범위"])
|
|
variables.append(one)
|
|
box = {"id": "in-design", "kind": "input", "name": "설계 값", "color": COLORS["input"]}
|
|
inp = {"mode": "variable", "elements": [], "variables": variables}
|
|
self.boxes["in-design"] = box | {"unit": "-", "x": 0, "y": 0, "input": inp}
|
|
|
|
# ── 바깥 값 알약 ──────────────────────────────────────────────────
|
|
def element(
|
|
self, group: str, ref: str, unit: str, label: str, rule: dict | None = None
|
|
) -> dict:
|
|
"""그룹 입력 상자의 요소 알약 — 그룹마다 상자 하나 · 같은 ref · 단위 · 조건이면 알약 하나
|
|
· rule = 조건 찾기(field 「키」 — ref 의 {…} 에 알약 값을 끼운 요소 · CfMaterialRule)."""
|
|
key = f"el|{group}|{ref}|{unit}|{json.dumps(rule, ensure_ascii=False, sort_keys=True)}"
|
|
if key in self.cache:
|
|
return self.cache[key]
|
|
gkey = f"group|{group}"
|
|
if gkey not in self.cache:
|
|
inp = {"mode": "group", "group": group, "elements": [], "variables": []}
|
|
self.cache[gkey] = self.box("input", group, input=inp)
|
|
els = self.cache[gkey]["input"]["elements"]
|
|
name = label
|
|
while any(e["name"] == name or e.get("slot") == name for e in els):
|
|
name += "·"
|
|
one = {"ref": ref, "name": name, "spec": "", "unit": unit or "-", "mode": "fixed"}
|
|
if rule:
|
|
one |= {"mode": "rule", "rule": rule, "slot": name}
|
|
self.link(rule["from"], {"box": self.cache[gkey]["id"], "name": name})
|
|
els.append(one)
|
|
self.cache[key] = {"box": self.cache[gkey]["id"], "name": name}
|
|
return self.cache[key]
|
|
|
|
def priced(self, group: str, ref: Any, unit: str, label: str, field: str | None = None):
|
|
"""요소 → 식 마디 「@port」 — 키에 {입력} 이 끼면 조건 찾기(field 키 · 같음 · from 알약)
|
|
· 자리 표시가 여럿이면 함수 상자가 글을 이어 만든 알약 하나로(정함 · 조사 ③)."""
|
|
if isinstance(ref, dict):
|
|
texts = {k: v for k, v in ref.items() if isinstance(v, str) and _HOLE.search(v)}
|
|
else:
|
|
texts = {None: ref} if _HOLE.search(ref) else {}
|
|
if not texts:
|
|
text = ref if isinstance(ref, str) else json.dumps(plain(ref), ensure_ascii=False)
|
|
return ("@port", self.element(group, self._key(text, field), unit, label))
|
|
if len(texts) > 1:
|
|
raise Unsupported(f"자리 표시가 여러 칸에 낀 요소 「{ref}」")
|
|
((at, text),) = texts.items()
|
|
# 찾는 값 = 원문번호(「EQ:{기계}」 → 기계 · 「MO:경유:{지역}」 → "경유:"&지역)
|
|
want = text.partition(":")[2] if at is None and ":" in text else text
|
|
if _HOLE.fullmatch(want):
|
|
src = self.hole(want[1:-1])
|
|
else:
|
|
src = self.func(f"{label} 찾는 말", ("@text", want))
|
|
if at is not None:
|
|
self.notes.append(f"재료 검색어에 입력이 낌 「{text}」 — 조건 찾기(키 같음)로 옮김")
|
|
tpl = text if at is None else json.dumps(plain(ref), ensure_ascii=False)
|
|
tpl += f"#{field}" if field else ""
|
|
rule = {"field": "키", "op": "같음", "from": src}
|
|
return ("@port", self.element(group, tpl, unit, f"{label} 자리", rule))
|
|
|
|
def hole(self, name: str) -> dict:
|
|
"""자리 표시 {이름} 의 값 알약 — 입력 · 중간."""
|
|
if name in self.specs:
|
|
return {"box": "in-design", "name": name}
|
|
if name in self.named:
|
|
return self.named[name]
|
|
raise Unsupported(f"자리 표시 {{{name}}} 의 값이 없음")
|
|
|
|
def _key(self, ref: str, field: str | None) -> str:
|
|
"""「ID:원문번호」 → 키 · 기계 칸은 「EQ키#칸」(입력 판 꼴)."""
|
|
if ":" in ref and not ref.lstrip().startswith("{"):
|
|
ref = str(self.master.get(ref).get("키"))
|
|
if field and ref[:2] != "EQ":
|
|
raise Unsupported(f"「{ref}」 의 칸 「{field}」 읽기")
|
|
return f"{ref}#{field}" if field else ref
|
|
|
|
def table(self, key: str) -> dict:
|
|
if f"tb|{key}" not in self.cache:
|
|
group = "소요량" if key[:2] in ("QF", "QC") else "계수"
|
|
inp = {"mode": "group", "group": group, "elements": [], "variables": []}
|
|
box = self.box("input", f"표 {key}", input=inp | {"table": {"key": key}})
|
|
self.cache[f"tb|{key}"] = {"box": box["id"], "name": TABLE_PORT}
|
|
return self.cache[f"tb|{key}"]
|
|
|
|
def logic(self, node: tuple) -> str:
|
|
"""로직(키, 넘길 값) → 다른 로직 입력 상자 id(같은 부르기는 하나)."""
|
|
key = "lg|" + repr(node)
|
|
if key not in self.cache:
|
|
head = node[1]
|
|
if "{" in head:
|
|
raise Unsupported(f"입력 따라 바뀌는 로직 「{head}」")
|
|
ref = str(self.master.logic(head).get("키"))
|
|
outs = [{"name": n, "unit": "원"} for n in (*COSTS, "결과")]
|
|
designs = [{"name": k, "unit": "-"} for k, _ in node[2]]
|
|
logic = {"key": ref, "outputs": outs, "designs": designs}
|
|
box = self.box("input", f"로직 {ref}", input={"mode": "logic", "elements": []})
|
|
box["input"] |= {"variables": [], "logic": logic}
|
|
for k, v in node[2]:
|
|
self.link(self.value(v), {"box": box["id"], "name": k})
|
|
self.cache[key] = {"box": box["id"]}
|
|
return self.cache[key]["box"]
|
|
|
|
def value(self, node: tuple) -> dict:
|
|
"""조건 · 넘길 값 → 내보내는 알약(입력 · 중간 · 아니면 함수 상자 · 같은 식은 하나)."""
|
|
if node[0] == "name" and node[1] in self.specs:
|
|
return {"box": "in-design", "name": node[1]}
|
|
if node[0] == "name" and node[1] in self.named:
|
|
return self.named[node[1]]
|
|
key = "val|" + repr(node)
|
|
if key not in self.cache:
|
|
label = literal(node[1]).strip('"') if node[0] in ("num", "str") else "값"
|
|
self.cache[key] = self.func(label, node)
|
|
return self.cache[key]
|
|
|
|
# ── 함수 상자 ─────────────────────────────────────────────────────
|
|
def func(self, name: str, node: tuple, scope: dict | None = None) -> dict:
|
|
"""식 하나 → 함수 상자(결과 하나) · 받는 알약마다 B2 · B3 … · 결과 칸은 그 아래."""
|
|
bid = f"fn-{len(self.boxes) + 1}"
|
|
self.boxes[bid] = {} # 자리 먼저 — 안에서 만드는 상자와 id 가 겹치지 않게
|
|
inputs: list[dict] = []
|
|
seen: dict[str, str] = {}
|
|
|
|
def take(src: dict, label: str, rule: dict | None = None) -> str:
|
|
k = json.dumps([src, rule], ensure_ascii=False, sort_keys=True, default=str)
|
|
if k not in seen:
|
|
port = label
|
|
while any(i["port"] == port for i in inputs):
|
|
port += "·"
|
|
seen[k] = f"B{len(inputs) + 2}"
|
|
inputs.append({"port": port, "cell": seen[k]})
|
|
self.link(src, {"box": bid, "name": port}, rule)
|
|
return seen[k]
|
|
|
|
formula = self.excel(node, take, scope or {})
|
|
out = f"B{len(inputs) + 2}"
|
|
cells: dict[str, Any] = {"A1": {"값": "알약"}, "B1": {"값": "값"}}
|
|
for i in inputs:
|
|
cells["A" + i["cell"][1:]] = {"값": i["port"]}
|
|
cells["A" + out[1:]] = {"값": name}
|
|
cells[out] = {"식": formula}
|
|
sheet = {"id": f"s-{bid}", "이름": name[:31] or bid, "칸": cells}
|
|
func = {"sheet": sheet, "inputs": inputs, "variables": []}
|
|
func["output"] = {"cell": out, "name": name, "unit": "-"}
|
|
body = {"id": bid, "kind": "function", "name": name, "color": COLORS["function"]}
|
|
self.boxes[bid] = body | {"unit": "-", "x": 0, "y": 0, "func": func}
|
|
return {"box": bid, "name": name}
|
|
|
|
def excel(self, node: tuple, take, scope: dict) -> str: # noqa: C901 — 마디 갈래 그대로
|
|
kind = node[0]
|
|
if kind in ("num", "str"):
|
|
return literal(node[1])
|
|
if kind == "neg":
|
|
return f"-({self.excel(node[1], take, scope)})"
|
|
if kind == "bin":
|
|
a, b = (self.excel(x, take, scope) for x in node[2:4])
|
|
if node[1] in JOINS:
|
|
return f"{JOINS[node[1]]}({a},{b})"
|
|
return f"({a}{OPS.get(node[1], node[1])}{b})"
|
|
if kind == "call":
|
|
items = [self.excel(x, take, scope) for x in node[2]]
|
|
if node[1] in ("올림", "버림", "반올림") and len(items) == 1:
|
|
items.append("0")
|
|
return f"{FUNCS[node[1]]}({','.join(items)})"
|
|
if kind == "name":
|
|
return self.name(node[1], take, scope)
|
|
if kind == "ref":
|
|
group = GROUP_OF.get(node[1][:2], "기계")
|
|
return self.excel(self.priced(group, node[1], "", node[1]), take, scope)
|
|
if kind == "field":
|
|
return self.field(node, take, scope)
|
|
if kind == "logic":
|
|
return take({"box": self.logic(node), "name": "결과"}, f"{node[1]} 결과")
|
|
if kind == "@port":
|
|
return take(node[1], node[1]["name"])
|
|
if kind == "@sum":
|
|
cells = [take(p, p["name"]) for p in node[1]]
|
|
return f"SUM({','.join(cells)})" if cells else "0"
|
|
if kind == "@text": # 자리 표시 글 「레미콘 {a}-{b}」 → "레미콘 "&B2&"-"&B3
|
|
bits = _HOLE.split(node[1])
|
|
out = [literal(b) if i % 2 == 0 else take(self.hole(b), b) for i, b in enumerate(bits)]
|
|
return "&".join(x for x in out if x != '""') or '""'
|
|
raise Unsupported(f"모르는 식 마디 「{kind}」")
|
|
|
|
def name(self, name: str, take, scope: dict) -> str:
|
|
if name in scope: # 덧줄의 비목 합 · 줄 금액
|
|
return self.excel(("@sum", scope[name]), take, scope)
|
|
if name in self.named:
|
|
return take(self.named[name], name)
|
|
if name in self.specs:
|
|
return take({"box": "in-design", "name": name}, name)
|
|
raise Unsupported(f"모르는 이름 「{name}」")
|
|
|
|
def field(self, node: tuple, take, scope: dict) -> str:
|
|
base, col = node[1], node[2]
|
|
if base[0] == "find":
|
|
if "{" in base[1]:
|
|
raise Unsupported(f"입력 따라 바뀌는 표 「{base[1]}」")
|
|
key = str(self.master.get(base[1]).get("키"))
|
|
conds = [{"col": k, "from": self.value(v)} for k, v in base[2]]
|
|
return take(self.table(key), col, {"kind": "lookup", "conds": conds, "col": col})
|
|
if base[0] == "ref":
|
|
group = GROUP_OF.get(base[1][:2], "기계")
|
|
node2 = self.priced(group, base[1], "", f"{base[1]} {col}", col)
|
|
return self.excel(node2, take, scope)
|
|
if base[0] == "logic" and col == "계": # 다른 로직 출력은 비목 셋 — 계 = 그 합
|
|
bid = self.logic(base)
|
|
return self.excel(("@sum", [{"box": bid, "name": k} for k in COSTS]), take, scope)
|
|
if base[0] == "logic":
|
|
return take({"box": self.logic(base), "name": col}, f"{base[1]} {col}")
|
|
if base[0] == "name" and base[1] == "줄" and f"줄.{col}" in scope:
|
|
return self.name(f"줄.{col}", take, scope)
|
|
raise Unsupported(f"칸 읽기 「{col}」")
|
|
|
|
|
|
# ── 한 로직 ──────────────────────────────────────────────────────────
|
|
def convert(row: dict, ctx: dict | None = None) -> dict:
|
|
"""옛 로직 줄 → {flow: CfLogic, notes: 못 옮긴 곳} · ctx = {key, file, master}(M01 저장소
|
|
/api/m01/logic/flow) · master 가 없으면 저장소가 읽어 둔 것."""
|
|
master = (ctx or {}).get("master") or store.loaded()[1]
|
|
f = _Flow(row, master)
|
|
f.design()
|
|
try:
|
|
_body(f, row)
|
|
except Unsupported as e:
|
|
f.notes.append(f"못 옮김 — {e}")
|
|
except mf.FormulaError as e:
|
|
f.notes.append(f"못 옮김 — {e}")
|
|
f.boxes = {k: b for k, b in f.boxes.items() if b} # 못 옮겨 비운 자리
|
|
_layout(f)
|
|
flow = {"key": row.get("키"), "name": row.get("이름") or ""}
|
|
flow |= {"resultUnit": row.get("결과단위") or "", "boxes": list(f.boxes.values())}
|
|
flow |= {"links": f.links, "userElements": []}
|
|
return {"flow": flow, "notes": list(dict.fromkeys(f.notes))}
|
|
|
|
|
|
def _output(f: _Flow, name: str, cost: str, src: dict, rule: dict | None, on: str) -> None:
|
|
out = {"cost": cost, "unit": "원" if cost != "값" else "-"}
|
|
if rule and rule.get("대상") == on:
|
|
out["round"] = {"places": int(rule.get("자리") or 0), "method": str(rule.get("방법"))}
|
|
box = f.box("output", name, output=out)
|
|
f.link(src, {"box": box["id"], "name": OUTPUT_PORT})
|
|
|
|
|
|
def _body(f: _Flow, row: dict) -> None:
|
|
rule = row.get("끝수") if isinstance(row.get("끝수"), dict) else None
|
|
for step in row.get("중간") or []:
|
|
f.named[step["이름"]] = f.func(step["이름"], mf.parse(step["식"]))
|
|
if "결과" in row:
|
|
_output(f, "결과", "값", f.func("결과", mf.parse(row["결과"]["식"])), rule, "결과")
|
|
return
|
|
parts: dict[str, list[dict]] = {c: [] for c in COSTS}
|
|
lines: dict[str, list[dict]] = {}
|
|
for i, item in enumerate(row.get("호표") or []):
|
|
name = str(item.get("이름") or f"줄 {i + 1}")
|
|
qty = mf.parse(str(item["수량"]))
|
|
if item.get("종류") == "로직":
|
|
call = mf.parse(str(item["요소"]))
|
|
if call[0] != "logic":
|
|
raise Unsupported(f"「{name}」 단가가 로직 부르기 꼴 아님")
|
|
bid = f.logic(call)
|
|
whole = ("@sum", [{"box": bid, "name": k} for k in COSTS])
|
|
for c in [item["비목"]] if item.get("비목") else COSTS:
|
|
price = whole if item.get("비목") else ("@port", {"box": bid, "name": c})
|
|
part = f.func(f"{name} {c}", ("bin", "*", qty, price))
|
|
parts[c].append(part)
|
|
lines.setdefault(f"줄.{name}", []).append(part)
|
|
continue
|
|
unit = str(item.get("단위") or "")
|
|
ref = item["요소"] # 그룹 = 키 앞 두 글자(「MO:경유:{지역}」 = 유가전력) · 없으면 줄 종류
|
|
group = GROUP_OF.get(ref[:2], str(item.get("종류"))) if isinstance(ref, str) else "재료"
|
|
price = f.priced(group, ref, unit, f"{name} 단가")
|
|
part = f.func(f"{name} 금액", ("bin", "*", qty, price))
|
|
parts[item["비목"]].append(part)
|
|
lines.setdefault(f"줄.{name}", []).append(part)
|
|
for extra in row.get("덧줄") or []:
|
|
scope = {c: list(parts[c]) for c in COSTS} | lines
|
|
part = f.func(extra["이름"], mf.parse(extra["식"]), scope)
|
|
f.named[extra["이름"]] = part
|
|
parts[extra["비목"]].append(part)
|
|
for c in COSTS:
|
|
if parts[c]:
|
|
_output(f, c, c, f.func(f"{c} 합", ("@sum", parts[c])), rule, "계")
|
|
|
|
|
|
# ── 자동 배치 ────────────────────────────────────────────────────────
|
|
def _layout(f: _Flow) -> None:
|
|
"""입력 왼쪽 → 함수 가운데(기댐 깊이마다 한 칸) → 출력 오른쪽 · 칸마다 위→아래(겹침 없음)."""
|
|
feeds: dict[str, set[str]] = {}
|
|
for ln in f.links:
|
|
srcs = [ln["from"], *[c["from"] for c in (ln.get("rule") or {}).get("conds", [])]]
|
|
feeds.setdefault(ln["to"]["box"], set()).update(s["box"] for s in srcs)
|
|
depth: dict[str, int] = {}
|
|
|
|
def level(bid: str, path: frozenset = frozenset()) -> int:
|
|
if bid not in depth:
|
|
ins = [x for x in feeds.get(bid, ()) if x not in path and x in f.boxes]
|
|
depth[bid] = 1 + max((level(x, path | {bid}) for x in ins), default=-1)
|
|
return depth[bid]
|
|
|
|
for bid in f.boxes:
|
|
level(bid)
|
|
last = max(depth.values(), default=0) + 1
|
|
rows: dict[int, int] = {}
|
|
for bid, box in f.boxes.items():
|
|
col = last if box["kind"] == "output" else depth[bid]
|
|
box["x"], box["y"] = 40 + col * 300, 40 + rows.get(col, 0) * 170
|
|
rows[col] = rows.get(col, 0) + 1
|