194 lines
8.0 KiB
Python
194 lines
8.0 KiB
Python
"""배수유역 세부 설계 API 라우터 (B05 — 일반 사용자용).
|
|
|
|
**분석하지 않는다.** B04가 미리 돌려 저장한 결과를 읽어 관을 보충하고 세부유역만 나눈다.
|
|
격자 해석은 30초가 걸려 일반 사용자를 붙잡아 두므로 여기서는 아예 돌리지 않는다
|
|
(2026-07-31 사용자 지시).
|
|
|
|
좌표는 사업지 CRS(m)에서 계산하고 응답만 WGS84로 바꿔 내보낸다.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
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 B05_wf2_Route.B05_wf2_Route_Engine_Drainage_Basin import build_drainage_detail
|
|
from B05_wf2_Route.B05_wf2_Route_Engine_Drainage_Store import (
|
|
apply_boundary_overrides,
|
|
load_boundary_overrides,
|
|
parse_overrides,
|
|
resample_boundary,
|
|
save_boundary_overrides,
|
|
)
|
|
from B05_wf2_Route.B05_wf2_Route_Repository import (
|
|
get_latest_route,
|
|
get_route_points,
|
|
get_surface_crs_epsg,
|
|
)
|
|
from common_util.common_util_route_geometry import StructureCandidate, build_route_vertices
|
|
from config.config_db import get_db_pool
|
|
from config.config_system import DRAINAGE_BOUNDARY_HANDLE_SPACING_M
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B05 Route Drainage"])
|
|
|
|
|
|
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:
|
|
"""확정 노선과 좌표 변환기를 준비한다. 도엽 피처는 읽지 않는다(분석을 안 하므로)."""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
route = await get_latest_route(connection, project_id)
|
|
if not route:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "확정된 경로가 없습니다."},
|
|
)
|
|
points = await get_route_points(connection, int(route["id"]))
|
|
surface_model_id = route.get("surface_model_id")
|
|
epsg = await get_surface_crs_epsg(
|
|
connection, project_id, int(surface_model_id) if surface_model_id else 0
|
|
)
|
|
|
|
vertices = build_route_vertices(points)
|
|
if len(vertices) < 2:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"status": "error", "message": "노선 좌표가 부족합니다."},
|
|
)
|
|
transformer = Transformer.from_crs(f"EPSG:{epsg or 5186}", "EPSG:4326", always_xy=True)
|
|
return {
|
|
"route_id": int(route["id"]),
|
|
"vertices": vertices,
|
|
"stored_path": stored_path,
|
|
"to_lonlat": lambda x, y: transformer.transform(x, y),
|
|
}
|
|
|
|
|
|
@router.post("/{project_id}/drainage/basins", response_model=None)
|
|
async def post_drainage_basins(
|
|
project_id: UUID,
|
|
payload: dict[str, Any] | None = None,
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""B04 분석 결과로 관을 보충하고 세부유역을 나눈다.
|
|
|
|
payload에 `chainages`(누가거리 목록)를 주면 그 위치로 관을 확정하고(사용자 편집),
|
|
없으면 B04의 기본 관에 최대 간격 규칙으로 최소 개수만 보충한다.
|
|
"""
|
|
prepared = await _prepare(project_id)
|
|
if isinstance(prepared, JSONResponse):
|
|
return prepared
|
|
raw = (payload or {}).get("chainages")
|
|
confirmed = _parse_chainages(raw) if isinstance(raw, list) else []
|
|
|
|
detail = await asyncio.to_thread(
|
|
build_drainage_detail, prepared["stored_path"], prepared["vertices"], confirmed
|
|
)
|
|
if detail is None:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={
|
|
"status": "error",
|
|
"message": "배수유역 분석 결과가 없습니다. B04에서 먼저 분석을 실행하세요.",
|
|
},
|
|
)
|
|
|
|
to_lonlat = prepared["to_lonlat"]
|
|
# 2차 전체 유역 외곽선 — 편집 핸들 간격으로 다시 찍고 저장된 편집분을 얹는다.
|
|
# 격자 경계라 원래 정점이 1m 간격이라 그대로는 손으로 잡을 수 없다.
|
|
boundary = resample_boundary(detail.basin_lonlat)
|
|
stored_overrides = load_boundary_overrides(prepared["stored_path"])
|
|
boundary, kept = apply_boundary_overrides(boundary, stored_overrides)
|
|
dropped = len(stored_overrides) - len(kept)
|
|
if dropped > 0:
|
|
# 새 유역 안쪽으로 들어갔거나 붙일 자리가 없어진 편집분은 파일에서도 지운다.
|
|
save_boundary_overrides(prepared["stored_path"], kept)
|
|
|
|
return {
|
|
"status": "success",
|
|
"project_id": str(project_id),
|
|
"route_id": prepared["route_id"],
|
|
# B04가 남긴 그대로 — 계획도로선. 외곽선만 편집분을 반영해 내보낸다.
|
|
"route_lonlat": detail.route_lonlat,
|
|
"main_polygon_lonlat": boundary,
|
|
"boundary_spacing_m": DRAINAGE_BOUNDARY_HANDLE_SPACING_M,
|
|
"boundary_overrides": [
|
|
{"base": list(item.base), "moved": list(item.moved)} for item in kept
|
|
],
|
|
"boundary_dropped": dropped,
|
|
"grid_cell_m": detail.grid_cell_m,
|
|
# 평균 흐름 화살표 — B04가 계산해 저장한 것을 그대로 넘긴다(사업지 CRS m).
|
|
"flow_arrows": detail.flow_arrows,
|
|
"arrow_spacing_m": detail.arrow_spacing_m,
|
|
# 유역 안쪽 상류 세류망 — 화면 강조 토글용(WGS84).
|
|
"upstream_lonlat": detail.upstream_lonlat,
|
|
# 계획선 위 배관 지점 — 유역이 없는 관도 마커로 표시해야 하므로 별도 목록.
|
|
"pipes": [_candidate_payload(pipe, to_lonlat) for pipe in detail.pipes],
|
|
"basins": [
|
|
{
|
|
"index": basin.index,
|
|
"chainage_m": round(basin.chainage_m, 2),
|
|
"outlet_lonlat": list(to_lonlat(basin.outlet_x, basin.outlet_y)),
|
|
"polygon_lonlat": [list(to_lonlat(x, y)) for x, y in basin.boundary_xy],
|
|
"area_m2": round(basin.area_m2, 1),
|
|
"relief_m": round(basin.relief_m, 2),
|
|
"flow_length_m": round(basin.flow_length_m, 1),
|
|
# 관경 수식 미확정 — None이면 프론트가 "미정"으로 표기한다.
|
|
"pipe_diameter_mm": basin.pipe_diameter_mm,
|
|
}
|
|
for basin in detail.basins
|
|
],
|
|
}
|
|
|
|
|
|
@router.put("/{project_id}/drainage/boundary", response_model=None)
|
|
async def put_drainage_boundary(
|
|
project_id: UUID,
|
|
payload: dict[str, Any] | None = None,
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""사용자가 옮긴 유역 외곽선 포인트만 저장한다(종단 경로 확정 시 모달 승인 후 호출).
|
|
|
|
폴리곤 전체가 아니라 이동한 포인트만 남긴다 — 노선이 바뀌어 유역을 다시 계산해도
|
|
좌표 근접으로 다시 붙일 수 있어야 하기 때문이다.
|
|
"""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection:
|
|
stored_path = await get_project_storage_relative_path(connection, project_id)
|
|
if not stored_path:
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={"status": "error", "message": "프로젝트 저장 경로가 없습니다."},
|
|
)
|
|
overrides = parse_overrides((payload or {}).get("points"))
|
|
saved = save_boundary_overrides(stored_path, overrides)
|
|
return {"status": "success", "project_id": str(project_id), "saved": saved}
|
|
|
|
|
|
def _parse_chainages(values: list[Any]) -> list[float]:
|
|
"""사용자가 확정·편집한 누가거리 목록을 숫자로 정리한다."""
|
|
parsed: list[float] = []
|
|
for value in values:
|
|
try:
|
|
parsed.append(float(value))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return parsed
|