auto: 2026-07-28 19:32 (EOMSANGDON-HOME)

B04_wf1_Surface/B04_wf1_Surface_UI_MapRender.ts

@@ -464,3 +464,60 @@ export function drawPreparedLabels(
     context.fillText(feature.labelText, x, y);
   }
 }
+
+/** 채움 폴리곤 오버레이(배수유역 등). 좌표는 lon/lat 링 1개. */
+export type FilledRing = {
+  ring: ReadonlyArray<readonly [number, number]>;
+  /** 면적 중심에 얹을 번호. 없으면 라벨을 그리지 않는다. */
+  label?: string;
+};
+
+/**
+ * lon/lat 폴리곤 링을 파스텔 채움 + 테두리 + 중심 번호로 그린다.
+ * 사전 투영 캐시를 쓰지 않는 소량(유역 수 개) 오버레이 전용이라 매 프레임 변환해도 부담이 없다.
+ */
+export function drawFilledRing(
+  context: CanvasRenderingContext2D,
+  entry: FilledRing,
+  normalizer: Normalizer,
+  view: ViewState,
+  color: string,
+): void {
+  if (entry.ring.length < 3) return;
+  const affine = affineOf(view);
+  let sumX = 0;
+  let sumY = 0;
+  context.beginPath();
+  entry.ring.forEach(([lon, lat], index) => {
+    const nx = (lon - normalizer.lonMin) / normalizer.lonRange;
+    const ny = 1 - (lat - normalizer.latMin) / normalizer.latRange;
+    const x = nx * affine.ax + affine.bx;
+    const y = ny * affine.ay + affine.by;
+    sumX += x;
+    sumY += y;
+    if (index === 0) context.moveTo(x, y);
+    else context.lineTo(x, y);
+  });
+  context.closePath();
+  context.fillStyle = color;
+  context.fill();
+  context.strokeStyle = color;
+  context.lineWidth = 1.6;
+  context.stroke();
+  if (!entry.label) return;
+  // 면적 중심(정점 평균)에 번호를 원형 배지로 얹는다.
+  const centerX = sumX / entry.ring.length;
+  const centerY = sumY / entry.ring.length;
+  context.beginPath();
+  context.arc(centerX, centerY, 11, 0, Math.PI * 2);
+  context.fillStyle = color;
+  context.fill();
+  context.strokeStyle = "rgba(255, 255, 255, 0.9)";
+  context.lineWidth = 1.5;
+  context.stroke();
+  context.font = "600 12px sans-serif";
+  context.textAlign = "center";
+  context.textBaseline = "middle";
+  context.fillStyle = "#1f2937";
+  context.fillText(entry.label, centerX, centerY);
+}

B05_wf2_Route/B05_wf2_Route_Api_Fetch.ts

@@ -247,3 +247,60 @@ export async function fetchLatestRoute(projectId: string): Promise<RouteLatestRe
     method: "GET",
   });
 }
+
+/* ── 배수유역도 (B05_wf2_Route_Router_Drainage.py) ───────────────────────── */
+
+/** 관 매설 구조물 측점 후보 1개. reason: stream=세류 교차, spacing=300m 보충. */
+export interface DrainageCandidate {
+  chainage_m: number;
+  x: number;
+  y: number;
+  lon: number;
+  lat: number;
+  reason: "stream" | "spacing" | "confirmed";
+  stream_name: string | null;
+}
+
+export interface DrainageCandidateResponse {
+  status: string;
+  project_id: string;
+  route_id: number;
+  candidates: DrainageCandidate[];
+}
+
+/** 배수유역 1개. 관경(pipe_diameter_mm)은 수식 미확정이라 당분간 항상 null이다. */
+export interface DrainageBasin {
+  index: number;
+  chainage_m: number;
+  polygon_lonlat: Array<[number, number]>;
+  area_m2: number;
+  relief_m: number;
+  flow_length_m: number;
+  pipe_diameter_mm: number | null;
+}
+
+export interface DrainageBasinResponse {
+  status: string;
+  project_id: string;
+  route_id: number;
+  basins: DrainageBasin[];
+}
+
+export async function fetchDrainageCandidates(
+  projectId: string,
+): Promise<DrainageCandidateResponse> {
+  return requestJson<DrainageCandidateResponse>(`/projects/${projectId}/drainage/candidates`, {
+    method: "GET",
+  });
+}
+
+/** chainages를 주면 그 위치로 확정 산정하고, 비우면 자동 제안분으로 산정한다. */
+export async function fetchDrainageBasins(
+  projectId: string,
+  chainages?: number[],
+): Promise<DrainageBasinResponse> {
+  return requestJson<DrainageBasinResponse>(`/projects/${projectId}/drainage/basins`, {
+    method: "POST",
+    body: JSON.stringify({ chainages: chainages ?? [] }),
+  });
+}

B05_wf2_Route/B05_wf2_Route_Engine_Drainage.py

@@ -0,0 +1,279 @@
+"""배수유역 산정 엔진.
+
+관 매설 구조물 측점 후보를 제안하고, 각 측점이 받는 배수유역 경계를 산정한다.
+지형 판단은 **도엽 등고선·세류선(하천중심선)·표고점**만 사용한다 — 3D 포인트클라우드나
+지형 메시는 쓰지 않는다(2026-07-28 사용자 지시).
+
+유역을 나누는 최종 목적은 각 지점의 파이프 관경 결정이다. 유역 경사면에 100년 강우빈도를
+적용해 모이는 물의 양을 산정하고 그 유량으로 관경을 정한다. 관경 수식은 아직 미확정이라
+`estimate_pipe_diameter_mm()`은 골격만 두고 비워 둔다.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass, field
+from typing import Any
+
+from shapely.geometry import LineString, Point, shape
+
+logger = logging.getLogger(__name__)
+
+# 구조물 측점 사이 최대 허용 간격(m). 세류 교차가 없어도 이 간격을 넘으면 절토부에 추가 배치한다.
+MAX_STRUCTURE_SPACING_M = 300.0
+# 같은 세류 교차로 볼 최소 이격(m). 이보다 가까운 교차점은 하나로 묶는다.
+MIN_STRUCTURE_SPACING_M = 20.0
+# 유역 경계 탐색 반경(m). 측점에서 이 거리를 넘는 지형은 해당 유역으로 보지 않는다.
+MAX_BASIN_RADIUS_M = 800.0
+
+
+@dataclass
+class RouteVertex:
+    """노선 폴리라인의 한 점. chainage는 시점 기준 누가거리(m)."""
+
+    x: float
+    y: float
+    z: float
+    chainage_m: float
+
+
+@dataclass
+class StructureCandidate:
+    """관 매설 구조물 측점 후보."""
+
+    chainage_m: float
+    x: float
+    y: float
+    # "stream"=세류 교차, "spacing"=300m 규칙에 따른 보충 배치
+    reason: str
+    stream_name: str | None = None
+
+
+@dataclass
+class DrainageBasin:
+    """한 구조물 측점이 받는 배수유역."""
+
+    index: int
+    chainage_m: float
+    outlet_x: float
+    outlet_y: float
+    polygon_lonlat: list[list[float]] = field(default_factory=list)
+    area_m2: float = 0.0
+    # 유역 최고 표고 − 측점 표고(m). 경사면 낙차.
+    relief_m: float = 0.0
+    # 유하거리: 측점에서 유역 최상단까지 물길 길이(m).
+    flow_length_m: float = 0.0
+    pipe_diameter_mm: float | None = None
+
+
+def build_route_vertices(points: list[dict[str, Any]]) -> list[RouteVertex]:
+    """DB route_points 행을 누가거리가 채워진 정점 목록으로 바꾼다."""
+    vertices: list[RouteVertex] = []
+    cumulative = 0.0
+    previous: tuple[float, float] | None = None
+    for row in points:
+        x = float(row["x"])
+        y = float(row["y"])
+        z = float(row.get("z") or 0.0)
+        if previous is not None:
+            cumulative += math.dist(previous, (x, y))
+        chainage = row.get("chainage_m")
+        vertices.append(
+            RouteVertex(
+                x=x,
+                y=y,
+                z=z,
+                chainage_m=float(chainage) if chainage is not None else cumulative,
+            )
+        )
+        previous = (x, y)
+    return vertices
+
+
+def _interpolate_vertex(
+    vertices: list[RouteVertex], chainage_m: float
+) -> tuple[float, float, float]:
+    """누가거리 위치의 (x, y, z)를 선형 보간한다."""
+    if not vertices:
+        return (0.0, 0.0, 0.0)
+    if chainage_m <= vertices[0].chainage_m:
+        return (vertices[0].x, vertices[0].y, vertices[0].z)
+    for previous, current in zip(vertices, vertices[1:]):
+        if chainage_m <= current.chainage_m:
+            span = current.chainage_m - previous.chainage_m
+            ratio = 0.0 if span <= 0 else (chainage_m - previous.chainage_m) / span
+            return (
+                previous.x + (current.x - previous.x) * ratio,
+                previous.y + (current.y - previous.y) * ratio,
+                previous.z + (current.z - previous.z) * ratio,
+            )
+    last = vertices[-1]
+    return (last.x, last.y, last.z)
+
+
+def is_uphill_at(vertices: list[RouteVertex], chainage_m: float, window_m: float = 20.0) -> bool:
+    """해당 위치가 오르막(절토부)인지 판정한다.
+
+    내리막(성토부)은 물이 노선 바깥으로 흘러나가므로 배수유역을 만들지 않는다
+    (2026-07-28 사용자 지시). 판정은 종단 계획선의 국소 기울기 부호로 한다.
+    """
+    _, _, back_z = _interpolate_vertex(vertices, max(0.0, chainage_m - window_m))
+    _, _, forward_z = _interpolate_vertex(vertices, chainage_m + window_m)
+    return forward_z >= back_z
+
+
+def find_stream_crossings(
+    vertices: list[RouteVertex],
+    stream_features: list[dict[str, Any]],
+) -> list[StructureCandidate]:
+    """노선 평면 선형과 세류선의 교차 지점을 찾는다."""
+    if len(vertices) < 2:
+        return []
+    route_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
+    candidates: list[StructureCandidate] = []
+    for feature in stream_features:
+        geometry = feature.get("geometry")
+        if not geometry:
+            continue
+        try:
+            stream = shape(geometry)
+        except Exception:  # noqa: BLE001 - 손상된 피처는 건너뛴다
+            continue
+        if stream.is_empty:
+            continue
+        intersection = route_line.intersection(stream)
+        if intersection.is_empty:
+            continue
+        name = _stream_name(feature)
+        for point in _collect_points(intersection):
+            candidates.append(
+                StructureCandidate(
+                    chainage_m=route_line.project(point),
+                    x=point.x,
+                    y=point.y,
+                    reason="stream",
+                    stream_name=name,
+                )
+            )
+    candidates.sort(key=lambda item: item.chainage_m)
+    return candidates
+
+
+def _stream_name(feature: dict[str, Any]) -> str | None:
+    properties = feature.get("properties") or {}
+    for key in ("명칭", "하천명", "NAME", "name"):
+        value = properties.get(key)
+        if value:
+            return str(value)
+    return None
+
+
+def _collect_points(geometry: Any) -> list[Point]:
+    """교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다."""
+    if geometry.geom_type == "Point":
+        return [geometry]
+    if geometry.geom_type in {"MultiPoint", "GeometryCollection"}:
+        points: list[Point] = []
+        for part in geometry.geoms:
+            points.extend(_collect_points(part))
+        return points
+    # 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다.
+    if geometry.geom_type in {"LineString", "MultiLineString"}:
+        return [geometry.interpolate(0.5, normalized=True)]
+    return []
+
+
+def propose_structure_stations(
+    vertices: list[RouteVertex],
+    stream_features: list[dict[str, Any]],
+) -> list[StructureCandidate]:
+    """구조물 측점 후보를 제안한다.
+
+    ① 세류 교차 지점 ② 내리막(성토부) 제외 ③ 직전 측점에서 300m 초과 시 절토부에 보충 배치.
+    """
+    if len(vertices) < 2:
+        return []
+    total_length = vertices[-1].chainage_m
+    crossings = [
+        candidate
+        for candidate in find_stream_crossings(vertices, stream_features)
+        if is_uphill_at(vertices, candidate.chainage_m)
+    ]
+
+    # 너무 가까운 교차는 하나로 본다(같은 계곡을 여러 선분이 지나는 경우).
+    merged: list[StructureCandidate] = []
+    for candidate in crossings:
+        if merged and candidate.chainage_m - merged[-1].chainage_m < MIN_STRUCTURE_SPACING_M:
+            continue
+        merged.append(candidate)
+
+    # 300m 규칙: 빈 구간에 절토부 지점을 찾아 보충한다.
+    filled: list[StructureCandidate] = []
+    previous_chainage = 0.0
+    for candidate in [*merged, None]:
+        boundary = candidate.chainage_m if candidate else total_length
+        filled.extend(_fill_spacing(vertices, previous_chainage, boundary))
+        if candidate:
+            filled.append(candidate)
+            previous_chainage = candidate.chainage_m
+        else:
+            previous_chainage = boundary
+    filled.sort(key=lambda item: item.chainage_m)
+    return filled
+
+
+def _fill_spacing(
+    vertices: list[RouteVertex],
+    start_m: float,
+    end_m: float,
+) -> list[StructureCandidate]:
+    """[start, end] 구간이 300m를 넘으면 절토부 지점에 보충 측점을 만든다."""
+    added: list[StructureCandidate] = []
+    cursor = start_m
+    while end_m - cursor > MAX_STRUCTURE_SPACING_M:
+        target = cursor + MAX_STRUCTURE_SPACING_M
+        placed = _nearest_uphill(vertices, target, end_m)
+        if placed is None:
+            break
+        x, y, _ = _interpolate_vertex(vertices, placed)
+        added.append(StructureCandidate(chainage_m=placed, x=x, y=y, reason="spacing"))
+        cursor = placed
+    return added
+
+
+def _nearest_uphill(
+    vertices: list[RouteVertex],
+    target_m: float,
+    limit_m: float,
+    step_m: float = 10.0,
+) -> float | None:
+    """목표 위치에서 가장 가까운 절토부(오르막) 지점을 찾는다. 없으면 None."""
+    if is_uphill_at(vertices, target_m):
+        return target_m
+    offset = step_m
+    while offset <= MAX_STRUCTURE_SPACING_M / 2:
+        for probe in (target_m - offset, target_m + offset):
+            if probe <= 0 or probe >= limit_m:
+                continue
+            if is_uphill_at(vertices, probe):
+                return probe
+        offset += step_m
+    return None
+
+
+def estimate_pipe_diameter_mm(
+    area_m2: float,
+    relief_m: float,
+    flow_length_m: float,
+    rainfall_mm_per_hour: 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

B05_wf2_Route/B05_wf2_Route_Engine_Drainage_Basin.py

@@ -0,0 +1,257 @@
+"""배수유역 경계 산정.
+
+구조물 측점(관 매설 지점)에서 산정상부까지 역추적해 밀폐된 유역 경계를 만든다.
+지형 판단 근거는 도엽 등고선·세류선·표고점뿐이다(3D 미사용, 2026-07-28 사용자 지시).
+
+정상부 판정 규칙(사용자 지시):
+    표고점 데이터는 산 정상부가 아닌 경우가 많다. 따라서 **등고선의 동심 폐합 패턴**
+    (안쪽으로 갈수록 표고가 높아지는 폐합 등고선의 최내곽)으로 정상부를 먼저 판단하고,
+    표고점은 그 판정을 보조·검증하는 용도로만 쓴다.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from typing import Any
+
+from shapely.geometry import LineString, Point, Polygon, shape
+from shapely.ops import unary_union
+
+from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
+    MAX_BASIN_RADIUS_M,
+    DrainageBasin,
+    RouteVertex,
+    StructureCandidate,
+    estimate_pipe_diameter_mm,
+)
+
+logger = logging.getLogger(__name__)
+
+# 등고선을 폐합으로 볼 때 허용하는 시종점 이격(m). 도엽 경계에서 잘린 선을 걸러낸다.
+CLOSED_TOLERANCE_M = 1.0
+# 폐합 등고선이 정상부 후보가 되는 최대 둘레(m). 이보다 크면 산체 전체라 정상부로 보지 않는다.
+MAX_SUMMIT_PERIMETER_M = 1200.0
+
+
+class ContourField:
+    """도엽 등고선 피처 모음을 표고 조회·정상부 판정에 쓸 수 있게 감싼 것."""
+
+    def __init__(self, features: list[dict[str, Any]], elevation_keys: tuple[str, ...]):
+        self.lines: list[tuple[LineString, float]] = []
+        self.closed: list[tuple[Polygon, float]] = []
+        for feature in features:
+            elevation = _read_elevation(feature.get("properties") or {}, elevation_keys)
+            if elevation is None:
+                continue
+            geometry = feature.get("geometry")
+            if not geometry:
+                continue
+            try:
+                geom = shape(geometry)
+            except Exception:  # noqa: BLE001 - 손상된 피처는 건너뛴다
+                continue
+            for line in _iter_lines(geom):
+                self.lines.append((line, elevation))
+                polygon = _as_closed_polygon(line)
+                if polygon is not None:
+                    self.closed.append((polygon, elevation))
+
+    def elevation_at(self, x: float, y: float, radius_m: float = 200.0) -> float | None:
+        """가장 가까운 등고선의 표고를 그 지점의 표고로 본다."""
+        point = Point(x, y)
+        best: tuple[float, float] | None = None
+        for line, elevation in self.lines:
+            distance = line.distance(point)
+            if distance > radius_m:
+                continue
+            if best is None or distance < best[0]:
+                best = (distance, elevation)
+        return None if best is None else best[1]
+
+    def find_summits(self, near: Point, radius_m: float) -> list[tuple[Polygon, float]]:
+        """주변의 정상부 후보를 찾는다.
+
+        동심 폐합 등고선 중 **자기보다 높은 폐합 등고선을 안에 품지 않은 것**이 최내곽,
+        즉 정상부다. 도엽 경계에서 잘린 선과 산체 전체를 감싸는 큰 폐합은 제외한다.
+        """
+        nearby = [
+            (polygon, elevation)
+            for polygon, elevation in self.closed
+            if polygon.length <= MAX_SUMMIT_PERIMETER_M and polygon.distance(near) <= radius_m
+        ]
+        summits: list[tuple[Polygon, float]] = []
+        for polygon, elevation in nearby:
+            has_higher_inside = any(
+                other_elevation > elevation and polygon.contains(other.representative_point())
+                for other, other_elevation in nearby
+                if other is not polygon
+            )
+            if not has_higher_inside:
+                summits.append((polygon, elevation))
+        return summits
+
+
+def _read_elevation(properties: dict[str, Any], keys: tuple[str, ...]) -> float | None:
+    for key in keys:
+        value = properties.get(key)
+        if value is None:
+            continue
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            continue
+    return None
+
+
+def _iter_lines(geometry: Any) -> list[LineString]:
+    if geometry.geom_type == "LineString":
+        return [geometry]
+    if geometry.geom_type == "MultiLineString":
+        return list(geometry.geoms)
+    return []
+
+
+def _as_closed_polygon(line: LineString) -> Polygon | None:
+    coords = list(line.coords)
+    if len(coords) < 4:
+        return None
+    if math.dist(coords[0], coords[-1]) > CLOSED_TOLERANCE_M:
+        return None
+    try:
+        polygon = Polygon(coords)
+    except Exception:  # noqa: BLE001
+        return None
+    return polygon if polygon.is_valid and polygon.area > 0 else None
+
+
+def build_basins(
+    vertices: list[RouteVertex],
+    candidates: list[StructureCandidate],
+    contours: ContourField,
+    stream_features: list[dict[str, Any]],
+    to_lonlat: Any,
+) -> list[DrainageBasin]:
+    """확정된 구조물 측점별 배수유역을 만든다.
+
+    측점에 물을 보내는 세류 가지를 따라 위로 올라가 정상부 폐합 등고선까지 닿는 범위를
+    유역으로 본다. 정상부는 ContourField.find_summits가 등고선 폐합 패턴으로 판정한다.
+    번호는 노선 시점에 가까운 순서(측점 누가거리 오름차순)로 1부터 매긴다.
+    """
+    route_line = (
+        LineString([(vertex.x, vertex.y) for vertex in vertices]) if len(vertices) > 1 else None
+    )
+    streams = _stream_lines(stream_features)
+    basins: list[DrainageBasin] = []
+    for index, candidate in enumerate(
+        sorted(candidates, key=lambda item: item.chainage_m), start=1
+    ):
+        outlet = Point(candidate.x, candidate.y)
+        uphill = _uphill_streams(outlet, streams, contours)
+        summits = contours.find_summits(outlet, MAX_BASIN_RADIUS_M)
+        boundary = _basin_polygon(outlet, uphill, summits, route_line)
+        if boundary is None or boundary.is_empty:
+            continue
+        outlet_elevation = contours.elevation_at(candidate.x, candidate.y) or 0.0
+        top_elevation = max((elevation for _, elevation in summits), default=outlet_elevation)
+        basin = DrainageBasin(
+            index=index,
+            chainage_m=candidate.chainage_m,
+            outlet_x=candidate.x,
+            outlet_y=candidate.y,
+            polygon_lonlat=[list(to_lonlat(x, y)) for x, y in boundary.exterior.coords],
+            area_m2=float(boundary.area),
+            relief_m=float(max(0.0, top_elevation - outlet_elevation)),
+            flow_length_m=_flow_length(outlet, uphill, boundary),
+        )
+        basin.pipe_diameter_mm = estimate_pipe_diameter_mm(
+            basin.area_m2, basin.relief_m, basin.flow_length_m
+        )
+        basins.append(basin)
+    return basins
+
+
+def _stream_lines(features: list[dict[str, Any]]) -> list[LineString]:
+    lines: list[LineString] = []
+    for feature in features:
+        geometry = feature.get("geometry")
+        if not geometry:
+            continue
+        try:
+            lines.extend(_iter_lines(shape(geometry)))
+        except Exception:  # noqa: BLE001
+            continue
+    return lines
+
+
+def _uphill_streams(
+    outlet: Point,
+    streams: list[LineString],
+    contours: ContourField,
+    tolerance_m: float = 30.0,
+) -> list[LineString]:
+    """측점에 연결된 세류 가지 중 위쪽(표고가 높아지는 방향)으로 뻗은 것만 모은다."""
+    connected = [line for line in streams if line.distance(outlet) <= tolerance_m]
+    uphill: list[LineString] = []
+    outlet_elevation = contours.elevation_at(outlet.x, outlet.y)
+    for line in connected:
+        far = _far_end(line, outlet)
+        far_elevation = contours.elevation_at(far.x, far.y)
+        if outlet_elevation is None or far_elevation is None or far_elevation >= outlet_elevation:
+            uphill.append(line)
+    return uphill
+
+
+def _far_end(line: LineString, outlet: Point) -> Point:
+    start = Point(line.coords[0])
+    end = Point(line.coords[-1])
+    return end if start.distance(outlet) <= end.distance(outlet) else start
+
+
+def _basin_polygon(
+    outlet: Point,
+    uphill: list[LineString],
+    summits: list[tuple[Polygon, float]],
+    route_line: LineString | None,
+) -> Polygon | None:
+    """유역 경계를 만든다.
+
+    측점 + 상류 세류 + 정상부 폐합 등고선을 함께 감싸는 볼록 껍질을 1차 경계로 삼고,
+    노선 아래쪽(성토부 방향)은 노선을 경계로 잘라낸다. 세류가 없으면 정상부까지의
+    반경 안에서 만들어지는 범위만 남는다.
+    """
+    parts: list[Any] = [outlet.buffer(5.0)]
+    parts.extend(uphill)
+    parts.extend(polygon for polygon, _ in summits)
+    if len(parts) <= 1:
+        return None
+    hull = unary_union(parts).convex_hull
+    if hull.geom_type != "Polygon":
+        return None
+    if route_line is not None:
+        hull = _clip_downhill(hull, route_line, outlet)
+    return hull if hull is not None and hull.geom_type == "Polygon" else None
+
+
+def _clip_downhill(hull: Polygon, route_line: LineString, outlet: Point) -> Polygon | None:
+    """노선을 경계로 유역을 잘라 산 쪽(상류) 조각만 남긴다."""
+    try:
+        pieces = hull.difference(route_line.buffer(0.5))
+    except Exception:  # noqa: BLE001
+        return hull
+    if pieces.is_empty:
+        return hull
+    parts = list(pieces.geoms) if pieces.geom_type == "MultiPolygon" else [pieces]
+    # 상류 조각 판별이 애매할 때를 대비해 면적이 가장 큰 조각을 채택한다.
+    best = max(parts, key=lambda part: part.area, default=None)
+    return best if best is not None and best.geom_type == "Polygon" else hull
+
+
+def _flow_length(outlet: Point, uphill: list[LineString], boundary: Polygon) -> float:
+    """유하거리: 측점에서 유역 최상단까지의 물길 길이(m).
+
+    상류 세류가 있으면 그 물길 길이의 최댓값을, 없으면 유역 안 최원점까지의 직선거리를 쓴다.
+    """
+    if uphill:
+        return float(max(line.length for line in uphill))
+    return float(max((outlet.distance(Point(xy)) for xy in boundary.exterior.coords), default=0.0))

B05_wf2_Route/B05_wf2_Route_Router_Drainage.py

@@ -0,0 +1,224 @@
+"""배수유역도 API 라우터.
+
+구조물 측점(관 매설) 후보 제안과 배수유역 산정을 제공한다. 지형 근거는 도엽 등고선·세류선·
+표고점 GeoJSON뿐이며, 좌표는 사업지 CRS(m)에서 계산하고 응답만 WGS84로 바꿔 내보낸다.
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from pyproj import Transformer
+
+from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
+    StructureCandidate,
+    build_route_vertices,
+    propose_structure_stations,
+)
+from B05_wf2_Route.B05_wf2_Route_Engine_Drainage_Basin import ContourField, build_basins
+from B05_wf2_Route.B05_wf2_Route_Repository import (
+    get_latest_route,
+    get_route_points,
+    get_surface_crs_epsg,
+)
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_db import get_db_pool
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/projects", tags=["B05 Route Drainage"])
+
+# 도엽 레이어 파일명 (B04 전처리 산출물과 동일 위치)
+_CONTOUR_FILE = "도엽_등고선.geojson"
+_STREAM_FILE = "도엽_하천중심선.geojson"
+# 도엽 등고선의 표고 속성 키. gpkg 등고선(CTRLN_HG)도 함께 본다.
+_ELEVATION_KEYS = ("등고수치", "CTRLN_HG", "elevation", "ELEV")
+
+
+def _sheet_dir(stored_path: str) -> Path:
+    return Path(resolve_stored_project_path(stored_path)) / "B04_wf1_Surface" / "processed"
+
+
+def _load_features(directory: Path, filename: str) -> list[dict[str, Any]]:
+    """도엽 GeoJSON을 읽어 피처 목록만 돌려준다. 없으면 빈 목록."""
+    path = directory / filename
+    if not path.exists():
+        return []
+    try:
+        with path.open("r", encoding="utf-8") as file:
+            data = json.load(file)
+    except (OSError, json.JSONDecodeError):
+        logger.warning("도엽 GeoJSON을 읽지 못했습니다: %s", path)
+        return []
+    features = data.get("features")
+    return features if isinstance(features, list) else []
+
+
+def _reproject_features(
+    features: list[dict[str, Any]],
+    transformer: Transformer | None,
+) -> list[dict[str, Any]]:
+    """WGS84 도엽 좌표를 사업지 CRS(m)로 바꾼다. 거리·면적을 미터로 계산하기 위함."""
+    if transformer is None:
+        return features
+    converted: list[dict[str, Any]] = []
+    for feature in features:
+        geometry = feature.get("geometry")
+        if not geometry:
+            continue
+        coordinates = _map_coordinates(geometry.get("coordinates"), transformer)
+        if coordinates is None:
+            continue
+        converted.append(
+            {
+                "type": "Feature",
+                "properties": feature.get("properties") or {},
+                "geometry": {"type": geometry.get("type"), "coordinates": coordinates},
+            }
+        )
+    return converted
+
+
+def _map_coordinates(coordinates: Any, transformer: Transformer) -> Any:
+    """중첩 좌표 배열을 재귀적으로 변환한다."""
+    if not isinstance(coordinates, list) or not coordinates:
+        return None
+    first = coordinates[0]
+    if isinstance(first, (int, float)):
+        x, y = transformer.transform(float(coordinates[0]), float(coordinates[1]))
+        return [x, y]
+    mapped = [_map_coordinates(item, transformer) for item in coordinates]
+    return [item for item in mapped if item is not None]
+
+
+def _candidate_payload(
+    candidate: StructureCandidate,
+    to_lonlat: Any,
+) -> dict[str, Any]:
+    lon, lat = to_lonlat(candidate.x, candidate.y)
+    return {
+        "chainage_m": round(candidate.chainage_m, 2),
+        "x": candidate.x,
+        "y": candidate.y,
+        "lon": lon,
+        "lat": lat,
+        "reason": candidate.reason,
+        "stream_name": candidate.stream_name,
+    }
+
+
+async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
+    """노선 정점·도엽 피처·좌표 변환기를 한 번에 준비한다."""
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        stored_path = await get_project_storage_relative_path(connection, project_id)
+        route = await get_latest_route(connection, project_id)
+        if not route:
+            return JSONResponse(
+                status_code=404,
+                content={"status": "error", "message": "확정된 경로가 없습니다."},
+            )
+        points = await get_route_points(connection, int(route["id"]))
+        epsg = await get_surface_crs_epsg(connection, project_id)
+
+    vertices = build_route_vertices(points)
+    if len(vertices) < 2:
+        return JSONResponse(
+            status_code=400,
+            content={"status": "error", "message": "노선 좌표가 부족합니다."},
+        )
+
+    source_crs = epsg or "EPSG:5186"
+    to_lonlat_transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True)
+    to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True)
+
+    directory = _sheet_dir(stored_path)
+    streams = _reproject_features(_load_features(directory, _STREAM_FILE), to_metric_transformer)
+    contour_features = _reproject_features(
+        _load_features(directory, _CONTOUR_FILE), to_metric_transformer
+    )
+    return {
+        "route_id": int(route["id"]),
+        "vertices": vertices,
+        "streams": streams,
+        "contours": contour_features,
+        "to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y),
+    }
+
+
+@router.get("/{project_id}/drainage/candidates", response_model=None)
+async def get_structure_candidates(project_id: UUID) -> dict[str, Any] | JSONResponse:
+    """관 매설 구조물 측점 후보를 제안한다(세류 교차 + 300m 보충, 성토부 제외)."""
+    prepared = await _prepare(project_id)
+    if isinstance(prepared, JSONResponse):
+        return prepared
+    candidates = propose_structure_stations(prepared["vertices"], prepared["streams"])
+    to_lonlat = prepared["to_lonlat"]
+    return {
+        "status": "success",
+        "project_id": str(project_id),
+        "route_id": prepared["route_id"],
+        "candidates": [_candidate_payload(candidate, to_lonlat) for candidate in candidates],
+    }
+
+
+@router.post("/{project_id}/drainage/basins", response_model=None)
+async def post_drainage_basins(
+    project_id: UUID,
+    payload: dict[str, Any] | None = None,
+) -> dict[str, Any] | JSONResponse:
+    """확정된 구조물 측점별 배수유역을 산정한다.
+
+    payload에 `chainages`(누가거리 목록)를 주면 그 위치로 확정하고, 없으면 자동 제안분을 쓴다.
+    """
+    prepared = await _prepare(project_id)
+    if isinstance(prepared, JSONResponse):
+        return prepared
+    vertices = prepared["vertices"]
+    chainages = (payload or {}).get("chainages")
+    if isinstance(chainages, list) and chainages:
+        candidates = _candidates_from_chainages(vertices, chainages)
+    else:
+        candidates = propose_structure_stations(vertices, prepared["streams"])
+
+    contours = ContourField(prepared["contours"], _ELEVATION_KEYS)
+    basins = build_basins(
+        vertices, candidates, contours, prepared["streams"], prepared["to_lonlat"]
+    )
+    return {
+        "status": "success",
+        "project_id": str(project_id),
+        "route_id": prepared["route_id"],
+        "basins": [
+            {
+                "index": basin.index,
+                "chainage_m": round(basin.chainage_m, 2),
+                "polygon_lonlat": basin.polygon_lonlat,
+                "area_m2": round(basin.area_m2, 1),
+                "relief_m": round(basin.relief_m, 2),
+                "flow_length_m": round(basin.flow_length_m, 1),
+                # 관경 수식 미확정 — 산정 함수가 None을 돌려주면 프론트가 "미정"으로 표기한다.
+                "pipe_diameter_mm": basin.pipe_diameter_mm,
+            }
+            for basin in basins
+        ],
+    }
+
+
+def _candidates_from_chainages(vertices: Any, chainages: list[Any]) -> list[StructureCandidate]:
+    """사용자가 확정한 누가거리 목록을 후보 구조로 되돌린다."""
+    from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import _interpolate_vertex
+
+    candidates: list[StructureCandidate] = []
+    for value in chainages:
+        try:
+            chainage = float(value)
+        except (TypeError, ValueError):
+            continue
+        x, y, _ = _interpolate_vertex(vertices, chainage)
+        candidates.append(StructureCandidate(chainage_m=chainage, x=x, y=y, reason="confirmed"))
+    return candidates

B05_wf2_Route/B05_wf2_Route_UI_Drainage_Panel.ts

@@ -8,15 +8,21 @@ import {
 import {
   computeMapRect,
   createNormalizer,
+  drawFilledRing,
   drawPreparedLayer,
   prepareLayer,
   prepareMetricPolyline,
   type GeoJsonCollection,
   type MapRect,
+  type Normalizer,
   type PreparedLayer,
   type ViewState,
 } from "../B04_wf1_Surface/B04_wf1_Surface_UI_MapRender";
-import type { RoutePoint } from "./B05_wf2_Route_Api_Fetch";
+import {
+  fetchDrainageBasins,
+  type DrainageBasin,
+  type RoutePoint,
+} from "./B05_wf2_Route_Api_Fetch";
 
 // 배수유역도 패널 — 하단 종단 패널 안쪽 우측에 도킹되는 2단 사이드 패널.
 // 하단 패널이 닫히면 이 패널도 함께 화면에서 사라진다(부모 안에 들어 있으므로 자동).
@@ -42,6 +48,18 @@ const LAYER_LABELS: Record<DrainageLayer, string> = {
 const ROUTE_COLOR = "#f97316";
 const COLLAPSED_KEY = "b05-route-drainage-collapsed";
 
+/** 유역 오버레이 파스텔 색상. 번호 순으로 돌려쓴다(사용자 지시: 파스텔톤). */
+const BASIN_COLORS = [
+  "rgba(167, 216, 199, 0.45)",
+  "rgba(247, 208, 168, 0.45)",
+  "rgba(186, 199, 240, 0.45)",
+  "rgba(241, 183, 199, 0.45)",
+  "rgba(214, 226, 168, 0.45)",
+  "rgba(202, 186, 227, 0.45)",
+  "rgba(168, 214, 232, 0.45)",
+  "rgba(240, 219, 168, 0.45)",
+] as const;
+
 export interface DrainagePanel {
   root: HTMLElement;
   /** 프로젝트가 정해지면 배경지도·도엽 레이어를 불러온다. */
@@ -64,6 +82,13 @@ export function createDrainagePanel(): DrainagePanel {
   layerButtons.className = "b05-drainage__layers";
   header.append(title, layerButtons);
 
+  // 유역 산정 실행 버튼 — 후보 제안·유역 산정을 한 번에 돌린다(자동 제안 + 사용자 확인 흐름).
+  const analyzeButton = document.createElement("button");
+  analyzeButton.type = "button";
+  analyzeButton.className = "b05-drainage__analyze";
+  analyzeButton.textContent = "유역 산정";
+  header.append(analyzeButton);
+
   const viewport = document.createElement("div");
   viewport.className = "b05-drainage__viewport";
   const backgroundImage = document.createElement("img");
@@ -76,7 +101,11 @@ export function createDrainagePanel(): DrainagePanel {
   status.className = "b05-drainage__status";
   status.textContent = "노선을 확정하면 배수유역 배경도가 표시됩니다.";
   viewport.append(backgroundImage, canvas, status);
-  root.append(panelHandle.root, header, viewport);
+  // 유역 제원 목록(면적·표고·유하거리·관경). 관경 수식 미확정이라 당분간 "미정"으로 나온다.
+  const basinList = document.createElement("div");
+  basinList.className = "b05-drainage__basins";
+  basinList.hidden = true;
+  root.append(panelHandle.root, header, viewport, basinList);
 
   let projectId: string | null = null;
   let meta: VWorldMeta | null = null;
@@ -84,6 +113,9 @@ export function createDrainagePanel(): DrainagePanel {
   const activeLayers = new Set<DrainageLayer>(DRAINAGE_LAYERS);
   let routeLayer: PreparedLayer | null = null;
   let routePoints: ReadonlyArray<RoutePoint> = [];
+  let normalizer: Normalizer | null = null;
+  let basins: DrainageBasin[] = [];
+  let selectedBasin: number | null = null;
   let scale = 1;
   let offsetX = 0;
   let offsetY = 0;
@@ -136,7 +168,22 @@ export function createDrainagePanel(): DrainagePanel {
     context.clearRect(0, 0, width, height);
     const mapRect: MapRect = computeMapRect(meta, width, height);
     const view: ViewState = { width, height, scale, offsetX, offsetY, mapRect };
-    // 등고선을 가장 아래에 얇게 깔고 세류·표고점을 그 위에, 노선을 맨 위에 둔다.
+    // 유역 채움을 가장 아래에 깔아 등고선·세류 판독을 가리지 않게 한다.
+    if (normalizer) {
+      basins.forEach((basin) => {
+        const color = BASIN_COLORS[(basin.index - 1) % BASIN_COLORS.length];
+        drawFilledRing(
+          context,
+          { ring: basin.polygon_lonlat, label: String(basin.index) },
+          normalizer!,
+          view,
+          selectedBasin === null || selectedBasin === basin.index
+            ? color
+            : color.replace(/0\.45\)$/, "0.18)"),
+        );
+      });
+    }
+    // 등고선을 얇게 깔고 세류·표고점을 그 위에, 노선을 맨 위에 둔다.
     DRAINAGE_LAYERS.forEach((layer) => {
       if (!activeLayers.has(layer)) return;
       const prepared = preparedLayers.get(layer);
@@ -161,6 +208,65 @@ export function createDrainagePanel(): DrainagePanel {
     });
   }
 
+  /** 유역 제원 목록을 다시 그린다. 항목을 누르면 해당 유역만 진하게 강조한다. */
+  function renderBasinList(): void {
+    basinList.textContent = "";
+    basinList.hidden = basins.length === 0;
+    basins.forEach((basin) => {
+      const row = document.createElement("button");
+      row.type = "button";
+      row.className = "b05-drainage__basin" + (selectedBasin === basin.index ? " is-selected" : "");
+      const badge = document.createElement("span");
+      badge.className = "b05-drainage__basin-index";
+      badge.textContent = String(basin.index);
+      badge.style.background = BASIN_COLORS[(basin.index - 1) % BASIN_COLORS.length];
+      const metrics = document.createElement("span");
+      metrics.className = "b05-drainage__basin-metrics";
+      // 관경은 수식 미확정이라 백엔드가 null을 주며, 확정 전까지 "미정"으로 표기한다.
+      const pipe =
+        basin.pipe_diameter_mm === null ? "미정" : `Ø${Math.round(basin.pipe_diameter_mm)}mm`;
+      metrics.textContent =
+        `면적 ${formatArea(basin.area_m2)} · 표고 ${basin.relief_m.toFixed(1)}m · ` +
+        `유하 ${Math.round(basin.flow_length_m)}m · 관경 ${pipe}`;
+      row.title = `측점 누가거리 ${basin.chainage_m.toFixed(1)}m`;
+      row.append(badge, metrics);
+      row.addEventListener("click", () => {
+        selectedBasin = selectedBasin === basin.index ? null : basin.index;
+        renderBasinList();
+        scheduleDraw();
+      });
+      basinList.append(row);
+    });
+  }
+
+  function formatArea(areaM2: number): string {
+    return areaM2 >= 10000 ? `${(areaM2 / 10000).toFixed(2)}ha` : `${Math.round(areaM2)}㎡`;
+  }
+
+  /** 구조물 측점 후보 제안 + 유역 산정을 백엔드에 요청한다(계산은 전부 백엔드). */
+  async function analyze(): Promise<void> {
+    if (!projectId) return;
+    analyzeButton.disabled = true;
+    status.hidden = false;
+    status.textContent = "배수유역을 산정하는 중…";
+    try {
+      const response = await fetchDrainageBasins(projectId);
+      basins = response.basins;
+      selectedBasin = null;
+      renderBasinList();
+      status.hidden = basins.length > 0;
+      if (basins.length === 0) status.textContent = "산정된 배수유역이 없습니다.";
+      scheduleDraw();
+    } catch (error) {
+      status.hidden = false;
+      status.textContent = error instanceof Error ? error.message : "배수유역 산정에 실패했습니다.";
+    } finally {
+      analyzeButton.disabled = false;
+    }
+  }
+
+  analyzeButton.addEventListener("click", () => void analyze());
+
   /** 노선 전체가 보이도록 배율·중심을 맞춘다. 노선이 없으면 도엽 전체를 그대로 보여준다. */
   function fitToRoute(): void {
     scale = 1;
@@ -216,12 +322,12 @@ export function createDrainagePanel(): DrainagePanel {
       );
       if (sequence !== loadSequence) return;
       meta = nextMeta;
-      const normalizer = createNormalizer(nextMeta);
+      normalizer = createNormalizer(nextMeta);
       let featureCount = 0;
       loaded.forEach(([layer, data]) => {
         if (!data) return;
         featureCount += data.features?.length ?? 0;
-        preparedLayers.set(layer, prepareLayer(data, normalizer));
+        preparedLayers.set(layer, prepareLayer(data, normalizer!));
       });
       backgroundImage.src = `${getVWorldMapUrl(activeProjectId, "satellite")}&_t=${Date.now()}`;
       if (routePoints.length > 1) routeLayer = prepareMetricPolyline(routePoints, nextMeta);

B05_wf2_Route/B05_wf2_Route_UI_Style.css

@@ -930,3 +930,74 @@
   font-size: var(--text-caption);
   pointer-events: none;
 }
+
+.b05-drainage__analyze {
+  margin-left: auto;
+  padding: 2px var(--spacing-8);
+  border: 1px solid
+    color-mix(in srgb, var(--color-royal-amethyst, rgb(109 40 217)) 55%, transparent);
+  border-radius: var(--radius-inputs);
+  background: var(--color-surface);
+  color: var(--color-text-body);
+  font-size: var(--text-caption);
+  cursor: pointer;
+}
+
+.b05-drainage__analyze:disabled {
+  opacity: 0.45;
+  cursor: default;
+}
+
+/* 유역 제원 목록 — 면적·유역표고·유하거리·관경(수식 확정 전까지 "미정"). */
+.b05-drainage__basins {
+  display: flex;
+  max-height: 34%;
+  flex: 0 0 auto;
+  flex-direction: column;
+  gap: 2px;
+  overflow-y: auto;
+  padding: var(--spacing-8);
+  border-top: 1px solid var(--color-border);
+}
+
+.b05-drainage__basin {
+  display: flex;
+  align-items: center;
+  gap: var(--spacing-8);
+  padding: var(--spacing-4) var(--spacing-8);
+  border: 1px solid transparent;
+  border-radius: var(--radius-inputs);
+  background: none;
+  color: var(--color-text-body);
+  font-size: var(--text-caption);
+  text-align: left;
+  cursor: pointer;
+}
+
+.b05-drainage__basin:hover,
+.b05-drainage__basin.is-selected {
+  border-color: var(--color-border);
+  background: var(--color-surface);
+}
+
+/* 지도 위 서클 번호와 같은 파스텔 색을 써서 목록 항목과 유역을 눈으로 잇는다. */
+.b05-drainage__basin-index {
+  display: inline-flex;
+  width: 20px;
+  height: 20px;
+  flex: 0 0 auto;
+  align-items: center;
+  justify-content: center;
+  border: 1px solid var(--color-border);
+  border-radius: 50%;
+  color: #1f2937;
+  font-size: 11px;
+  font-weight: 600;
+}
+
+.b05-drainage__basin-metrics {
+  min-width: 0;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}

main.py

@@ -32,6 +32,7 @@ from B04_wf1_Surface.B04_wf1_Surface_Router_Contour import router as b04_surface
 from B04_wf1_Surface.B04_wf1_Surface_Router_GIS import router as b04_surface_gis_router
 from B04_wf1_Surface.B04_wf1_Surface_Router_GIS import tiles_router
 from B05_wf2_Route.B05_wf2_Route_Router import router as b05_route_router
+from B05_wf2_Route.B05_wf2_Route_Router_Drainage import router as b05_drainage_router
 from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Router import router as b06_section_router
 from B07_wf4_DesignDetail.B07_wf4_DesignDetail_Router import router as b07_design_router
 from common_util.common_util_auth import require_company, verify_session
@@ -274,6 +275,7 @@ app.include_router(b04_surface_contour_router, dependencies=protected_with_compa
 app.include_router(b04_surface_gis_router, dependencies=protected_with_company)
 app.include_router(tiles_router, dependencies=protected_with_company)
 app.include_router(b05_route_router, dependencies=protected_with_company)
+app.include_router(b05_drainage_router, dependencies=protected_with_company)
 app.include_router(b06_section_router, dependencies=protected_with_company)
 app.include_router(b07_design_router, dependencies=protected_with_company)