feat(B08): 수리 표 측점마다 B05 배수유역 연결값 · 표준값 B05 배수 설정 하나로(96-1 · 96-3)
- 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
This commit is contained in:
Claude Opus 5.5
@@ -1,6 +1,7 @@
|
||||
"""B08 4탭 홍수량 계산서 — 기본값 · 선택지 한 곳 (PLAN 62-3 · 73-11 · 73-12).
|
||||
"""B08 4탭 홍수량 계산서 — 기본값 · 선택지 · B05 연결값 한 곳 (PLAN 62-3 · 73-11 · 73-12 · 96).
|
||||
|
||||
임의로 정한 값 — 사용자가 나중에 고침. B05/B06 · 배수유역 값과 잇지 않음(자기 기본값만).
|
||||
표준값(유출계수 · 할증 · 재현기간 · 최소 도달시간)은 B05 배수 설정(config DRAINAGE_*)과 같은 값.
|
||||
측점 고유 값은 B05 배수유역 저장본의 연결값(linked · 96-1) · 하상경사 · 관 종류 · n 은 안 이음.
|
||||
근거 후보: docs/raw/조사/2026-10-06_62-1_홍수량_방법.md (별표2 2.0배 · KDS 표 2.7-1 · 실무 견본).
|
||||
"""
|
||||
|
||||
@@ -8,7 +9,14 @@ from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from common_util.common_util_drainage_pipes import carry_indices
|
||||
from config.config_system import FORD_MANNING_N, FORD_SLOPE
|
||||
from config.config_system_design import (
|
||||
DRAINAGE_DESIGN_FLOW_FACTOR,
|
||||
DRAINAGE_DESIGN_RETURN_PERIOD_YR,
|
||||
DRAINAGE_RUNOFF_COEFFICIENT,
|
||||
DRAINAGE_TC_MIN_MINUTES,
|
||||
)
|
||||
|
||||
#: 할증(Qd = Q × 할증) — 2.0 별표2 법정 · 1.5 견본(울진 2공구) · 1.2 구기준
|
||||
FACTORS = (2.0, 1.5, 1.2)
|
||||
@@ -22,15 +30,15 @@ PIPE_KINDS = {"파형강관": 0.024, "흄관": 0.013, "콘크리트관": 0.015}
|
||||
#: 처음 값 = 표준 값만(69-6) · 측점 고유 값(유역면적 · L · H · D · S)은 None(빈칸)
|
||||
#: R24 · 지역은 공사지 자료(B08_Quantity_Hydro_Site) — 자료가 없을 때만 화면 입력
|
||||
DEFAULTS: dict[str, Any] = {
|
||||
"factor": 2.0,
|
||||
"runoff": 0.80, # 유출계수 f — KDS 가파른 산지
|
||||
"factor": DRAINAGE_DESIGN_FLOW_FACTOR,
|
||||
"runoff": DRAINAGE_RUNOFF_COEFFICIENT, # 유출계수 f — KDS 가파른 산지
|
||||
"area_km2": None,
|
||||
"length_m": None, # 산정에서 하천시점까지 거리
|
||||
"height_m": None, # 산정에서 하천시점까지 표고차
|
||||
"tc_min_minutes": 5.0, # 최소 도달시간(KDS 2.4)
|
||||
"tc_min_minutes": DRAINAGE_TC_MIN_MINUTES, # 최소 도달시간(KDS 2.4)
|
||||
"r24_mm": None,
|
||||
"region": "",
|
||||
"return_period": 100,
|
||||
"return_period": DRAINAGE_DESIGN_RETURN_PERIOD_YR,
|
||||
"pipe_kind": "파형강관",
|
||||
"n": None, # None = 관 종류 값
|
||||
"diameter_mm": None,
|
||||
@@ -58,15 +66,62 @@ SHEET_DEFAULTS: dict[str, dict[str, Any]] = {
|
||||
}
|
||||
|
||||
|
||||
def station_rows(points: list[Any]) -> list[dict[str, Any]]:
|
||||
"""관 지점 → 보조 탭 줄(측점 차례) · 장 종류(pipe · box · ford)."""
|
||||
def _positive(value: Any) -> float | None:
|
||||
"""양수만 — 글 '1000' 도 숫자로 · 0 · 빈 값 · 글자는 None(빈칸)."""
|
||||
try:
|
||||
number = float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return number if number > 0 else None
|
||||
|
||||
|
||||
def _linked(sheet: str, options: dict[str, Any], basin: dict[str, Any] | None) -> dict[str, Any]:
|
||||
"""B05 값 → 계산서 칸(96-1) — 값이 있는 칸만 · A · L · H 는 유역 · 치수는 시설 옵션."""
|
||||
values: dict[str, Any] = {}
|
||||
if basin is not None:
|
||||
area_m2 = _positive(basin.get("area_m2"))
|
||||
values["area_km2"] = area_m2 / 1e6 if area_m2 else None
|
||||
values["length_m"] = _positive(basin.get("flow_length_m"))
|
||||
values["height_m"] = _positive(basin.get("relief_m"))
|
||||
if sheet == "pipe":
|
||||
values["diameter_mm"] = _positive(options.get("pipe_diameter_mm"))
|
||||
elif sheet == "box":
|
||||
values["box_width_m"] = _positive(options.get("body_width_m"))
|
||||
values["box_height_m"] = _positive(options.get("body_height_m"))
|
||||
else:
|
||||
values["overflow_width_m"] = _positive(options.get("ford_width_m"))
|
||||
return {k: v for k, v in values.items() if v is not None}
|
||||
|
||||
|
||||
def station_rows(
|
||||
points: list[Any], basins: list[dict[str, Any]] | None = None
|
||||
) -> list[dict[str, Any]]:
|
||||
"""관 지점 → 보조 탭 줄(측점 차례) · 장 종류(pipe · box · ford).
|
||||
|
||||
basins(B05 세부유역 속성) 를 주면 측점마다 연결값 `linked` · 비교용 `b05`(Qd · T) — 측점 짝은
|
||||
`carry_indices`(±0.05 m) · 안 주면 둘 다 None.
|
||||
"""
|
||||
points = sorted(points, key=lambda p: p.chainage_m)
|
||||
picks = (
|
||||
carry_indices([p.chainage_m for p in points], [float(b["chainage_m"]) for b in basins])
|
||||
if basins is not None
|
||||
else [None] * len(points)
|
||||
)
|
||||
rows = []
|
||||
for p in points:
|
||||
for p, pick in zip(points, picks):
|
||||
sheet, kind = SHEETS.get(p.facility, ("pipe", ""))
|
||||
rows.append(
|
||||
{"chainage_m": p.chainage_m, "facility": p.facility, "kind": kind, "sheet": sheet}
|
||||
)
|
||||
return sorted(rows, key=lambda s: s["chainage_m"])
|
||||
row = {"chainage_m": p.chainage_m, "facility": p.facility, "kind": kind, "sheet": sheet}
|
||||
row["linked"] = row["b05"] = None
|
||||
if basins is not None:
|
||||
basin = basins[pick] if pick is not None else None
|
||||
row["linked"] = _linked(sheet, getattr(p, "options", None) or {}, basin)
|
||||
if basin is not None:
|
||||
row["b05"] = {
|
||||
"design_flow_m3s": basin.get("design_flow_m3s"),
|
||||
"tc_minutes": basin.get("tc_minutes"),
|
||||
}
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
def options() -> dict[str, Any]:
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
"""B08 4탭 홍수량 계산서 — 공사지 지역 · 확률강우량 · 저장 파일 (PLAN 69-5 · 69-6).
|
||||
|
||||
읽기만: 전처리 강우량표(`B04_PreProcess/drainage/rainfall_table.json` · 관측소 확률강우량)
|
||||
· 행정구역(`B04_PreProcess/processed/행정구역_*_bounds.geojson` · WGS84) · 프로젝트 등록 지역(DB).
|
||||
· 행정구역(`B04_PreProcess/processed/행정구역_*_bounds.geojson` · WGS84) · 프로젝트 등록 지역(DB)
|
||||
· B05 세부유역(`B04_PreProcess/drainage/04_detailed_basins.geojson` · B05 [저장] 때 씀 · 96-1).
|
||||
배수유역 계산 코드는 부르지 않음 — 이미 저장된 파일만 읽음.
|
||||
|
||||
쓰기: B08 자기 파일 `B08_Quantity/quantities/hydro_inputs.json` 하나([저장] 때만).
|
||||
@@ -16,7 +17,10 @@ from typing import Any
|
||||
|
||||
from shapely.geometry import Point, shape
|
||||
|
||||
RAINFALL_FILE = Path("B04_PreProcess") / "drainage" / "rainfall_table.json"
|
||||
from config.config_system_design import DRAINAGE_CACHE_DIRNAME, DRAINAGE_DETAIL_FILENAME
|
||||
|
||||
RAINFALL_FILE = Path("B04_PreProcess") / DRAINAGE_CACHE_DIRNAME / "rainfall_table.json"
|
||||
BASINS_FILE = Path("B04_PreProcess") / DRAINAGE_CACHE_DIRNAME / DRAINAGE_DETAIL_FILENAME
|
||||
PROCESSED_DIR = Path("B04_PreProcess") / "processed"
|
||||
EMD_FILE = "행정구역_읍면동_bounds.geojson"
|
||||
SGG_FILE = "행정구역_시군구_bounds.geojson"
|
||||
@@ -105,6 +109,18 @@ def site_r24(root: Path, return_period: float) -> float | None:
|
||||
return _r24_table(_read_json(root / RAINFALL_FILE)).get(int(return_period))
|
||||
|
||||
|
||||
def read_basins(root: Path) -> list[dict[str, Any]]:
|
||||
"""B05 세부유역 속성(kind detail_basin) — 파일이 없으면 빈 목록(연결값은 시설 옵션만)."""
|
||||
basins = []
|
||||
for feature in _features(root / BASINS_FILE):
|
||||
props = feature.get("properties") or {}
|
||||
if props.get("kind") == "detail_basin" and isinstance(
|
||||
props.get("chainage_m"), (int, float)
|
||||
):
|
||||
basins.append(props)
|
||||
return basins
|
||||
|
||||
|
||||
def read_saved(root: Path) -> dict[str, dict[str, Any]]:
|
||||
data = _read_json(root / SAVED_FILE) or {}
|
||||
stations = data.get("stations")
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
"""B08 4탭 홍수량 계산서 길 (PLAN 62-2 · 62-3 · 69-5 · 69-6 · 73-11 · 73-12).
|
||||
|
||||
GET /api/projects/{id}/quantity/hydro 배수지점 측점(저장된 pipe_points) + 기본값 · 선택지
|
||||
+ 공사지(지역 · 재현기간별 R24) + 저장한 입력
|
||||
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 자기 파일에만 씀
|
||||
|
||||
측점 목록만 관 지점 정본에서 읽음 — 관경 · 관 종류 등 B05/B06 값은 잇지 않음(자기 기본값).
|
||||
B05/B06 · 배수유역 파일에는 쓰지 않음.
|
||||
측점 목록 · 시설 치수는 관 지점 정본 · A · L · H 는 B05 세부유역 저장본에서 연결값(96-1) —
|
||||
화면이 빈칸만 채움 · 하상경사 · 관 종류 · n 은 잇지 않음. B05/B06 · 배수유역 파일에는 쓰지 않음.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -22,7 +22,13 @@ 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_saved, site_info, site_r24, write_saved
|
||||
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
|
||||
@@ -75,7 +81,7 @@ async def _project_root(project_id: UUID) -> Path:
|
||||
@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)))
|
||||
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,
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
"""B08 수리 표 ← B05 배수유역 연결값 (PLAN 96-1 · 96-3).
|
||||
|
||||
견본 세부유역 + 관 지점 → 측점마다 linked(A · L · H · 관경 · 박스 내공 · 월류 폭) ·
|
||||
그 값으로 계산한 Qd ≈ B05 설계유량 · 표준값은 B05 배수 설정 하나.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from B08_Quantity.B08_Quantity_Engine_Hydro import compute # noqa: E402
|
||||
from B08_Quantity.B08_Quantity_Hydro_Defaults import DEFAULTS, station_rows # noqa: E402
|
||||
from B08_Quantity.B08_Quantity_Hydro_Site import BASINS_FILE, read_basins # noqa: E402
|
||||
from common_util.common_util_drainage_detail import size_pipe # noqa: E402
|
||||
from common_util.common_util_drainage_pipes import PipePoint # noqa: E402
|
||||
from common_util.common_util_wamis_station import mononobe_idf # noqa: E402
|
||||
from config import config_system_design as cfg # noqa: E402
|
||||
|
||||
R24 = 231.56
|
||||
BASINS = [
|
||||
# B05 저장 꼴 — 측점 0.01 반올림 · 면적 0.1 · 표고차 0.01 · 길이 0.1
|
||||
{"kind": "detail_basin", "chainage_m": 85.05, "area_m2": 10717.0, "flow_length_m": 163.9,
|
||||
"relief_m": 120.0},
|
||||
{"kind": "detail_basin", "chainage_m": 173.09, "area_m2": 141145.0, "flow_length_m": 603.3,
|
||||
"relief_m": 320.0},
|
||||
{"kind": "detail_basin", "chainage_m": 300.0, "area_m2": 52000.0, "flow_length_m": 410.0,
|
||||
"relief_m": 0.0},
|
||||
] # fmt: skip
|
||||
for _b in BASINS:
|
||||
_out = size_pipe(
|
||||
_b["area_m2"], max(_b["relief_m"], 0.1), _b["flow_length_m"], mononobe_idf(R24)
|
||||
)
|
||||
_b.update(design_flow_m3s=_out["design_flow_m3s"], tc_minutes=_out["tc_minutes"])
|
||||
|
||||
POINTS = [
|
||||
PipePoint(chainage_m=300.004, facility="box_culvert",
|
||||
options={"body_width_m": 2.0, "body_height_m": 1.5}),
|
||||
PipePoint(chainage_m=85.052, options={"pipe_diameter_mm": "1000", "pipe_kind": "파형강관"}),
|
||||
PipePoint(chainage_m=173.09, facility="ford_bridge", options={"ford_width_m": 10}),
|
||||
PipePoint(chainage_m=500.0, options={"pipe_diameter_mm": 800}), # 유역 짝 없음
|
||||
] # fmt: skip
|
||||
|
||||
|
||||
def test_측점마다_연결값() -> None:
|
||||
rows = station_rows(POINTS, BASINS)
|
||||
assert [r["chainage_m"] for r in rows] == [85.052, 173.09, 300.004, 500.0]
|
||||
pipe, ford, box, lone = (r["linked"] for r in rows)
|
||||
assert pipe == {
|
||||
"area_km2": 0.010717,
|
||||
"length_m": 163.9,
|
||||
"height_m": 120.0,
|
||||
"diameter_mm": 1000.0,
|
||||
}
|
||||
assert ford == {"area_km2": 0.141145, "length_m": 603.3, "height_m": 320.0,
|
||||
"overflow_width_m": 10.0} # fmt: skip
|
||||
# 표고차 0 은 빈칸(저장 검사 gt 0) · 박스는 내공
|
||||
assert box == {"area_km2": 0.052, "length_m": 410.0, "box_width_m": 2.0, "box_height_m": 1.5}
|
||||
# 유역 짝이 없으면 시설 치수만 · B05 참고값 없음
|
||||
assert lone == {"diameter_mm": 800.0} and rows[3]["b05"] is None
|
||||
assert rows[0]["b05"] == {"design_flow_m3s": BASINS[0]["design_flow_m3s"], "tc_minutes": 5.0}
|
||||
|
||||
|
||||
def test_잇지_않는_칸() -> None:
|
||||
for row in station_rows(POINTS, BASINS):
|
||||
assert not {"slope_denominator", "pipe_kind", "n"} & set(row["linked"])
|
||||
|
||||
|
||||
def test_basins_를_안_주면_linked_없음() -> None:
|
||||
rows = station_rows(POINTS)
|
||||
assert all(r["linked"] is None and r["b05"] is None for r in rows)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("index", [0, 1])
|
||||
def test_연결값으로_계산한_Qd_는_B05_설계유량(index: int) -> None:
|
||||
row = station_rows(POINTS, BASINS)[index]
|
||||
out = compute({**DEFAULTS, **row["linked"], "r24_mm": R24, "sheet": row["sheet"]})
|
||||
assert out["design_flow"] == pytest.approx(row["b05"]["design_flow_m3s"], abs=1e-4)
|
||||
assert out["tc_hr"] * 60 == pytest.approx(row["b05"]["tc_minutes"], abs=0.05)
|
||||
|
||||
|
||||
def test_표준값은_B05_배수_설정() -> None:
|
||||
assert DEFAULTS["runoff"] == cfg.DRAINAGE_RUNOFF_COEFFICIENT == 0.8
|
||||
assert DEFAULTS["factor"] == cfg.DRAINAGE_DESIGN_FLOW_FACTOR == 2.0
|
||||
assert DEFAULTS["return_period"] == cfg.DRAINAGE_DESIGN_RETURN_PERIOD_YR == 100
|
||||
assert DEFAULTS["tc_min_minutes"] == cfg.DRAINAGE_TC_MIN_MINUTES == 5.0
|
||||
|
||||
|
||||
def test_세부유역_파일_읽기(tmp_path: Path) -> None:
|
||||
assert read_basins(tmp_path) == [] # 분석 저장 전
|
||||
features = [{"type": "Feature", "properties": b, "geometry": None} for b in BASINS]
|
||||
features.append({"type": "Feature", "properties": {"kind": "pipe", "chainage_m": 85.05}})
|
||||
path = tmp_path / BASINS_FILE
|
||||
path.parent.mkdir(parents=True)
|
||||
path.write_text(json.dumps({"type": "FeatureCollection", "features": features}), "utf-8")
|
||||
assert [b["chainage_m"] for b in read_basins(tmp_path)] == [85.05, 173.09, 300.0]
|
||||
Reference in New Issue
Block a user