Aislo/common_util/common_util_route_geometry.py

"""계획 노선 기하 공용 유틸 — 정점·누가거리·세류 교차점.

배수유역 분석(B04)과 관 편집·세부유역(B05)이 같은 노선 표현을 써야 하므로 여기 한 곳에만
정의한다. 어느 한쪽 페이지 폴더에 두면 반대 방향 import가 생긴다.

노선 원천은 두 가지다.
  · B03에 업로드된 **계획 노선 파일**(CSV) — 배수유역 분석의 입력
  · DB `route_points` — B05에서 탐색·확정한 노선
둘 다 같은 `RouteVertex` 목록으로 바꿔 아래 함수들이 그대로 받는다.
"""

from __future__ import annotations

import csv
import logging
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from shapely.geometry import LineString, Point, shape

logger = logging.getLogger(__name__)

# 계획 노선 CSV 열 이름 후보. B03이 여러 형식을 받게 되므로 흔한 표기를 모두 받아 준다.
_X_KEYS = ("x", "X", "동", "easting", "EASTING")
_Y_KEYS = ("y", "Y", "북", "northing", "NORTHING")
_Z_KEYS = ("z", "Z", "표고", "elevation", "ELEV")
_ORDER_KEYS = ("sequence", "order", "seq", "no", "번호")
_EPSG_KEYS = ("crs_epsg", "epsg", "EPSG")


@dataclass
class RouteVertex:
    """노선 폴리라인의 한 점. chainage는 시점 기준 누가거리(m)."""

    x: float
    y: float
    z: float
    chainage_m: float


@dataclass
class StructureCandidate:
    """관 매설 구조물 측점 후보."""

    chainage_m: float
    x: float
    y: float
    # "stream"=세류 교차, "spacing"=최대 간격 규칙 보충, "confirmed"=사용자 확정
    reason: str
    stream_name: str | None = None


@dataclass
class PlannedRoute:
    """계획 노선 파일에서 읽은 노선."""

    vertices: list[RouteVertex]
    epsg: int | None
    name: str | None
    source: Path

    @property
    def line(self) -> LineString:
        return LineString([(vertex.x, vertex.y) for vertex in self.vertices])


def read_planned_route_csv(path: Path) -> PlannedRoute | None:
    """계획 노선 CSV를 읽어 정점 목록으로 바꾼다.

    열 이름은 대소문자·한글 표기를 함께 받아 준다(B03이 여러 형식을 수용할 예정).
    `sequence`가 있으면 그 순서로 정렬하고, 없으면 파일에 적힌 순서를 그대로 쓴다.
    """
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as file:
            rows = list(csv.DictReader(file))
    except (OSError, csv.Error, UnicodeDecodeError):
        logger.warning("계획 노선 CSV를 읽지 못했습니다: %s", path)
        return None
    if not rows:
        return None

    epsg = _first_int(rows[0], _EPSG_KEYS)
    name = _first_text(rows[0], ("route_name", "name", "노선명"))
    parsed: list[tuple[float, float, float, float]] = []  # (정렬키, x, y, z)
    for index, row in enumerate(rows):
        x = _first_float(row, _X_KEYS)
        y = _first_float(row, _Y_KEYS)
        if x is None or y is None:
            continue
        order = _first_float(row, _ORDER_KEYS)
        parsed.append(
            (float(index) if order is None else order, x, y, _first_float(row, _Z_KEYS) or 0.0)
        )
    if len(parsed) < 2:
        logger.warning("계획 노선 CSV에 좌표가 2점 미만입니다: %s", path)
        return None

    parsed.sort(key=lambda item: item[0])
    vertices: list[RouteVertex] = []
    cumulative = 0.0
    previous: tuple[float, float] | None = None
    for _, x, y, z in parsed:
        if previous is not None:
            cumulative += math.dist(previous, (x, y))
        vertices.append(RouteVertex(x=x, y=y, z=z, chainage_m=cumulative))
        previous = (x, y)
    logger.info(
        "계획 노선 %s: 정점 %d개, 연장 %.0fm, EPSG %s", path.name, len(vertices), cumulative, epsg
    )
    return PlannedRoute(vertices=vertices, epsg=epsg, name=name, source=path)


def find_planned_route_file(input_dir: Path) -> Path | None:
    """B03 입력 폴더에서 계획 노선 파일을 찾는다. 여러 개면 가장 최근 것."""
    if not input_dir.exists():
        return None
    candidates = sorted(
        input_dir.rglob("*.csv"), key=lambda item: item.stat().st_mtime, reverse=True
    )
    return candidates[0] if candidates else None


def build_route_vertices(points: list[dict[str, Any]]) -> list[RouteVertex]:
    """DB route_points 행을 누가거리가 채워진 정점 목록으로 바꾼다."""
    vertices: list[RouteVertex] = []
    cumulative = 0.0
    previous: tuple[float, float] | None = None
    for row in points:
        x = float(row["x"])
        y = float(row["y"])
        z = float(row.get("z") or 0.0)
        if previous is not None:
            cumulative += math.dist(previous, (x, y))
        chainage = row.get("chainage_m")
        vertices.append(
            RouteVertex(
                x=x, y=y, z=z, chainage_m=float(chainage) if chainage is not None else cumulative
            )
        )
        previous = (x, y)
    return vertices


def interpolate_vertex(
    vertices: list[RouteVertex], chainage_m: float
) -> tuple[float, float, float]:
    """누가거리 위치의 (x, y, z)를 선형 보간한다. 범위 밖은 끝점으로 당긴다."""
    if not vertices:
        return (0.0, 0.0, 0.0)
    if chainage_m <= vertices[0].chainage_m:
        return (vertices[0].x, vertices[0].y, vertices[0].z)
    for previous, current in zip(vertices, vertices[1:]):
        if chainage_m <= current.chainage_m:
            span = current.chainage_m - previous.chainage_m
            ratio = 0.0 if span <= 0 else (chainage_m - previous.chainage_m) / span
            return (
                previous.x + (current.x - previous.x) * ratio,
                previous.y + (current.y - previous.y) * ratio,
                previous.z + (current.z - previous.z) * ratio,
            )
    last = vertices[-1]
    return (last.x, last.y, last.z)


def is_uphill_at(vertices: list[RouteVertex], chainage_m: float, window_m: float = 20.0) -> bool:
    """해당 위치가 오르막(절토부)인지 종단 계획선의 국소 기울기 부호로 판정한다."""
    _, _, back_z = interpolate_vertex(vertices, max(0.0, chainage_m - window_m))
    _, _, forward_z = interpolate_vertex(vertices, chainage_m + window_m)
    return forward_z >= back_z


def find_stream_crossings(
    vertices: list[RouteVertex],
    stream_features: list[dict[str, Any]],
) -> list[StructureCandidate]:
    """노선 평면 선형과 세류선의 교차 지점을 누가거리 순으로 찾는다."""
    if len(vertices) < 2:
        return []
    route_line = LineString([(vertex.x, vertex.y) for vertex in vertices])
    candidates: list[StructureCandidate] = []
    for feature in stream_features:
        geometry = feature.get("geometry")
        if not geometry:
            continue
        try:
            stream = shape(geometry)
        except Exception:  # noqa: BLE001 - 손상된 피처는 건너뛴다
            continue
        if stream.is_empty:
            continue
        intersection = route_line.intersection(stream)
        if intersection.is_empty:
            continue
        name = _stream_name(feature)
        for point in _collect_points(intersection):
            candidates.append(
                StructureCandidate(
                    chainage_m=route_line.project(point),
                    x=point.x,
                    y=point.y,
                    reason="stream",
                    stream_name=name,
                )
            )
    candidates.sort(key=lambda item: item.chainage_m)
    return candidates


def _stream_name(feature: dict[str, Any]) -> str | None:
    properties = feature.get("properties") or {}
    for key in ("명칭", "하천명", "NAME", "name"):
        value = properties.get(key)
        if value:
            return str(value)
    return None


def _collect_points(geometry: Any) -> list[Point]:
    """교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다."""
    if geometry.geom_type == "Point":
        return [geometry]
    if geometry.geom_type in {"MultiPoint", "GeometryCollection"}:
        points: list[Point] = []
        for part in geometry.geoms:
            points.extend(_collect_points(part))
        return points
    # 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다.
    if geometry.geom_type in {"LineString", "MultiLineString"}:
        return [geometry.interpolate(0.5, normalized=True)]
    return []


def _first_text(row: dict[str, Any], keys: tuple[str, ...]) -> str | None:
    for key in keys:
        value = row.get(key)
        if value not in (None, ""):
            return str(value).strip()
    return None


def _first_float(row: dict[str, Any], keys: tuple[str, ...]) -> float | None:
    text = _first_text(row, keys)
    if text is None:
        return None
    try:
        return float(text)
    except ValueError:
        return None


def _first_int(row: dict[str, Any], keys: tuple[str, ...]) -> int | None:
    value = _first_float(row, keys)
    return None if value is None else int(value)