eomsangdonandClaude Opus 5
f1bcee7857
단일 링 자료형이 곧 중첩·빈공간이었다. 아래 유역이 위 유역을 감싸면 구멍이 사라져 위 유역을 통째로 덮고(합성 실측 256㎡ 전량), 한 관의 유역이 두 조각이면 작은 쪽이 사라져 빈공간이 됐다(144㎡). - `polygon_parts()` 신설 — 조각마다 [외곽 링, 구멍 링...], 넓은 조각부터. - `WatershedBasin.boundary_parts` 로 교체. `boundary_xy`(가장 넓은 조각의 외곽)와 `boundary_rings`(편 링 목록)는 파생 속성이라 옛 소비처는 그대로 동작한다. - API 에 `polygon_rings_lonlat` 추가. 저장 GeoJSON 은 구멍을 가진 Polygon, 조각이 여럿이면 MultiPolygon. - 캔버스 4곳(B04 유역화면 채움·선택, B05 배수유역도 채움·선택)은 even-odd 로 한 번에 채우고 `pointInRings()` 로 판정 — 구멍 안을 눌러도 바깥 유역이 잡히지 않는다. 중복이던 지역 `pointInRing` 은 삭제하고 공용 것으로 통일. 검증: 단위 5건 신규(도넛 구멍 보존·조각 2개 보존·파생 속성·빈 경계·조립 경로 전체에서 링 2개), 전체 168 passed, typecheck 통과. 브라우저 — API 가 유역 22개에 링 목록을 실어 보내고, 캔버스 even-odd 실측(가운데 알파 0 / 고리 255)으로 구멍이 실제로 뚫린다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
338 lines
16 KiB
Python
338 lines
16 KiB
Python
"""상세 배수유역 API 라우터 (B04 — 관리자 검토용).
|
|
|
|
격자 해석은 하지 않는다. 배수유역 분석이 저장해 둔 `03_road_routing` 산출물을 읽어
|
|
|
|
· 기본 관(도로 × 상류 세류선 교차점) + 최대 간격 자동 보충으로 관 목록을 만들고
|
|
· 계획노선 종단 Z로 노면 물이 어느 관으로 가는지 정해 세부유역을 나눈다
|
|
|
|
계산 알고리즘은 B05 사용자 화면과 공용이다(`common_util_drainage_detail`) — 같은 이름의
|
|
버튼은 같은 결과를 내야 하기 때문이다(2026-08-01 사용자 지시).
|
|
|
|
편집분(관 지점)은 "이 모델 확정" 시점에만 파일로 남는다. 그 전까지는 화면 메모리에만 있고,
|
|
확정 없이 나가면 저장된 값으로 되돌아온다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
|
|
from common_util.common_util_drainage_context import DrainageContext, load_drainage_context
|
|
from common_util.common_util_drainage_detail import DrainageDetail, build_detail
|
|
from common_util.common_util_drainage_pipes import (
|
|
PIPE_FACILITY_PIPE,
|
|
PIPE_SOURCE_USER,
|
|
PipePoint,
|
|
carry_facility_attributes,
|
|
clear_pipe_points,
|
|
load_pipe_points,
|
|
parse_pipe_points,
|
|
route_signature,
|
|
save_detail_basins,
|
|
save_pipe_points,
|
|
)
|
|
from common_util.common_util_route_geometry import StructureCandidate
|
|
from config.config_system import DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/projects", tags=["B04 Surface Drainage Basins"])
|
|
|
|
_NO_ANALYSIS = "배수유역 분석 결과가 없습니다. [유역 분석]을 먼저 실행하세요."
|
|
|
|
|
|
def _build(
|
|
stored_path: str, context: DrainageContext, points: list[PipePoint] | None
|
|
) -> tuple[DrainageDetail | None, list[PipePoint]]:
|
|
"""세부유역을 계산하고, 그 결과에 쓰인 관 목록을 함께 돌려준다.
|
|
|
|
`points`가 없으면 B04 기본 관에 최대 간격 규칙으로 자동 보충한 목록이 만들어진다.
|
|
있으면 그 자리를 그대로 쓰되, 관이 왜 거기 있는지(기본/자동/수동)는 넘겨받은 목록의
|
|
표시를 유지한다 — 계산기는 확정 좌표만 보므로 그대로 두면 전부 "수동"이 된다.
|
|
"""
|
|
directory = drainage_dir(stored_path)
|
|
chainages = [point.chainage_m for point in points] if points else None
|
|
detail = build_detail(directory, context.vertices, chainages)
|
|
if detail is None:
|
|
return None, []
|
|
if points:
|
|
_retag(detail.pipes, points)
|
|
rebuilt = [PipePoint(chainage_m=pipe.chainage_m, source=pipe.reason) for pipe in detail.pipes]
|
|
# 계산기는 chainage만 다뤄 시설 종류·구간·옵션이 사라진다 — 원본에서 되붙인다
|
|
# (2026-08-17 계곡 통과 시설: 배관/BOX암거/물넘이/세월교).
|
|
carried = carry_facility_attributes(rebuilt, points or [])
|
|
# 되붙인 뒤에 추천을 얹는다 — 순서가 뒤바뀌면 승계된 사용자 값이 추천에 덮인다.
|
|
_apply_recommendations(detail, carried)
|
|
return detail, carried
|
|
|
|
|
|
def _retag(pipes: list[StructureCandidate], points: list[PipePoint]) -> None:
|
|
"""확정 좌표로 되돌아온 관에 원래의 생성 사유를 다시 붙인다."""
|
|
by_chainage = {round(point.chainage_m, 2): point.source for point in points}
|
|
for pipe in pipes:
|
|
source = by_chainage.get(round(pipe.chainage_m, 2))
|
|
if source is None and points:
|
|
nearest = min(points, key=lambda item: abs(item.chainage_m - pipe.chainage_m))
|
|
source = nearest.source
|
|
pipe.reason = source or PIPE_SOURCE_USER
|
|
|
|
|
|
def _apply_recommendations(detail: DrainageDetail, points: list[PipePoint]) -> None:
|
|
"""유역별 추천(시설 종류·관경)을 관 지점 옵션의 **빈칸에만** 채운다.
|
|
|
|
"초기 전처리 계산에서 해당값으로 설정"(2026-08-17 사용자 지시) — 자동 배치된 관이
|
|
유역 유량에 맞는 규격을 처음부터 갖고 있어야 폼을 열지 않아도 하류가 값을 받는다.
|
|
|
|
덮어쓰지 않는 두 경우: ① 사용자가 직접 놓거나 옮긴 관(`source == "user"`) ②
|
|
그 옵션 키가 이미 있는 관. 설계자가 고른 값을 재계산이 되돌리면 안 되기 때문이다.
|
|
"""
|
|
by_chainage = {round(basin.chainage_m, 2): basin for basin in detail.basins}
|
|
for point in points:
|
|
if point.source == PIPE_SOURCE_USER:
|
|
continue
|
|
basin = by_chainage.get(round(point.chainage_m, 2))
|
|
if basin is None:
|
|
continue
|
|
options = dict(point.options or {})
|
|
# 시설 종류는 아직 손대지 않은 기본 배관일 때만 추천으로 바꾼다.
|
|
if (
|
|
point.facility == PIPE_FACILITY_PIPE
|
|
and basin.recommended_facility != PIPE_FACILITY_PIPE
|
|
):
|
|
point.facility = basin.recommended_facility
|
|
if (
|
|
point.facility == PIPE_FACILITY_PIPE
|
|
and basin.recommended_diameter_mm is not None
|
|
and "pipe_diameter_mm" not in options
|
|
):
|
|
options["pipe_diameter_mm"] = basin.recommended_diameter_mm
|
|
if options:
|
|
point.options = options
|
|
|
|
|
|
def _payload(
|
|
project_id: UUID,
|
|
context: DrainageContext,
|
|
detail: DrainageDetail,
|
|
points: list[PipePoint],
|
|
saved: bool,
|
|
) -> dict[str, Any]:
|
|
"""관 목록과 세부유역을 화면 좌표(WGS84)로 정리한다."""
|
|
to_lonlat = context.to_lonlat
|
|
return {
|
|
"status": "success",
|
|
"project_id": str(project_id),
|
|
# 종단 Z를 어디서 가져왔는지 — 관 담당 구간이 갈리는 근거라 화면에서 확인 가능해야 한다.
|
|
"z_source": context.z_source,
|
|
"route_length_m": round(context.vertices[-1].chainage_m, 2),
|
|
"max_spacing_m": DRAINAGE_PIPE_MAX_SPACING_M,
|
|
"min_spacing_m": DRAINAGE_PIPE_MIN_SPACING_M,
|
|
"saved": saved,
|
|
# 도로 1m 구간별 유입 면적 — 계획선을 색으로 칠하는 데 쓴다(B05와 같은 값).
|
|
"strength_profile": detail.strength_profile,
|
|
# 유입 집중점 — 관을 어디에 둘지 판단하는 근거. 화면에서 토글로 켠다.
|
|
"inflow_hotspots": detail.inflow_hotspots,
|
|
# ── 아래는 B05 배수유역도가 그리는 데 필요한 값. B04 지도는 자체 오버레이가 있어
|
|
# 쓰지 않지만, 두 화면이 같은 응답을 받아야 결과가 갈리지 않는다(2026-08-01 일원화).
|
|
"route_lonlat": detail.route_lonlat,
|
|
# 2차 전체 유역 외곽선 = 분수령. 해석 결과 그대로(손으로 고치는 기능 없음).
|
|
"main_polygon_lonlat": detail.basin_lonlat,
|
|
"grid_cell_m": detail.grid_cell_m,
|
|
"flow_arrows": detail.flow_arrows,
|
|
"arrow_spacing_m": detail.arrow_spacing_m,
|
|
"upstream_lonlat": detail.upstream_lonlat,
|
|
# 시설 종류·구간·옵션까지 함께 싣는다(as_dict — 기본 배관·폭 0은 필드 생략).
|
|
"pipe_points": [
|
|
{
|
|
**point.as_dict(),
|
|
"chainage_m": round(pipe.chainage_m, 2),
|
|
"lonlat": list(to_lonlat(pipe.x, pipe.y)),
|
|
"source": pipe.reason,
|
|
}
|
|
for pipe, point in zip(detail.pipes, points)
|
|
],
|
|
"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],
|
|
# 조각·구멍을 모두 편 링 목록 — 도넛 유역과 떨어진 조각을 그대로 그린다.
|
|
"polygon_rings_lonlat": [
|
|
[list(to_lonlat(x, y)) for x, y in ring] for ring in basin.boundary_rings
|
|
],
|
|
"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,
|
|
# 산출 근거 — 도달시간(분)·설계강우강도(mm/hr)·설계유량(m³/s, 2.0배 반영).
|
|
"tc_minutes": basin.tc_minutes,
|
|
"intensity_mm_hr": basin.intensity_mm_hr,
|
|
"design_flow_m3s": basin.design_flow_m3s,
|
|
# 유효직경이 관 최대 규격 초과 — 세월교·물넘이·교량 검토 대상(임도설치규정 제12조).
|
|
"bridge_required": basin.bridge_required,
|
|
# 필요 통수단면적(㎡) — 물넘이·세월교 개략 단면의 출발값.
|
|
"required_area_m2": basin.required_area_m2,
|
|
# 유량 근거 추천(2026-08-17 사용자 확정) — 배관이면 규격 스냅 관경이 붙는다.
|
|
"recommended_facility": basin.recommended_facility,
|
|
"recommended_diameter_mm": basin.recommended_diameter_mm,
|
|
}
|
|
for basin in detail.basins
|
|
],
|
|
"pipe_count": len(points),
|
|
}
|
|
|
|
|
|
def _basin_features(
|
|
context: DrainageContext, detail: DrainageDetail, points: list[PipePoint]
|
|
) -> list[dict[str, Any]]:
|
|
"""세부유역을 저장용 GeoJSON 피처로 바꾼다(관 지점도 같은 파일에 함께 남긴다)."""
|
|
to_lonlat = context.to_lonlat
|
|
features: list[dict[str, Any]] = []
|
|
for basin in detail.basins:
|
|
# GeoJSON 규격 그대로 — 조각마다 [외곽, 구멍...], 조각이 여럿이면 MultiPolygon.
|
|
parts = [
|
|
[[list(to_lonlat(x, y)) for x, y in ring] for ring in part if len(ring) >= 4]
|
|
for part in basin.boundary_parts
|
|
]
|
|
parts = [part for part in parts if part]
|
|
if not parts:
|
|
continue
|
|
geometry = (
|
|
{"type": "Polygon", "coordinates": parts[0]}
|
|
if len(parts) == 1
|
|
else {"type": "MultiPolygon", "coordinates": parts}
|
|
)
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
"properties": {
|
|
"kind": "detail_basin",
|
|
"index": basin.index,
|
|
"chainage_m": round(basin.chainage_m, 2),
|
|
"area_m2": round(basin.area_m2, 1),
|
|
"relief_m": round(basin.relief_m, 2),
|
|
"flow_length_m": round(basin.flow_length_m, 1),
|
|
"pipe_diameter_mm": basin.pipe_diameter_mm,
|
|
# 규격관(도면 배수규격 표기용) — 소요 관경과 구분해 함께 남긴다.
|
|
"recommended_diameter_mm": basin.recommended_diameter_mm,
|
|
"tc_minutes": basin.tc_minutes,
|
|
"intensity_mm_hr": basin.intensity_mm_hr,
|
|
"design_flow_m3s": basin.design_flow_m3s,
|
|
"bridge_required": basin.bridge_required,
|
|
},
|
|
"geometry": geometry,
|
|
}
|
|
)
|
|
for pipe, point in zip(detail.pipes, points):
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
# 시설 종류·구간·옵션도 함께 남긴다(as_dict — 기본값 생략) — GeoJSON만
|
|
# 보고도 세월교인지 배관인지 알 수 있어야 한다.
|
|
"properties": {"kind": "pipe", **point.as_dict()},
|
|
"geometry": {"type": "Point", "coordinates": list(to_lonlat(pipe.x, pipe.y))},
|
|
}
|
|
)
|
|
return features
|
|
|
|
|
|
async def _resolve(
|
|
project_id: UUID, payload: dict[str, Any] | None, *, use_stored: bool
|
|
) -> tuple[DrainageContext, DrainageDetail, list[PipePoint], bool] | JSONResponse:
|
|
"""요청 본문 → 저장분 → 자동 생성 순으로 관 목록을 정하고 세부유역까지 계산한다."""
|
|
context, reason = await load_drainage_context(project_id)
|
|
if context is None:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": reason})
|
|
|
|
requested = parse_pipe_points((payload or {}).get("points"))
|
|
signature = route_signature(context.vertices)
|
|
stored = (
|
|
load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None
|
|
)
|
|
points = requested or stored or None
|
|
|
|
result = await asyncio.to_thread(_build, context.stored_path, context, points)
|
|
detail, resolved = result
|
|
if detail is None:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": _NO_ANALYSIS})
|
|
return context, detail, resolved, bool(stored) and not requested
|
|
|
|
|
|
@router.get("/{project_id}/drainage/pipe-points", response_model=None)
|
|
async def get_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
|
"""저장된 관 지점과 그 세부유역을 돌려준다. 저장분이 없으면 자동 생성한 목록."""
|
|
resolved = await _resolve(project_id, None, use_stored=True)
|
|
if isinstance(resolved, JSONResponse):
|
|
return resolved
|
|
context, detail, points, saved = resolved
|
|
return _payload(project_id, context, detail, points, saved)
|
|
|
|
|
|
@router.post("/{project_id}/drainage/detail-basins", response_model=None)
|
|
async def post_detail_basins(
|
|
project_id: UUID, payload: dict[str, Any] | None = None
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""화면에서 편집 중인 관 목록으로 세부유역을 다시 나눈다(저장하지 않는다).
|
|
|
|
`points`를 비우고 부르면 저장분을 무시하고 기본 관 + 자동 보충으로 되돌린다 —
|
|
노선이 바뀌어 전체 분석을 다시 돌린 직후의 경로다.
|
|
"""
|
|
resolved = await _resolve(project_id, payload, use_stored=False)
|
|
if isinstance(resolved, JSONResponse):
|
|
return resolved
|
|
context, detail, points, _ = resolved
|
|
return _payload(project_id, context, detail, points, saved=False)
|
|
|
|
|
|
@router.delete("/{project_id}/drainage/pipe-points", response_model=None)
|
|
async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
|
|
"""저장된 관 지점을 버리고 기본 관 + 자동 보충 배치로 되돌린다("초기화").
|
|
|
|
화면만 되돌리면 다시 들어왔을 때 옛 관이 살아난다 — 저장분과 파생 산출물까지 지운다
|
|
(2026-08-02 사용자 보고).
|
|
"""
|
|
context, reason = await load_drainage_context(project_id)
|
|
if context is None:
|
|
return JSONResponse(status_code=404, content={"status": "error", "message": reason})
|
|
await asyncio.to_thread(clear_pipe_points, context.stored_path)
|
|
|
|
resolved = await _resolve(project_id, None, use_stored=False)
|
|
if isinstance(resolved, JSONResponse):
|
|
return resolved
|
|
context, detail, points, _ = resolved
|
|
return _payload(project_id, context, detail, points, saved=False)
|
|
|
|
|
|
@router.put("/{project_id}/drainage/pipe-points", response_model=None)
|
|
async def put_pipe_points(
|
|
project_id: UUID, payload: dict[str, Any] | None = None
|
|
) -> dict[str, Any] | JSONResponse:
|
|
"""관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점)."""
|
|
resolved = await _resolve(project_id, payload, use_stored=True)
|
|
if isinstance(resolved, JSONResponse):
|
|
return resolved
|
|
context, detail, points, _ = resolved
|
|
|
|
signature = route_signature(context.vertices)
|
|
# 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
|
|
saved = await asyncio.to_thread(
|
|
save_pipe_points, context.stored_path, signature, points, context.vertices
|
|
)
|
|
await asyncio.to_thread(
|
|
save_detail_basins,
|
|
context.stored_path,
|
|
_basin_features(context, detail, points),
|
|
context.crs,
|
|
)
|
|
result = _payload(project_id, context, detail, points, saved=True)
|
|
result["saved_count"] = saved
|
|
return result
|