Files
Aislo/B08_Quantity/B08_Quantity_Router_Hydro.py
T
eomsangdonandClaude Opus 5.5 426e574cd2 feat(B08): 수리집수면적표 우리 양식 · 공사지 지역 · 재현기간 드롭다운 · 처음 값 빈칸 · [저장](PLAN 69-4~69-6)
- 엑셀 모양(흰 종이 · 노랑 칸 · 쪽 표시) 대신 공용 카드 · 입력 · 드롭다운 · 테마 토큰 색
- 테두리는 입력 · 선택 칸만 · 굵게 + 붉은 계열은 Qd · 유출량 · 판정 · 유출율만 · 관 단면 그림 테마 색
- 지역 = 전처리 행정구역(노선 대표점) > 시군구 하나 > 프로젝트 등록 지역 · R24 = 전처리 강우량표의 재현기간별 24시간 값(없으면 입력 칸)
- 계산 때 R24 는 서버가 공사지 표로 정함 · 빈칸에 기대는 계산 칸은 「—」
- 처음 값 = 표준 값만(할증 2.0 · 관 종류 · 단면 가정 · 유출계수 0.8 · 재현기간 100년) · 측점 고유 값 빈칸
- [저장] = PUT …/quantity/hydro → B08_Quantity/quantities/hydro_inputs.json 만 · 성공 뒤 캐시 비움 · 캐시 판 올림(v2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W4uKaV4ZBhsJN4GHG7yRj4
2026-10-06 23:09:21 +09:00

108 lines
4.5 KiB
Python

"""B08 4탭 홍수량 계산서 길 (PLAN 62-2 · 62-3 · 69-5 · 69-6).
GET /api/projects/{id}/quantity/hydro 배수지점 측점(저장된 pipe_points) + 기본값 · 선택지
+ 공사지(지역 · 재현기간별 R24) + 저장한 입력
POST /api/projects/{id}/quantity/hydro/calc 화면 입력 → 계산서 값(B08_Quantity_Engine_Hydro)
PUT /api/projects/{id}/quantity/hydro [저장] — 측점별 입력을 B08 자기 파일에만 씀
측점 목록만 관 지점 정본에서 읽음 — 관경 · 관 종류 등 B05/B06 값은 잇지 않음(자기 기본값).
B05/B06 · 배수유역 파일에는 쓰지 않음.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Literal
from uuid import UUID
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from B02_ProjRegister.B02_ProjRegister_Repository import get_project_by_id
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B08_Quantity.B08_Quantity_Engine_Hydro import compute
from B08_Quantity.B08_Quantity_Hydro_Defaults import FACTORS, PIPE_KINDS, options
from B08_Quantity.B08_Quantity_Hydro_Site import read_saved, site_info, site_r24, write_saved
from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
from common_util.common_util_storage import resolve_stored_project_path
from config.config_db import get_db_pool
router = APIRouter(prefix="/api/projects", tags=["B08 Hydro"])
class HydroInputs(BaseModel):
"""빈칸(None) = 측점 고유 값을 아직 안 넣음 — 그 값에 기대는 계산 칸만 비움."""
factor: float = Field(gt=0)
runoff: float = Field(gt=0, le=1)
area_km2: float | None = Field(default=None, gt=0)
length_m: float | None = Field(default=None, gt=0)
height_m: float | None = Field(default=None, gt=0)
tc_min_minutes: float = Field(ge=0)
r24_mm: float | None = Field(default=None, gt=0)
region: str = ""
return_period: float = Field(gt=0)
pipe_kind: Literal[tuple(PIPE_KINDS)] # type: ignore[valid-type]
n: float | None = Field(default=None, gt=0)
diameter_mm: float | None = Field(default=None, gt=0)
slope_denominator: float | None = Field(default=None, gt=0)
section: Literal["full", "arc240", "eff70"]
class HydroSave(BaseModel):
"""측점 키(chainage 소수 셋째 자리) → 그 측점 입력 전부."""
stations: dict[str, HydroInputs]
async def _project_root(project_id: UUID) -> Path:
try:
async with get_db_pool().acquire() as connection:
stored = await get_project_storage_relative_path(connection, project_id)
except LookupError:
stored = None
if not stored:
raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.")
return Path(resolve_stored_project_path(stored))
@router.get("/{project_id}/quantity/hydro")
async def hydro_sheet(project_id: UUID) -> dict[str, Any]:
root = await _project_root(project_id)
points = read_pipe_points_file(pipe_points_path_in(root))
stations = sorted(
({"chainage_m": p.chainage_m, "facility": p.facility} for p in points),
key=lambda s: s["chainage_m"],
)
project = await get_project_by_id(str(project_id)) or {}
return {
"stations": stations,
**options(),
"site": site_info(root, project.get("region")),
"saved": read_saved(root),
}
@router.post("/{project_id}/quantity/hydro/calc")
async def hydro_calc(project_id: UUID, body: HydroInputs) -> dict[str, Any]:
root = await _project_root(project_id) # 없는 프로젝트는 404
if body.factor not in FACTORS:
raise HTTPException(status_code=400, detail=f"할증은 {FACTORS} 중 하나")
inputs = body.model_dump()
# 공사지 확률강우량이 있으면 그 값이 정본 — 화면 값을 받아 적지 않음
r24 = site_r24(root, body.return_period)
if r24 is not None:
inputs["r24_mm"] = r24
return compute(inputs)
@router.put("/{project_id}/quantity/hydro")
async def hydro_save(project_id: UUID, body: HydroSave) -> dict[str, Any]:
root = await _project_root(project_id)
for key, inputs in body.stations.items():
if inputs.factor not in FACTORS:
raise HTTPException(status_code=400, detail=f"{key}: 할증은 {FACTORS} 중 하나")
saved = {key: inputs.model_dump() for key, inputs in body.stations.items()}
write_saved(root, saved)
return {"saved": saved}