Merge remote-tracking branches 'origin/main_laptop_1' and 'origin/sub_laptop_3' into sub_laptop_7

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("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.")

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)

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

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

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

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<GroundStageDeps, "grid" | "remote"> {
   return {
     grid: () => designTerrainGridNow(projectId),
@@ -88,7 +88,7 @@ export function groundStageDeps(projectId: string): Pick<GroundStageDeps, "grid"
             method: "POST",
             credentials: "include",
             headers: { "Content-Type": "application/json" },
-            body: JSON.stringify({ points: chunk, terrain }),
+            body: JSON.stringify({ points: chunk, terrain, decimals: 6 }),
             signal,
           });
           if (!response.ok) throw new Error(`지반고 요청 실패: HTTP ${response.status}`);

B05_Profile/B05_Profile_Router_Terrain.py

@@ -62,6 +62,8 @@ class ElevationRequest(BaseModel):
 
     points: list[tuple[float, float]] = Field(..., min_length=1, max_length=MAX_POINTS)
     terrain: AnalysisTerrain | None = None
+    #: 표고 자릿수 — 화면 값은 mm(3) · 계산 묶음 서버 길은 6(격자 표본 · 서버 생성과 같은 자리).
+    decimals: int = Field(3, ge=0, le=6)
 
 
 def _sample(project_root: Path, params: dict, points: list[tuple[float, float]]):
@@ -112,7 +114,9 @@ async def read_route_elevations(project_id: UUID, request: ElevationRequest) ->
     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],
     }
 

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

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

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)]

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)