From fa6953ce6dc43586358b415f9a89dff1cad13596 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Wed, 5 Aug 2026 18:19:15 +0900 Subject: [PATCH] =?UTF-8?q?feat(B04/B05):=20=ED=99=95=EB=A5=A0=EA=B0=95?= =?UTF-8?q?=EC=9A=B0=EB=9F=89=20=EC=88=98=EA=B8=89=C2=B7=EB=B0=B0=EC=88=98?= =?UTF-8?q?=20=EC=9C=A0=ED=9A=A8=EC=A7=81=EA=B2=BD=20=ED=95=A9=EB=A6=AC?= =?UTF-8?q?=EC=8B=9D=20=EA=B3=84=EC=82=B0=20(=EA=B3=84=ED=9A=8D=EC=84=9C?= =?UTF-8?q?=20Phase=201=C2=B72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - common_util_wamis_rainfall.py 신설: map.wamis.go.kr 등우선도 urllib 수급, 전역 캐시(resources/wamis_contours), 거리비례 내삽, General형 IDF 적합 - /drainage/rainfall API + primary-region 백그라운드 강우량표 생성 훅 - estimate_pipe_diameter_mm TODO 구현: Kirpich 도달시간(하한5분) + 합리식(C=0.8, 2.0배) + Manning(경사10도, n=0.024, V<=3.0) + 통수단면 70% -> 배수 유효직경 - basins 응답·B05 유역 제원에 유효면적/유효직경·산출근거(tc/I/Qd) 표기 - config_system: 계산 계수·구조물 옵션(관종/직경/대피로) 기본값 등록 Co-Authored-By: Claude Fable 5 --- B04_wf1_Surface/B04_wf1_Surface_Api_Fetch.ts | 6 +- .../B04_wf1_Surface_Router_Basins.py | 9 +- .../B04_wf1_Surface_Router_Watershed.py | 108 +++++- .../B05_wf2_Route_UI_Drainage_Parts.ts | 11 +- common_util/common_util_drainage_detail.py | 107 +++++- common_util/common_util_wamis_rainfall.py | 324 ++++++++++++++++++ config/config_system.py | 43 +++ ui_template/ui_template_locale_b2.ts | 11 +- 8 files changed, 600 insertions(+), 19 deletions(-) create mode 100644 common_util/common_util_wamis_rainfall.py diff --git a/B04_wf1_Surface/B04_wf1_Surface_Api_Fetch.ts b/B04_wf1_Surface/B04_wf1_Surface_Api_Fetch.ts index 5ddf75a6..1703f021 100644 --- a/B04_wf1_Surface/B04_wf1_Surface_Api_Fetch.ts +++ b/B04_wf1_Surface/B04_wf1_Surface_Api_Fetch.ts @@ -451,8 +451,12 @@ export interface DetailBasin { area_m2: number; relief_m: number; flow_length_m: number; - /** 관경 수식 미확정 — null이면 "미정"으로 표기한다. */ + /** 배수 유효직경(합리식 산출, mm). 강우량표가 아직 없으면 null → "미정" 표기. */ pipe_diameter_mm: number | null; + /** 산출 근거 — 홍수도달시간(분), 설계강우강도(mm/hr), 설계유량(m³/s, 2.0배 반영). */ + tc_minutes?: number | null; + intensity_mm_hr?: number | null; + design_flow_m3s?: number | null; } export interface DetailBasinResponse { diff --git a/B04_wf1_Surface/B04_wf1_Surface_Router_Basins.py b/B04_wf1_Surface/B04_wf1_Surface_Router_Basins.py index 97e26438..63cf5f12 100644 --- a/B04_wf1_Surface/B04_wf1_Surface_Router_Basins.py +++ b/B04_wf1_Surface/B04_wf1_Surface_Router_Basins.py @@ -124,8 +124,12 @@ def _payload( "area_m2": round(basin.area_m2, 1), "relief_m": round(basin.relief_m, 2), "flow_length_m": round(basin.flow_length_m, 1), - # 관경 수식 미확정 — None이면 프론트가 "미정"으로 표기한다. + # 배수 유효직경(합리식 산출). 강우량표가 아직 없으면 None → "미정" 표기. "pipe_diameter_mm": basin.pipe_diameter_mm, + # 산출 근거 — 도달시간(분)·설계강우강도(mm/hr)·설계유량(m³/s, 2.0배 반영). + "tc_minutes": basin.tc_minutes, + "intensity_mm_hr": basin.intensity_mm_hr, + "design_flow_m3s": basin.design_flow_m3s, } for basin in detail.basins ], @@ -154,6 +158,9 @@ def _basin_features( "relief_m": round(basin.relief_m, 2), "flow_length_m": round(basin.flow_length_m, 1), "pipe_diameter_mm": basin.pipe_diameter_mm, + "tc_minutes": basin.tc_minutes, + "intensity_mm_hr": basin.intensity_mm_hr, + "design_flow_m3s": basin.design_flow_m3s, }, "geometry": {"type": "Polygon", "coordinates": [ring]}, } diff --git a/B04_wf1_Surface/B04_wf1_Surface_Router_Watershed.py b/B04_wf1_Surface/B04_wf1_Surface_Router_Watershed.py index 3bbd1ae1..5ac447c5 100644 --- a/B04_wf1_Surface/B04_wf1_Surface_Router_Watershed.py +++ b/B04_wf1_Surface/B04_wf1_Surface_Router_Watershed.py @@ -42,8 +42,18 @@ from common_util.common_util_route_geometry import ( read_planned_route_csv, ) from common_util.common_util_storage import resolve_stored_project_path +from common_util.common_util_wamis_rainfall import ( + build_rainfall_table, + ensure_contour_cache, +) from config.config_db import get_db_pool -from config.config_system import DRAINAGE_ARROW_SPACING_M, DRAINAGE_RESPONSE_FILENAME +from config.config_system import ( + DRAINAGE_ARROW_SPACING_M, + DRAINAGE_DESIGN_RETURN_PERIOD_YR, + DRAINAGE_RAINFALL_FILENAME, + DRAINAGE_RESPONSE_FILENAME, + WAMIS_CONTOUR_CACHE_DIR, +) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B04 Surface Watershed"]) @@ -218,6 +228,99 @@ def _save_response(stored_path: str, payload: dict[str, Any]) -> None: logger.warning("배수유역: 분석 응답을 저장하지 못했습니다 (%s).", path) +def _rainfall_path(stored_path: str) -> Path: + return drainage_dir(stored_path) / DRAINAGE_RAINFALL_FILENAME + + +def _route_center_lonlat(stored_path: str, fallback_epsg: int | None) -> tuple[float, float] | None: + """계획 노선 중간점(WGS84). 강우량 내삽 기준 좌표 — 유역 규모 대비 등우선 간격이 + 훨씬 넓어 노선 대표점 하나로 고정한다(설계기준.md 4절, 2026-08-05 협의).""" + route_file = find_planned_route_file(_route_input_dir(stored_path)) + if route_file is None: + return None + planned = read_planned_route_csv(route_file) + if planned is None or not planned.vertices: + return None + middle = planned.vertices[len(planned.vertices) // 2] + source_crs = f"EPSG:{planned.epsg or fallback_epsg or 5186}" + lon, lat = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True).transform( + middle.x, middle.y + ) + return lat, lon + + +def _build_rainfall_sync(lat: float, lon: float) -> dict[str, Any]: + """등우선 전역 캐시 확보(없는 파일만 다운로드) 후 지점 강우량표를 만든다.""" + cached, failures = ensure_contour_cache(WAMIS_CONTOUR_CACHE_DIR) + table = build_rainfall_table( + lat, + lon, + WAMIS_CONTOUR_CACHE_DIR, + design_return_period=DRAINAGE_DESIGN_RETURN_PERIOD_YR, + ) + table["contour_cache_files"] = cached + if failures: + table["failures"] = (table.get("failures") or []) + failures + return table + + +async def _ensure_rainfall_table(stored_path: str, fallback_epsg: int | None) -> None: + """rainfall_table.json이 없으면 백그라운드로 만든다. 실패는 비치명(로그만).""" + path = _rainfall_path(stored_path) + if path.exists(): + return + center = _route_center_lonlat(stored_path, fallback_epsg) + if center is None: + return + try: + table = await asyncio.to_thread(_build_rainfall_sync, *center) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8") + logger.info("확률강우량표 저장: %s (값 %d개)", path, len(table.get("values") or [])) + except Exception: # noqa: BLE001 — 외부망 차단 등. 배수유역 해석은 계속되어야 한다. + logger.warning("확률강우량표 생성 실패 (%s)", stored_path, exc_info=True) + + +@router.get("/{project_id}/drainage/rainfall", response_model=None) +async def get_drainage_rainfall( + project_id: UUID, refresh: bool = False +) -> dict[str, Any] | JSONResponse: + """프로젝트 지점의 확률강우량표(설계빈도 IDF 적합계수 포함)를 돌려준다. + + 저장분이 있으면 그대로 주고, 없으면 즉석 생성한다(최초 1회는 등우선 96개 + 다운로드로 수십 초 걸릴 수 있다). B05 세션 캐시가 이 응답을 물고 다닌다. + """ + pool = get_db_pool() + async with pool.acquire() as connection: + stored_path = await get_project_storage_relative_path(connection, project_id) + epsg = await get_surface_crs_epsg(connection, project_id, 0) + path = _rainfall_path(stored_path) + if not refresh and path.exists(): + try: + return {**json.loads(path.read_text(encoding="utf-8")), "from_cache": True} + except (OSError, json.JSONDecodeError): + logger.warning("강우량표 저장분을 읽지 못해 다시 만듭니다 (%s).", path) + center = _route_center_lonlat(stored_path, epsg) + if center is None: + return JSONResponse( + status_code=404, + content={"status": "error", "message": "계획 노선 파일이 업로드되지 않았습니다."}, + ) + try: + table = await asyncio.to_thread(_build_rainfall_sync, *center) + except Exception as exc: # noqa: BLE001 + return JSONResponse( + status_code=502, + content={ + "status": "error", + "message": f"확률강우량 자료를 받지 못했습니다: {exc}", + }, + ) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8") + return {**table, "from_cache": False} + + @router.get("/{project_id}/drainage/primary-region", response_model=None) async def get_primary_region( project_id: UUID, refresh: bool = False @@ -231,6 +334,9 @@ async def get_primary_region( pool = get_db_pool() async with pool.acquire() as connection: stored_path = await get_project_storage_relative_path(connection, project_id) + epsg = await get_surface_crs_epsg(connection, project_id, 0) + # 확률강우량표가 없으면 백그라운드로 만들어 둔다 — 유효직경 계산(B05)이 이 파일을 쓴다. + asyncio.create_task(_ensure_rainfall_table(stored_path, epsg)) if not refresh: saved = _load_saved_response(stored_path) if saved is not None: diff --git a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Parts.ts b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Parts.ts index d9d39ce6..ed990963 100644 --- a/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Parts.ts +++ b/B05_wf2_Route/B05_wf2_Route_UI_Drainage_Parts.ts @@ -129,7 +129,7 @@ export function renderBasinRows( badge.style.background = basinColor(basin.index); const metrics = document.createElement("span"); metrics.className = "b05-drainage__basin-metrics"; - // 관경은 수식 미확정이라 백엔드가 null을 주며, 확정 전까지 "미정"으로 표기한다. + // 유효직경은 강우량표(rainfall_table.json)가 생기기 전까지 null → "미정" 표기. const pipe = basin.pipe_diameter_mm === null ? L("B05_Drainage_Basin_Undecided") @@ -140,6 +140,15 @@ export function renderBasinRows( .replace("{flow}", String(Math.round(basin.flow_length_m))) .replace("{pipe}", pipe); row.title = L("B05_Drainage_Basin_Chainage").replace("{chainage}", basin.chainage_m.toFixed(1)); + // 산출 근거(도달시간·강우강도·설계유량)는 툴팁 둘째 줄로 붙인다 — 행이 길어지지 않게. + if (basin.tc_minutes != null && basin.design_flow_m3s != null) { + row.title += + "\n" + + L("B05_Drainage_Basin_Basis") + .replace("{tc}", String(basin.tc_minutes)) + .replace("{i}", String(basin.intensity_mm_hr ?? "-")) + .replace("{q}", String(basin.design_flow_m3s)); + } row.append(badge, metrics); row.addEventListener("click", () => onPick(basin.index)); container.append(row); diff --git a/common_util/common_util_drainage_detail.py b/common_util/common_util_drainage_detail.py index d02dc870..b67dc928 100644 --- a/common_util/common_util_drainage_detail.py +++ b/common_util/common_util_drainage_detail.py @@ -25,6 +25,7 @@ from __future__ import annotations import json import logging +import math from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -41,10 +42,20 @@ from common_util.common_util_route_geometry import ( interpolate_vertex, is_uphill_at, ) +from common_util.common_util_wamis_rainfall import idf_intensity from config.config_system import ( + DRAINAGE_DESIGN_FLOW_FACTOR, DRAINAGE_DITCH_SAMPLE_M, + DRAINAGE_FLOW_AREA_RATIO, + DRAINAGE_MANNING_N, DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M, + DRAINAGE_PIPE_SLOPE_DEG, + DRAINAGE_RAINFALL_FILENAME, + DRAINAGE_RUNOFF_COEFFICIENT, + DRAINAGE_TC_MIN_MINUTES, + DRAINAGE_VELOCITY_MAX_MS, + DRAINAGE_VELOCITY_MIN_MS, ) logger = logging.getLogger(__name__) @@ -89,6 +100,10 @@ class WatershedBasin: relief_m: float = 0.0 flow_length_m: float = 0.0 pipe_diameter_mm: float | None = None + # 유효직경 산출 근거(합리식) — 도달시간·설계강우강도·설계유량. 미산출이면 None. + tc_minutes: float | None = None + intensity_mm_hr: float | None = None + design_flow_m3s: float | None = None @dataclass @@ -172,7 +187,7 @@ def build_detail( if not pipes: return detail pipe_of_slot = assign_road_cells_to_pipes(vertices, pipes, routing.road_chainage) - detail.basins = assemble_basins(routing, pipes, pipe_of_slot) + detail.basins = assemble_basins(routing, pipes, pipe_of_slot, load_rainfall_idf(directory)) logger.info( "배수유역: 세부 설계 — 관 %d개(기본 %d + 보충 %d), 세부유역 %d개", len(pipes), @@ -476,6 +491,7 @@ def assemble_basins( routing: RoadRouting, pipes: list[StructureCandidate], pipe_of_slot: np.ndarray, + idf: dict[str, float] | None = None, ) -> list[WatershedBasin]: """셀이 도달한 도로 셀의 담당 관을 그대로 유역 번호로 삼아 세부유역을 만든다.""" spec = routing.spec @@ -498,6 +514,7 @@ def assemble_basins( area = count * cell_area relief = max(0.0, highest - outlet_z) flow_length = float(routing.path_length[member].max()) + sizing = size_pipe(area, relief, flow_length, idf) basins.append( WatershedBasin( index=len(basins) + 1, @@ -508,7 +525,10 @@ def assemble_basins( area_m2=area, relief_m=relief, flow_length_m=flow_length, - pipe_diameter_mm=estimate_pipe_diameter_mm(area, relief, flow_length), + pipe_diameter_mm=sizing["diameter_mm"] if sizing else None, + tc_minutes=sizing["tc_minutes"] if sizing else None, + intensity_mm_hr=sizing["intensity_mm_hr"] if sizing else None, + design_flow_m3s=sizing["design_flow_m3s"] if sizing else None, ) ) return basins @@ -524,18 +544,81 @@ def _outlet_elevation(routing: RoadRouting, pipe_order: int, pipe_of_slot: np.nd return float(finite.min()) if finite.size else 0.0 +def load_rainfall_idf(directory: Path) -> dict[str, float] | None: + """배수유역 폴더의 rainfall_table.json에서 설계빈도 IDF 적합계수를 읽는다. + + 파일은 B04 전처리(`/drainage/rainfall`)가 만든다. 없으면 None — 유효직경은 + "미정"으로 남고, 강우량표가 생기는 순간 다음 세부유역 계산부터 채워진다. + """ + path = Path(directory) / DRAINAGE_RAINFALL_FILENAME + if not path.exists(): + return None + try: + table = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + logger.warning("강우량표를 읽지 못했습니다 (%s) — 유효직경 미산출.", path) + return None + idf = table.get("idf_design") + if not isinstance(idf, dict) or not all(k in idf for k in ("a", "b", "n")): + return None + return idf + + +def size_pipe( + area_m2: float, + relief_m: float, + flow_length_m: float, + idf: dict[str, float] | None, +) -> dict[str, float] | None: + """유역 제원 + 설계빈도 IDF로 배수 유효직경(mm)과 산출 근거를 계산한다. + + 시행규칙 별표2 (가) 1호 방법: 100년빈도 확률강우량 + 홍수도달시간 → 합리식. + 전 과정·계수 근거: docs/raw/guidelines/2026-08-05_임도_배수설계_규정_조사.md 4·6절. + + ① 도달시간 tc = 0.0663·L^0.77·(H/L)^-0.385 [Kirpich, hr] — 하한 5분 + ② 강우강도 I = General형 적합식 I(tc) [mm/hr] + ③ 합리식 Q = (1/3.6)·C·I·A[km²], 설계유량 = 2.0·Q (별표2) + ④ 유속 V = Manning(경사 10도 고정, n=0.024) → 0.8~3.0m/s 클램프 + ⑤ 통수단면 70%만 유효 → D = √(4·(Q설계/V)/(0.7π)) + """ + if idf is None or area_m2 <= 0: + return None + length_km = max(flow_length_m, 1.0) / 1000.0 + slope = max(relief_m, 0.1) / max(flow_length_m, 1.0) + tc_hr = 0.0663 * (length_km**0.77) * (slope**-0.385) + tc_min = max(DRAINAGE_TC_MIN_MINUTES, tc_hr * 60.0) + intensity = idf_intensity(idf, tc_min) + flow = DRAINAGE_RUNOFF_COEFFICIENT * intensity * (area_m2 / 1e6) / 3.6 + design_flow = DRAINAGE_DESIGN_FLOW_FACTOR * flow + if design_flow <= 0: + return None + pipe_slope = math.tan(math.radians(DRAINAGE_PIPE_SLOPE_DEG)) + diameter = 0.5 + for _ in range(20): + velocity = (1.0 / DRAINAGE_MANNING_N) * (diameter / 4.0) ** (2.0 / 3.0) * math.sqrt( + pipe_slope + ) + velocity = min(DRAINAGE_VELOCITY_MAX_MS, max(DRAINAGE_VELOCITY_MIN_MS, velocity)) + required_area = design_flow / velocity + updated = math.sqrt(4.0 * required_area / (DRAINAGE_FLOW_AREA_RATIO * math.pi)) + if abs(updated - diameter) < 1e-4: + diameter = updated + break + diameter = updated + return { + "tc_minutes": round(tc_min, 1), + "intensity_mm_hr": round(intensity, 1), + "design_flow_m3s": round(design_flow, 4), + "diameter_mm": round(diameter * 1000.0, 1), + } + + def estimate_pipe_diameter_mm( area_m2: float, relief_m: float, flow_length_m: float, - rainfall_mm_per_hour: float | None = None, + idf: dict[str, float] | None = None, ) -> float | None: - """유역 제원으로 배수 파이프 관경(mm)을 산정한다. - - 100년 강우빈도와 유역 경사면을 곱해 유출량을 구하고, 그 유량으로 관경을 정하는 것이 - 목적이다. **수식은 아직 확정되지 않았다** — 사용자가 로직을 제공하면 여기를 채운다. - 그때까지는 None을 돌려 호출부가 "미정"으로 표기하게 한다. - """ - # TODO(사용자 로직 대기): 100년 강우강도 × 유역면적 × 유출계수 → 유량 Q → 관경 D 산정. - _ = (area_m2, relief_m, flow_length_m, rainfall_mm_per_hour) - return None + """유역 제원으로 배수 유효직경(mm)만 돌려주는 축약형 — 상세는 size_pipe().""" + sizing = size_pipe(area_m2, relief_m, flow_length_m, idf) + return sizing["diameter_mm"] if sizing else None diff --git a/common_util/common_util_wamis_rainfall.py b/common_util/common_util_wamis_rainfall.py new file mode 100644 index 00000000..7831cdca --- /dev/null +++ b/common_util/common_util_wamis_rainfall.py @@ -0,0 +1,324 @@ +"""map.wamis.go.kr 확률강우량 등우선도 수급·내삽 (표준 라이브러리만 사용). + +「전국 하천수계 홍수량 및 확률강우량 정보」가 공개하는 등우선 GeoJSON에서 +지점의 지속시간×재현기간 확률강우량표를 만든다. 인증 불필요. + + https://map.wamis.go.kr/data/gis/contour/{재현기간3자리}yr_{지속시간}.json + +파일 안 등우선의 ``ELEV`` 값이 그 선의 확률강우량(mm), 좌표계는 EPSG:3857. +지점값은 값이 다른 인접 등우선 두 개까지의 거리비례 내부삽입으로 얻는다. + +- 원본 96개(8빈도×12지속시간)는 전국 공통이라 전역 폴더에 1회 캐시한다. +- 1시간 미만 자료는 사이트에 없다(2026-08 실측). 5분 강우강도는 100년빈도 + 12점을 General형 강우강도식으로 적합해 얻는다(설계 근거: + docs/raw/guidelines/2026-08-05_임도_배수설계_규정_조사.md). + +검증: 2026-08-05 실서버 실측 — 문경 100년 1시간 142.5mm / 24시간 387.5mm. +""" + +from __future__ import annotations + +import json +import logging +import math +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any, Callable, Iterable, Sequence + +logger = logging.getLogger(__name__) + +BASE_URL = "https://map.wamis.go.kr" +CONTOUR_PATH = "/data/gis/contour" + +#: 사이트가 제공하는 재현기간(년). 2026-08 실측 8종. +RETURN_PERIODS: tuple[int, ...] = (2, 10, 30, 50, 80, 100, 200, 500) + +#: 사이트가 제공하는 지속시간 (파일명 표기, 분). 2026-08 실측 12종 — 1시간 미만 없음. +DURATIONS: tuple[tuple[str, int], ...] = ( + ("01hr", 60), + ("02hr", 120), + ("03hr", 180), + ("04hr", 240), + ("05hr", 300), + ("06hr", 360), + ("08hr", 480), + ("09hr", 540), + ("10hr", 600), + ("12hr", 720), + ("18hr", 1080), + ("24hr", 1440), +) + +REQUEST_TIMEOUT = 120.0 +EARTH_RADIUS_M = 6378137.0 + +_HEADERS = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", + "Accept": "application/json, text/plain, */*", +} + + +# -------------------------------------------------------------------------- +# 좌표·주소 +# -------------------------------------------------------------------------- +def to_web_mercator(lat: float, lon: float) -> tuple[float, float]: + """위경도(WGS84) → 웹지도 좌표(EPSG:3857) 미터.""" + if not -90.0 < lat < 90.0: + raise ValueError(f"위도 범위를 벗어났습니다: {lat}") + if not -180.0 <= lon <= 180.0: + raise ValueError(f"경도 범위를 벗어났습니다: {lon}") + x = EARTH_RADIUS_M * math.radians(lon) + y = EARTH_RADIUS_M * math.log(math.tan(math.pi / 4.0 + math.radians(lat) / 2.0)) + return x, y + + +def contour_url(return_period: int, duration_label: str, base_url: str = BASE_URL) -> str: + """``100yr_01hr.json`` 형태의 주소. 재현기간은 3자리 0채움.""" + return f"{base_url.rstrip('/')}{CONTOUR_PATH}/{return_period:03d}yr_{duration_label}.json" + + +def cache_filename(return_period: int, duration_label: str) -> str: + return f"{return_period:03d}yr_{duration_label}.json" + + +# -------------------------------------------------------------------------- +# 등우선 읽기·내삽 +# -------------------------------------------------------------------------- +def parse_contours(payload: dict) -> list[tuple[float, list[tuple[float, float]]]]: + """GeoJSON에서 ``(강우량mm, 좌표열)`` 목록을 뽑는다.""" + if not isinstance(payload, dict) or "features" not in payload: + raise RuntimeError("등우선 자료 형식이 아닙니다 (features 없음)") + lines: list[tuple[float, list[tuple[float, float]]]] = [] + for feature in payload.get("features") or []: + if not isinstance(feature, dict): + continue + props = feature.get("properties") or {} + elev = props.get("ELEV", props.get("elev")) + try: + level = float(elev) + except (TypeError, ValueError): + continue + geometry = feature.get("geometry") or {} + for coords in _iter_linestrings(geometry): + points = [ + (float(p[0]), float(p[1])) + for p in coords + if isinstance(p, (list, tuple)) and len(p) >= 2 + ] + if len(points) >= 2: + lines.append((level, points)) + if not lines: + raise RuntimeError("등우선을 하나도 읽지 못했습니다 (ELEV 값 확인 필요)") + return lines + + +def _iter_linestrings(geometry: dict) -> Iterable[list]: + kind = geometry.get("type") + coords = geometry.get("coordinates") or [] + if kind == "LineString": + yield coords + elif kind == "MultiLineString": + yield from coords + elif kind == "Polygon": + yield from coords + elif kind == "MultiPolygon": + for polygon in coords: + yield from polygon + + +def value_at( + lines: Sequence[tuple[float, Sequence[tuple[float, float]]]], x: float, y: float +) -> float: + """등우선에서 지점 ``(x, y)``의 확률강우량(mm) — 거리비례 내부삽입.""" + if not lines: + raise RuntimeError("등우선이 비어 있습니다") + distances = sorted((_distance_to_polyline(x, y, points), elev) for elev, points in lines) + d1, e1 = distances[0] + if d1 == 0.0: + return e1 + for d2, e2 in distances[1:]: + if e2 != e1: + total = d1 + d2 + return e1 if total == 0 else e1 + (e2 - e1) * (d1 / total) + return e1 + + +def _distance_to_polyline(x: float, y: float, points: Sequence[tuple[float, float]]) -> float: + return min(_distance_to_segment(x, y, p0, p1) for p0, p1 in zip(points, points[1:])) + + +def _distance_to_segment(x, y, p0, p1) -> float: + x0, y0 = p0 + x1, y1 = p1 + dx, dy = x1 - x0, y1 - y0 + if dx == 0.0 and dy == 0.0: + return math.hypot(x - x0, y - y0) + t = ((x - x0) * dx + (y - y0) * dy) / (dx * dx + dy * dy) + t = max(0.0, min(1.0, t)) + return math.hypot(x - (x0 + t * dx), y - (y0 + t * dy)) + + +# -------------------------------------------------------------------------- +# 내려받기·전역 캐시 +# -------------------------------------------------------------------------- +def _download(url: str, timeout: float = REQUEST_TIMEOUT) -> bytes: + request = urllib.request.Request(url, headers=_HEADERS) + with urllib.request.urlopen(request, timeout=timeout) as response: + return response.read() + + +def ensure_contour_cache( + cache_dir: Path, + return_periods: Sequence[int] = RETURN_PERIODS, + durations: Sequence[tuple[str, int]] = DURATIONS, + base_url: str = BASE_URL, + progress: Callable[[str], None] | None = None, +) -> tuple[int, list[str]]: + """전역 캐시 폴더에 없는 등우선 원본만 내려받는다. + + Returns (캐시 확보 파일 수, 실패 메시지 목록). 실패는 비치명 — 있는 것만으로 진행. + """ + cache_dir = Path(cache_dir) + cache_dir.mkdir(parents=True, exist_ok=True) + ok, failures = 0, [] + for period in return_periods: + for label, _minutes in durations: + path = cache_dir / cache_filename(period, label) + if path.exists() and path.stat().st_size > 0: + ok += 1 + continue + url = contour_url(period, label, base_url) + if progress: + progress(f"등우선 내려받기 {period}년/{label}") + try: + raw = _download(url) + json.loads(raw) # JSON 검증 후에만 저장 + tmp = path.with_suffix(".part") + tmp.write_bytes(raw) + tmp.replace(path) + ok += 1 + except (urllib.error.URLError, OSError, ValueError) as exc: + failures.append(f"{period}년/{label}: {exc}") + return ok, failures + + +def _load_cached_lines( + cache_dir: Path, period: int, label: str +) -> list[tuple[float, list[tuple[float, float]]]] | None: + path = Path(cache_dir) / cache_filename(period, label) + if not path.exists(): + return None + try: + return parse_contours(json.loads(path.read_text(encoding="utf-8"))) + except (OSError, ValueError, RuntimeError) as exc: + logger.warning("등우선 캐시 파일을 읽지 못했습니다 (%s): %s", path, exc) + return None + + +# -------------------------------------------------------------------------- +# General형 강우강도식 적합 — I(t) = a / (t^n + b) [t: 분, I: mm/hr] +# -------------------------------------------------------------------------- +def fit_general_idf(points: Sequence[tuple[float, float]]) -> dict[str, float] | None: + """``(지속시간 분, 강우강도 mm/hr)`` 점들로 General형 계수를 적합한다. + + scipy가 있으면 curve_fit, 없거나 실패하면 b·n 격자탐색 + a 최소제곱 폴백. + 1시간 미만(도달시간 5분~) 외삽의 근거식이므로 적합 잔차(rmse)를 함께 기록한다. + """ + pts = [(float(t), float(i)) for t, i in points if t > 0 and i > 0] + if len(pts) < 4: + return None + t = [p[0] for p in pts] + i = [p[1] for p in pts] + + def rmse(a: float, b: float, n: float) -> float: + return math.sqrt(sum((a / (tk**n + b) - ik) ** 2 for tk, ik in zip(t, i)) / len(t)) + + best: tuple[float, float, float, float] | None = None + try: + from scipy.optimize import curve_fit # 프로젝트 GIS 스택에 이미 포함 + + popt, _ = curve_fit( + lambda tt, a, b, n: a / (tt**n + b), + t, + i, + p0=(3000.0, 15.0, 0.7), + bounds=((1.0, 0.0, 0.05), (1e6, 1e4, 1.5)), + maxfev=20000, + ) + a, b, n = (float(v) for v in popt) + best = (rmse(a, b, n), a, b, n) + except Exception: # noqa: BLE001 — 폴백 격자탐색으로 계속 + pass + if best is None: + for n in (x / 100.0 for x in range(30, 121, 5)): + for b in (0.0, 1.0, 2.0, 5.0, 10.0, 20.0, 40.0, 80.0): + denom = [tk**n + b for tk in t] + # 최소제곱: min Σ(a/d − i)² → a = Σ(i/d) / Σ(1/d²) + a = sum(ik / dk for ik, dk in zip(i, denom)) / sum(1.0 / (dk * dk) for dk in denom) + err = rmse(a, b, n) + if best is None or err < best[0]: + best = (err, a, b, n) + if best is None: + return None + err, a, b, n = best + return {"a": round(a, 4), "b": round(b, 4), "n": round(n, 5), "rmse_mm_hr": round(err, 3)} + + +def idf_intensity(idf: dict[str, float], duration_min: float) -> float: + """General형 계수로 지속시간(분)의 강우강도(mm/hr)를 계산한다.""" + return float(idf["a"]) / (duration_min ** float(idf["n"]) + float(idf["b"])) + + +# -------------------------------------------------------------------------- +# 프로젝트 강우량표 생성 +# -------------------------------------------------------------------------- +def build_rainfall_table( + lat: float, + lon: float, + cache_dir: Path, + design_return_period: int = 100, + return_periods: Sequence[int] = RETURN_PERIODS, + durations: Sequence[tuple[str, int]] = DURATIONS, +) -> dict[str, Any]: + """전역 캐시된 등우선에서 지점 강우량표 + 설계빈도 IDF 적합계수를 만든다.""" + x, y = to_web_mercator(lat, lon) + values: list[dict[str, float]] = [] + failures: list[str] = [] + design_points: list[tuple[float, float]] = [] + for period in return_periods: + for label, minutes in durations: + lines = _load_cached_lines(cache_dir, period, label) + if lines is None: + failures.append(f"{period}년/{label}: 캐시 없음") + continue + try: + depth = value_at(lines, x, y) + except RuntimeError as exc: + failures.append(f"{period}년/{label}: {exc}") + continue + values.append( + {"return_period_yr": period, "duration_min": minutes, "depth_mm": round(depth, 2)} + ) + if period == design_return_period: + design_points.append((minutes, depth / (minutes / 60.0))) + idf = fit_general_idf(design_points) if design_points else None + return { + "source": f"{BASE_URL} 확률강우량 등우선도 (거리비례 내부삽입)", + "lat": lat, + "lon": lon, + "design_return_period_yr": design_return_period, + "values": values, + # 설계빈도 IDF: I(t)=a/(t^n+b), t분 → mm/hr. 도달시간(≥5분) 강우강도 산출용. + "idf_design": idf, + "failures": failures, + } + + +def design_depth_mm(table: dict[str, Any], duration_min: int) -> float | None: + """저장된 표에서 설계빈도·지속시간 강우량(mm)을 찾는다.""" + period = table.get("design_return_period_yr") + for row in table.get("values") or []: + if row.get("return_period_yr") == period and row.get("duration_min") == duration_min: + return float(row["depth_mm"]) + return None diff --git a/config/config_system.py b/config/config_system.py index c1a1be18..c7691314 100644 --- a/config/config_system.py +++ b/config/config_system.py @@ -316,6 +316,49 @@ DRAINAGE_ARROW_MIN_AGREEMENT = float(os.getenv("DRAINAGE_ARROW_MIN_AGREEMENT", " # 단계별 검증 산출물은 같은 폴더에 `{번호}_{단계}.geojson` + `manifest.json`으로 쌓인다. # 파일명 규칙은 B05_wf2_Route_Engine_Watershed_Export.STAGES가 유일한 정의처다. +# ── 확률강우량 (map.wamis.go.kr 등우선도) ── +# 전국 공통 등우선 GeoJSON 원본을 이 폴더에 1회 내려받아 영구 캐시한다(약 20~30MB). +# 근거·검증: docs/raw/guidelines/2026-08-05_임도_배수설계_규정_조사.md 5절. +WAMIS_CONTOUR_CACHE_DIR = PROJECT_ROOT / "resources" / "wamis_contours" +# 프로젝트별 내삽 결과 파일명. drainage/ 아래에 놓인다. +DRAINAGE_RAINFALL_FILENAME = "rainfall_table.json" + +# ── 배수 유효직경 계산 계수 (2026-08-05 사용자 확정, 위 규정 조사 문서 6절) ── +# 합리식 유출계수 C. 산악지 준용 0.8 — 필요 시 사용자가 이 값을 고친다. +DRAINAGE_RUNOFF_COEFFICIENT = float(os.getenv("DRAINAGE_RUNOFF_COEFFICIENT", "0.8")) +# 설계유량 배수. 시행규칙 별표2 (가): 최대홍수유출량의 2.0배 이상. +DRAINAGE_DESIGN_FLOW_FACTOR = float(os.getenv("DRAINAGE_DESIGN_FLOW_FACTOR", "2.0")) +# 설계빈도(년). 별표2 (가) 1호: 100년빈도 확률강우량. +DRAINAGE_DESIGN_RETURN_PERIOD_YR = int(os.getenv("DRAINAGE_DESIGN_RETURN_PERIOD_YR", "100")) +# 홍수도달시간 하한(분). 국도건설공사 설계실무요령·도로배수지침 "강우지속기간 5분 원칙". +DRAINAGE_TC_MIN_MINUTES = float(os.getenv("DRAINAGE_TC_MIN_MINUTES", "5.0")) +# 관 경사(도). 시공 중 수시 변경되므로 지형 산출 대신 실무 대푯값 10도로 고정(사용자 제시). +DRAINAGE_PIPE_SLOPE_DEG = float(os.getenv("DRAINAGE_PIPE_SLOPE_DEG", "10.0")) +# Manning 조도계수(파형강관). +DRAINAGE_MANNING_N = float(os.getenv("DRAINAGE_MANNING_N", "0.024")) +# 관내 유속 허용범위(m/s). 도로배수지침 0.8~3.0 — 상한 클램프가 안전측(관이 커진다). +DRAINAGE_VELOCITY_MIN_MS = float(os.getenv("DRAINAGE_VELOCITY_MIN_MS", "0.8")) +DRAINAGE_VELOCITY_MAX_MS = float(os.getenv("DRAINAGE_VELOCITY_MAX_MS", "3.0")) +# 원형관 통수단면 비율. 도로설계요령: 관 단면의 70%만 통수 고려. +DRAINAGE_FLOW_AREA_RATIO = float(os.getenv("DRAINAGE_FLOW_AREA_RATIO", "0.7")) + +# ── B05 구조물 옵션 (드롭다운 목록·기본값, 2026-08-05 사용자 확정) ── +# 프론트는 이 값을 /api 설정 응답 또는 빌드타임 복사로 받아 쓴다. 수정은 여기서만 한다. +STRUCTURE_TYPES = ("배관", "기성막이", "대피로", "기타") +STRUCTURE_DEFAULT_TYPE = "배관" +PIPE_DIAMETERS_MM = { + "이중벽관": (150, 200, 250, 300, 400, 450, 500, 600, 700, 800, 900, 1000, 1200, 1500), + "삼중벽관": (200, 250, 300, 400, 450, 500, 600, 700, 800, 900, 1000, 1200, 1500), + "파형강관": (150, 200, 250, 300, 350, 400, 450, 500, 600, 700, 800, 900, 1000, + 1200, 1350, 1500, 1650, 1800, 2000), +} +PIPE_DEFAULT_TYPE = "파형강관" +# 자동 지정 기본 관경. 별표2 (나) 예외 하한 800mm(사용자 결정 — 계산값이 더 크면 바로 위 규격). +PIPE_DEFAULT_DIAMETER_MM = int(os.getenv("PIPE_DEFAULT_DIAMETER_MM", "800")) +ESCAPE_ROUTE_WIDTHS_M = (1.5, 2.0, 2.5, 3.0) +ESCAPE_ROUTE_DEFAULT_WIDTH_M = 2.0 +STRUCTURE_ETC_DEFAULT_NAME = "기타 구조물" + # ───────────────────────────────────────────────────────────────────────── # 5-4. 종횡단 생성 파라미터 (B06 WF3) diff --git a/ui_template/ui_template_locale_b2.ts b/ui_template/ui_template_locale_b2.ts index 89d640f8..0b6b8f4a 100644 --- a/ui_template/ui_template_locale_b2.ts +++ b/ui_template/ui_template_locale_b2.ts @@ -77,10 +77,15 @@ export const ui_locales_b2 = { ], B05_Drainage_Status_LoadFailed: ["배경도를 불러오지 못했습니다.", "Failed to load the basemap."], B05_Drainage_Basin_Undecided: ["미정", "TBD"], - /* {area}=면적, {relief}=표고차, {flow}=유하장, {pipe}=관경 */ + /* {area}=배수 유효면적, {relief}=표고차, {flow}=유하장, {pipe}=배수 유효직경 */ B05_Drainage_Basin_Metrics: [ - "면적 {area} · 표고 {relief}m · 유하 {flow}m · 관경 {pipe}", - "Area {area} · Relief {relief}m · Flow {flow}m · Pipe {pipe}", + "유효면적 {area} · 표고 {relief}m · 유하 {flow}m · 유효직경 {pipe}", + "Eff. area {area} · Relief {relief}m · Flow {flow}m · Eff. dia {pipe}", + ], + /* {tc}=도달시간(분), {i}=설계강우강도(mm/hr), {q}=설계유량(m³/s) — 유효직경 산출 근거 */ + B05_Drainage_Basin_Basis: [ + "도달시간 {tc}분 · 강우강도 {i}mm/hr · 설계유량 {q}m³/s (100년빈도·2.0배)", + "Tc {tc}min · I {i}mm/hr · Qd {q}m³/s (100yr, ×2.0)", ], /* {chainage}=측점 누가거리(m) */ B05_Drainage_Basin_Chainage: ["측점 누가거리 {chainage}m", "Station chainage {chainage}m"],