diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py new file mode 100644 index 00000000..21768ec8 --- /dev/null +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Export.py @@ -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) diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py index b8c2c2e3..7f30333e 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py @@ -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) diff --git a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py index acb8ab9c..1f2569e5 100644 --- a/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py +++ b/B05_wf2_Route/B05_wf2_Route_Router_Drainage.py @@ -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]]: diff --git a/config/config_system.py b/config/config_system.py index 29852717..53547279 100644 --- a/config/config_system.py +++ b/config/config_system.py @@ -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가 유일한 정의처다. # ─────────────────────────────────────────────────────────────────────────