Files
Aislo/B04_PreProcess/B04_PreProcess_Router_Watershed.py
T
eomsangdonandClaude Opus 5 48e03e9d8a perf(B04): 유역 준비의 순차 질의 3건을 함께 보냄 + 헬퍼 한 곳으로
DB 가 원격이라 질의 하나가 곧 왕복 12ms 임. 서로 기다릴 이유가 없는 읽기를
한 커넥션에서 순차로 내면 그 왕복이 그대로 더해짐.

- config_db.run_with_connection: 저장소 함수를 자기 커넥션으로 돌려 gather 로
  묶을 수 있게 하는 공용 헬퍼. 정의를 한 곳에 둠(drainage_context 의 _query 는
  이 함수를 가리키는 이름으로 정리).
  머리에 경고 적음 — 순서가 필요한 쓰기는 이걸로 묶으면 트랜잭션이 깨짐.
- B04_PreProcess_Router_Watershed._prepare: 저장경로·좌표계·지표면확정값
  세 건을 gather 로. 약 24ms 절약.

common_util_auth_repository.py:318(decide_join_request)은 FOR UPDATE + 순서 있는
UPDATE 라 묶지 않음 — 읽기 블록만 대상.

자체검증 — 관 드래그 왕복 305 -> 259ms(최소 193), 관 11 · 유역 11 로 값 동일.
전체 484 통과.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-06 21:57:10 +09:00

521 lines
23 KiB
Python

"""배수유역 분석 API 라우터 (B04 — 관리자 확인용).
계획 노선(B03 업로드 CSV)과 도엽 등고선·세류선으로 배수유역을 끝까지 분석하고, 결과를
`storage/{프로젝트}/B04_PreProcess/drainage/`에 남긴다. 30초 안팎이 걸리므로 여기서 한 번만
돌리고, 일반 사용자가 쓰는 B05는 저장분을 읽어 쓴다(2026-07-31 사용자 지시).
좌표는 사업지 CRS(m)에서 계산하고 응답만 WGS84로 바꿔 내보낸다.
"""
import asyncio
import json
import logging
import math
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 shapely.geometry import Point
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,
write_stage,
)
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
mask_row_spans,
)
from B04_PreProcess.B04_PreProcess_Router_Watershed_Output import (
_as_polygons,
_boundary_geometry,
_flow_payload,
_grid_bbox_lonlat,
_grid_bbox_polygon,
_line_lonlat,
_polygon_rings,
_write_road_routing,
_write_stage_arrays,
)
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 (
StructureCandidate,
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_ARROW_SPACING_M,
DRAINAGE_DESIGN_RETURN_PERIOD_YR,
DRAINAGE_RAINFALL_FILENAME,
DRAINAGE_RAINFALL_IDF_METHOD,
DRAINAGE_RAINFALL_STATION_DIRNAME,
DRAINAGE_RESPONSE_FILENAME,
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"
# 응답 형식 판(版). 응답에 항목을 더하거나 값의 의미를 바꾸면 이 값을 올린다 —
# 저장해 둔 옛 응답을 그대로 돌려주면 화면이 조용히 어긋나기 때문이다.
# 2 = 강도 곡선 간격 1m(구간 합) + 유입 집중점(inflow_hotspots) 추가 (2026-08-01)
# 3 = 강도 곡선 구간 기준을 round→floor로 바로잡고(2m 주기 빗살 제거),
# 유입 집중점이 기본 관 옆 최소간격 안쪽을 피하도록 수정 (2026-08-01)
RESPONSE_SCHEMA_VERSION = 3
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]
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:
"""계획 노선 파일과 도엽 피처, 좌표 변환기를 준비한다.
노선은 **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}",
},
)
# 설계 계통과 같은 프로젝트 좌표계로 맞춘다. 도엽 재투영도 이 좌표계로 간다.
source_crs = resolve_project_crs(project_root, route_crs_input=planned.crs_input, db_epsg=epsg)
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_source": route_file.name,
"vertices": planned.vertices,
"route_line": planned.line,
"streams": streams,
"contours": contour_features,
"stored_path": stored_path,
"to_lonlat": lambda x, y: to_lonlat_transformer.transform(x, y),
}
def _response_path(stored_path: str) -> Path:
"""분석 응답 캐시 경로. 재산정하지 않는 한 이 파일을 그대로 돌려준다."""
return drainage_dir(stored_path) / DRAINAGE_RESPONSE_FILENAME
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 _save_response(stored_path: str, payload: dict[str, Any]) -> None:
path = _response_path(stored_path)
try:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as file:
json.dump(payload, file, ensure_ascii=False)
except OSError:
logger.warning("배수유역: 분석 응답을 저장하지 못했습니다 (%s).", path)
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}
prepared = await _prepare(project_id)
if isinstance(prepared, JSONResponse):
return prepared
preview = await asyncio.to_thread(
preview_stages,
prepared["vertices"],
prepared["contours"],
prepared["streams"],
)
if preview is None:
return JSONResponse(
status_code=400,
content={"status": "error", "message": "1차 배수유역을 정할 등고선이 없습니다."},
)
region = preview.region
to_lonlat = prepared["to_lonlat"]
# 확장을 거치면 격자·해석 영역이 1차 영역보다 커진다. 최종본을 써야 화면과 어긋나지 않는다.
spec = preview.spec or region.spec
domain = preview.domain if preview.domain is not None else region.cell_mask
payload = {
"status": "success",
"schema_version": RESPONSE_SCHEMA_VERSION,
"project_id": str(project_id),
"route_source": prepared["route_source"],
"radius_m": region.radius_m,
# 채택된 상류 세류망 = 1차 영역의 기준선.
"upstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.upstream],
# 도로 아래로 이어진 하류망 — 판정이 맞는지 눈으로 대조하기 위해 함께 준다.
"downstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.downstream],
"no_contact_count": region.split.no_contact,
# 노선이 1차 영역 밖으로 나간 길이(m). 크면 반경을 올려야 한다는 신호.
"road_outside_m": round(region.road_outside_m, 1),
# 1차 영역(버퍼 합집합) 외곽 링 목록.
"region_rings": _polygon_rings(region.area, to_lonlat),
"grid": {
"cell_m": spec.cell_m,
"rows": spec.n_rows,
"cols": spec.n_cols,
# bbox 전체 셀 수와, 해석 영역에 실제로 생성된 셀 수(확장 반영).
"bbox_cells": spec.size,
"cells": int(domain.sum()) if domain is not None else 0,
"width_m": round(spec.n_cols * spec.cell_m, 1),
"height_m": round(spec.n_rows * spec.cell_m, 1),
# 격자 bbox 링. 프론트는 이 사각형을 rows×cols로 나눠 행·열 좌표를 얻는다.
"bbox_lonlat": _grid_bbox_lonlat(spec, to_lonlat),
# 실제 생성된 셀을 행별 연속 구간 [행, 시작열, 끝열]으로 압축해 보낸다.
# 셀을 낱개로 보내면 수십만 건이라 응답이 감당되지 않는다.
"row_spans": [list(span) for span in mask_row_spans(domain)]
if domain is not None
else [],
},
# 최외곽 적색 셀 주변 확장 결과.
"expansion": {
"rounds": preview.expand_rounds,
"closed": preview.expand_closed,
"added_cells": preview.expand_added_cells,
"initial_cells": region.active_cells,
},
# 셀별 흐름 방향과 도로 도달 여부. row_spans 순서(행 → 구간 → 열 오름차순)로 1바이트씩.
"flow": _flow_payload(preview, domain),
# ⑦ 2차 전체 배수유역 외곽선(= 분수령)과 면적.
"basin_polygon_lonlat": [list(to_lonlat(x, y)) for x, y in preview.basin_boundary_xy],
"basin_area_m2": round(preview.basin_area_m2, 1),
# ⑥ 도로 위 흐름 강도 — [누가거리 m, 그 구간으로 모이는 상류 면적 ㎡].
"strength_profile": [
[round(chainage, 1), round(area, 1)] for chainage, area in preview.strength_profile
],
# ⑥-1 유입 집중점 — 기본 관으로 나눈 구역마다 물이 많이 모이는 자리.
# [누가거리 m, 유입면적 ㎡, 구역 번호, 구역 내 순위]. 화면 마커·하이라이트 대상이다.
"inflow_hotspots": [
[round(chainage, 1), round(area, 1), zone, rank]
for chainage, area, zone, rank in preview.inflow_hotspots
],
# ⑧ 기본 관 매설 위치 — 도로 × 세류선 교차점.
"pipes": [_candidate_payload(pipe, to_lonlat) for pipe in preview.pipes],
# B05용 평균 흐름 화살표 — [lon, lat, 방위(도), 도로도달, 셀 수].
# 세류·도로 셀을 뺀 블록 평균이라 사면 경향만 남는다.
"flow_arrows": [
[*to_lonlat(x, y), round(math.degrees(angle), 1), reaches, cells]
for x, y, angle, reaches, cells in preview.flow_arrows
],
# 화살표 간격(m). 화면이 화살표 크기를 정할 때 쓴다 — 서로 닿지 않게 이 값보다 짧게 그린다.
"arrow_spacing_m": DRAINAGE_ARROW_SPACING_M,
"from_cache": False,
}
# 단계 산출물을 영구저장소에 남긴다 — 기능을 붙일 때마다 여기에 단계가 하나씩 늘어난다.
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)],
# ⑦ 2차 전체 배수유역 외곽선, ⑧ 기본 관 위치(누가거리·근거 포함).
"basin_boundary": _boundary_geometry(preview.basin_boundary_xy),
"pipe": [
(
Point(pipe.x, pipe.y),
{
"chainage_m": round(pipe.chainage_m, 2),
"reason": pipe.reason,
"stream_name": pipe.stream_name,
},
)
for pipe in preview.pipes
],
},
{
"radius_m": region.radius_m,
"road_outside_m": payload["road_outside_m"],
"no_contact_count": region.split.no_contact,
"basin_area_m2": payload["basin_area_m2"],
"pipe_count": len(preview.pipes),
# 기하(bbox·셀 구간)는 파일 본문에 있으므로 요약 수치만 남긴다.
"grid": {
key: value
for key, value in payload["grid"].items()
if key not in {"bbox_lonlat", "row_spans"}
},
},
to_lonlat,
)
_write_stage_arrays(prepared["stored_path"], preview, domain, spec)
_write_road_routing(prepared["stored_path"], preview, spec, prepared["route_line"], to_lonlat)
# 응답 자체를 캐시로 남긴다 — 다음 조회는 배열을 재조립하지 않고 이 파일을 그대로 준다.
_save_response(prepared["stored_path"], payload)
return payload