Aislo/common_util/common_util_drainage_context.py

"""배수유역 세부 설계 입력 준비 (B04 관리자 화면 · B05 사용자 화면 공용).

두 화면이 같은 관 목록과 같은 세부유역을 보여 주려면 **입력이 한 글자도 달라선 안 된다**
(2026-08-01 사용자 지시). 그래서 노선·종단 Z·좌표계를 여기 한 곳에서 만들어 양쪽에 넘긴다.

노선 기준선은 B05가 푼 최적 경로가 아니라 **B03이 업로드한 원청 계획노선 CSV**다. 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_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
    epsg: int = 5186
    route_id: int | None = None
    to_lonlat: Callable[[float, float], tuple[float, float]] = lambda x, y: (x, y)


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,
    )

    epsg = int(planned.epsg or db_epsg or 5186)
    transformer = Transformer.from_crs(f"EPSG:{epsg}", "EPSG:4326", always_xy=True)
    return (
        DrainageContext(
            stored_path=stored_path,
            project_root=project_root,
            vertices=vertices,
            z_source=z_source,
            epsg=epsg,
            route_id=int(route["id"]) if route else None,
            to_lonlat=lambda x, y: transformer.transform(x, y),
        ),
        "",
    )


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