Aislo/B05_Profile/B05_Profile_Route_Save_Parts.py

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

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

  - 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
    (사용자 관 편집 지킴).
  - 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
  - 관 최대 간격 보충 없음(369) — 간격 보충은 배수유역 [자동 제안]만.
  - 측점 열쇠 = `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` 일곱 + `ROUTE_CARRY_KEYS`(사면 쉬프트 ·
    소단 깊이 · 측점 폭 · 메모 · 상세 표 · 417) · 관 · 구조물 열쇠면
    `STRUCTURE_ADJUST_KEYS` 도(구조물 조정은 구조물 따라) · 열쇠가 사라진 측점 수 =
    「노선 밖 고침」.
"""

from __future__ import annotations

import copy
import math
from typing import Any, Sequence

from shapely.geometry import LineString, Point

from common_util.common_util_drainage_pipes import (
    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 curve_moves, station_key
from common_util.common_util_structure_station import plan_stations, snap_structure_station
from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M

# 옛 노선에서 이만큼(m) 넘게 떨어진 자리 = 노선이 바뀐 자리(교차점 · 보충 대상).
MOVED_AWAY_M = 0.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 _facility_owners(
    pipes: Sequence[PipePoint], structures: Sequence[Any]
) -> dict[float, tuple[str, str | None, str]]:
    """측점 자리(0.01 m) → (관 · 구조물, id, 역할 anchor · start · end). 같은 자리면 관이 먼저
    (`plan_stations` 이름 규칙)."""
    owners: dict[float, tuple[str, str | None, str]] = {}
    for pipe in pipes:
        for mark in plan_stations([pipe], []):
            owners.setdefault(
                round(float(mark["chainage_m"]), 2),
                ("pipe", getattr(pipe, "id", None), str(mark.get("role") or "anchor")),
            )
    for item in structures:
        for mark in plan_stations([], [item]):
            owners.setdefault(
                round(float(mark["chainage_m"]), 2),
                (
                    "structure",
                    getattr(item, "structure_id", None),
                    str(mark.get("role") or "anchor"),
                ),
            )
    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` 한 벌) → 누가거리.

    시종점(맨 앞 · 맨 끝 = 관 · 구조물이 얹혀도 `end:BP/EP` · 화면 `buildStations` · B06
    `stationKeysOf` 와 같음 · 452) > 관 · 구조물 자리(구간형 시작 · 끝 `:start` · `:end`) >
    곡선 측점(`curve:` · 252-2) > 일반 격자(간격 배수인 것만) · id 없는 관 · 구조물은 `at:` ·
    같은 열쇠가 둘이면 뒤엣것은 `at:`(B06 `stationKeysOf` 와 같음).
    """
    owners = _facility_owners(pipes, structures)
    rows = [row for row in stations if row.get("chainage_m") is not None]
    values = [float(row["chainage_m"]) for row in rows]
    out: dict[str, float] = {}
    for index, chainage in enumerate(values):
        owner = owners.get(round(chainage, 2))
        if index == 0:
            key = station_key("bp", chainage, interval_m)
        elif index == len(values) - 1:
            key = station_key("ep", chainage, interval_m)
        elif owner is not None:
            key = station_key(owner[0], chainage, interval_m, owner[1], role=owner[2])
        elif rows[index].get("kind") == "curve":
            key = station_key("curve", 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}. 버린 수 = 설계가 있던 옛 측점 중 새 노선에
    같은 열쇠가 없는 것(「노선 밖 고침」) · MC 가 움직인 곡선 열쇠는 가까운 새 곡선 측점
    (`curve_moves`).
    """
    moves = curve_moves(list(old_keys.items()), list(new_keys.items()))
    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(moves.get(key, 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