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,
    same_seat,
    station_key,
    station_seat,
)
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 station_seats(
    stations: Sequence[dict[str, Any]], keys: dict[str, float]
) -> dict[str, tuple[float, float, tuple[float, float] | None] | None]:
    """측점 열쇠 → 그 측점 자리(`station_seat` · 같은 누가거리 0.001 m 줄 · 모르면 None)."""
    rows = {
        round(float(row["chainage_m"]), 3): row
        for row in stations
        if isinstance(row, dict) and row.get("chainage_m") is not None
    }
    return {key: station_seat(rows.get(round(at, 3))) for key, at in keys.items()}


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],
    seat_keys: Sequence[str] = (),
    seats: tuple[dict[str, Any], dict[str, Any]] | None = None,
) -> tuple[dict[float, dict[str, Any]], dict[float, dict[str, Any]], int]:
    """같은 열쇠 측점으로 횡단 입력을 넘김 → (입력{새 누가거리 0.001: 값}, 구조물 조정{…}, 버린 수).

    `old_designs` = 옛 노선 행 {누가거리 0.001: design}. 버린 수 = 설계가 있던 옛 측점 중 새 노선에
    같은 열쇠가 없는 것(「노선 밖 고침」) · MC 가 움직인 곡선 열쇠는 가까운 새 곡선 측점
    (`curve_moves`). 자리 칸(`seat_keys` · 요구 275-3)은 옛 · 새 자리(`seats` = 열쇠 → 자리)가
    같은(`same_seat`) 측점만 넘김 · 새 자리 = 기본값(TS `carryKeyedStates` 와 같은 규칙).
    """
    moves = curve_moves(list(old_keys.items()), list(new_keys.items()))
    old_seats, new_seats = seats or ({}, {})
    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_key = moves.get(key, key)
        new_at = new_keys.get(new_key)
        if new_at is None:
            dropped += 1
            continue
        seated = same_seat(old_seats.get(key), new_seats.get(new_key))
        names = [n for n in input_keys if seated or n not in seat_keys]
        names += [n for n in seat_keys if seated and n not in names]
        carried = {name: design[name] for name in names 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


def route_moved_stations(
    stations: Sequence[dict[str, Any]],
    old_xy: Sequence[Sequence[float]],
    new_xy: Sequence[Sequence[float]],
) -> list[float]:
    """노선 바뀜으로 옮긴 시설 측점 — 관 · 구조물 측점(`structure` 이름 있음) 중 옛 선과
    같은 자리가 아닌 것(`line_matcher` 짝 없음 · 459). 같은 자리로 따라간 시설은 뺌(옛 높이
    그대로). TS 짝 = `routeMovedStations`(B05_Profile/logic/B05_Profile_Pipeline_Grade.ts) ·
    거울 시험 = test_459_route_seat_mirror.py."""
    from B05_Profile.B05_Profile_Engine_Grade_Carry import line_matcher

    match = line_matcher([tuple(p) for p in old_xy], [tuple(p) for p in new_xy])
    return sorted(
        float(station["chainage_m"])
        for station in stations
        if station.get("structure") and match(float(station["chainage_m"])) is None
    )


def seat_route_moved(
    alignment: dict[str, Any], born: Sequence[float], held: Sequence[float] = ()
) -> list[dict[str, Any]]:
    """옮긴 시설 측점(`born`)에 427 규칙으로 꼭짓점을 다시 세움(459 · 요구 270) — 그 자리
    옛 꼭짓점(옮겨 붙은 지반 · 옛 높이)을 걷고 원지반 지남 · 이웃 1 mm 맞춤 · 균형 종단이면
    옮기기 전 계획선 높이(`_seat_born_nodes`). `alignment` = 새 지반 표본 · 정책 · 옮겨 붙인
    `base_pvi` · `held` = 높이 고친 측점. TS 짝 = `seatRouteMoved`."""
    from B05_Profile.B05_Profile_Router_Confirm import _SAME_STATION_M, _seat_born_nodes

    base = alignment.get("base_pvi") or []
    if not born:
        return base
    kept = [
        node
        for node in base
        if all(abs(float(node["chainage_m"]) - value) >= _SAME_STATION_M for value in born)
    ]
    return _seat_born_nodes(alignment, kept, list(born), list(held))