eomsangdonandClaude Opus 5.5
0ba4653212
- B04 Router_Watershed — 분석 응답 · 단계 산출물 저장을 Output.write_analysis 로 뗌(primary-region 같은 길) · 도엽 좌표 저장본 메모리 보관(2 벌) - B04 Router_Watershed_Live — 저장 노선 기준(초안 없음 · 지문 같음) 답이면 마지막 끊김 확인 뒤 분석을 남김 · design/save 잠금 중이면 안 남김 · 남길 때 같은 잠금 쥠 · POST /drainage/live/cancel(그 차례가 마지막일 때만 지움) · 셈 멈춤 로그 - B05 Drainage_Live — 요청 끊을 때 서버에 끊김 알림 · 저장 노선 [업데이트] 답이 저장분 낡음 걷음 · 되돌리기로 초안이 없어지면 저장분의 마지막 낡음으로 표시 바로 - 시험 test_349_drainage_live_flow 고침 — 되돌리기 직후 낡음 기대가 옛 동작(답 올 때까지 낡음) · 새 동작 = 저장분 마지막 낡음(여기선 꺼짐) + 끊김 알림 확인 - 시험 test_347_2 — 「작업본에 안 씀」 옛 설명 고침 · 남김 · 초안 안 남김 · 잠금 중 안 남김 · 끊김 알림 시험 더함 · test_347_1 메모리 보관 확인 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
432 lines
19 KiB
Python
432 lines
19 KiB
Python
"""배수유역 분석 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
|