eomsangdonandClaude Opus 5.5
311c43d83a
- 04_detailed_basins + pipe_points → linked(A · L · H · 관경 · 박스 내공 · 월류 폭) · 참고 b05(Qd · T) - 측점 짝 carry_indices(±0.05 m) · 하상경사 · 관 종류 · n 은 안 이음 - 유출계수 · 할증 · 재현기간 · 최소 도달시간 = config DRAINAGE_* import - 시험 test_96_b08_hydro_linked(연결값 · Qd ≈ B05 설계유량) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BVPp2o6Mh2FZJr72yW3ofZ
116 lines
4.9 KiB
Python
116 lines
4.9 KiB
Python
"""B08 4탭 홍수량 계산서 길 (PLAN 62-2 · 62-3 · 69-5 · 69-6 · 73-11 · 73-12).
|
|
|
|
GET /api/projects/{id}/quantity/hydro 배수지점 측점(저장된 pipe_points) + B05 연결값
|
|
+ 기본값 · 선택지 + 공사지(지역 · R24) + 저장본
|
|
POST /api/projects/{id}/quantity/hydro/calc 화면 입력 → 계산서 값(B08_Quantity_Engine_Hydro)
|
|
PUT /api/projects/{id}/quantity/hydro [저장] — 측점별 입력을 B08 자기 파일에만 씀
|
|
|
|
측점 목록 · 시설 치수는 관 지점 정본 · A · L · H 는 B05 세부유역 저장본에서 연결값(96-1) —
|
|
화면이 빈칸만 채움 · 하상경사 · 관 종류 · n 은 잇지 않음. 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, station_rows
|
|
from B08_Quantity.B08_Quantity_Hydro_Site import (
|
|
read_basins,
|
|
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"]
|
|
# 장 — 박스(사각 단면 · 73-11) · 세월교 · 물넘이(월류 높이 · 73-12) · 그 밖 관
|
|
sheet: Literal["pipe", "box", "ford"] = "pipe"
|
|
box_width_m: float | None = Field(default=None, gt=0)
|
|
box_height_m: float | None = Field(default=None, gt=0)
|
|
depth_ratio: float | None = Field(default=None, gt=0, le=1)
|
|
overflow_width_m: float | None = Field(default=None, gt=0)
|
|
|
|
|
|
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)
|
|
stations = station_rows(read_pipe_points_file(pipe_points_path_in(root)), read_basins(root))
|
|
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}
|