feat(B05): 1차 영역 반경 100m + 노선 버퍼 복귀, 단계별 산출물 저장 모듈 분리

- DRAINAGE_INITIAL_RADIUS_M 50m -> 100m - 1차 영역 = 상류 세류망 buffer(100m) UNION 계획 노선 buffer(100m). 세류망 선정이 정확해져 노선 버퍼를 다시 넣어도 영역이 폭발하지 않는다. - 저장을 B05_wf2_Route_Engine_Watershed_Export.py 로 분리. STAGES 딕셔너리에 단계 이름을 추가하고 write_stage 를 부르면 drainage/{번호}_{단계}.geojson + manifest.json 이 함께 갱신된다. 앞으로 기능을 붙일 때마다 이 자리에 단계가 하나씩 쌓인다. - primary_region 단계에 route 레이어 추가(도로 대조용). - 라우터의 임시 GeoJSON 작성 코드 제거. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-31 17:53:43 +09:00
parent e1b410c391
commit 578ada6d84
4 changed files with 191 additions and 60 deletions
@@ -0,0 +1,146 @@
+"""배수유역 단계별 검증 산출물을 영구저장소에 남긴다.
+
+기능을 하나씩 붙일 때마다 그 단계의 결과를 파일로 남겨 사람이 QGIS 등으로 직접 열어
+대조할 수 있게 하는 것이 목적이다(2026-07-31 사용자 지시). 새 단계를 추가할 때는
+`STAGES`에 이름을 하나 더 넣고 `write_stage()`를 호출하면 된다 — 파일명 규칙과 매니페스트
+갱신은 여기서 일괄로 처리한다.
+
+저장 위치: `storage/{회사}/{사용자}/{프로젝트}/B05_wf2_Route/drainage/`
+  - `{단계번호}_{단계이름}.geojson` — WGS84 FeatureCollection, 피처마다 `kind` 속성
+  - `manifest.json` — 지금까지 남긴 단계 목록과 요약값
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Sequence
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable
+
+from shapely.geometry.base import BaseGeometry
+
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_system import DRAINAGE_CACHE_DIRNAME
+
+logger = logging.getLogger(__name__)
+
+# 단계 이름 → 파일 접두 번호. 순서대로 읽으면 파이프라인 진행 순서가 된다.
+STAGES: dict[str, str] = {
+    "primary_region": "01",
+}
+
+_MANIFEST_FILENAME = "manifest.json"
+LonLat = Callable[[float, float], tuple[float, float]]
+
+
+def drainage_dir(stored_path: str) -> Path:
+    return Path(resolve_stored_project_path(stored_path)) / "B05_wf2_Route" / DRAINAGE_CACHE_DIRNAME
+
+
+def write_stage(
+    stored_path: str,
+    stage: str,
+    layers: dict[str, Sequence[BaseGeometry]],
+    properties: dict[str, Any],
+    to_lonlat: LonLat,
+) -> str | None:
+    """한 단계의 기하 산출물을 GeoJSON으로 저장하고 매니페스트를 갱신한다.
+
+    `layers`는 {레이어이름: 사업지 CRS(m) 기하 목록}이며 피처 `kind` 속성이 된다.
+    좌표는 여기서 WGS84로 바꾼다 — 저장 파일은 어떤 도구로 열어도 바로 보여야 한다.
+    """
+    prefix = STAGES.get(stage)
+    if prefix is None:
+        logger.warning("배수유역: 등록되지 않은 저장 단계 '%s' — 저장을 건너뜁니다.", stage)
+        return None
+
+    features: list[dict[str, Any]] = []
+    counts: dict[str, int] = {}
+    for kind, geometries in layers.items():
+        for index, geometry in enumerate(geometries):
+            feature = _to_feature(kind, index, geometry, to_lonlat)
+            if feature is not None:
+                features.append(feature)
+        counts[kind] = len(geometries)
+
+    filename = f"{prefix}_{stage}.geojson"
+    directory = drainage_dir(stored_path)
+    target = directory / filename
+    document = {
+        "type": "FeatureCollection",
+        "crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}},
+        "properties": {**properties, "counts": counts},
+        "features": features,
+    }
+    try:
+        directory.mkdir(parents=True, exist_ok=True)
+        with target.open("w", encoding="utf-8") as file:
+            json.dump(document, file, ensure_ascii=False)
+    except OSError:
+        logger.warning("배수유역: %s 저장 실패 (%s)", stage, target)
+        return None
+
+    _update_manifest(directory, stage, filename, {**properties, "counts": counts})
+    logger.info("배수유역: %s 저장 — %s (피처 %d개)", stage, target, len(features))
+    return str(target)
+
+
+def _to_feature(
+    kind: str, index: int, geometry: BaseGeometry, to_lonlat: LonLat
+) -> dict[str, Any] | None:
+    coordinates = _to_lonlat_coords(geometry, to_lonlat)
+    if coordinates is None:
+        return None
+    return {
+        "type": "Feature",
+        "properties": {"kind": kind, "index": index},
+        "geometry": {"type": geometry.geom_type, "coordinates": coordinates},
+    }
+
+
+def _to_lonlat_coords(geometry: BaseGeometry, to_lonlat: LonLat) -> Any:
+    """shapely 기하를 WGS84 GeoJSON 좌표 배열로 바꾼다."""
+    if geometry.is_empty:
+        return None
+    kind = geometry.geom_type
+    if kind == "Point":
+        return list(to_lonlat(geometry.x, geometry.y))
+    if kind == "LineString":
+        return [list(to_lonlat(x, y)) for x, y in geometry.coords]
+    if kind == "Polygon":
+        return [
+            [list(to_lonlat(x, y)) for x, y in ring.coords]
+            for ring in (geometry.exterior, *geometry.interiors)
+        ]
+    if kind in {"MultiPoint", "MultiLineString", "MultiPolygon", "GeometryCollection"}:
+        parts = [_to_lonlat_coords(part, to_lonlat) for part in geometry.geoms]
+        return [part for part in parts if part is not None]
+    return None
+
+
+def _update_manifest(
+    directory: Path, stage: str, filename: str, properties: dict[str, Any]
+) -> None:
+    """지금까지 남긴 단계 목록을 한 파일에 모아 둔다 — 무엇이 저장돼 있는지 한눈에 본다."""
+    manifest_path = directory / _MANIFEST_FILENAME
+    manifest: dict[str, Any] = {}
+    if manifest_path.exists():
+        try:
+            with manifest_path.open("r", encoding="utf-8") as file:
+                loaded = json.load(file)
+            if isinstance(loaded, dict):
+                manifest = loaded
+        except (OSError, json.JSONDecodeError):
+            logger.warning("배수유역: manifest를 읽지 못해 새로 만듭니다 (%s).", manifest_path)
+    manifest[stage] = {
+        "file": filename,
+        "saved_at": datetime.now().isoformat(timespec="seconds"),
+        "properties": properties,
+    }
+    try:
+        with manifest_path.open("w", encoding="utf-8") as file:
+            json.dump(manifest, file, ensure_ascii=False, indent=2)
+    except OSError:
+        logger.warning("배수유역: manifest 저장 실패 (%s).", manifest_path)
@@ -459,19 +459,19 @@ def build_primary_region(
    radius_m: float,
    cell_m: float = DRAINAGE_GRID_SIZE_M,
 ) -> PrimaryRegion:
-    """**상류 세류망만** 반경 버퍼한 범위 = 1차 배수유역, 그 bbox = 해석 격자.
+    """**상류 세류망 + 계획 노선**을 반경 버퍼해 합친 범위 = 1차 배수유역, bbox = 해석 격자.

-    노선은 버퍼하지 않는다(2026-07-31 사용자 지시). 1차 영역의 기준은 도로 교차점 상류로
-    이어진 세류선 그 자체이며, 도로를 버퍼하면 도로 아래쪽(하류)까지 영역이 퍼져 의미가 없다.
+    노선 버퍼는 세류 교차가 없는 구간의 도로도 격자 안에 들어오게 한다 — 그래야 그 구간
+    사면이 유역으로 잡힌다. 상류 세류망 선정이 정확해진 뒤 다시 포함했다(2026-07-31 사용자 지시).

    노선이 이 영역 밖으로 나가는 길이는 따로 재서 남긴다 — 그 구간은 도로 셀이 격자에
    없어 유역이 잡히지 않으므로 반경을 올릴지 판단하는 근거가 된다.
    """
    split = split_streams_at_road(route_line, stream_features, cloud)
-    geometries = [line.buffer(radius_m) for line in split.upstream]
-    if not geometries:
-        logger.warning("배수유역: 상류 세류망이 없어 노선 버퍼로 대체합니다.")
-        geometries = [route_line.buffer(radius_m)]
+    geometries = [route_line.buffer(radius_m)]
+    geometries.extend(line.buffer(radius_m) for line in split.upstream)
+    if not split.upstream:
+        logger.warning("배수유역: 상류 세류망이 없어 노선 버퍼만으로 1차 영역을 잡습니다.")
    area = unary_union(geometries)
    x_min, y_min, x_max, y_max = area.bounds
    spec = _spec_from_bounds(x_min, y_min, x_max, y_max, cell_m)
@@ -15,6 +15,7 @@ from uuid import UUID
 from fastapi import APIRouter
 from fastapi.responses import JSONResponse
 from pyproj import Transformer
+from shapely.geometry import LineString, Polygon, box

 from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
 from B05_wf2_Route.B05_wf2_Route_Engine_Drainage import (
@@ -26,6 +27,7 @@ from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Basin import (
    build_drainage_watershed,
    preview_primary_region,
 )
+from B05_wf2_Route.B05_wf2_Route_Engine_Watershed_Export import write_stage
 from B05_wf2_Route.B05_wf2_Route_Repository import (
    get_latest_route,
    get_route_points,
@@ -33,11 +35,7 @@ from B05_wf2_Route.B05_wf2_Route_Repository import (
 )
 from common_util.common_util_storage import resolve_stored_project_path
 from config.config_db import get_db_pool
-from config.config_system import (
-    DRAINAGE_CACHE_DIRNAME,
-    DRAINAGE_CACHE_FILENAME,
-    DRAINAGE_REGION_FILENAME,
-)
+from config.config_system import DRAINAGE_CACHE_DIRNAME, DRAINAGE_CACHE_FILENAME

 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects", tags=["B05 Route Drainage"])
@@ -161,6 +159,7 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
    return {
        "route_id": int(route["id"]),
        "vertices": vertices,
+        "route_line": LineString([(vertex.x, vertex.y) for vertex in vertices]),
        "streams": streams,
        "contours": contour_features,
        "stored_path": stored_path,
@@ -234,54 +233,38 @@ async def get_primary_region(project_id: UUID) -> dict[str, Any] | JSONResponse:
            "bbox_lonlat": _grid_bbox_lonlat(spec, to_lonlat),
        },
    }
-    payload["saved_to"] = _save_region_geojson(prepared["stored_path"], payload)
+    # 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다.
+    payload["saved_to"] = write_stage(
+        prepared["stored_path"],
+        "primary_region",
+        {
+            "primary_region": _as_polygons(region.area),
+            "upstream": region.split.upstream,
+            "downstream": region.split.downstream,
+            "route": [prepared["route_line"]],
+            "grid_bbox": [_grid_bbox_polygon(spec)],
+        },
+        {
+            "radius_m": region.radius_m,
+            "road_outside_m": payload["road_outside_m"],
+            "no_contact_count": region.split.no_contact,
+            "grid": {key: value for key, value in payload["grid"].items() if key != "bbox_lonlat"},
+        },
+        to_lonlat,
+    )
    return payload


-def _save_region_geojson(stored_path: str, payload: dict[str, Any]) -> str | None:
-    """1차 영역 검증 산출물을 영구저장소에 GeoJSON(WGS84)으로 남긴다."""
-    features: list[dict[str, Any]] = []
-    for index, ring in enumerate(payload["region_rings"]):
-        features.append(_geojson_feature("primary_region", index, "Polygon", [ring]))
-    for index, line in enumerate(payload["upstream_lines"]):
-        features.append(_geojson_feature("upstream", index, "LineString", line))
-    for index, line in enumerate(payload["downstream_lines"]):
-        features.append(_geojson_feature("downstream", index, "LineString", line))
-    features.append(_geojson_feature("grid_bbox", 0, "Polygon", [payload["grid"]["bbox_lonlat"]]))
-    document = {
-        "type": "FeatureCollection",
-        "crs": {"type": "name", "properties": {"name": "urn:ogc:def:crs:OGC:1.3:CRS84"}},
-        "properties": {
-            "radius_m": payload["radius_m"],
-            "road_outside_m": payload["road_outside_m"],
-            "no_contact_count": payload["no_contact_count"],
-            "grid": {key: value for key, value in payload["grid"].items() if key != "bbox_lonlat"},
-        },
-        "features": features,
-    }
-    target = (
-        Path(resolve_stored_project_path(stored_path))
-        / "B05_wf2_Route"
-        / DRAINAGE_CACHE_DIRNAME
-        / DRAINAGE_REGION_FILENAME
-    )
-    try:
-        target.parent.mkdir(parents=True, exist_ok=True)
-        with target.open("w", encoding="utf-8") as file:
-            json.dump(document, file, ensure_ascii=False)
-    except OSError:
-        logger.warning("배수유역: 1차 영역 GeoJSON을 저장하지 못했습니다 (%s).", target)
-        return None
-    logger.info("배수유역: 1차 영역 GeoJSON 저장 — %s (피처 %d개)", target, len(features))
-    return str(target)
+def _as_polygons(geometry: Any) -> list[Any]:
+    if geometry is None or geometry.is_empty:
+        return []
+    return list(geometry.geoms) if geometry.geom_type == "MultiPolygon" else [geometry]


-def _geojson_feature(kind: str, index: int, geom_type: str, coordinates: Any) -> dict[str, Any]:
-    return {
-        "type": "Feature",
-        "properties": {"kind": kind, "index": index},
-        "geometry": {"type": geom_type, "coordinates": coordinates},
-    }
+def _grid_bbox_polygon(spec: Any) -> Polygon:
+    x_max = spec.x_min + spec.n_cols * spec.cell_m
+    y_min = spec.y_max - spec.n_rows * spec.cell_m
+    return box(spec.x_min, y_min, x_max, spec.y_max)


 def _line_lonlat(line: Any, to_lonlat: Any) -> list[list[float]]:
@@ -242,9 +242,11 @@ SKELETON_NODE_SPACING_M = float(os.getenv("SKELETON_NODE_SPACING_M", "10.0"))
 # ─────────────────────────────────────────────────────────────────────────
 # 해석 격자 한 변(m). 작을수록 정밀하나 셀 수가 제곱으로 늘어난다.
 DRAINAGE_GRID_SIZE_M = float(os.getenv("DRAINAGE_GRID_SIZE_M", "1.0"))
-# 1차 배수유역 반경(m). 도로 교차점 상류로 이어진 세류망을 이 반경으로 버퍼한 범위가
-# 1차 영역이며 그 bbox가 해석 격자다. 노선은 버퍼하지 않는다(2026-07-31 사용자 지시).
-DRAINAGE_INITIAL_RADIUS_M = float(os.getenv("DRAINAGE_INITIAL_RADIUS_M", "50.0"))
+# 1차 배수유역 반경(m). 도로 교차점 상류로 이어진 세류망과 계획 노선을 각각 이 반경으로
+# 버퍼해 합친 범위가 1차 영역이며, 그 bbox가 해석 격자다.
+# 노선 버퍼가 필요한 이유: 세류 교차가 없는 구간의 도로도 격자 안에 있어야 그 구간 사면이
+# 유역으로 잡힌다. 세류망 선정이 정확해진 뒤 다시 포함했다(2026-07-31 사용자 지시).
+DRAINAGE_INITIAL_RADIUS_M = float(os.getenv("DRAINAGE_INITIAL_RADIUS_M", "100.0"))
 # 활성 셀이 격자 최외곽에 닿았을 때 한 번에 넓히는 폭(m).
 DRAINAGE_EXPAND_STEP_M = float(os.getenv("DRAINAGE_EXPAND_STEP_M", "200.0"))
 # 확장 반복 상한. 경계 링이 전부 비활성이 되면 그 전에 스스로 멈춘다(안전핀).
@@ -278,8 +280,8 @@ DRAINAGE_MIN_BASIN_AREA_M2 = float(os.getenv("DRAINAGE_MIN_BASIN_AREA_M2", "100.
 # 격자 해석 결과 캐시 파일명. 프로젝트 저장소의 B05_wf2_Route/drainage/ 아래에 놓인다.
 DRAINAGE_CACHE_DIRNAME = "drainage"
 DRAINAGE_CACHE_FILENAME = "watershed_grid.npz"
-# 1차 영역 검증 산출물. 버튼을 누를 때마다 덮어써서 사람이 QGIS 등으로 직접 열어볼 수 있게 한다.
-DRAINAGE_REGION_FILENAME = "primary_region.geojson"
+# 단계별 검증 산출물은 같은 폴더에 `{번호}_{단계}.geojson` + `manifest.json`으로 쌓인다.
+# 파일명 규칙은 B05_wf2_Route_Engine_Watershed_Export.STAGES가 유일한 정의처다.


 # ─────────────────────────────────────────────────────────────────────────