Aislo/B05_Profile/B05_Profile_Route_Save_Parts.py

"""노선 저장 길(346-3) 순수 부품 — 관 따라감 · 새 교차점 기본 관 · 오목 지점 보충 · 측점 열쇠 ·
횡단 이월.

DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름). 차례 · 저장 · 되살림은
`B05_Profile_Route_Save.py`.

  - 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
    (사용자 관 편집 지킴).
  - 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
  - 관 최대 간격 보충 = 옛 노선에서 벗어난 자리의 **오목 지점**(지반 종단 낮은 곳)에만 — 격자 없음.
  - 측점 열쇠 = `common_util_station_key.station_key` 한 벌(sub2 짝) — 일반 `grid:k` ·
    관 `pipe:<id>` · 구조물 `structure:<id>` · 시종점 `end:BP` · `end:EP` ·
    id 없으면 `at:<누가거리 0.01 m>`.
  - 횡단 이월 = 같은 열쇠 측점에 `CARRY_INPUT_KEYS` 일곱 · 관 · 구조물 열쇠면
    `STRUCTURE_ADJUST_KEYS` 도(구조물 조정은 구조물 따라) · 열쇠가 사라진 측점 수 =
    「노선 밖 고침」.
"""

from __future__ import annotations

import copy
import math
from typing import Any, Sequence

import numpy as np
from shapely.geometry import LineString, Point

from common_util.common_util_drainage_pipes import (
    PIPE_SOURCE_SPACING,
    PIPE_SOURCE_STREAM,
    PipePoint,
    project_pipe_points,
)
from common_util.common_util_route_geometry import RouteVertex
from common_util.common_util_station_key import chainage_key as station_chainage_key
from common_util.common_util_station_key import station_key
from common_util.common_util_structure_station import plan_stations, snap_structure_station
from config.config_system import DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M

# 옛 노선에서 이만큼(m) 넘게 떨어진 자리 = 노선이 바뀐 자리(교차점 · 보충 대상).
MOVED_AWAY_M = 0.5
# 오목 지점을 찾을 지반 표본 간격(m) · 고름 창(표본 수) — DTM 잔물결을 오목으로 안 읽게.
# ponytail: 이동 평균 + 국소 최저 · 계곡이 얕아 놓치면 창을 줄이거나 흐름 강도와 섞을 것.
CONCAVE_STEP_M = 2.0
CONCAVE_WINDOW = 5
_SAME_M = 0.01


def vertices_of(xy: Sequence[Sequence[float]]) -> list[RouteVertex]:
    """평면 점열 → 누가거리 붙은 정점(z 0)."""
    out: list[RouteVertex] = []
    walked = 0.0
    for index, point in enumerate(xy):
        x, y = float(point[0]), float(point[1])
        if index:
            walked += math.hypot(x - out[-1].x, y - out[-1].y)
        out.append(RouteVertex(x=x, y=y, z=0.0, chainage_m=walked))
    return out


def follow_pipes(
    old_pipes: list[PipePoint], vertices: list[RouteVertex]
) -> list[tuple[int, PipePoint]]:
    """관이 새 노선 좌표를 따라감 → [(옛 차례 · 더한 관 -1, 새 관)] 누가거리 순 · 옛 목록 그대로."""
    pairs = [(index, copy.deepcopy(pipe)) for index, pipe in enumerate(old_pipes)]
    for _, pipe in pairs:
        project_pipe_points([pipe], vertices)
    pairs.sort(key=lambda pair: pair[1].chainage_m)
    return pairs


def _moved_away(old_line: LineString | None, x: float, y: float) -> bool:
    return old_line is None or old_line.distance(Point(x, y)) > MOVED_AWAY_M


def add_crossing_pipes(
    pairs: list[tuple[int, PipePoint]],
    crossings: Sequence[Any],
    old_xy: Sequence[Sequence[float]] | None,
    total_m: float,
) -> int:
    """옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관을 더함 · 더한 수."""
    old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
    added = 0
    for crossing in crossings:
        if not _moved_away(old_line, crossing.x, crossing.y):
            continue  # 옛 교차점 — 사용자가 지운 관을 되살리지 않음
        at = snap_structure_station(float(crossing.chainage_m), [total_m])
        if any(abs(pipe.chainage_m - at) < DRAINAGE_PIPE_MIN_SPACING_M for _, pipe in pairs):
            continue
        pairs.append(
            (-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_STREAM, x=crossing.x, y=crossing.y))
        )
        added += 1
    pairs.sort(key=lambda pair: pair[1].chainage_m)
    return added


def concave_chainages(chainage: np.ndarray, ground: np.ndarray) -> list[tuple[float, float]]:
    """지반 종단의 오목 지점(이동 평균 국소 최저) → [(누가거리, 고른 지반고)]."""
    if ground.size < 3:
        return []
    kernel = np.ones(CONCAVE_WINDOW) / CONCAVE_WINDOW
    pad = CONCAVE_WINDOW // 2
    smooth = np.convolve(np.pad(ground, pad, mode="edge"), kernel, mode="valid")
    return [
        (float(chainage[i]), float(smooth[i]))
        for i in range(1, smooth.size - 1)
        if smooth[i] < smooth[i - 1] and smooth[i] <= smooth[i + 1]
    ]


def fill_concave_pipes(
    pairs: list[tuple[int, PipePoint]],
    concave: list[tuple[float, float]],
    vertices: list[RouteVertex],
    old_xy: Sequence[Sequence[float]] | None,
) -> int:
    """관 사이(시종점 포함)가 최대 간격을 넘으면 옛 노선 밖 오목 지점 중 가장 낮은 곳에 관."""
    line = LineString([(vertex.x, vertex.y) for vertex in vertices])
    old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
    total = float(line.length)
    added = 0
    while True:
        edges = [0.0, *(pipe.chainage_m for _, pipe in pairs), total]
        best: tuple[float, float] | None = None
        for start, end in zip(edges, edges[1:]):
            if end - start <= DRAINAGE_PIPE_MAX_SPACING_M:
                continue
            for at, z in concave:
                if not (
                    start + DRAINAGE_PIPE_MIN_SPACING_M < at < end - DRAINAGE_PIPE_MIN_SPACING_M
                ):
                    continue
                point = line.interpolate(at)
                if _moved_away(old_line, point.x, point.y) and (best is None or z < best[1]):
                    best = (at, z)
        if best is None:
            return added
        at = snap_structure_station(best[0], [total])
        point = line.interpolate(at)
        pairs.append(
            (-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_SPACING, x=point.x, y=point.y))
        )
        pairs.sort(key=lambda pair: pair[1].chainage_m)
        concave = [item for item in concave if abs(item[0] - best[0]) > _SAME_M]
        added += 1


def _facility_owners(
    pipes: Sequence[PipePoint], structures: Sequence[Any]
) -> dict[float, tuple[str, str | None]]:
    """측점 자리(0.01 m) → (관 · 구조물, id). 같은 자리면 관이 먼저(`plan_stations` 이름 규칙)."""
    owners: dict[float, tuple[str, str | None]] = {}
    for pipe in pipes:
        for mark in plan_stations([pipe], []):
            owners.setdefault(
                round(float(mark["chainage_m"]), 2), ("pipe", getattr(pipe, "id", None))
            )
    for item in structures:
        for mark in plan_stations([], [item]):
            owners.setdefault(
                round(float(mark["chainage_m"]), 2),
                ("structure", getattr(item, "structure_id", None)),
            )
    return owners


def station_keys(
    stations: Sequence[dict[str, Any]],
    pipes: Sequence[PipePoint],
    structures: Sequence[Any],
    interval_m: float,
) -> dict[str, float]:
    """측점 열쇠(`common_util_station_key.station_key` 한 벌) → 누가거리.

    관 · 구조물 자리 > 시종점 > 일반 격자(간격 배수인 것만) · id 없는 관 · 구조물은 `at:` ·
    같은 열쇠가 둘이면 뒤엣것은 `at:`(B06 `stationKeysOf` 와 같음).
    """
    owners = _facility_owners(pipes, structures)
    values = [float(row["chainage_m"]) for row in stations if row.get("chainage_m") is not None]
    out: dict[str, float] = {}
    for index, chainage in enumerate(values):
        owner = owners.get(round(chainage, 2))
        if owner is not None:
            key = station_key(owner[0], chainage, interval_m, owner[1])
        elif index == 0:
            key = station_key("bp", chainage, interval_m)
        elif index == len(values) - 1:
            key = station_key("ep", chainage, interval_m)
        elif interval_m > 0 and abs(chainage - round(chainage / interval_m) * interval_m) < _SAME_M:
            key = station_key("regular", chainage, interval_m)
        else:
            continue
        if key in out:
            key = f"at:{station_chainage_key(chainage)}"
        out.setdefault(key, chainage)
    return out


def carry_cross_inputs(
    old_keys: dict[str, float],
    new_keys: dict[str, float],
    old_designs: dict[float, dict[str, Any]],
    input_keys: Sequence[str],
    adjust_keys: Sequence[str],
) -> tuple[dict[float, dict[str, Any]], dict[float, dict[str, Any]], int]:
    """같은 열쇠 측점으로 횡단 입력을 넘김 → (입력{새 누가거리 0.001: 값}, 구조물 조정{…}, 버린 수).

    `old_designs` = 옛 노선 행 {누가거리 0.001: design}. 버린 수 = 설계가 있던 옛 측점 중 새 노선에
    같은 열쇠가 없는 것(「노선 밖 고침」).
    """
    inputs: dict[float, dict[str, Any]] = {}
    adjusts: dict[float, dict[str, Any]] = {}
    dropped = 0
    for key, old_at in old_keys.items():
        design = old_designs.get(round(old_at, 3))
        if not isinstance(design, dict):
            continue
        new_at = new_keys.get(key)
        if new_at is None:
            dropped += 1
            continue
        carried = {name: design[name] for name in input_keys if design.get(name) is not None}
        if carried:
            inputs[round(new_at, 3)] = carried
        if key.startswith(("pipe:", "structure:")):
            adjust = {name: design[name] for name in adjust_keys if design.get(name) is not None}
            if adjust:
                adjusts[round(new_at, 3)] = adjust
    return inputs, adjusts, dropped