diff --git a/B08_Quantity/B08_Quantity_Hydro_Defaults.py b/B08_Quantity/B08_Quantity_Hydro_Defaults.py index d226def3f..8a8931ab9 100644 --- a/B08_Quantity/B08_Quantity_Hydro_Defaults.py +++ b/B08_Quantity/B08_Quantity_Hydro_Defaults.py @@ -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]: diff --git a/B08_Quantity/B08_Quantity_Hydro_Site.py b/B08_Quantity/B08_Quantity_Hydro_Site.py index 5ce27adf7..6a6f83234 100644 --- a/B08_Quantity/B08_Quantity_Hydro_Site.py +++ b/B08_Quantity/B08_Quantity_Hydro_Site.py @@ -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") diff --git a/B08_Quantity/B08_Quantity_Router_Hydro.py b/B08_Quantity/B08_Quantity_Router_Hydro.py index 1c442f995..d51f3294c 100644 --- a/B08_Quantity/B08_Quantity_Router_Hydro.py +++ b/B08_Quantity/B08_Quantity_Router_Hydro.py @@ -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, diff --git a/resources/tester/test_96_b08_hydro_linked.py b/resources/tester/test_96_b08_hydro_linked.py new file mode 100644 index 000000000..ffa10618d --- /dev/null +++ b/resources/tester/test_96_b08_hydro_linked.py @@ -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]