eomsangdonandClaude Opus 5.5
e3cb11185c
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01F3mdb91125mPoC6tpsQx4P
612 lines
25 KiB
Python
612 lines
25 KiB
Python
"""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`)
|
||
BOX암거 › {가로×세로} › 본체 m B06 새 키 `box_length_m`(두 끝 offset 차 · 120)
|
||
› 날개벽 › 유입 · 유출 개소
|
||
세월교 › Φ{관경}×{n}련 › 월류 폭 m · 본체 개소 · 날개벽 › 유입 · 유출 개소
|
||
관 몫 = 관 공 › 관매설 · 커플링에 더함(B06 새 키 `ford_pipe_length_m` × 련 · 131)
|
||
물넘이포장 › 월류 폭 › H={월류 높이} m
|
||
노출형 횡단수로 · 개거 · 맹암거(B05 구조물) 개소
|
||
흙막이 · 돌붙임 열은 관 · 박스 · 세월교 공용(실무 꼴) — 빈 높이: 관 = 관경 최소 ·
|
||
박스 = 박스 높이 · 세월교 = 「미입력」(본체 끝 높이는 기하에만)
|
||
열 id 접두 `pipe:` = 이 집계표(횡단배수) 이름공간 · 같은 측점 박스 + 관은 종류로 갈라 셈.
|
||
머리 글 = M02 정의 이름(`구조물정의.json` types.name · 종류.{종류}.이름 — 사용자가 M02 에서 고침) ·
|
||
열마다 `combos` = 그 값을 낸 컨테이너에 이은 조합 id(`{층}|{주인}|{키}` · 95-3 모달 · 맨 윗 머리).
|
||
설계 = B08 토공과 같은 길(확정 노선 먼저) · 관 · 컨테이너 줄 = M02 채움 `collect_items` 재사용.
|
||
`pipes` = 관마다 길이(커플링 열 · 측점) — 좌측 패널 1본 길이를 바꾸면 화면이 같은 식으로 바로 다시 셈
|
||
(짝 = `couplings` ↔ `B08_Quantity_Options_Calc.ts` `couplings` · `recountCouplings` · PLAN 122).
|
||
"""
|
||
|
||
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_height_m, 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 = "기슭막이" # 저장 값(화면 이름 흙막이)
|
||
GUARD_SOIL = "흙막이" # 박스 · 세월교 유입 · 유출 구조 값
|
||
WING = "날개벽"
|
||
#: 박스 · 세월교 필수 컨테이너 잇기 키(`design_structure_box` · `_ford`) · 물넘이 경사각도
|
||
BOX_LINKS = {"body": "gboxbody", "inlet": "gboxin", "outlet": "gboxout"}
|
||
FORD_LINKS = {"body": "gford", "inlet": "gfordin", "outlet": "gfordout"}
|
||
PAVE_LINKS = {"body": "gfordpaveslope"}
|
||
#: 옛 날개벽 설치 칸(「있음」 · 「없음」)
|
||
WING_KEYS = {"inlet": "wing_in", "outlet": "wing_out"}
|
||
SIDE_NAMES = {"inlet": "유입", "outlet": "유출"}
|
||
#: 개소 1 로 세는 B05 구조물(structures.json · 옵션 거의 없음)
|
||
SIMPLE = ("cross_drain_exposed", "open_ditch", "cross_underdrain")
|
||
#: 이 집계표가 펼치는 종류 — 화면은 이 종류를 조합 열에서 뺌(응답 `types`)
|
||
DRAIN_TYPES = ("pipe", "box_culvert", "ford_bridge", "ford_pavement", *SIMPLE)
|
||
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 _station_values(designs: Any, key: str) -> dict[float, float]:
|
||
"""설계 → `{측점: 양수 값}` — 값 없는 측점은 안 담음."""
|
||
found: dict[float, float] = {}
|
||
for row in designs or []:
|
||
design = row.get("design") if isinstance(row, dict) else None
|
||
value = _num(design.get(key)) if isinstance(design, dict) else None
|
||
if value and value > 0:
|
||
found[round(float(row.get("chainage_m") or 0.0), 3)] = value
|
||
return found
|
||
|
||
|
||
def _guard(type_id: str, options: dict[str, Any], side: str) -> str:
|
||
"""박스 · 세월교 유입 · 유출 구조(없음 · 날개벽 · 흙막이) — 옛 설치 칸 「있음」 = 날개벽.
|
||
짝(TS): 박스 `boxOptionsIn` · 세월교 `withLegacyGuards`(`A00_Common/design/`)."""
|
||
wing = options.get(WING_KEYS[side])
|
||
if type_id == "box_culvert":
|
||
return WING if wing == "있음" else str(wing or "")
|
||
kind = str(options.get(f"{side}_type") or "")
|
||
if wing == "있음":
|
||
return WING
|
||
if wing == "없음" and (kind or WING) == WING:
|
||
return "없음"
|
||
return kind
|
||
|
||
|
||
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 _revets(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
options: dict[str, Any],
|
||
links: dict[str, str | None],
|
||
sides: list[tuple[str, str | None]],
|
||
empty_height: float | None,
|
||
names: dict[str, str],
|
||
) -> None:
|
||
"""흙막이 — 유입 · 유출(흙막이일 때 · `sides`) + 흙막이 n(container_rows 가 선 단까지만 냄).
|
||
|
||
저장 키 = 필수는 `{쪽}_revet_{칸}_m` · 선택 흙막이는 `soil_add_{n}_{칸}`(명세 인스턴스 키) ·
|
||
빈 높이 = `empty_height`(None 이면 「미입력」 열) · 머리 글은 관 정의 이름(공용 열).
|
||
"""
|
||
sources = [(f"{side}_revet_", "_m", side, combo) for side, combo in sides]
|
||
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 = empty_height
|
||
height_text = fill._label("height_m", height)
|
||
sheet.add(
|
||
at,
|
||
f"revet|{form or fill.EMPTY_LABEL}|{height_text}",
|
||
(0, _order(FORMS, form), form, math.inf if height is None else height),
|
||
[
|
||
names.get(f"outlet.structure={SOIL}", SOIL),
|
||
names.get(f"{container}.form={form}", form) or fill.EMPTY_TEXT,
|
||
fill.EMPTY_TEXT if height is None else f"H={height_text}",
|
||
],
|
||
"m",
|
||
_revet_length(options, head, unit),
|
||
[combo],
|
||
)
|
||
|
||
|
||
def _chutes(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
options: dict[str, Any],
|
||
links: dict[str, str | None],
|
||
names: dict[str, str],
|
||
) -> None:
|
||
"""돌붙임 — 붙임 형식 · 면적이 다 있을 때만(설계자가 넣음)."""
|
||
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],
|
||
)
|
||
|
||
|
||
def _wings(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
type_id: str,
|
||
sort: int,
|
||
title: str,
|
||
options: dict[str, Any],
|
||
combos: dict[str, str | None],
|
||
names: dict[str, str],
|
||
) -> list[tuple[str, str | None]]:
|
||
"""박스 · 세월교 날개벽(유입 · 유출 개소) — 흙막이 쪽은 돌려줌(공용 흙막이 열로)."""
|
||
soils: list[tuple[str, str | None]] = []
|
||
for index, side in enumerate(("inlet", "outlet")):
|
||
kind = _guard(type_id, options, side)
|
||
if kind == GUARD_SOIL:
|
||
soils.append((side, combos.get(side)))
|
||
elif kind == WING:
|
||
sheet.add(
|
||
at,
|
||
f"{type_id}|wing|{side}",
|
||
(sort, 9, index),
|
||
[title, names.get(f"{side}.structure={WING}", WING), SIDE_NAMES[side]],
|
||
"개소",
|
||
1.0,
|
||
[combos.get(side)],
|
||
)
|
||
return soils
|
||
|
||
|
||
def _box(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
base: dict[str, Any],
|
||
combos: dict[str, str | None],
|
||
names: dict[str, str],
|
||
length: float | None,
|
||
) -> float | None:
|
||
"""BOX암거 본체(가로 × 세로마다 m) — 흙막이 빈 높이로 쓸 박스 높이를 돌려줌."""
|
||
width = _num(base["options"].get("body_width_m"))
|
||
height = _num(base["options"].get("body_height_m"))
|
||
size = (
|
||
f"{fill._label('width_m', width)}×{fill._label('height_m', height)}"
|
||
if width or height
|
||
else fill.EMPTY_LABEL
|
||
)
|
||
sheet.add(
|
||
at,
|
||
f"box|body|{size}",
|
||
(3, 0, width or 0.0, height or 0.0),
|
||
[
|
||
names["종류"],
|
||
names.get(f"body.size={size}", fill.EMPTY_TEXT if size == fill.EMPTY_LABEL else size),
|
||
"본체",
|
||
],
|
||
"m",
|
||
length,
|
||
[_combo(base.get("summary_item")), combos.get("body")],
|
||
)
|
||
return height
|
||
|
||
|
||
def _ford(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
base: dict[str, Any],
|
||
combos: dict[str, str | None],
|
||
names: dict[str, str],
|
||
) -> None:
|
||
"""세월교 — Φ관경×n련마다 월류 폭 m · 본체 개소."""
|
||
diameter = int(_num(base["options"].get("pipe_diameter_mm")) or 1000)
|
||
count = int(_num(base["options"].get("pipe_count")) or 1)
|
||
spec = f"Φ{names.get(f'ford.diameter={diameter}', diameter)}×{count}련"
|
||
width_name = names.get("ford.width", "월류 폭").split(" (")[0]
|
||
cells = (
|
||
("width", width_name, "m", _num(base["options"].get("ford_width_m"))),
|
||
("count", "본체", "개소", 1.0),
|
||
)
|
||
for index, (key, label, unit, value) in enumerate(cells):
|
||
sheet.add(
|
||
at,
|
||
f"ford|{diameter}|{count}|{key}",
|
||
(4, 0, diameter, count, index),
|
||
[names["종류"], spec, label],
|
||
unit,
|
||
value,
|
||
[_combo(base.get("summary_item")), combos.get("body")],
|
||
)
|
||
|
||
|
||
def _ford_pipes(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
base: dict[str, Any],
|
||
length: float | None,
|
||
one_m: float | None,
|
||
names: dict[str, str],
|
||
pipes: list[dict[str, Any]],
|
||
) -> None:
|
||
"""세월교 관 몫 — 한 련 길이 × 련 수를 관매설 · 커플링 열에 더함(관 공과 같은 열 · 131).
|
||
|
||
`pipes` 에는 련마다 한 줄(좌측 패널 1본 길이로 커플링을 다시 셀 때 련 수만큼 셈).
|
||
"""
|
||
diameter = int(_num(base["options"].get("pipe_diameter_mm")) or 1000)
|
||
count = int(_num(base["options"].get("pipe_count")) or 1)
|
||
head = ["관 공", f"Φ{names.get(f'pipe.diameter={diameter}', diameter)}m/m"]
|
||
combos = [_combo(base.get("summary_item"))]
|
||
each = couplings(length, one_m)
|
||
sheet.add(
|
||
at,
|
||
f"len|{diameter}",
|
||
(2, diameter, 0),
|
||
[*head, "관매설"],
|
||
"m",
|
||
length * count if length else None,
|
||
combos,
|
||
)
|
||
sheet.add(
|
||
at,
|
||
f"cp|{diameter}",
|
||
(2, diameter, 1),
|
||
[*head, "커플링"],
|
||
"ea",
|
||
None if each is None else each * count,
|
||
combos,
|
||
)
|
||
pipes += [{"chainage_m": at, "column": f"pipe:cp|{diameter}", "length_m": length}] * count
|
||
|
||
|
||
def _pavement(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
base: dict[str, Any],
|
||
combos: dict[str, str | None],
|
||
names: dict[str, str],
|
||
) -> None:
|
||
"""물넘이 포장 — 월류 높이(H)마다 월류 폭 m(실무 「물넘이 포장 H=1.0」)."""
|
||
height = _num(base["options"].get("ford_height_m"))
|
||
height_text = fill._label("height_m", height)
|
||
sheet.add(
|
||
at,
|
||
f"pave|{height_text}",
|
||
(5, math.inf if height is None else height),
|
||
[
|
||
names["종류"],
|
||
names.get("slope.width", "월류 폭").split(" (")[0],
|
||
fill.EMPTY_TEXT if height is None else f"H={height_text}",
|
||
],
|
||
"m",
|
||
_num(base["options"].get("ford_width_m")),
|
||
[_combo(base.get("summary_item")), combos.get("body")],
|
||
)
|
||
|
||
|
||
def _drain(
|
||
sheet: _Sheet,
|
||
at: float,
|
||
type_id: str,
|
||
base: dict[str, Any],
|
||
options: dict[str, Any],
|
||
links: dict[str, str | None],
|
||
group: dict[str, Any],
|
||
kinds: dict[str, Any],
|
||
pipe_names: dict[str, str],
|
||
box_length: float | None,
|
||
) -> None:
|
||
"""관 밖 횡단배수 한 곳 — 종류마다 본체 열 + 박스 · 세월교는 날개벽 · 흙막이 · 돌붙임(공용)."""
|
||
title = next(
|
||
(t["name"] for t in kinds.get("types") or [] if t.get("type_id") == type_id), type_id
|
||
)
|
||
names = {**(((kinds.get("종류") or {}).get(type_id) or {}).get("이름") or {}), "종류": title}
|
||
if type_id in SIMPLE:
|
||
combos = [_combo(base.get("summary_item")), *links.values()]
|
||
sheet.add(at, type_id, (6, SIMPLE.index(type_id)), [title], "개소", 1.0, combos)
|
||
return
|
||
# 필수 컨테이너 잇기 — 정의 그룹에 없는 것(세월교)은 저장된 잇기에서
|
||
required = {"box_culvert": BOX_LINKS, "ford_bridge": FORD_LINKS}.get(type_id, PAVE_LINKS)
|
||
combos = {part: links.get(link) or _combo(group.get(link)) for part, link in required.items()}
|
||
if type_id == "ford_pavement":
|
||
_pavement(sheet, at, base, combos, names)
|
||
return
|
||
if type_id == "box_culvert":
|
||
empty_height = _box(sheet, at, base, combos, names, box_length)
|
||
else:
|
||
_ford(sheet, at, base, combos, names)
|
||
empty_height = None # 본체 끝 높이는 기하에만 — 「미입력」 열
|
||
sort = 3 if type_id == "box_culvert" else 4
|
||
soils = _wings(sheet, at, type_id, sort, title, base["options"], combos, names)
|
||
_revets(sheet, at, options, links, soils, empty_height, pipe_names)
|
||
_chutes(sheet, at, options, links, pipe_names)
|
||
|
||
|
||
def pipe_summary(
|
||
project_root: str | Path,
|
||
designs: list[dict[str, Any]],
|
||
names: dict[str, str],
|
||
one_m: float | None,
|
||
kinds: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""횡단배수 한 곳마다 컨테이너 값 → 집계표 열 · 줄 · 계(저장 안 함).
|
||
|
||
`names` = 관 정의 이름(공용 흙막이 · 돌붙임 머리) · `kinds` = M02 정의(비면 읽음).
|
||
"""
|
||
kinds = kinds or definition.read()
|
||
items = fill.collect_items(
|
||
project_root,
|
||
fill.pipe_lengths_from_designs(designs),
|
||
soil_tiers=soil_tiers_from_designs(designs),
|
||
)
|
||
# 높이 등 빈 칸은 정의 기본값이 아니라 그림 규칙을 따라야 해 저장 그대로를 따로 읽음
|
||
# 같은 측점 박스 + 관은 복사본 id(`pipe@측점`)가 같음 — 종류로 가름
|
||
raw = {
|
||
(point.facility or "pipe", f"pipe@{round(float(point.chainage_m), 3)}"): point
|
||
for point in read_pipe_points_file(pipe_points_path_in(Path(project_root)))
|
||
}
|
||
basins = _basins(designs)
|
||
box_lengths = _station_values(designs, "box_length_m")
|
||
ford_lengths = _station_values(designs, "ford_pipe_length_m")
|
||
by_owner: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for item in items:
|
||
if item["type_id"] not in DRAIN_TYPES:
|
||
continue
|
||
owner, _, link = item["복사본"].partition(GROUP_JOIN)
|
||
entry = by_owner.setdefault((item["type_id"], owner), {"links": {}})
|
||
if link:
|
||
entry["links"][link] = _combo(item.get("summary_item"))
|
||
else:
|
||
entry["base"] = item
|
||
|
||
sheet = _Sheet()
|
||
pipes: list[dict[str, Any]] = []
|
||
no_length = no_basin = no_box = no_ford = 0
|
||
for (type_id, owner), entry in by_owner.items():
|
||
base = entry.get("base")
|
||
if base is None:
|
||
continue
|
||
links = entry["links"]
|
||
point = raw.get((type_id, owner))
|
||
options = dict(point.options or {}) if point else {}
|
||
at = float(base["chainage_m"])
|
||
if type_id != "pipe":
|
||
box_length = fill._nearest(box_lengths, at) if type_id == "box_culvert" else None
|
||
no_box += type_id == "box_culvert" and not box_length
|
||
group = (point.group_items if point else None) or {}
|
||
_drain(sheet, at, type_id, base, options, links, group, kinds, names, box_length)
|
||
if type_id == "ford_bridge":
|
||
ford_length = fill._nearest(ford_lengths, at)
|
||
no_ford += not ford_length
|
||
_ford_pipes(sheet, at, base, ford_length, one_m, names, pipes)
|
||
continue
|
||
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,
|
||
)
|
||
pipes.append({"chainage_m": at, "column": f"pipe:cp|{diameter}", "length_m": length})
|
||
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"])],
|
||
)
|
||
|
||
sides = [side for side in ("inlet", "outlet") if options.get(f"{side}_type") == SOIL]
|
||
sources = [(side, links.get(LINKS[side])) for side in sides]
|
||
# 옛 기본 1.5 · 관경 > 1.0 = 관경 기준 값(139-1 · 그림과 한 값)
|
||
moved = {
|
||
f"{side}_revet_height_m": pipe_wall_height_m(
|
||
_num(options.get(f"{side}_revet_height_m")), diameter / 1000
|
||
)
|
||
for side in sides
|
||
}
|
||
wall_options = {**options, **moved}
|
||
_revets(
|
||
sheet, at, wall_options, links, sources, pipe_wall_min_height_m(diameter / 1000), names
|
||
)
|
||
_chutes(sheet, at, options, links, names)
|
||
|
||
out = sheet.result()
|
||
notes = []
|
||
if no_length:
|
||
notes.append(f"관 연장 없음 {no_length}곳 — B06 횡단 [저장] 뒤 채워짐")
|
||
if no_basin:
|
||
notes.append(f"집수정 모양 기록 없음 {no_basin}곳 — B06 횡단 [저장] 뒤 채워짐")
|
||
if no_box:
|
||
notes.append(f"BOX 본체 길이 없음 {no_box}곳 — B06 횡단 [저장] 뒤 채워짐")
|
||
if no_ford:
|
||
notes.append(f"세월교 관 길이 없음 {no_ford}곳 — B06 횡단 [저장] 뒤 채워짐")
|
||
if one_m is None:
|
||
notes.append(f"1본 길이 변수({ONE_LENGTH_VAR}) 빈칸 — 커플링 빈칸")
|
||
return {
|
||
**out,
|
||
"notes": notes,
|
||
"one_length_m": one_m,
|
||
"types": list(DRAIN_TYPES),
|
||
"pipes": pipes,
|
||
}
|
||
|
||
|
||
def _summary(root: Path, designs: list[dict[str, Any]]) -> dict[str, Any]:
|
||
kinds = definition.read()
|
||
names = ((kinds.get("종류") or {}).get("pipe") or {}).get("이름") or {}
|
||
return pipe_summary(root, designs, names, one_pipe_length(root), kinds)
|
||
|
||
|
||
@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)
|