Aislo/B05_Profile/B05_Profile_Engine_Grade_Carry.py

"""노선이 바뀔 때 계획고 옮김 — 같은 자리는 옛 계획고 · 바뀐 자리만 새 지반 첫 계획선(344-2 · 400).

사용자 확정(10-16 · 341 결정 1): 노선 끝쪽을 고쳤는데 앞쪽 계획고가 날아가지 않게 한다.
요구 250-2둘째(400): 평면을 고쳐도 크게 안 바뀐 자리의 종단 편집(직선화 · 자유지정 · 쉬프트 ·
▲▼)은 이어받음.

  - `changed_span` = 옛 · 새 노선 정점열의 같은 앞머리 · 같은 꼬리를 뺀 가운데 —
    「노선 그대로?」 · 범위.
  - 같은 자리(`line_matcher`) = 새 선 점을 옛 선에 투영 · 옆 `SAME_LINE_TOL_M` 안 · 방향 30° 안
    (측점 스냅이 뒤 교점을 밀어도 이어받음).
  - 자동 꼭짓점(`carry_base`) · 편집(`carry_edits`)은 **새 측점 위에** 다시 놓음 — 같은 자리면
    옛 선 그 자리 높이 · 아니면 새 지반 첫 계획선 · 편집은 옛 사용자 꼭짓점이 닿은 자리만 ·
    R 은 가까운 측점.

짝 = TS `B05_Profile/logic/B05_Profile_Grade_Carry.ts`(같은 규칙) · 거울 시험 =
`resources/tester/test_400_profile_carry.py` · `test_344_2_route_pairs.py`.
서버 정본은 이쪽(346-3 [저장]이 부름).
"""

from __future__ import annotations

import math
from collections.abc import Callable
from typing import Any

import numpy as np

from B05_Profile.B05_Profile_Engine_Grade_Alignment import chainage_key, resolve_pvi

# 같은 자리로 볼 옆 어긋남(m) — 브레인 정함 10-17 · 바꿀 수 있음 · TS `SAME_LINE_TOL_M` 짝.
SAME_LINE_TOL_M = 1.0
# 같은 자리로 볼 방향 차(cos 30°) — 엇갈려 지나는 선 · 머리핀 맞은편을 뺌.
_SAME_DIRECTION_COS = 0.8660254037844386
# 꼭짓점 위로 볼 거리(m) — `SAME_STATION_SNAP_M` 짝.
_ON_NODE_M = 0.006
# 같은 정점으로 볼 거리(m) — 저장 좌표 0.0001 m 자리 · 원호 다시 뽑기 잡음보다 크게.
SAME_VERTEX_M = 1e-3
_EPS = 1e-6


def _cumulative(xy: list[tuple[float, float]]) -> list[float]:
    total = [0.0]
    for (x0, y0), (x1, y1) in zip(xy, xy[1:]):
        total.append(total[-1] + math.hypot(x1 - x0, y1 - y0))
    return total


def _same(a: tuple[float, float], b: tuple[float, float]) -> bool:
    return abs(a[0] - b[0]) <= SAME_VERTEX_M and abs(a[1] - b[1]) <= SAME_VERTEX_M


def changed_span(
    old_xy: list[tuple[float, float]], new_xy: list[tuple[float, float]]
) -> dict[str, Any] | None:
    """옛 · 새 노선 정점열 → 바뀐 누가거리 구간. 같으면 None.

    `start_m` = 같은 앞머리 끝(옛 = 새) · `old_end_m` · `new_end_m` = 같은 꼬리 시작 ·
    `delta_m` = 새 연장 − 옛 연장 · `start_fixed` · `end_fixed` = 앞머리 · 꼬리가 한 점이라도
    같은가.
    """
    old_xy = [(float(x), float(y)) for x, y in old_xy]
    new_xy = [(float(x), float(y)) for x, y in new_xy]
    limit = min(len(old_xy), len(new_xy))
    prefix = 0
    while prefix < limit and _same(old_xy[prefix], new_xy[prefix]):
        prefix += 1
    if prefix == len(old_xy) == len(new_xy):
        return None
    suffix = 0
    while suffix < limit - prefix and _same(old_xy[-1 - suffix], new_xy[-1 - suffix]):
        suffix += 1
    old_cum, new_cum = _cumulative(old_xy), _cumulative(new_xy)
    return {
        "start_m": old_cum[prefix - 1] if prefix else 0.0,
        "old_end_m": old_cum[len(old_xy) - suffix] if suffix else old_cum[-1],
        "new_end_m": new_cum[len(new_xy) - suffix] if suffix else new_cum[-1],
        "delta_m": new_cum[-1] - old_cum[-1],
        "start_fixed": prefix > 0,
        "end_fixed": suffix > 0,
    }


def _segments(xy: list[tuple[float, float]]) -> np.ndarray:
    """토막 표 [x, y, ux, uy, 길이, 시작 누가거리] — 길이 0 토막은 뺌(누가거리는 셈)."""
    rows: list[tuple[float, ...]] = []
    start = 0.0
    for (x0, y0), (x1, y1) in zip(xy, xy[1:]):
        dx, dy = float(x1) - float(x0), float(y1) - float(y0)
        length = math.sqrt(dx * dx + dy * dy)
        if length > _EPS:
            rows.append((float(x0), float(y0), dx / length, dy / length, length, start))
        start += length
    return np.array(rows, dtype=np.float64).reshape(-1, 6)


def line_matcher(
    old_xy: list[tuple[float, float]],
    new_xy: list[tuple[float, float]],
    tol_m: float = SAME_LINE_TOL_M,
) -> Callable[[float], float | None]:
    """새 누가거리 → 같은 자리 옛 누가거리(옆 `tol_m` · 방향 30° 안) · 아니면 None."""
    olds, news = _segments(old_xy), _segments(new_xy)

    def match(chainage: float) -> float | None:
        if not len(olds) or not len(news):
            return None
        ahead = np.nonzero(chainage <= news[:, 5] + news[:, 4])[0]
        x, y, ux, uy, length, start = news[ahead[0] if len(ahead) else -1]
        along = min(max(chainage - start, 0.0), length)
        qx, qy = x + ux * along, y + uy * along
        facing = olds[olds[:, 2] * ux + olds[:, 3] * uy >= _SAME_DIRECTION_COS]
        if not len(facing):
            return None
        ox, oy, oux, ouy, olen, ostart = facing.T
        p = np.minimum(np.maximum((qx - ox) * oux + (qy - oy) * ouy, 0.0), olen)
        dx, dy = qx - (ox + oux * p), qy - (oy + ouy * p)
        gap = np.sqrt(dx * dx + dy * dy)
        best = int(np.argmin(gap))
        return float(ostart[best] + p[best]) if gap[best] <= tol_m else None

    return match


def carry_base(
    old_xy: list[tuple[float, float]],
    new_xy: list[tuple[float, float]],
    old_base: list[dict[str, float]],
    ground_pvi: list[dict[str, float]],
    tol_m: float = SAME_LINE_TOL_M,
) -> list[dict[str, float]]:
    """새 자동 꼭짓점 — 지반 첫 계획선 측점마다 같은 자리면 옛 자동선 높이 · 아니면 지반."""
    if not old_base or not ground_pvi:
        return [dict(node) for node in old_base]
    match = line_matcher(old_xy, new_xy, tol_m)
    base_s = np.array([float(n["chainage_m"]) for n in old_base], dtype=np.float64)
    base_z = np.array([float(n["elevation_m"]) for n in old_base], dtype=np.float64)
    out = []
    for node in ground_pvi:
        s = match(float(node["chainage_m"]))
        z = (
            float(node["elevation_m"])
            if s is None
            else round(float(np.interp(s, base_s, base_z)), 6)
        )
        out.append({"chainage_m": float(node["chainage_m"]), "elevation_m": z})
    return out


def carry_edits(
    old_xy: list[tuple[float, float]],
    new_xy: list[tuple[float, float]],
    old_base: list[dict[str, float]],
    edits: dict[str, Any] | None,
    stations: list[float],
    half_m: float,
    tol_m: float = SAME_LINE_TOL_M,
) -> dict[str, dict[str, float]]:
    """새 편집 — 새 측점마다 같은 자리 · 옛 사용자 꼭짓점이 닿은 자리면 옛 계획선 − 옛 자동선.

    닿은 자리 = 꼭짓점 위(그 꼭짓점이 사용자) · 앞뒤 꼭짓점 하나가 사용자. 종단곡선 R = 옛 키에서
    `half_m` 안 가장 가까운 같은 자리 측점. 새 자동 꼭짓점이 같은 자리에서 옛 자동선
    높이(`carry_base`)라 계획고가 그대로.
    """
    out: dict[str, dict[str, float]] = {"station_offsets": {}, "curve_radii": {}}
    if not old_base:
        return out
    match = line_matcher(old_xy, new_xy, tol_m)
    mapped: list[tuple[float, float]] = []
    for t in sorted({round(float(v), 3) for v in stations if math.isfinite(float(v))}):
        s = match(t)
        if s is not None:
            mapped.append((t, s))
    base_s = np.array([float(n["chainage_m"]) for n in old_base], dtype=np.float64)
    base_z = np.array([float(n["elevation_m"]) for n in old_base], dtype=np.float64)
    pvi_s, pvi_z, sources = resolve_pvi(base_s, base_z, (edits or {}).get("station_offsets") or {})

    def touched(s: float) -> bool:
        on = np.nonzero(np.abs(pvi_s - s) < _ON_NODE_M)[0]
        if len(on):
            return sources[int(on[0])] == "user"
        ahead = np.nonzero(pvi_s > s)[0]
        nxt = int(ahead[0]) if len(ahead) else -1
        return nxt > 0 and (sources[nxt - 1] == "user" or sources[nxt] == "user")

    for t, s in mapped:
        if touched(s):
            out["station_offsets"][chainage_key(t)] = round(
                float(np.interp(s, pvi_s, pvi_z)) - float(np.interp(s, base_s, base_z)), 6
            )
    gaps: dict[str, float] = {}
    for key, value in ((edits or {}).get("curve_radii") or {}).items():
        try:
            chainage = float(key)
        except (TypeError, ValueError):
            continue
        if not math.isfinite(chainage) or value is None:
            continue
        slot, best = None, math.inf
        for t, s in mapped:
            gap = round(abs(s - chainage), 3)  # mm — 가운데 같은 거리면 앞 측점
            if gap <= half_m and gap < best:
                slot, best = chainage_key(t), gap
        if slot is None or best >= gaps.get(slot, math.inf):
            continue
        gaps[slot] = best
        out["curve_radii"][slot] = float(value)
    return out


def carry_policy(
    old_xy: list[tuple[float, float]],
    new_xy: list[tuple[float, float]],
    old_alignment: dict[str, Any],
    new_alignment: dict[str, Any],
) -> dict[str, Any]:
    """노선 [저장] 정책(432) — 새 정책에 옛 기본 L · 옛 정책 없으면 새 정책. 바뀐 자리 새 꼭짓점은
    새 지반 첫 계획선(직선화 438 · `carry_base`)."""
    new_policy = dict(new_alignment.get("policy") or {})
    old_policy = old_alignment.get("policy") or {}
    if not old_policy:
        return new_policy
    return {**new_policy, "default_curve_length_m": old_policy.get("default_curve_length_m")}