"""B08 구조물 집계표 길(PLAN 98-1) — 읽기만 · 계산 서버 단독(CLAUDE.md 5장 ③). GET /api/projects/{id}/quantity/summary 배관(횡단배수관) 한 곳마다 컨테이너 값 → 실무 구조물집계표 꼴 열 · 측점 줄 · 계 열 = 설계에 쓰인 값만 펼침 · 머리 = `group`[구조물, 형식, 규격] + `label`(단위 줄): 흙막이 › 형태 › H=높이 m 유입(흙막이일 때) · 유출 · 흙막이 n(실제 선 단까지) · 길이 = 전 + 후 · 빈 높이 = 횡단도가 그리는 관경 최소 높이 돌붙임 › 찰붙임 · 메붙임 ㎡ 도수로/돌붙임 붙임 형식 · 면적(빈칸이면 안 셈 · 98-3) 관 공 › Φ{관경}m/m › 관매설 m B06 관 길이(1 m 올림 그대로) › 커플링 ea 올림(L ÷ 1본) − 1 · 1본 = 프로젝트 층 M02 구조물집계표 변수 › 관보호공 모양 실제로 선 집수정 모양마다 1 개소(98-2 `inlet_basin`) 머리 글 = M02 정의 이름(`구조물정의.json` 종류.pipe.이름 — 사용자가 M02 에서 고침) · 열마다 `combos` = 그 값을 낸 컨테이너에 이은 조합 id(`{층}|{주인}|{키}` · 95-3 모달 · 맨 윗 머리). 설계 = B08 토공과 같은 길(확정 노선 먼저) · 관 · 컨테이너 줄 = M02 채움 `collect_items` 재사용. """ from __future__ import annotations import asyncio import math from pathlib import Path from typing import Any from uuid import UUID from fastapi import APIRouter from B06_Section.B06_Section_Engine_Culvert import pipe_wall_min_height_m from B06_Section.B06_Section_Repository import ( get_confirmed_route_context, get_cross_section_designs, get_workflow_route_context, ) from B08_Quantity.B08_Quantity_Router_Hydro import _project_root from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file from common_util.common_util_summary_item import GROUP_JOIN from config.config_db import get_db_pool from M02_MasterTemplete import M02_Table_Fill as fill from M02_MasterTemplete import M02_Template_Layers as layers from M02_MasterTemplete.M02_Table_Fill_Spec import soil_tiers_from_designs from M02_MasterTemplete.StructOptions import M02_SO_Definition as definition router = APIRouter(prefix="/api/projects", tags=["B08 Summary"]) TABLE_NAME = "구조물집계표" ONE_LENGTH_VAR = "파형강관_1본_m" #: 필수 컨테이너 잇기 키(`design_structure_pipe.ts` link) — 관 · 유입 · 유출 LINKS = {"pipe": "grv6lny", "inlet": "gnbf9m9", "outlet": "g9lq9uk"} SOIL = "기슭막이" # 저장 값(화면 이름 흙막이) FORMS = ("돌쌓기(찰)", "돌쌓기(메)", "콘크리트", "돌망태", "통나무·목재틀", "바자") SHAPES = ("I", "L", "U") BONDS = ("찰붙임", "메붙임") def _num(value: Any) -> float | None: if isinstance(value, bool) or value in (None, ""): return None try: return float(value) except (TypeError, ValueError): return None def _combo(item: Any) -> str | None: """집계 항목 → 조합 id(화면 `comboId` 와 같은 꼴) · 없으면 None.""" if not isinstance(item, dict) or not item.get("key"): return None return f"{item.get('layer')}|{item.get('owner') or ''}|{item['key']}" def one_pipe_length(project_root: str | Path) -> float | None: """1본 길이(m) — 프로젝트 층 구조물집계표 변수 · 없으면 시스템 층 · 빈칸이면 None.""" for folder in (layers.project_dir(project_root), layers.system_dir()): found = layers.read_template(folder, "table", TABLE_NAME) if found is not None: value = _num((found["문서"].get("변수") or {}).get(ONE_LENGTH_VAR)) return value if value and value > 0 else None return None def couplings(length_m: float | None, one_m: float | None) -> int | None: """커플링 = 올림(L ÷ 1본) − 1(0 아래 없음) · 길이나 1본이 빈칸이면 None.""" if not length_m or not one_m: return None return max(math.ceil(length_m / one_m - 1e-9) - 1, 0) def _basins(designs: Any) -> dict[float, str]: """설계 → `{측점: 실제로 선 집수정 모양}` — 안 섰으면 "" · 기록 없는 측점은 안 담음(98-2).""" found: dict[float, str] = {} for row in designs or []: design = row.get("design") if isinstance(row, dict) else None if isinstance(design, dict) and "inlet_basin" in design: found[round(float(row.get("chainage_m") or 0.0), 3)] = str(design["inlet_basin"] or "") return found def _revet_length(options: dict[str, Any], head: str, unit: str) -> float | None: """흙막이 길이 = 전 + 후 · 둘 다 비면 길이 칸.""" before = _num(options.get(f"{head}before{unit}")) after = _num(options.get(f"{head}after{unit}")) if before is None and after is None: return _num(options.get(f"{head}length{unit}")) return (before or 0.0) + (after or 0.0) class _Sheet: """열 · 측점 줄 · 계를 모음 — 같은 측점(1 cm)의 같은 열은 더함.""" def __init__(self) -> None: self.columns: dict[str, dict[str, Any]] = {} self.rows: dict[int, dict[str, Any]] = {} def add( self, chainage: float, key: str, sort: tuple[Any, ...], group: list[str], label: str, value: float | None, combos: list[str | None], ) -> None: cid = f"pipe:{key}" column = self.columns.setdefault( cid, {"id": cid, "group": group, "label": label, "combos": [], "_sort": sort} ) for combo in combos: if combo and combo not in column["combos"]: column["combos"].append(combo) row = self.rows.setdefault(round(chainage * 100), {"chainage_m": chainage, "values": {}}) if value is not None: # 열은 세움 — 놓였는데 값이 빈칸인 것이 보여야 함 row["values"][cid] = round((row["values"].get(cid) or 0.0) + value, 3) def result(self) -> dict[str, Any]: columns = sorted(self.columns.values(), key=lambda column: column["_sort"]) rows = [self.rows[key] for key in sorted(self.rows)] total: dict[str, float] = {} for row in rows: for cid, value in row["values"].items(): total[cid] = round(total.get(cid, 0.0) + value, 3) return { "columns": [{k: v for k, v in c.items() if k != "_sort"} for c in columns], "rows": rows, "total": total, } def _order(values: tuple[str, ...], value: str) -> int: return values.index(value) if value in values else len(values) def pipe_summary( project_root: str | Path, designs: list[dict[str, Any]], names: dict[str, str], one_m: float | None, ) -> dict[str, Any]: """배관 한 곳마다 컨테이너 값 → 집계표 열 · 줄 · 계(저장 안 함).""" items = fill.collect_items( project_root, fill.pipe_lengths_from_designs(designs), soil_tiers=soil_tiers_from_designs(designs), ) # 높이 등 빈 칸은 정의 기본값이 아니라 그림 규칙을 따라야 해 저장 그대로를 따로 읽음 raw = { f"pipe@{round(float(point.chainage_m), 3)}": dict(point.options or {}) for point in read_pipe_points_file(pipe_points_path_in(Path(project_root))) if (point.facility or "pipe") == "pipe" } basins = _basins(designs) by_owner: dict[str, dict[str, Any]] = {} for item in items: if item["type_id"] != "pipe": continue owner, _, link = item["복사본"].partition(GROUP_JOIN) entry = by_owner.setdefault(owner, {"links": {}}) if link: entry["links"][link] = _combo(item.get("summary_item")) else: entry["base"] = item sheet = _Sheet() no_length = no_basin = 0 soil_name = names.get(f"outlet.structure={SOIL}", SOIL) for owner, entry in by_owner.items(): base = entry.get("base") if base is None: continue links = entry["links"] options = raw.get(owner, {}) at = float(base["chainage_m"]) diameter = int(_num(base["options"].get("pipe_diameter_mm")) or 1000) pipe_head = ["관 공", f"Φ{names.get(f'pipe.diameter={diameter}', diameter)}m/m"] pipe_combos = [_combo(base.get("summary_item")), links.get(LINKS["pipe"])] length = base.get("관연장") no_length += not length sheet.add( at, f"len|{diameter}", (2, diameter, 0), [*pipe_head, "관매설"], "m", length, pipe_combos, ) sheet.add( at, f"cp|{diameter}", (2, diameter, 1), [*pipe_head, "커플링"], "ea", couplings(length, one_m), pipe_combos, ) shape = fill._nearest(basins, at) no_basin += shape is None if shape: sheet.add( at, f"basin|{diameter}|{shape}", (2, diameter, 2 + _order(SHAPES, shape), shape), [*pipe_head, "관보호공"], names.get(f"inlet.form={shape}", shape), 1.0, [links.get(LINKS["inlet"])], ) # 흙막이 — 유입(흙막이일 때) · 유출 · 흙막이 n(container_rows 가 실제 선 단까지만 냄) # 저장 키 = 필수는 `{쪽}_revet_{칸}_m` · 선택 흙막이는 `soil_add_{n}_{칸}`(명세 인스턴스 키) sources = [ (f"{side}_revet_", "_m", side, links.get(LINKS[side])) for side in ("inlet", "outlet") if options.get(f"{side}_type") == SOIL ] sources += [ (f"soil_add_{link[len('gsoiladd') :]}_", "", "soil_add", combo) for link, combo in links.items() if link.startswith("gsoiladd") ] for head, unit, container, combo in sources: form = str(options.get(f"{head}form") or "") height = _num(options.get(f"{head}height{unit}")) if height is None: height = pipe_wall_min_height_m(diameter / 1000) height_text = fill._label("height_m", height) sheet.add( at, f"revet|{form or fill.EMPTY_LABEL}|{height_text}", (0, _order(FORMS, form), form, height), [ soil_name, names.get(f"{container}.form={form}", form) or fill.EMPTY_TEXT, f"H={height_text}", ], "m", _revet_length(options, head, unit), [combo], ) # 돌붙임 — 붙임 형식 · 면적이 다 있을 때만(설계자가 넣음) for link, combo in links.items(): if not link.startswith("gchuteadd"): continue head = f"chute_add_{link[len('gchuteadd') :]}_" bond = str(options.get(f"{head}bond") or "") area = _num(options.get(f"{head}area")) if options.get(f"{head}structure") != "돌붙임" or not bond or not area: continue sheet.add( at, f"chute|{bond}", (1, _order(BONDS, bond), bond), [ names.get("chute_add.structure=돌붙임", "돌붙임"), names.get(f"chute_add.bond={bond}", bond), ], "㎡", area, [combo], ) out = sheet.result() notes = [] if no_length: notes.append(f"관 연장 없음 {no_length}곳 — B06 횡단 [저장] 뒤 채워짐") if no_basin: notes.append(f"집수정 모양 기록 없음 {no_basin}곳 — B06 횡단 [저장] 뒤 채워짐") if one_m is None: notes.append(f"1본 길이 변수({ONE_LENGTH_VAR}) 빈칸 — 커플링 빈칸") return {**out, "notes": notes, "one_length_m": one_m} def _summary(root: Path, designs: list[dict[str, Any]]) -> dict[str, Any]: names = ((definition.read().get("종류") or {}).get("pipe") or {}).get("이름") or {} return pipe_summary(root, designs, names, one_pipe_length(root)) @router.get("/{project_id}/quantity/summary") async def structure_summary(project_id: UUID) -> dict[str, Any]: root = await _project_root(project_id) async with get_db_pool().acquire() as connection: context = await get_confirmed_route_context(connection, project_id) context = context or await get_workflow_route_context(connection, project_id) # 노선 전이면 설계 없이 — 관은 서 있고 연장 · 집수정 모양만 빈칸(알림) designs = ( await get_cross_section_designs(connection, int(context["route_id"])) if context else [] ) return await asyncio.to_thread(_summary, root, designs)