"""배수유역 분석 API 라우터 (B04 — 관리자 확인용). 계획 노선(B03 업로드 CSV)과 도엽 등고선·세류선으로 배수유역을 끝까지 분석하고, 결과를 `storage/{프로젝트}/B04_PreProcess/drainage/`에 남긴다. 30초 안팎이 걸리므로 여기서 한 번만 돌리고, 일반 사용자가 쓰는 B05는 저장분을 읽어 쓴다(2026-07-31 사용자 지시). 좌표는 사업지 CRS(m)에서 계산하고 응답만 WGS84로 바꿔 내보낸다. """ import asyncio import hashlib import json import logging import pickle import time 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 B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir from B04_PreProcess.B04_PreProcess_Router_Watershed_Output import ( RESPONSE_SCHEMA_VERSION, _response_path, write_analysis, ) from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg from common_util.common_util_crs import resolve_project_crs from common_util.common_util_route_geometry import ( find_planned_route_file, load_design_route, read_planned_route, ) from common_util.common_util_storage import resolve_stored_project_path from common_util.common_util_surface_confirmation import get_surface_confirmation_params from common_util.common_util_wamis_rainfall import ( build_rainfall_table, ensure_contour_cache, ) from common_util.common_util_wamis_station import ( build_station_rainfall_table, is_jeju, ) from config.config_db import get_db_pool, run_with_connection from config.config_system import ( DRAINAGE_DESIGN_RETURN_PERIOD_YR, DRAINAGE_RAINFALL_FILENAME, DRAINAGE_RAINFALL_IDF_METHOD, DRAINAGE_RAINFALL_STATION_DIRNAME, WAMIS_CONTOUR_CACHE_DIR, WAMIS_STATION_RADIUS_M, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B04 Surface Watershed"]) # 도엽 레이어 파일명 (B04 전처리 산출물과 같은 위치) # B07 유역도가 같은 배경을 그리므로 파일명은 공개 상수로 둔다(사본 금지). CONTOUR_FILE = "도엽_등고선.geojson" STREAM_FILE = "도엽_하천중심선.geojson" def _sheet_dir(stored_path: str) -> Path: return Path(resolve_stored_project_path(stored_path)) / "B04_PreProcess" / "processed" def _route_input_dir(stored_path: str) -> Path: """B03 업로드 폴더 — 계획 노선 파일이 여기 들어온다.""" return Path(resolve_stored_project_path(stored_path)) / "B03_FileInput" / "input" 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] # 도엽을 사업지 좌표로 바꾼 저장본(347-1) — 원본 파일 자국 · 좌표계가 같으면 읽기만. SHEET_METRIC_FILENAME = "00_sheet_metric.pkl" # 메모리 보관(520-2) — (저장본 경로 · 지문) → (하천, 등고선) · 피클 33 MB 읽기(0.3~0.6 s)를 건너뜀. # 부르는 쪽은 피처를 읽기만 함(고치지 않음) · 최근 몇 벌만. _SHEETS: dict[str, tuple[list[dict[str, Any]], list[dict[str, Any]]]] = {} _SHEETS_SIZE = 2 def _keep_sheets(slot: str, value: tuple[list[dict[str, Any]], list[dict[str, Any]]]) -> None: _SHEETS.pop(slot, None) _SHEETS[slot] = value while len(_SHEETS) > _SHEETS_SIZE: _SHEETS.pop(next(iter(_SHEETS))) def _metric_sheets( directory: Path, cache_dir: Path, source_crs: str, transformer: Transformer ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], str]: """(하천, 등고선, 도엽 등고선 지문). 피처 4,200개 재투영(3.8 s)을 한 번만 한다.""" marks = [] for name in (STREAM_FILE, CONTOUR_FILE): try: info = (directory / name).stat() marks.append(f"{name}:{info.st_mtime_ns}:{info.st_size}") except OSError: marks.append(f"{name}:-") text = "|".join([*marks, str(source_crs)]) key = hashlib.sha1(text.encode(), usedforsecurity=False).hexdigest()[:16] path = cache_dir / SHEET_METRIC_FILENAME slot = f"{path}|{key}" kept = _SHEETS.get(slot) if kept is not None: _keep_sheets(slot, kept) return *kept, key try: with path.open("rb") as file: saved = pickle.load(file) if saved.get("key") == key: _keep_sheets(slot, (saved["streams"], saved["contours"])) return saved["streams"], saved["contours"], key except (OSError, pickle.PickleError, EOFError, AttributeError, KeyError, TypeError): pass streams = _reproject_features(_load_features(directory, STREAM_FILE), transformer) contours = _reproject_features(_load_features(directory, CONTOUR_FILE), transformer) _keep_sheets(slot, (streams, contours)) try: cache_dir.mkdir(parents=True, exist_ok=True) with path.open("wb") as file: pickle.dump({"key": key, "streams": streams, "contours": contours}, file) except OSError: logger.warning("배수유역: 도엽 좌표 저장본을 남기지 못했습니다 (%s).", path) return streams, contours, key async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse: """계획 노선 파일과 도엽 피처, 좌표 변환기를 준비한다. 노선은 **B03에 업로드된 계획 노선 파일**에서 읽는다 — B05의 확정 경로가 아니다. 배수유역 분석은 노선 설계보다 먼저 끝나 있어야 하기 때문이다(2026-07-31 사용자 지시). 다만 설계 계통과 **같은 노선**이어야 한다 — `load_design_route()`가 지표면 밖 구간을 잘라 프로젝트 좌표계로 돌려준다. 원본을 그대로 쓰면 유역·관이 확정 노선 밖에도 찍히고 좌표계마저 갈린다(2026-09-01 실측: 관은 5179, 노선은 5176이었다). """ # 셋은 서로 기다릴 이유가 없다 — DB 가 원격이라 순차로 내면 왕복 12ms 가 세 번 붙는다 # (2026-09-06 실측). 커넥션을 갈라 같이 보낸다. stored_path, epsg, surface_params = await asyncio.gather( run_with_connection(get_project_storage_relative_path, project_id), run_with_connection(get_surface_crs_epsg, project_id, 0), run_with_connection(get_surface_confirmation_params, str(project_id)), ) project_root = Path(resolve_stored_project_path(stored_path)) route_file = find_planned_route_file(_route_input_dir(stored_path)) if route_file is None: return JSONResponse( status_code=404, content={"status": "error", "message": "계획 노선 파일이 업로드되지 않았습니다."}, ) planned = await asyncio.to_thread(load_design_route, project_root, surface_params) if planned is None or len(planned.vertices) < 2: return JSONResponse( status_code=400, content={ "status": "error", "message": f"계획 노선을 읽지 못했거나 지표면과 겹치지 않습니다: {route_file.name}", }, ) # 설계 계통과 같은 프로젝트 좌표계로 맞춘다. 도엽 재투영도 이 좌표계로 간다. def sheets() -> tuple[Transformer, list[dict[str, Any]], list[dict[str, Any]], str]: source_crs = resolve_project_crs( project_root, route_crs_input=planned.crs_input, db_epsg=epsg ) to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True) return ( Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True), *_metric_sheets( _sheet_dir(stored_path), drainage_dir(stored_path), source_crs, to_metric_transformer, ), ) # 재투영(저장본 없으면 3.8 s) · 피클 읽기는 스레드로 — 이벤트 루프를 안 막음(505-1) to_lonlat_transformer, streams, contour_features, contour_key = await asyncio.to_thread(sheets) return { "route_source": route_file.name, "vertices": planned.vertices, "route_line": planned.line, "streams": streams, "contours": contour_features, "contour_key": contour_key, "stored_path": stored_path, "to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y), } def _load_saved_response(stored_path: str) -> dict[str, Any] | None: path = _response_path(stored_path) if not path.exists(): return None try: with path.open("r", encoding="utf-8") as file: saved = json.load(file) except (OSError, json.JSONDecodeError): logger.warning("배수유역: 저장된 분석 응답을 읽지 못했습니다 (%s).", path) return None # 응답 형식이 바뀌면 옛 저장분을 그대로 주면 안 된다 — 화면이 없는 항목을 그리려다 # 조용히 어긋난다(예: 강도 곡선 간격 5m→1m, 유입 집중점 신설). 다시 계산하게 둔다. if saved.get("schema_version") != RESPONSE_SCHEMA_VERSION: logger.info("배수유역: 저장분이 옛 형식이라 다시 분석합니다 (%s).", path) return None return saved def _rainfall_path(stored_path: str) -> Path: return drainage_dir(stored_path) / DRAINAGE_RAINFALL_FILENAME def _route_center_lonlat(stored_path: str, fallback_epsg: int | None) -> tuple[float, float] | None: """계획 노선 중간점(WGS84). 강우량 내삽 기준 좌표 — 유역 규모 대비 등우선 간격이 훨씬 넓어 노선 대표점 하나로 고정한다(설계기준.md 4절, 2026-08-05 협의).""" route_file = find_planned_route_file(_route_input_dir(stored_path)) if route_file is None: return None planned = read_planned_route(route_file) if planned is None or not planned.vertices: return None middle = planned.vertices[len(planned.vertices) // 2] # 여기 좌표는 **원본 파일 그대로**(트림 전)라 파일이 밝힌 좌표계로 읽는다 — 창구 사다리. source_crs = resolve_project_crs( Path(resolve_stored_project_path(stored_path)), route_crs_input=planned.crs_input, file_label_epsg=planned.epsg, db_epsg=fallback_epsg, ) lon, lat = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True).transform( middle.x, middle.y ) return lat, lon def _build_rainfall_sync(lat: float, lon: float, stored_path: str) -> dict[str, Any]: """계획노선 좌표로 지역을 분별해 강우량표를 만든다 (PLAN.md A, 2026-08-13). 제주: 등우선 내삽(서버 원본이 제주만 커버) / 본토: 관측소 방식(최근접+10km 최대값). 본토에서 관측소 수급이 실패하면 그대로 예외를 올린다 — 제주 등우선으로 폴백하면 "조용히 틀린 값"이 재발하므로 금지. """ if is_jeju(lat, lon): cached, failures = ensure_contour_cache(WAMIS_CONTOUR_CACHE_DIR) table = build_rainfall_table( lat, lon, WAMIS_CONTOUR_CACHE_DIR, design_return_period=DRAINAGE_DESIGN_RETURN_PERIOD_YR, ) table["contour_cache_files"] = cached if failures: table["failures"] = (table.get("failures") or []) + failures table["region_mode"] = "jeju_contour" return table table = build_station_rainfall_table( lat, lon, drainage_dir(stored_path) / DRAINAGE_RAINFALL_STATION_DIRNAME, design_return_period=DRAINAGE_DESIGN_RETURN_PERIOD_YR, radius_m=WAMIS_STATION_RADIUS_M, idf_method=DRAINAGE_RAINFALL_IDF_METHOD, ) table["region_mode"] = "mainland_station" return table async def _ensure_rainfall_table(stored_path: str, fallback_epsg: int | None) -> None: """rainfall_table.json이 없으면 백그라운드로 만든다. 실패는 비치명(로그만).""" path = _rainfall_path(stored_path) if path.exists(): return center = _route_center_lonlat(stored_path, fallback_epsg) if center is None: return try: table = await asyncio.to_thread(_build_rainfall_sync, *center, stored_path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8") logger.info("확률강우량표 저장: %s (값 %d개)", path, len(table.get("values") or [])) except Exception: # noqa: BLE001 — 외부망 차단 등. 배수유역 해석은 계속되어야 한다. logger.warning("확률강우량표 생성 실패 (%s)", stored_path, exc_info=True) @router.get("/{project_id}/drainage/rainfall", response_model=None) async def get_drainage_rainfall( project_id: UUID, refresh: bool = False ) -> dict[str, Any] | JSONResponse: """프로젝트 지점의 확률강우량표(설계빈도 IDF 적합계수 포함)를 돌려준다. 저장분이 있으면 그대로 주고, 없으면 즉석 생성한다(최초 1회는 등우선 96개 다운로드로 수십 초 걸릴 수 있다). B05 세션 캐시가 이 응답을 물고 다닌다. """ pool = get_db_pool() async with pool.acquire() as connection: stored_path = await get_project_storage_relative_path(connection, project_id) epsg = await get_surface_crs_epsg(connection, project_id, 0) path = _rainfall_path(stored_path) if not refresh and path.exists(): try: return {**json.loads(path.read_text(encoding="utf-8")), "from_cache": True} except (OSError, json.JSONDecodeError): logger.warning("강우량표 저장분을 읽지 못해 다시 만듭니다 (%s).", path) center = _route_center_lonlat(stored_path, epsg) if center is None: return JSONResponse( status_code=404, content={"status": "error", "message": "계획 노선 파일이 업로드되지 않았습니다."}, ) try: table = await asyncio.to_thread(_build_rainfall_sync, *center, stored_path) except Exception as exc: # noqa: BLE001 return JSONResponse( status_code=502, content={ "status": "error", "message": f"확률강우량 자료를 받지 못했습니다: {exc}", }, ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8") return {**table, "from_cache": False} @router.get("/{project_id}/drainage/primary-region", response_model=None) async def get_primary_region( project_id: UUID, refresh: bool = False ) -> dict[str, Any] | JSONResponse: """배수유역 분석 결과를 돌려준다. 기본은 **영구저장소에 남은 결과를 그대로** 준다 — 분석이 30초 걸리므로 화면을 열 때마다 다시 돌릴 이유가 없다. `refresh=true`면 처음부터 다시 계산하고 덮어쓴다 (2026-07-31 사용자 지시). """ pool = get_db_pool() async with pool.acquire() as connection: stored_path = await get_project_storage_relative_path(connection, project_id) epsg = await get_surface_crs_epsg(connection, project_id, 0) # 확률강우량표가 없으면 백그라운드로 만들어 둔다 — 유효직경 계산(B05)이 이 파일을 쓴다. asyncio.create_task(_ensure_rainfall_table(stored_path, epsg)) if not refresh: saved = _load_saved_response(stored_path) if saved is not None: logger.info("배수유역: 저장된 분석 결과를 그대로 돌려줍니다 (%s).", stored_path) return {**saved, "from_cache": True} from B03_FileInput.B03_FileInput_Service_Chain import _log_steps marks = [("시작", time.perf_counter())] prepared = await _prepare(project_id) if isinstance(prepared, JSONResponse): return prepared marks.append(("_prepare(도엽 읽기·좌표변환)", time.perf_counter())) preview = await asyncio.to_thread( preview_stages, prepared["vertices"], prepared["contours"], prepared["streams"], # 흐름 방향 저장본(347-1) — 노선이 저장 범위 안이면 읽기만 (drainage_dir(prepared["stored_path"]), prepared["contour_key"]), ) if preview is None: return JSONResponse( status_code=400, content={"status": "error", "message": "1차 배수유역을 정할 등고선이 없습니다."}, ) marks.append(("preview_stages(유역 산정)", time.perf_counter())) payload = write_analysis(project_id, prepared, preview, prepared["vertices"]) marks.append(("응답 만들기·저장", time.perf_counter())) _log_steps("배수유역 분석 내부", marks) return payload # 「지금 노선으로 셈」(347-2) — 같은 router 에 붙임(그 파일이 이 router 를 읽음). from B04_PreProcess import B04_PreProcess_Router_Watershed_Live # noqa: E402, F401