diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Api.ts b/M02_MasterTemplete/DataFlow/M02_DF_Api.ts index 83706b0fb..35596a0c9 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Api.ts +++ b/M02_MasterTemplete/DataFlow/M02_DF_Api.ts @@ -1,13 +1,13 @@ /* ============================================================================= * M02_DF_Api.ts - * 단가산출 다이어그램 Api — 목록(내 것 + 견본) · 열기 · 저장(서버가 입력 값을 마스터에서 다시 받아 - * 함수 상자 시트를 같은 엔진으로 셈하고 받는 칸에 값만 되돌려 넣음) · 지우기 · 입력 값 한 번에 받기 · - * 사용자 요소. 모양 = sub7 Model(CfLogic · CfGroup · CfUserElement) · 아래는 Api 몫만. - * 서버 = `M02_DF_Router.py`(`/api/m02/df/*`) · 정함 `docs/raw/M01_쉬운단가/원가다이어그램_정함.md` + * 단가산출 흐름도 Api — 목록(내 것 + 견본) · 열기 · 저장(서버가 입력 값을 마스터에서 다시 받아 화면과 + * 같은 차례 · 같은 규칙으로 셈하고 받는 칸에 값만 되돌려 넣음) · 지우기 · 입력 값 한 번에 받기 · 사용자 요소 · + * 자재 찾기 · 다른 로직 값 · 설계 때 정할 것 · 단위 · 금액 표준 표. 모양 = sub7 Model · 아래는 Api 몫만. + * 서버 = `M02_DF_Router.py`(`/api/m02/df/*`) · 정함 `docs/raw/M01_쉬운단가/흐름도_정함.md` * ========================================================================== */ import { API_BASE_URL } from "@config/config_frontend"; -import type { CfGroup, CfLogic, CfUserElement } from "./M02_DF_Model"; +import type { CfDesign, CfElementRef, CfGroup, CfLogic, CfUserElement } from "./M02_DF_Model"; /** 돈 로직 결과에서 꺼낼 칸(기계 시간당 경비) */ export type CfField = "노무비" | "재료비" | "경비" | "계"; @@ -17,7 +17,9 @@ export type CfField = "노무비" | "재료비" | "경비" | "계"; * 기존 로직 재료 줄과 같은 단가) * · 유가전력 · 환율 = 그 줄 값 · 요율 = 값 묶음의 field(기본 rate_percent · %) * · 기계 ref = EQ 키 + field(값 칸 — 가격 · 손료계수 …) 또는 `로직(GC…, 기계='…', 지역=지역)` + inputs → 한 시간 값 - * · 소요량 · 계수 table = {key, row?} → 표 블록(머리 줄 포함 · 조건 칸 → 값 칸 · 범위는 「·아래」 「·위」) */ + * · 소요량 · 계수 table = {key, row?} → 표 블록(머리 줄 포함 · 조건 칸 → 값 칸 · 범위는 「·아래」 「·위」) + * · 재료 자리 slot = 요소 그대로(mode fixed · choice · rule · scope · default) + field · op(조건 찾기 규칙) · + * value = 이어진 알약 값(조건 찾기) · pick = 고른 품목 키(선택) → 단가 · 여럿이면 candidates */ export interface CfQuery { id: string; group: CfGroup; @@ -26,6 +28,9 @@ export interface CfQuery { unit?: string; field?: string; inputs?: Record; + slot?: Partial & { field?: "규격" | "이름"; op?: "포함" | "같음" }; + value?: unknown; + pick?: string; } /** 입력 값 답 — 표는 `rows` · 그 밖은 `value` · 못 구하면 ok false + 쉬운 말 */ @@ -37,10 +42,32 @@ export interface CfAnswer { name?: string; unit?: string; reason?: string; + /** 재료 조건 찾기가 여럿 맞을 때 후보 다섯 */ + candidates?: CfMaterial[]; } -/** 서버 계산 결과 — 저장 답 */ +/** 자재 찾기 한 줄 */ +export interface CfMaterial { + ref: string; + name: string; + spec: string; + unit: string; + price: number | null; + 구분: string; + 상세구분: string; +} + +/** 단위 · 금액 표준 표 — 고친 것(`_data`)이 있으면 saved */ +export interface CfStandards { + unit: { 이름: string; 줄: Record[]; [k: string]: unknown }; + money: { 이름: string; 줄: Record[]; [k: string]: unknown }; + saved: boolean; +} + +/** 서버 계산 결과 — 저장 답(화면 Eval 과 같은 차례 · 같은 규칙 · arrived = 상자마다 받은 값 · breaks = 끊긴 곳) */ export interface CfComputed { + arrived?: Record>; + breaks?: { kind: string; box: string; port?: string; link?: string; text: string }[]; /** 상자 id → 내보내는 알약 이름 → 값(표는 칸 덩어리) */ ports: Record>; /** 출력 상자 id → 값 */ @@ -82,11 +109,11 @@ export class CfError extends Error { } } -async function send(path: string, body?: unknown): Promise { +async function send(path: string, body?: unknown, method?: string): Promise { let response: Response; try { response = await fetch(`${API_BASE_URL}/m02/df${path}`, { - method: body === undefined ? "GET" : "POST", + method: method ?? (body === undefined ? "GET" : "POST"), credentials: "include", headers: { "Content-Type": "application/json" }, body: body === undefined ? undefined : JSON.stringify(body), @@ -156,3 +183,37 @@ export function cfPlain(e: unknown): string { if (e.status === 400 && text && !/[A-Z]{2}\d{6}|[{}]/.test(text)) return text; return PLAIN[e.status] ?? "저장하지 못함 — 잠시 뒤 다시 해 주세요"; } + +/** 자재 찾기 — 띄어쓰기로 나눈 말을 모두 포함(이름 · 규격 · 키 · 원문번호) · 쪽 넘기기 */ +export async function cfMaterials(q: { + q?: string; + sub?: string; + detail?: string; + page?: number; + size?: number; +}): Promise<{ total: number; page: number; size: number; items: CfMaterial[] }> { + const qs = new URLSearchParams( + Object.entries(q).flatMap(([k, v]) => (v === undefined || v === "" ? [] : [[k, String(v)]])), + ); + return send(`/materials?${qs.toString()}`); +} + +/** 다른 로직 값 — 저장 로직 · 견본을 받은 설계 값(없으면 예시값)으로 서버가 셈 · 출력 이름 → 값 */ +export async function cfLogicValue( + key: string, + designs: Record = {}, +): Promise<{ outputs: Record; cost: CfComputed["cost"]; errors: string[] }> { + return send("/logic-value", { key, designs }); +} + +/** 설계 때 정할 것 — 변수 · 재료 자리 · 다른 로직 받는 알약(프로젝트 일위대가가 쓸 꼴) */ +export async function cfDesignInputs(key: string): Promise { + const qs = new URLSearchParams({ key }).toString(); + return (await send<{ designs: CfDesign[] }>(`/design-inputs?${qs}`)).designs; +} + +/** 단위 · 금액 표준 표 읽기 · 고치기(시스템 관리자) · 처음 값으로 */ +export const cfStandards = (): Promise => send("/standards"); +export const cfStandardsSave = (body: Omit): Promise => + send("/standards", body, "PUT"); +export const cfStandardsReset = (): Promise => send("/standards", undefined, "DELETE"); diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py index fc88da736..c50191906 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py @@ -1,18 +1,17 @@ """단가산출 흐름도 서버 계산(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 계열 · 원가 = 출력 상자 비용 종류별 합 -· 다른 로직 = 저장 로직 · 견본을 받은 설계 값(없으면 그 로직 예시값)으로 셈 · 서로 부르면 막음 +모양 = 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 과 같음 +· 다른 로직 = 저장 로직 · 견본을 받은 설계 값(선 · 검증 칸 · 없으면 그 로직 예시값)으로 셈 · + 서로 부르면 막음 마스터 파일은 읽기만. """ @@ -20,7 +19,7 @@ from __future__ import annotations import copy import re -from decimal import Decimal +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 @@ -143,6 +142,12 @@ def _element_value( 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") @@ -151,8 +156,8 @@ def _input_ports(lg: dict, box: dict, values: dict, ports: dict, errors: list, c 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 + 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"]) @@ -237,7 +242,7 @@ def design_inputs(lg: dict) -> list[dict]: out.append( { "box": box["id"], - "port": el.get("name"), + "port": _port(el), "kind": "재료 자리", "unit": el.get("unit") or "-", "element": el, @@ -258,261 +263,292 @@ def design_inputs(lg: dict) -> list[dict]: 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] +# ── 선 규칙(`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 _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 +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]) + 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, filled}(filled = 받는 칸에 값만 되돌려 넣음).""" + """셈 — {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 []} - 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} + 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 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("값") + 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, + "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 - 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(): + 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 - at = outs.get("out:" + b["id"]) - value = read(*at) if at else None - value = None if value == "" else value + 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 in COSTS - and isinstance(value, (int, float, Decimal)) - and not isinstance(value, bool) - ): - cost[kind] += Decimal(str(value)) - elif kind == "값": + 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) # 받는 칸에 값만 되돌려 넣음 — 수식 · 참조는 셈한 값으로 + 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} + 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 diff --git a/resources/tester/test_m02_df_router.py b/resources/tester/test_m02_df_router.py index 9f4277ee3..833a40096 100644 --- a/resources/tester/test_m02_df_router.py +++ b/resources/tester/test_m02_df_router.py @@ -1,3 +1,4 @@ +# ruff: noqa: E501 — 견본 상자 · 선 덩어리는 한눈에 보이게 한 줄로 둠 """단가산출 다이어그램 시험 길(`M02_MasterTemplete/DataFlow/M02_DF_Router.py`) 시험 — PLAN 35-7. · Model CfLogic(boxes · links · userElements) 저장 → 열기 → 목록 → 지우기 · 견본(`_sample`)은 읽기만 @@ -482,22 +483,17 @@ def test_다른_로직_값_설계_값_순환(client: TestClient) -> None: @pytest.mark.skipif(NO_NODE, reason="node 가 없음") def test_선_규칙_끝수_재료_조건_설계_때_정할_것(client: TestClient, folder: Path) -> None: + """Eval 과 같은 뜻 — 규칙은 값 기준 · 막힌 선(통과 안 됨 · 표에 없는 조건)은 받는 상자를 멈춤.""" keys = _add_pipes(folder) scope = {"구분": "시험구분", "상세구분": "시험상세", "찾는말": "견본관"} - vars_ = [{"name": "돌", "type": "choice", "unit": "-", "example": "견치돌"}, - {"name": "높이", "type": "number", "unit": "m", "example": 4}, + vars_ = [{"name": "돌", "type": "choice", "unit": "-", "example": "깬돌"}, + {"name": "높이", "type": "number", "unit": "m", "example": 2}, {"name": "관경", "type": "text", "unit": "-", "example": "D300"}] # fmt: skip - pipe = {"ref": "", "name": "관", "spec": "", "unit": "m", "mode": "rule", "scope": scope, + pipe = {"ref": "", "name": "관", "slot": "관 자리", "spec": "", "unit": "m", "mode": "rule", "scope": scope, "rule": {"field": "규격", "op": "포함", "from": {"box": "v", "name": "관경"}}} # fmt: skip - pick = { - "ref": "", - "name": "골라 쓰는 관", - "spec": "", - "unit": "m", - "mode": "choice", - "default": keys[1], - } - sheet = {"id": "s", "이름": "셈", "칸": {"B1": {"식": "A1+A2+A3+A4"}}} + pick = {"ref": "", "name": "관2", "slot": "골라 쓰는 관", "spec": "", "unit": "m", "mode": "choice", + "default": keys[1]} # fmt: skip + sheet = {"id": "s", "이름": "셈", "칸": {"B1": {"식": "A1+A2+A3"}}} lg = { "key": None, "name": "규칙 견본", "resultUnit": "원", "boxes": [ @@ -506,41 +502,58 @@ def test_선_규칙_끝수_재료_조건_설계_때_정할_것(client: TestClien "variables": [], "table": {"key": SAMPLE}}), _box("m", "input", "재료", input={"mode": "group", "group": "재료", "elements": [pipe, pick], "variables": []}), - _box("f", "function", "셈", func={**_func(sheet, [{"port": "품", "cell": "A1"}, - {"port": "관", "cell": "A2"}, {"port": "통과", "cell": "A3"}, {"port": "막힘", "cell": "A4"}], - {"cell": "B1", "name": "합", "unit": "원", "round": {"places": 0, "method": "버림"}})}), - _box("o", "output", "재료비", output={"cost": "재료비", "unit": "원", "round": {"places": -1, "method": "버림"}}), + _box("f", "function", "셈", func=_func(sheet, [{"port": "품", "cell": "A1"}, + {"port": "관", "cell": "A2"}, {"port": "높이", "cell": "A3"}], + {"cell": "B1", "name": "합", "unit": "원", "round": {"places": 0, "method": "버림"}})), + _box("o", "output", "재료비", output={"cost": "재료비", "unit": "원", + "round": {"places": -1, "method": "버림"}}), _box("o2", "output", "골라 쓴 관", output={"cost": "값", "unit": "원"}), + _box("o3", "output", "낮을 때만", output={"cost": "값", "unit": "m"}), ], "links": [ {**_link("l1", "t", "표", "f", "품"), "rule": {"kind": "lookup", "col": "석공", "conds": [ {"col": "돌", "from": {"box": "v", "name": "돌"}}, {"col": "높이", "from": {"box": "v", "name": "높이"}}]}}, - {**_link("l2", "m", "관", "f", "관"), "rule": {"kind": "scale", "op": "+%", "value": 10}}, - {**_link("l3", "v", "높이", "f", "통과"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"}, "op": "≥", "value": 3}}, - {**_link("l4", "v", "높이", "f", "막힘"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"}, "op": "<", "value": 3}}, - _link("l5", "f", "합", "o", "값"), - {**_link("l6", "m", "골라 쓰는 관", "o2", "값"), "rule": {"kind": "round", "places": -3, "method": "반올림"}}, + {**_link("l2", "m", "관 자리", "f", "관"), "rule": {"kind": "scale", "op": "+%", "value": 10}}, + {**_link("l3", "v", "높이", "f", "높이"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"}, + "op": "≥", "value": 1}}, + _link("l4", "f", "합", "o", "값"), + {**_link("l5", "m", "골라 쓰는 관", "o2", "값"), "rule": {"kind": "round", "places": -3, "method": "반올림"}}, + {**_link("l6", "v", "높이", "o3", "값"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"}, + "op": "<", "value": 1}}, ], "userElements": [], } # fmt: skip got = _post(client, "/api/m02/df/save", {"logic": lg}) v = got["values"] - assert v["ports"]["m"] == {"관": 30000, "골라 쓰는 관": 45000} # 조건 찾기 D300 · 선택 = 기본 - total = Decimal("0.7") + 30000 * Decimal("1.1") + 4 # 표에서 찾기 + 10% + 통과(4) + 막힘("") + assert v["ports"]["m"] == { + "관 자리": 30000, + "골라 쓰는 관": 45000, + } # 조건 찾기 D300 · 선택 = 기본 + total = Decimal("0.5") + 30000 * Decimal("1.1") + 2 # 깬돌 · 높이 2 → 「~3」 줄 · +10% · 통과 assert _d(v["ports"]["f"]["합"]) == int(total) # 함수 출력 끝수 0 자리 버림 assert _d(v["cost"]["재료비"]) == int(total) // 10 * 10 # 출력 끝수 −1 자리 버림 - assert _d(v["cost"]["값들"]["골라 쓴 관"]) == 45000 and v["errors"] == [] + assert _d(v["cost"]["값들"]["골라 쓴 관"]) == 45000 and v["cost"]["값들"]["낮을 때만"] is None + assert [b["kind"] for b in v["breaks"]] == ["통과 안 됨"] and v["errors"] == [] cells = {b["id"]: b for b in got["logic"]["boxes"]}["f"]["func"]["sheet"]["칸"] - assert _d(cells["A1"]["값"]) == Decimal("0.7") and "A4" not in cells # 받는 칸에 값만 + assert _d(cells["A1"]["값"]) == Decimal("0.5") and cells["B1"] == { + "식": "A1+A2+A3" + } # 받는 칸에 값만 designs = client.get("/api/m02/df/design-inputs", params={"key": got["key"]}).json()["designs"] assert [(d["port"], d["kind"]) for d in designs] == [ ("돌", "변수"), ("높이", "변수"), ("관경", "변수"), ("골라 쓰는 관", "재료 자리")] # fmt: skip - lg["boxes"][0]["input"]["variables"][2]["example"] = "D999" # 맞는 재료 없음 · 표에 없는 조건 - lg["boxes"][0]["input"]["variables"][0]["example"] = "없는돌" - bad = _post(client, "/api/m02/df/save", {"logic": lg})["values"]["errors"] - assert any("맞는 재료 없음" in e for e in bad) and any("표에 없는 조건" in e for e in bad) + lg["boxes"][0]["input"]["variables"][1]["example"] = 3 # 「~3」 줄(위 끝 포함) + assert _d( + _post(client, "/api/m02/df/save", {"logic": lg})["values"]["ports"]["f"]["합"] + ) == int(Decimal("0.5") + 33000 + 3) + lg["boxes"][0]["input"]["variables"][2]["example"] = "D999" # 맞는 재료 없음 + lg["boxes"][0]["input"]["variables"][0]["example"] = "없는돌" # 표에 없는 조건 → 함수 멈춤 + bad = _post(client, "/api/m02/df/save", {"logic": lg})["values"] + assert any("맞는 재료 없음" in e for e in bad["errors"]) and any( + "표에 없는 조건" in e for e in bad["errors"] + ) + assert bad["ports"]["f"]["합"] is None and _d(bad["cost"]["재료비"]) == 0 def test_표준_표_읽기_쓰기_처음값(client: TestClient, folder: Path) -> None: