Aislo/common_util/common_util_route_profile.py

"""계획노선 종단 Z 해석 (공용).

배수유역 세부 설계는 **노면 물이 어느 쪽으로 흐르는가**로 관 담당 구간을 나눈다. 그래서
노선 정점의 Z가 무엇이냐에 따라 결과가 통째로 달라진다. B04(관리자)와 B05(사용자)가 서로
다른 Z를 쓰면 같은 프로젝트에서 세부유역이 갈라지므로, 어느 Z를 쓸지는 여기 한 곳에서만
정한다(2026-08-01 사용자 지시).

우선순위
  ① B05 종단 계획고(`longitudinal_sections.data.design_profiles`) — 사용자가 편집을 마친 정본
  ② B05 경로 정점 Z(`route_points.z`) — 계획고 편집 전 단계
  ③ 확정 지표면 모델 샘플링 — B05를 아직 지나지 않은 B04 시점의 기본값
  ④ 원청 계획노선 CSV의 z — 위 셋이 모두 없을 때의 최후 폴백

①②는 B05가 푼 **최적 경로** 기하 위의 값이다. 원청 계획노선과 노선 자체가 다르면 같은
누가거리라도 다른 지점이므로, 노선이 실질적으로 같을 때만 채택하고 아니면 ③으로 내려간다.

DB나 페이지 모듈은 여기서 건드리지 않는다 — 호출부(B04/B05 라우터)가 읽어서 넘긴다.
"""

from __future__ import annotations

import logging
import math
from dataclasses import replace
from typing import Any

import numpy as np

from common_util.common_util_route_geometry import RouteVertex, interpolate_vertex
from common_util.common_util_surface_sampler import SurfaceElevationSampler

logger = logging.getLogger(__name__)

# 노선 동일성 판정 — 시·종점이 이 거리 안이고 연장 차이가 아래 비율 안이면 같은 노선으로 본다.
# 최적 경로는 원청 노선을 따라가되 격자 해상도만큼 흔들리므로 여유를 둔다.
ROUTE_MATCH_ENDPOINT_TOLERANCE_M = 20.0
ROUTE_MATCH_LENGTH_TOLERANCE = 0.05

# Z 출처 표시 — 화면과 로그가 어느 종단을 쓴 결과인지 알 수 있어야 한다.
Z_SOURCE_DESIGN = "design_profile"
Z_SOURCE_ROUTE_POINTS = "route_points"
Z_SOURCE_SURFACE = "surface"
Z_SOURCE_CSV = "csv"


def design_elevation_from_longitudinal(
    longitudinal: dict[str, Any], chainage_m: float
) -> float | None:
    """종단 계획선(design_profiles) 샘플을 chainage 기준 선형보간해 계획고를 구한다.

    프론트 designElevationAt과 동일 규칙(범위 밖 양 끝값 클램프). 계획선이 없으면
    None을 반환해 지반고 폴백/오류 처리를 호출부에 맡긴다.
    """
    profiles = longitudinal.get("design_profiles") if isinstance(longitudinal, dict) else None
    if not isinstance(profiles, list) or not profiles:
        return None
    samples = [
        s
        for s in profiles[0].get("samples", [])
        if isinstance(s.get("elevation_m"), (int, float))
        and isinstance(s.get("chainage_m"), (int, float))
    ]
    if not samples:
        return None
    if chainage_m <= samples[0]["chainage_m"]:
        return float(samples[0]["elevation_m"])
    last = samples[-1]
    if chainage_m >= last["chainage_m"]:
        return float(last["elevation_m"])
    for index in range(1, len(samples)):
        previous = samples[index - 1]
        current = samples[index]
        if chainage_m > current["chainage_m"]:
            continue
        span = current["chainage_m"] - previous["chainage_m"]
        if span <= 0:
            return float(current["elevation_m"])
        ratio = (chainage_m - previous["chainage_m"]) / span
        return float(
            previous["elevation_m"] + (current["elevation_m"] - previous["elevation_m"]) * ratio
        )
    return float(last["elevation_m"])


def routes_match(left: list[RouteVertex], right: list[RouteVertex]) -> bool:
    """두 노선이 실질적으로 같은 노선인지 본다(시·종점 근접 + 연장 유사).

    최적 경로는 원청 노선을 격자 위에서 다시 그은 것이라 정점이 하나도 겹치지 않을 수
    있다. 그래서 정점 대조가 아니라 끝점과 연장으로만 판단한다.
    """
    if len(left) < 2 or len(right) < 2:
        return False
    start_gap = math.dist((left[0].x, left[0].y), (right[0].x, right[0].y))
    end_gap = math.dist((left[-1].x, left[-1].y), (right[-1].x, right[-1].y))
    if max(start_gap, end_gap) > ROUTE_MATCH_ENDPOINT_TOLERANCE_M:
        return False
    left_length = left[-1].chainage_m
    right_length = right[-1].chainage_m
    if left_length <= 0 or right_length <= 0:
        return False
    return abs(left_length - right_length) / left_length <= ROUTE_MATCH_LENGTH_TOLERANCE


def resolve_route_profile(
    vertices: list[RouteVertex],
    *,
    route_vertices: list[RouteVertex] | None = None,
    longitudinal: dict[str, Any] | None = None,
    sampler: SurfaceElevationSampler | None = None,
) -> tuple[list[RouteVertex], str]:
    """계획노선 정점에 종단 Z를 채워 돌려준다. (정점 목록, Z 출처) 형태.

    `vertices`는 원청 계획노선(B04·B05 공용 기준선)이고, `route_vertices`는 B05가 푼 최적
    경로다. 노선이 같을 때만 ①②를 쓰고, 아니면 ③(지표면 샘플링)으로 내려간다.
    """
    if len(vertices) < 2:
        return vertices, Z_SOURCE_CSV

    matched = bool(route_vertices) and routes_match(vertices, route_vertices or [])
    if matched and longitudinal:
        elevations = [
            design_elevation_from_longitudinal(longitudinal, vertex.chainage_m)
            for vertex in vertices
        ]
        if all(value is not None for value in elevations):
            logger.info("노선 종단 Z: B05 계획고(design_profiles) 채택 — 정점 %d개", len(vertices))
            return (
                [replace(v, z=float(z)) for v, z in zip(vertices, elevations)],
                Z_SOURCE_DESIGN,
            )

    if matched and route_vertices:
        logger.info("노선 종단 Z: B05 경로 정점(route_points) 채택 — 정점 %d개", len(vertices))
        return (
            [replace(v, z=interpolate_vertex(route_vertices, v.chainage_m)[2]) for v in vertices],
            Z_SOURCE_ROUTE_POINTS,
        )

    if sampler is not None:
        sampled = _sample_z(vertices, sampler)
        if sampled is not None:
            logger.info("노선 종단 Z: 확정 지표면 샘플링 채택 — 정점 %d개", len(vertices))
            return sampled, Z_SOURCE_SURFACE

    logger.info("노선 종단 Z: 원청 CSV z 유지 — 정점 %d개", len(vertices))
    return vertices, Z_SOURCE_CSV


def _sample_z(
    vertices: list[RouteVertex], sampler: SurfaceElevationSampler
) -> list[RouteVertex] | None:
    """확정 지표면에서 노선 정점의 지반고를 뽑는다.

    모델 밖으로 나간 정점은 유효한 이웃 정점 값으로 메운다 — 노선 한두 점이 DTM 가장자리를
    벗어났다고 종단 전체를 버리면 세부유역을 못 나눈다. 유효한 값이 하나도 없으면 None.
    """
    xy = np.array([[vertex.x, vertex.y] for vertex in vertices], dtype=np.float64)
    try:
        z, valid = sampler.sample_xy(xy)
    except (ValueError, OSError) as exc:
        logger.warning("노선 종단 Z: 지표면 샘플링 실패 — %s", exc)
        return None
    if not bool(valid.any()):
        logger.warning("노선 종단 Z: 노선이 확정 지표면 범위 밖입니다.")
        return None
    if not bool(valid.all()):
        index = np.arange(z.size)
        known = index[valid]
        z = np.interp(index, known, z[valid])
        logger.info(
            "노선 종단 Z: 지표면 밖 정점 %d개를 이웃 값으로 메웠습니다.", int((~valid).sum())
        )
    return [replace(vertex, z=float(value)) for vertex, value in zip(vertices, z)]