141 lines
6.4 KiB
Python
141 lines
6.4 KiB
Python
"""배수유역 세부 설계 입력 준비 (B04 관리자 화면 · B05 사용자 화면 공용).
|
|
|
|
두 화면이 같은 관 목록과 같은 세부유역을 보여 주려면 **입력이 한 글자도 달라선 안 된다**
|
|
(2026-08-01 사용자 지시). 그래서 노선·종단 Z·좌표계를 여기 한 곳에서 만들어 양쪽에 넘긴다.
|
|
|
|
노선 기준선은 B05가 푼 최적 경로가 아니라 **B03이 받은 원청 계획노선(정본)**이다. B04 격자
|
|
해석이 그 노선으로 도로 셀을 구웠으므로, 다른 노선의 누가거리를 쓰면 도로 셀과 관 위치가
|
|
어긋난다. 종단 Z만 상황에 따라 갈아 끼운다 → [[common_util_route_profile]].
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from pyproj import Transformer
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from B05_Profile.B05_Profile_Repository import (
|
|
get_latest_route,
|
|
get_route_points,
|
|
get_surface_crs_epsg,
|
|
)
|
|
from B06_Section.B06_Section_Repository import get_longitudinal_section
|
|
from common_util.common_util_route_geometry import (
|
|
RouteVertex,
|
|
build_route_vertices,
|
|
load_design_route,
|
|
)
|
|
from common_util.common_util_route_profile import Z_SOURCE_CSV, resolve_route_profile
|
|
from common_util.common_util_crs import resolve_project_crs
|
|
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_surface_sampler import build_surface_sampler
|
|
from config.config_db import get_db_pool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_INPUT_SUBDIR = Path("B03_FileInput") / "input"
|
|
_MODELS_SUBDIR = Path("B04_PreProcess") / "models"
|
|
|
|
|
|
@dataclass
|
|
class DrainageContext:
|
|
"""세부 설계 한 번에 필요한 입력 묶음."""
|
|
|
|
stored_path: str
|
|
project_root: Path
|
|
vertices: list[RouteVertex] = field(default_factory=list)
|
|
z_source: str = Z_SOURCE_CSV
|
|
# 사업지 좌표계 — pyproj 입력 문자열(`EPSG:n` 또는 .prj WKT). 노선 CSV의 `crs_epsg`
|
|
# 열은 표시용 라벨이라 실좌표계와 다를 수 있다(2026-09-01 실측: 라벨 5179, 실제 5176).
|
|
crs: str = "EPSG:5186"
|
|
route_id: int | None = None
|
|
to_lonlat: Callable[[float, float], tuple[float, float]] = lambda x, y: (x, y)
|
|
# 1단계에서 확정한 지표면 선택(source_filter·method·smooth). B07 라이다 계획평면도가
|
|
# 어느 DTM 격자로 음영기복을 만들지 고르는 데 쓴다(2026-09-04).
|
|
surface_params: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
async def load_drainage_context(project_id: UUID) -> tuple[DrainageContext | None, str]:
|
|
"""노선·종단 Z·좌표계를 준비한다. 실패하면 (None, 사용자에게 보일 사유).
|
|
|
|
B05 확정 경로는 **있으면 쓰고 없으면 넘어간다** — B04는 WF1 화면이라 아직 경로가 없다.
|
|
"""
|
|
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 None, "프로젝트 저장 경로가 없습니다."
|
|
route = await get_latest_route(connection, project_id)
|
|
route_points: list[dict[str, Any]] = []
|
|
longitudinal: dict[str, Any] | None = None
|
|
if route:
|
|
route_points = await get_route_points(connection, int(route["id"]))
|
|
section = await get_longitudinal_section(connection, project_id, int(route["id"]))
|
|
longitudinal = (section or {}).get("data")
|
|
surface_model_id = (route or {}).get("surface_model_id")
|
|
db_epsg = await get_surface_crs_epsg(
|
|
connection, project_id, int(surface_model_id) if surface_model_id else 0
|
|
)
|
|
surface_params = await get_surface_confirmation_params(connection, str(project_id))
|
|
|
|
project_root = Path(resolve_stored_project_path(stored_path))
|
|
# 설계 계통과 **같은 노선**을 쓴다 — 지표면 밖 구간을 자른 뒤의 노선이다. 원본을 그대로
|
|
# 쓰면 유역·관이 확정 노선 밖에도 찍혀 종단 계획선이 그 관을 버린다(2026-09-01).
|
|
planned = await asyncio.to_thread(_read_planned_route, project_root, surface_params)
|
|
if planned is None or len(planned.vertices) < 2:
|
|
return None, "계획노선을 읽지 못했습니다. B03에서 노선 파일을 확인하세요."
|
|
|
|
sampler = await asyncio.to_thread(_open_sampler, project_root, surface_params)
|
|
vertices, z_source = await asyncio.to_thread(
|
|
resolve_route_profile,
|
|
planned.vertices,
|
|
route_vertices=build_route_vertices(route_points) if route_points else None,
|
|
longitudinal=longitudinal,
|
|
sampler=sampler,
|
|
)
|
|
|
|
# 노선을 실제로 담고 있는 좌표계를 쓴다 — `load_design_route()`가 .prj 좌표계로
|
|
# 재투영하며 `crs_input`만 갱신하고 `epsg` 라벨은 CSV 값 그대로 남긴다(라벨은 안 씀).
|
|
crs = resolve_project_crs(project_root, route_crs_input=planned.crs_input, db_epsg=db_epsg)
|
|
transformer = Transformer.from_crs(crs, "EPSG:4326", always_xy=True)
|
|
return (
|
|
DrainageContext(
|
|
stored_path=stored_path,
|
|
project_root=project_root,
|
|
vertices=vertices,
|
|
z_source=z_source,
|
|
crs=crs,
|
|
route_id=int(route["id"]) if route else None,
|
|
to_lonlat=lambda x, y: transformer.transform(x, y),
|
|
surface_params=dict(surface_params),
|
|
),
|
|
"",
|
|
)
|
|
|
|
|
|
def _read_planned_route(project_root: Path, surface_params: dict[str, Any] | None = None):
|
|
"""설계용 계획노선을 읽는다(파일 접근이라 스레드에서 돈다)."""
|
|
return load_design_route(project_root, surface_params)
|
|
|
|
|
|
def _open_sampler(project_root: Path, surface_params: dict[str, Any]):
|
|
"""확정 지표면 표고 sampler를 연다. 모델이 없으면 None(종단 Z가 다른 경로로 폴백)."""
|
|
try:
|
|
return build_surface_sampler(
|
|
project_root / _MODELS_SUBDIR,
|
|
str(surface_params["source_filter"]),
|
|
str(surface_params["method"]),
|
|
bool(surface_params["smooth"]),
|
|
)
|
|
except (FileNotFoundError, ValueError, OSError) as exc:
|
|
logger.warning("배수유역: 확정 지표면 sampler를 열지 못했습니다 — %s", exc)
|
|
return None
|