refactor(B04/B05): 배수유역 분석을 B04로 이관, B05는 저장분 소비만

분석이 30초 걸리는데 B05는 일반 사용자 화면이다. 관리자 확인용 B04에서 한 번 돌려 저장하고, B05는 그 결과를 읽어 관 보충과 세부유역만 처리한다 (2026-07-31 사용자 지시). 노선 원천 변경 - B05 확정 경로 -> B03 업로드 계획 노선 파일(CSV). 분석이 노선 설계보다 먼저 끝나 있어야 하기 때문. 샘플 planned_route_sample_epsg5187.csv 로 검증. - common_util_route_geometry.py 신설 — RouteVertex/StructureCandidate/누가거리 보간/세류 교차점/계획 노선 CSV 리더. B04와 B05가 같은 표현을 쓰도록 공용화. 열 이름은 대소문자·한글 표기를 함께 받는다(B03이 여러 형식 수용 예정). B04 (관리자 확인용, 신규) - Engine_Watershed_{Grid,Stream,Descent,Flow,Expand,Export} — B05에서 git mv - Engine_Watershed_Analyze.py — 1~8단계 오케스트레이션 - Router_Watershed.py — GET /drainage/primary-region - UI_Watershed.ts — 2D 지도 GIS 레이어 그룹에 "배수유역" 토글 추가. 격자/화살표/세류망/1차영역/2차유역/기본관을 겹쳐 그린다. - 저장 위치 B05_wf2_Route/drainage -> B04_wf1_Surface/drainage - 03_road_routing 단계 추가: B05가 세부유역을 나눌 최소 배열(셀->도로셀 귀속, 유하장, 강도, 도로셀 제원, 셀 표고) + 계획도로선/기본배관/2차유역 기하 B05 (일반 사용자용, 축소) - Engine_Drainage_Basin.py — B04 산출물 로더 + 관 보충(9) + 측구 라우팅/세부유역(10,11) - Engine_Drainage.py 는 관경 산정만 남기고 322 -> 27줄 - Router_Drainage.py 509 -> 142줄. POST /drainage/basins 만 남김 - 화살표·격자·강도 띠 렌더 제거. 계획도로선/기본배관/2차유역만 받는다 삭제 - _legacy_watershed/ 4파일 (능선 행진 방식 원본 보관본) - Engine_Watershed_Basin.py (B04 Analyze + B05 Drainage_Basin 으로 분할) - GET /drainage/candidates 와 propose_structure_stations (구방식 후보 제안) E2E 검증 (실데이터) B04 분석 28.2s -> 저장(geojson 11KB + npz 2.6MB) B05 로드 + 세부 설계 0.11s <-- 30초가 0.1초로 면적 457,404m2 로 B04 2차 유역과 정확히 일치 관 편집 재산정 0.12s, 관 3개 -> 세부유역 3개, 면적 보존 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-31 20:57:09 +09:00
parent b0747e74b6
commit 8a0d640e0e
38 changed files with 2440 additions and 3560 deletions
@@ -0,0 +1,255 @@
+"""계획 노선 기하 공용 유틸 — 정점·누가거리·세류 교차점.
+
+배수유역 분석(B04)과 관 편집·세부유역(B05)이 같은 노선 표현을 써야 하므로 여기 한 곳에만
+정의한다. 어느 한쪽 페이지 폴더에 두면 반대 방향 import가 생긴다.
+
+노선 원천은 두 가지다.
+  · B03에 업로드된 **계획 노선 파일**(CSV) — 배수유역 분석의 입력
+  · DB `route_points` — B05에서 탐색·확정한 노선
+둘 다 같은 `RouteVertex` 목록으로 바꿔 아래 함수들이 그대로 받는다.
+"""
+
+from __future__ import annotations
+
+import csv
+import logging
+import math
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from shapely.geometry import LineString, Point, shape
+
+logger = logging.getLogger(__name__)
+
+# 계획 노선 CSV 열 이름 후보. B03이 여러 형식을 받게 되므로 흔한 표기를 모두 받아 준다.
+_X_KEYS = ("x", "X", "동", "easting", "EASTING")
+_Y_KEYS = ("y", "Y", "북", "northing", "NORTHING")
+_Z_KEYS = ("z", "Z", "표고", "elevation", "ELEV")
+_ORDER_KEYS = ("sequence", "order", "seq", "no", "번호")
+_EPSG_KEYS = ("crs_epsg", "epsg", "EPSG")
+
+
+@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"=최대 간격 규칙 보충, "confirmed"=사용자 확정
+    reason: str
+    stream_name: str | None = None
+
+
+@dataclass
+class PlannedRoute:
+    """계획 노선 파일에서 읽은 노선."""
+
+    vertices: list[RouteVertex]
+    epsg: int | None
+    name: str | None
+    source: Path
+
+    @property
+    def line(self) -> LineString:
+        return LineString([(vertex.x, vertex.y) for vertex in self.vertices])
+
+
+def read_planned_route_csv(path: Path) -> PlannedRoute | None:
+    """계획 노선 CSV를 읽어 정점 목록으로 바꾼다.
+
+    열 이름은 대소문자·한글 표기를 함께 받아 준다(B03이 여러 형식을 수용할 예정).
+    `sequence`가 있으면 그 순서로 정렬하고, 없으면 파일에 적힌 순서를 그대로 쓴다.
+    """
+    try:
+        with path.open("r", encoding="utf-8-sig", newline="") as file:
+            rows = list(csv.DictReader(file))
+    except (OSError, csv.Error, UnicodeDecodeError):
+        logger.warning("계획 노선 CSV를 읽지 못했습니다: %s", path)
+        return None
+    if not rows:
+        return None
+
+    epsg = _first_int(rows[0], _EPSG_KEYS)
+    name = _first_text(rows[0], ("route_name", "name", "노선명"))
+    parsed: list[tuple[float, float, float, float]] = []  # (정렬키, x, y, z)
+    for index, row in enumerate(rows):
+        x = _first_float(row, _X_KEYS)
+        y = _first_float(row, _Y_KEYS)
+        if x is None or y is None:
+            continue
+        order = _first_float(row, _ORDER_KEYS)
+        parsed.append(
+            (float(index) if order is None else order, x, y, _first_float(row, _Z_KEYS) or 0.0)
+        )
+    if len(parsed) < 2:
+        logger.warning("계획 노선 CSV에 좌표가 2점 미만입니다: %s", path)
+        return None
+
+    parsed.sort(key=lambda item: item[0])
+    vertices: list[RouteVertex] = []
+    cumulative = 0.0
+    previous: tuple[float, float] | None = None
+    for _, x, y, z in parsed:
+        if previous is not None:
+            cumulative += math.dist(previous, (x, y))
+        vertices.append(RouteVertex(x=x, y=y, z=z, chainage_m=cumulative))
+        previous = (x, y)
+    logger.info(
+        "계획 노선 %s: 정점 %d개, 연장 %.0fm, EPSG %s", path.name, len(vertices), cumulative, epsg
+    )
+    return PlannedRoute(vertices=vertices, epsg=epsg, name=name, source=path)
+
+
+def find_planned_route_file(input_dir: Path) -> Path | None:
+    """B03 입력 폴더에서 계획 노선 파일을 찾는다. 여러 개면 가장 최근 것."""
+    if not input_dir.exists():
+        return None
+    candidates = sorted(
+        input_dir.rglob("*.csv"), key=lambda item: item.stat().st_mtime, reverse=True
+    )
+    return candidates[0] if candidates else 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:
+    """해당 위치가 오르막(절토부)인지 종단 계획선의 국소 기울기 부호로 판정한다."""
+    _, _, 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 _first_text(row: dict[str, Any], keys: tuple[str, ...]) -> str | None:
+    for key in keys:
+        value = row.get(key)
+        if value not in (None, ""):
+            return str(value).strip()
+    return None
+
+
+def _first_float(row: dict[str, Any], keys: tuple[str, ...]) -> float | None:
+    text = _first_text(row, keys)
+    if text is None:
+        return None
+    try:
+        return float(text)
+    except ValueError:
+        return None
+
+
+def _first_int(row: dict[str, Any], keys: tuple[str, ...]) -> int | None:
+    value = _first_float(row, keys)
+    return None if value is None else int(value)