diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py index 298df74b3..4d8981143 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py +++ b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py @@ -23,6 +23,7 @@ from __future__ import annotations import logging import math import time +from collections.abc import Callable from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -31,7 +32,10 @@ import numpy as np from shapely.geometry import LineString from B04_PreProcess.B04_PreProcess_Engine_Watershed_Descent import ContourDescent -from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import expand_by_red_boundary +from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import ( + AnalysisStopped, + expand_by_red_boundary, +) from B04_PreProcess.B04_PreProcess_Engine_Watershed_Flow import ( FlowClassification, RoadRaster, @@ -154,6 +158,7 @@ def preview_stages( contour_features: list[dict[str, Any]], stream_features: list[dict[str, Any]], descent_store: tuple[Path, str] | None = None, + should_stop: Callable[[], bool] | None = None, ) -> StagePreview | None: """지금까지 구현·검증된 단계를 순서대로 돌려 결과를 모은다. @@ -176,6 +181,8 @@ def preview_stages( marks.append(("1차 영역(resolve_primary_region)", time.perf_counter())) if region is None: return None + if should_stop is not None and should_stop(): + raise AnalysisStopped started = time.perf_counter() floor = route_elevation_floor([vertex.z for vertex in vertices]) @@ -187,6 +194,7 @@ def preview_stages( region.split.upstream, floor, descent_store=descent_store, + should_stop=should_stop, ) if expansion is None: logger.warning("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.") diff --git a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py index d1e09fa39..ebcc2c8fc 100644 --- a/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py +++ b/B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py @@ -21,6 +21,7 @@ from __future__ import annotations import logging import time +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from typing import Any @@ -58,6 +59,10 @@ logger = logging.getLogger(__name__) DESCENT_SPARE_M = 400.0 +class AnalysisStopped(Exception): + """새 차례가 와서 계산을 그만둠(「지금 노선으로 셈」 · 347-2) — 단계 사이에서만 확인.""" + + @dataclass class GridAnalysis: """한 회차 분석 결과 — 격자·해석 영역·방향·색까지 한 묶음.""" @@ -142,6 +147,7 @@ def expand_by_red_boundary( band_m: float = DRAINAGE_RED_EXPAND_BAND_M, max_rounds: int = DRAINAGE_RED_EXPAND_MAX_ROUNDS, descent_store: tuple[Path, str] | None = None, + should_stop: Callable[[], bool] | None = None, ) -> RedExpansion | None: """최외곽 적색 셀 주변으로 넓히며, 새로 추가한 셀에 적색이 없을 때까지 반복한다.""" _expand_started = time.perf_counter() @@ -194,6 +200,8 @@ def expand_by_red_boundary( rounds = 0 closed = False for attempt in range(max_rounds): + if should_stop is not None and should_stop(): + raise AnalysisStopped _round_started = time.perf_counter() current = analysis.spec reaches = analysis.flow.reaches_road.reshape(current.n_rows, current.n_cols) diff --git a/B04_PreProcess/B04_PreProcess_Router_Basins.py b/B04_PreProcess/B04_PreProcess_Router_Basins.py index 867b3dd55..2b27942d4 100644 --- a/B04_PreProcess/B04_PreProcess_Router_Basins.py +++ b/B04_PreProcess/B04_PreProcess_Router_Basins.py @@ -73,7 +73,11 @@ STALE_MESSAGE = ( def _build( - stored_path: str, context: DrainageContext, points: list[PipePoint] | None + stored_path: str, + context: DrainageContext, + points: list[PipePoint] | None, + routing: Any = None, + upstream_lonlat: list | None = None, ) -> tuple[DrainageDetail | None, list[PipePoint]]: """세부유역을 계산하고, 그 결과에 쓰인 관 목록을 함께 돌려준다. @@ -83,7 +87,7 @@ def _build( """ directory = drainage_dir(stored_path) chainages = [point.chainage_m for point in points] if points is not None else None - detail = build_detail(directory, context.vertices, chainages) + detail = build_detail(directory, context.vertices, chainages, routing, upstream_lonlat) if detail is None: return None, [] if points is not None: @@ -377,6 +381,11 @@ def _raw_payload( for basin in detail.basins ], "pipe_count": len(points), + # 관 자리 후보(347-3) — 제안만 · [적용] 때 화면이 관 초안에 더함. + "pipe_candidates": [ + {"chainage_m": round(pipe.chainage_m, 2), "lonlat": list(to_lonlat(pipe.x, pipe.y))} + for pipe in detail.pipe_candidates + ], } @@ -479,6 +488,13 @@ async def _resolve( return context, detail, resolved, saved, False +async def _analysis_stale(project_id: UUID) -> bool: + context, _ = await load_drainage_context(project_id) + if context is None: + return False + return analysis_route_signature(context.stored_path) != route_signature(context.vertices) + + @router.get("/{project_id}/drainage/pipe-points", response_model=None) async def get_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse: """저장된 관 지점과 그 세부유역을 돌려준다. 저장분이 없으면 자동 생성한 목록.""" @@ -529,8 +545,17 @@ async def put_pipe_points( project_id: UUID, payload: dict[str, Any] | None = None, session: dict[str, Any] | None = Depends(get_optional_session), # 가져온이(142-2) + recompute: bool = False, ) -> dict[str, Any] | JSONResponse: """관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점).""" + # [확정](recompute) = 낡았으면 지금 노선으로 배수유역부터 다시 셈(정본 · 작업본에 씀 · 347-2) · + # [저장] = 낡음 표시 달고 관만 저장. + if recompute and await _analysis_stale(project_id): + from B04_PreProcess.B04_PreProcess_Router_Watershed import get_primary_region + + region = await get_primary_region(project_id, refresh=True) + if isinstance(region, JSONResponse): + return region resolved = await _resolve(project_id, payload, use_stored=True, fresh=True) if isinstance(resolved, JSONResponse): return resolved diff --git a/B04_PreProcess/B04_PreProcess_Router_Watershed.py b/B04_PreProcess/B04_PreProcess_Router_Watershed.py index c1d55c4a4..b36c07fee 100644 --- a/B04_PreProcess/B04_PreProcess_Router_Watershed.py +++ b/B04_PreProcess/B04_PreProcess_Router_Watershed.py @@ -569,3 +569,7 @@ async def get_primary_region( _log_steps("배수유역 분석 내부", marks) _save_response(prepared["stored_path"], payload) return payload + + +# 「지금 노선으로 셈」(347-2) — 같은 router 에 붙임(그 파일이 이 router 를 읽음). +from B04_PreProcess import B04_PreProcess_Router_Watershed_Live # noqa: E402, F401 diff --git a/B04_PreProcess/B04_PreProcess_Router_Watershed_Live.py b/B04_PreProcess/B04_PreProcess_Router_Watershed_Live.py new file mode 100644 index 000000000..471da8f43 --- /dev/null +++ b/B04_PreProcess/B04_PreProcess_Router_Watershed_Live.py @@ -0,0 +1,152 @@ +"""「지금 노선으로 셈」(347-2) — 노선 초안 + 관 자리로 배수유역 · 세부 유역을 한 요청에 셈. + +서버 단독(③) · **작업본에 안 씀** — `00_watershed_response.json` · 격자 배열 · 관 지점은 그대로이고 +답으로만 돌려준다(파생 저장본 = 흐름 방향 · 도엽 좌표만 · 347-1). 세부 유역 나눔까지 같은 요청에서 +메모리 격자로 해 격자를 화면에 넘기지 않는다. + +프로젝트마다 「마지막 차례」를 적어 두고 단계 사이(1차 영역 뒤 · 확장 회차마다 · 세부 유역 앞)에서 +차례가 바뀌었으면 그만둔다(`superseded`). 화면은 늦은 답을 차례 · 노선 지문으로 한 번 더 거른다. +라우터는 `B04_PreProcess_Router_Watershed.router` 에 붙는다(그 파일 끝에서 읽음). +""" + +from __future__ import annotations + +import asyncio +import math +from dataclasses import replace +from typing import Any +from uuid import UUID + +from fastapi.responses import JSONResponse + +from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages +from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import AnalysisStopped +from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir +from B04_PreProcess.B04_PreProcess_Router_Basins import _build, _payload +from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare, router +from common_util.common_util_drainage_context import _open_sampler, load_drainage_context +from common_util.common_util_drainage_detail import RoadRouting +from common_util.common_util_drainage_pipes import parse_pipe_points +from common_util.common_util_route_geometry import RouteVertex, build_route_vertices +from common_util.common_util_route_polyline import build_planned_polyline +from common_util.common_util_route_profile import resolve_route_profile +from config.config_system import DRAINAGE_ARROW_SPACING_M + +# 프로젝트 → 마지막으로 받은 차례 번호. ponytail: 서버 프로세스 하나 전제(여럿이면 DB · 파일로). +_LATEST_GENERATION: dict[str, int] = {} +SUPERSEDED = {"status": "superseded"} + + +def draft_vertices(draft: dict[str, Any], min_radius_m: float) -> list[RouteVertex]: + """노선 초안(RouteDraft) → 그린 노선 정점(z 없음) — 노선 [확인]과 같은 폴리라인 규칙. + + ponytail: 붙든 곡선 길이(curveLock · curveArc) · 측점 스냅은 안 봄 — 곡선 중점이 조금 다를 뿐 + 배수유역 결과엔 셀 몇 개 차이. 필요하면 노선 [확인] 길과 같이 넘김. + """ + nodes = [(float(x), float(y)) for x, y in draft["nodes"]] + flags = [bool(flag) for flag in draft.get("curveOn") or [True] * len(nodes)] + radii = [None if r is None else float(r) for r in draft.get("curveRadius") or []] + polyline = build_planned_polyline( + nodes, min_radius_m=min_radius_m, simplify=False, curve_flags=flags, radii=radii or None + ) + return build_route_vertices([{"x": x, "y": y} for x, y in polyline.vertices]) + + +def live_routing(preview: Any, route_xy: list[tuple[float, float]], to_lonlat: Any) -> RoadRouting: + """분석 결과(메모리) → 세부 유역 나눔 입력 — 파일 길(`_write_road_routing` → 읽기)과 같은 값.""" + routing, road = preview.routing, preview.road + return RoadRouting( + spec=preview.spec, + road_slot=routing.road_slot, + path_length=routing.path_length, + elevation=preview.terrain.elevation.reshape(-1), + road_cell_index=road.cell_index, + road_chainage=road.chainage, + strength=routing.strength, + route_lonlat=[list(to_lonlat(x, y)) for x, y in route_xy], + basin_lonlat=[list(to_lonlat(x, y)) for x, y in preview.basin_boundary_xy], + base_pipes=list(preview.pipes), + flow_arrows=[ + [round(x, 2), round(y, 2), round(math.degrees(angle), 1), reaches, cells] + for x, y, angle, reaches, cells in preview.flow_arrows + ], + arrow_spacing_m=DRAINAGE_ARROW_SPACING_M, + ) + + +@router.post("/{project_id}/drainage/live", response_model=None) +async def post_drainage_live( + project_id: UUID, payload: dict[str, Any] | None = None +) -> dict[str, Any] | JSONResponse: + """몸통 = {generation, route: RouteDraft | null(저장 노선), points: 관 목록 | null(자동)}.""" + body = payload or {} + generation = int(body.get("generation") or 0) + key = str(project_id) + _LATEST_GENERATION[key] = generation + + def superseded() -> bool: + return _LATEST_GENERATION.get(key) != generation + + try: + raw = body.get("points") + points = None if raw is None else parse_pipe_points(raw) + except ValueError as error: + return JSONResponse(status_code=400, content={"status": "error", "message": str(error)}) + prepared = await _prepare(project_id) + if isinstance(prepared, JSONResponse): + return prepared + context, reason = await load_drainage_context(project_id) + if context is None: + return JSONResponse(status_code=404, content={"status": "error", "message": reason}) + + vertices = context.vertices + draft = body.get("route") + if draft and len(draft.get("nodes") or []) >= 2: + from B05_Profile.B05_Profile_Router_Replan import _plan_criteria + + radius_m, *_ = await _plan_criteria(project_id) + drawn = await asyncio.to_thread(draft_vertices, draft, radius_m) + sampler = await asyncio.to_thread( + _open_sampler, context.project_root, context.surface_params + ) + vertices, _ = await asyncio.to_thread(resolve_route_profile, drawn, sampler=sampler) + if superseded(): + return SUPERSEDED + + try: + preview = await asyncio.to_thread( + preview_stages, + vertices, + prepared["contours"], + prepared["streams"], + (drainage_dir(context.stored_path), prepared["contour_key"]), + superseded, + ) + except AnalysisStopped: + return SUPERSEDED + if preview is None or preview.routing is None: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "지금 노선으로 배수유역을 정하지 못했습니다."}, + ) + if superseded(): + return SUPERSEDED + + live = replace(context, vertices=vertices) + to_lonlat = context.to_lonlat + routing = live_routing(preview, [(v.x, v.y) for v in vertices], to_lonlat) + upstream = [ + [list(to_lonlat(x, y)) for x, y in line.coords] for line in preview.region.split.upstream + ] + detail, carried = await asyncio.to_thread( + _build, context.stored_path, live, points, routing, upstream + ) + if detail is None: + return JSONResponse( + status_code=400, + content={"status": "error", "message": "세부 유역을 나누지 못했습니다."}, + ) + result = _payload(project_id, live, detail, carried, False) + result["generation"] = generation + result["live"] = True + return result diff --git a/B05_Profile/B05_Profile_Api_TerrainGrid.ts b/B05_Profile/B05_Profile_Api_TerrainGrid.ts index 647f5dcad..14d5387c6 100644 --- a/B05_Profile/B05_Profile_Api_TerrainGrid.ts +++ b/B05_Profile/B05_Profile_Api_TerrainGrid.ts @@ -73,7 +73,7 @@ export function designTerrainGridNow(projectId: string): DesignTerrainGrid | nul /** 한 번에 물을 수 있는 점 수 — 서버 `MAX_POINTS` 와 같은 값. */ const ELEVATION_CHUNK = 4000; -/** 계산 묶음 ground 단계에 넣을 격자 · 서버 길(PLAN 346-1). 서버 길 값은 mm 자리(응답 그대로). */ +/** 계산 묶음 ground 단계에 넣을 격자 · 서버 길(PLAN 346-1). 서버 길도 소수 6자리 — 격자 길과 같은 값. */ export function groundStageDeps(projectId: string): Pick { return { grid: () => designTerrainGridNow(projectId), @@ -88,7 +88,7 @@ export function groundStageDeps(projectId: string): Pick return { "status": "success", "project_id": str(project_id), - "z": [None if not ok else round(float(value), 3) for value, ok in zip(z, valid)], + "z": [ + None if not ok else round(float(value), request.decimals) for value, ok in zip(z, valid) + ], "valid": [bool(ok) for ok in valid], } diff --git a/common_util/common_util_drainage_detail.py b/common_util/common_util_drainage_detail.py index f369dcdc7..05e215273 100644 --- a/common_util/common_util_drainage_detail.py +++ b/common_util/common_util_drainage_detail.py @@ -25,7 +25,6 @@ from __future__ import annotations import json import logging -import math from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -36,10 +35,11 @@ from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import find_inflow_h from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import STAGES from B04_PreProcess.B04_PreProcess_Engine_Watershed_Flow import polygon_parts, polygonize_labels from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec -from common_util.common_util_drainage_pipes import ( - PIPE_FACILITY_BOX, - PIPE_FACILITY_FORD_BRIDGE, - PIPE_FACILITY_PIPE, +from common_util.common_util_drainage_sizing import ( # noqa: F401 — 옛 자리에서 읽는 곳 + estimate_pipe_diameter_mm, + load_rainfall_idf, + recommend_structure, + size_pipe, ) from common_util.common_util_route_geometry import ( RouteVertex, @@ -48,24 +48,11 @@ from common_util.common_util_route_geometry import ( is_uphill_at, ) from common_util.common_util_structure_station import snap_candidates -from common_util.common_util_wamis_rainfall import idf_intensity from config.config_system import ( - DRAINAGE_BOX_THRESHOLD_MM, DRAINAGE_BRIDGE_THRESHOLD_MM, - 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_RECOMMEND_DIAMETERS_MM, - DRAINAGE_RUNOFF_COEFFICIENT, - DRAINAGE_TC_MIN_MINUTES, - DRAINAGE_VELOCITY_MAX_MS, - DRAINAGE_VELOCITY_MIN_MS, - PIPE_DEFAULT_DIAMETER_MM, ) logger = logging.getLogger(__name__) @@ -98,6 +85,8 @@ class DrainageDetail: strength_profile: list[list[float]] = field(default_factory=list) # 유입 집중점 — [누가거리, 유입면적, 구역번호, 구역 내 순위]. 관 자리를 판단하는 근거. inflow_hotspots: list[list[float]] = field(default_factory=list) + # 관 자리 후보(347-3) — 관 사이가 최대 간격을 넘는 구간의 흐름 세기 + 오목 자리 · 제안만. + pipe_candidates: list[StructureCandidate] = field(default_factory=list) @dataclass @@ -176,18 +165,23 @@ def build_detail( directory: Path, vertices: list[RouteVertex], confirmed_chainages: list[float] | None = None, + routing: RoadRouting | None = None, + upstream_lonlat: list[list[list[float]]] | None = None, ) -> DrainageDetail | None: """B04 분석 결과를 읽어 관을 보충하고 세부유역을 나눈다. `confirmed_chainages`를 주면 그 위치를 관으로 확정하고(사용자 편집 · 빈 목록 = 관 0개), None 이면 B04의 기본 관에 최대 간격 규칙으로 최소 개수만 보충한다. 어느 쪽이든 격자 해석은 하지 않는다. + `routing` · `upstream_lonlat` 을 주면 파일 대신 그것(「지금 노선으로 셈」 메모리 결과 · 347-2). """ - routing = read_road_routing(directory) + routing = routing or read_road_routing(directory) if routing is None or len(vertices) < 2: return None + candidates: list[StructureCandidate] = [] if confirmed_chainages is not None: pipes = pipes_from_chainages(vertices, confirmed_chainages) + candidates = suggest_pipe_positions(vertices, pipes, routing.strength_curve) else: # 저장분의 기본 관은 누가거리만 신뢰한다 — 좌표는 현재 노선 위로 다시 찍는다. base = [ @@ -209,7 +203,9 @@ def build_detail( grid_cell_m=routing.spec.cell_m, flow_arrows=routing.flow_arrows, arrow_spacing_m=routing.arrow_spacing_m, - upstream_lonlat=read_upstream_lines(directory), + upstream_lonlat=read_upstream_lines(directory) + if upstream_lonlat is None + else upstream_lonlat, strength_profile=build_strength_profile(routing), inflow_hotspots=[ [chainage, area, float(zone), float(rank)] @@ -219,6 +215,7 @@ def build_detail( vertices[-1].chainage_m, ) ], + pipe_candidates=candidates, ) if not pipes: return detail @@ -475,6 +472,24 @@ def _best_position( return float(positions[int(np.argmax(score))]) +def suggest_pipe_positions( + vertices: list[RouteVertex], + pipes: list[StructureCandidate], + strength_curve: np.ndarray, +) -> list[StructureCandidate]: + """관 자리 후보(347-3) — 지금 관 사이가 최대 간격을 넘는 구간만 간격 보충 규칙으로 고른 자리. + + `_best_position`(흐름 세기 + 오목 · 성토 덜 선호) · 정수 m · 관 목록엔 안 넣음(제안). + """ + taken = [pipe.chainage_m for pipe in pipes] + filled = snap_candidates(place_pipes(vertices, pipes, strength_curve), vertices) + return [ + pipe + for pipe in filled + if pipe.reason == "spacing" and all(abs(pipe.chainage_m - other) > 0.5 for other in taken) + ] + + def pipes_from_chainages( vertices: list[RouteVertex], chainages: list[float] ) -> list[StructureCandidate]: @@ -610,112 +625,3 @@ def _outlet_elevation(routing: RoadRouting, pipe_order: int, pipe_of_slot: np.nd elevations = routing.elevation[routing.road_cell_index[slots]] finite = elevations[np.isfinite(elevations)] 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), - # 물넘이·세월교는 관이 아니라 개수로라 직경 대신 이 단면적에서 출발한다. - "required_area_m2": round(required_area, 4), - } - - -def recommend_structure(diameter_mm: float | None) -> tuple[str, int | None]: - """유효직경으로 추천 구조물과 관경을 고른다 (2026-08-17 사용자 확정: 유량 근거만). - - 지식DB 선정 트리(개거_세월시설 §5·횡단배수관_암거 §5)에서 유량 조건만 취했다 — - 계곡 횡단경사·하천 차수는 지형 계산이 필요해 화면이 "현장 확인"으로 안내한다. - - D ≤ 1,500㎜ 배관 (레지스트리 선택지로 스냅, 하한 800㎜) - 1,500 < D ≤ 2,000 BOX암거 후보 - D > 2,000㎜ 세월교·물넘이 검토 (관 최대 규격 초과) - - 유효직경을 못 구한 유역(강우량표 없음)은 배관·관경 미정으로 둔다. - """ - if diameter_mm is None: - return PIPE_FACILITY_PIPE, None - if diameter_mm > DRAINAGE_BRIDGE_THRESHOLD_MM: - return PIPE_FACILITY_FORD_BRIDGE, None - if diameter_mm > DRAINAGE_BOX_THRESHOLD_MM: - return PIPE_FACILITY_BOX, None - # 별표2 (나) 예외 하한 800㎜ — 계산값이 더 작아도 그 아래로는 내리지 않는다. - need = max(float(PIPE_DEFAULT_DIAMETER_MM), diameter_mm) - size = next( - (value for value in DRAINAGE_RECOMMEND_DIAMETERS_MM if value >= need), - DRAINAGE_RECOMMEND_DIAMETERS_MM[-1], - ) - return PIPE_FACILITY_PIPE, int(size) - - -def estimate_pipe_diameter_mm( - area_m2: float, - relief_m: float, - flow_length_m: float, - idf: dict[str, float] | None = None, -) -> float | 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_drainage_sizing.py b/common_util/common_util_drainage_sizing.py new file mode 100644 index 000000000..b2d121514 --- /dev/null +++ b/common_util/common_util_drainage_sizing.py @@ -0,0 +1,145 @@ +"""배수 관경 산정 — 강우량표 읽기 · 합리식 유효직경 · 추천 구조물. + +세부 유역 엔진(`common_util_drainage_detail` · 700줄로 나눔)이 유역 제원으로 부름 · +이름은 그 파일에서도 그대로 읽힘. +""" + +from __future__ import annotations + +import json +import logging +import math +from pathlib import Path + +from common_util.common_util_drainage_pipes import ( + PIPE_FACILITY_BOX, + PIPE_FACILITY_FORD_BRIDGE, + PIPE_FACILITY_PIPE, +) +from common_util.common_util_wamis_rainfall import idf_intensity +from config.config_system import ( + DRAINAGE_BOX_THRESHOLD_MM, + DRAINAGE_BRIDGE_THRESHOLD_MM, + DRAINAGE_DESIGN_FLOW_FACTOR, + DRAINAGE_FLOW_AREA_RATIO, + DRAINAGE_MANNING_N, + DRAINAGE_PIPE_SLOPE_DEG, + DRAINAGE_RAINFALL_FILENAME, + DRAINAGE_RECOMMEND_DIAMETERS_MM, + DRAINAGE_RUNOFF_COEFFICIENT, + DRAINAGE_TC_MIN_MINUTES, + DRAINAGE_VELOCITY_MAX_MS, + DRAINAGE_VELOCITY_MIN_MS, + PIPE_DEFAULT_DIAMETER_MM, +) + +logger = logging.getLogger(__name__) + + +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), + # 물넘이·세월교는 관이 아니라 개수로라 직경 대신 이 단면적에서 출발한다. + "required_area_m2": round(required_area, 4), + } + + +def recommend_structure(diameter_mm: float | None) -> tuple[str, int | None]: + """유효직경으로 추천 구조물과 관경을 고른다 (2026-08-17 사용자 확정: 유량 근거만). + + 지식DB 선정 트리(개거_세월시설 §5·횡단배수관_암거 §5)에서 유량 조건만 취했다 — + 계곡 횡단경사·하천 차수는 지형 계산이 필요해 화면이 "현장 확인"으로 안내한다. + + D ≤ 1,500㎜ 배관 (레지스트리 선택지로 스냅, 하한 800㎜) + 1,500 < D ≤ 2,000 BOX암거 후보 + D > 2,000㎜ 세월교·물넘이 검토 (관 최대 규격 초과) + + 유효직경을 못 구한 유역(강우량표 없음)은 배관·관경 미정으로 둔다. + """ + if diameter_mm is None: + return PIPE_FACILITY_PIPE, None + if diameter_mm > DRAINAGE_BRIDGE_THRESHOLD_MM: + return PIPE_FACILITY_FORD_BRIDGE, None + if diameter_mm > DRAINAGE_BOX_THRESHOLD_MM: + return PIPE_FACILITY_BOX, None + # 별표2 (나) 예외 하한 800㎜ — 계산값이 더 작아도 그 아래로는 내리지 않는다. + need = max(float(PIPE_DEFAULT_DIAMETER_MM), diameter_mm) + size = next( + (value for value in DRAINAGE_RECOMMEND_DIAMETERS_MM if value >= need), + DRAINAGE_RECOMMEND_DIAMETERS_MM[-1], + ) + return PIPE_FACILITY_PIPE, int(size) + + +def estimate_pipe_diameter_mm( + area_m2: float, + relief_m: float, + flow_length_m: float, + idf: dict[str, float] | None = None, +) -> float | 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/resources/tester/test_344_1_terrain_grid_route.py b/resources/tester/test_344_1_terrain_grid_route.py index 69476e550..05c363bc3 100644 --- a/resources/tester/test_344_1_terrain_grid_route.py +++ b/resources/tester/test_344_1_terrain_grid_route.py @@ -114,3 +114,30 @@ def test_받을_자리_서명은_manifest_구성과_평활(tmp_path): ) assert terrain_router._design_grid({**PARAMS, "method": "tin"}, models, confirmed) is None assert terrain_router._design_grid(PARAMS, models, None) is None + + +def test_지반고_자릿수_기본_3_묶음은_6(tmp_path, monkeypatch): + root = tmp_path + models = root / "B04_PreProcess" / "models" + models.mkdir(parents=True) + x = np.arange(0.0, 11.0) + np.savez( + models / "dtm_csf_smooth.npz", + x=x, + y=x, + z=np.add.outer(x * 0.0, x * 0.1234567), + valid_mask=np.ones((11, 11), dtype=bool), + ) + _patch(monkeypatch, root, None) + + async def params(_connection, _project_id): + return dict(PARAMS) + + monkeypatch.setattr(terrain_router, "get_surface_confirmation_params", params) + ask = terrain_router.ElevationRequest + plain = asyncio.run(terrain_router.read_route_elevations(PID, ask(points=[(5.5, 5.0)]))) + fine = asyncio.run( + terrain_router.read_route_elevations(PID, ask(points=[(5.5, 5.0)], decimals=6)) + ) + assert plain["z"] == [round(5.5 * 0.1234567, 3)] + assert fine["z"] == [round(5.5 * 0.1234567, 6)] diff --git a/resources/tester/test_347_2_drainage_live.py b/resources/tester/test_347_2_drainage_live.py new file mode 100644 index 000000000..948ff924a --- /dev/null +++ b/resources/tester/test_347_2_drainage_live.py @@ -0,0 +1,169 @@ +"""347-2 「지금 노선으로 셈」 서버 길 · 347-3 관 자리 후보. + +- 늦은 차례: 계산 중 새 차례가 오면 단계 사이에서 그만둠(superseded) · 작업본에 안 씀. +- 메모리 격자로 세부 유역(파일 안 읽음) · 노선 초안 → 그린 노선. +- 관 자리 후보: 최대 간격(100 m)을 넘는 구간에만 · 기존 관 자리엔 안 냄 · 관 목록엔 안 넣음. +- [확정](recompute) = 낡았으면 배수유역부터 다시 셈 · [저장] = 안 셈. +""" + +import asyncio +import math +from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4 + +import numpy as np +import pytest +from shapely.geometry import LineString + +from B04_PreProcess import B04_PreProcess_Engine_Watershed_Expand as expand +from B04_PreProcess import B04_PreProcess_Router_Basins as basins +from B04_PreProcess import B04_PreProcess_Router_Watershed_Live as live +from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec +from common_util.common_util_drainage_context import DrainageContext +from common_util.common_util_drainage_detail import ( + RoadRouting, + build_detail, + pipes_from_chainages, + suggest_pipe_positions, +) +from common_util.common_util_route_geometry import RouteVertex + +VERTICES = [RouteVertex(0.0, 0.0, 100.0, 0.0), RouteVertex(300.0, 0.0, 90.0, 300.0)] + + +def test_draft_vertices_follow_nodes(): + draft = {"nodes": [[0, 0], [100, 0], [100, 100]], "curveOn": [False, False, False]} + vertices = live.draft_vertices(draft, 30.0) + assert vertices[0].x == 0 and vertices[-1].y == 100 + assert math.isclose(vertices[-1].chainage_m, 200.0) + + +def test_candidates_only_in_long_gaps(): + strength = np.zeros(301) + strength[150] = 50.0 # 흐름이 모이는 자리 + pipes = pipes_from_chainages(VERTICES, [0.0, 300.0]) + picked = suggest_pipe_positions(VERTICES, pipes, strength) + assert picked and all(0 < pipe.chainage_m < 300 for pipe in picked) + assert all(pipe.chainage_m == round(pipe.chainage_m) for pipe in picked) # 정수 m + dense = pipes_from_chainages(VERTICES, [0.0, 90.0, 180.0, 270.0, 300.0]) + assert suggest_pipe_positions(VERTICES, dense, strength) == [] + + +def _routing() -> RoadRouting: + spec = GridSpec(x_min=-5.0, y_max=5.0, cell_m=1.0, n_rows=10, n_cols=310) + road_index = np.arange(5 * 310 + 5, 5 * 310 + 305, dtype=np.int32) + road_slot = np.full(spec.size, -1, dtype=np.int32) + road_slot[road_index] = np.arange(road_index.size) + return RoadRouting( + spec=spec, + road_slot=road_slot, + path_length=np.zeros(spec.size, np.float32), + elevation=np.full(spec.size, 95.0, np.float32), + road_cell_index=road_index, + road_chainage=np.arange(road_index.size, dtype=np.float64), + strength=np.ones(road_index.size, np.int64), + ) + + +def test_build_detail_uses_memory_routing(tmp_path): + missing = tmp_path / "없는 폴더" # 파일을 읽으면 None 이 됨 + detail = build_detail(missing, VERTICES, [0.0, 300.0], _routing(), []) + assert detail is not None and detail.pipe_candidates + assert build_detail(missing, VERTICES, [0.0, 300.0]) is None + + +def _patch_live(monkeypatch, preview_fn): + async def prepare(project_id): + return {"contours": [], "streams": [], "contour_key": "k"} + + async def load(project_id): + return DrainageContext(stored_path="x", project_root=Path("."), vertices=VERTICES), "" + + monkeypatch.setattr(live, "_prepare", prepare) + monkeypatch.setattr(live, "load_drainage_context", load) + monkeypatch.setattr(live, "drainage_dir", lambda stored: Path(".")) + monkeypatch.setattr(live, "preview_stages", preview_fn) + calls = SimpleNamespace(build=0) + + def build(*args): + calls.build += 1 + return None, [] + + monkeypatch.setattr(live, "_build", build) + return calls + + +def test_newer_generation_stops_between_stages(monkeypatch): + pid = uuid4() + + def preview(vertices, contours, streams, store, should_stop): + live._LATEST_GENERATION[str(pid)] = 8 # 계산 중 새 조작 → 새 요청이 차례를 올림 + if should_stop(): + raise expand.AnalysisStopped + + calls = _patch_live(monkeypatch, preview) + result = asyncio.run(live.post_drainage_live(pid, {"generation": 7, "points": []})) + assert result == live.SUPERSEDED and calls.build == 0 + + +def test_newer_generation_after_preview_skips_split(monkeypatch): + pid = uuid4() + + def preview(vertices, contours, streams, store, should_stop): + assert not should_stop() + live._LATEST_GENERATION[str(pid)] = 3 + return SimpleNamespace(routing=object()) + + calls = _patch_live(monkeypatch, preview) + result = asyncio.run(live.post_drainage_live(pid, {"generation": 2})) + assert result == live.SUPERSEDED and calls.build == 0 + + +def test_engine_stops_in_expansion_round(): + contours = [ + { + "geometry": { + "type": "LineString", + "coordinates": [ + [150 + r * math.cos(t), -150 + r * math.sin(t)] + for t in np.linspace(0, 2 * math.pi, 121) + ], + }, + "properties": {"elevation": 300 - r}, + } + for r in range(10, 200, 10) + ] + spec = GridSpec(x_min=100.0, y_max=-100.0, cell_m=1.0, n_rows=100, n_cols=100) + domain = np.zeros((100, 100), bool) + domain[20:80, 20:80] = True + route = LineString([(110.0, -150.0), (190.0, -150.0)]) + with pytest.raises(expand.AnalysisStopped): + expand.expand_by_red_boundary( + spec, domain, contours, route, [], None, should_stop=lambda: True + ) + + +@pytest.mark.parametrize( + "recompute, stale, expected", [(True, True, 1), (True, False, 0), (False, True, 0)] +) +def test_confirm_recomputes_only_when_stale(monkeypatch, recompute, stale, expected): + import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed + + calls = [] + + async def is_stale(project_id): + return stale + + async def region(project_id, refresh=False): + calls.append(refresh) + return {"status": "success"} + + async def resolve(*args, **kwargs): + return basins.JSONResponse(status_code=404, content={}) + + monkeypatch.setattr(basins, "_analysis_stale", is_stale) + monkeypatch.setattr(watershed, "get_primary_region", region) + monkeypatch.setattr(basins, "_resolve", resolve) + asyncio.run(basins.put_pipe_points(uuid4(), {"points": []}, None, recompute)) + assert len(calls) == expected and all(calls)