Aislo/B06_Section/B06_Section_Router_DesignSave_Route.py

"""노선 바뀐 [저장] 한 본문 몫(PLAN 472-6 6-B) — `B06_Section_Router_DesignSave` 가 부름.

노선 초안이 오면 차례 맨 앞:

    노선 저장(`/route/replan` 체인 · 관 · 구조물 · 추가 측점 좌표 따라감 · 종횡단 · 계획고 ·
    횡단 이월)
    → 관 · 구조물 초안 다시 얹기(Node `facility.rebase` · 서버 따라감 위에)
    → 6-A 차례(새 노선 · 계획고 = 서버 옮김 위에 사용자 손댐만 · 498-2)
    → 횡단 = 옛 노선 세션 칸을 새 측점에 넘긴 patch(Node `stations.carry`)

브라우저가 하던 다시 얹기(`rebasePipeDraft` · `carryAfterRouteSave`)와 같은 TS 를 서버가 부름 —
브라우저는 노선 초안 + 사용자 칸만 보냄. 손 상단측은 안 넘김(요구 274 · 새 지형이 정함).
"""

from __future__ import annotations

from typing import Any
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field

from common_util.common_util_calc_runner import run_calc


class RouteCarryRequest(BaseModel):
    """옛 노선 세션 칸 — 브라우저 `routeCarryInputs` 꼴 그대로(넘김 규칙은 TS 한 벌)."""

    model_config = ConfigDict(extra="forbid")

    #: 칸 이름 → 세션 원본(측점 열쇠 칸 다섯 · 누가거리 칸 · `kindreset`)
    states: dict[str, Any] = Field(default_factory=dict)
    #: 옛 노선 측점 열쇠 목록 [{key, chainageM}] · 브라우저가 못 세웠으면 null
    from_keys: list[dict[str, Any]] | None = None
    #: 측점을 덮는 관 · 구조물 [{id, kind, at, role}]
    owners: list[dict[str, Any]] = Field(default_factory=list)
    #: 관 세션 장부 {next, ids: [{id, at}]}
    pipe_ids: dict[str, Any] | None = None


class AlignmentCleared(BaseModel):
    """노선 바뀐 [저장]에서 사용자가 지운 계획고 열쇠(498-2 · 브라우저 `gradeTouch`)."""

    model_config = ConfigDict(extra="forbid")

    station_offsets: list[str] = Field(default_factory=list)
    curve_radii: list[str] = Field(default_factory=list)


class StepFailed(Exception):
    def __init__(self, step: str, message: str) -> None:
        super().__init__(message)
        self.step = step


def _node(ids: list[str], inputs: dict[str, Any], step: str) -> dict[str, Any]:
    out = run_calc(ids, inputs)
    if not isinstance(out, dict) or ids[0] not in out:
        raise StepFailed(step, "다시 얹기 계산(Node)을 못 돌렸습니다")
    return out[ids[0]]


def _pipes(listing: Any) -> list[dict[str, Any]]:
    """관 GET 응답 → 관 목록(지도용 lonlat 뺌 · 브라우저 `fetchDetailPipePoints` 짝)."""
    points = listing.get("pipe_points") if isinstance(listing, dict) else None
    return [{k: v for k, v in point.items() if k != "lonlat"} for point in points or []]


async def saved_facilities(project_id: UUID) -> dict[str, Any]:
    """노선 저장 직전 저장 관 · 구조물(다시 얹기 · 넘김의 `before`)."""
    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points
    from B05_Profile.B05_Profile_Structures_Router import read_structures

    pipes = await get_pipe_points(project_id)
    structures = await read_structures(project_id)
    return {
        "pipes": _pipes(pipes),
        "structures": [
            item.model_dump(mode="json") for item in getattr(structures, "structures", None) or []
        ],
    }


async def rebase_drafts(project_id: UUID, request: Any, before: dict[str, Any]) -> None:
    """관 · 구조물 초안을 서버가 따라 옮긴 정본 위에(Node `facility.rebase`) — 본문을 제자리에서
    고침.

    구조물 판번호 = 노선 저장 뒤 지금 판(따라감이 판을 올림 · 옛 길 = 저장 뒤 받은 판).
    """
    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points
    from B05_Profile.B05_Profile_Structures_Router import read_structures

    rebase: dict[str, Any] = {}
    points = (request.pipes or {}).get("points") if isinstance(request.pipes, dict) else None
    if points is not None:
        after = _pipes(await get_pipe_points(project_id))
        rebase["pipes"] = {"draft": points, "before": before["pipes"], "after": after}
    stored = None
    if request.structures is not None:
        stored = await read_structures(project_id)
        draft = [item.model_dump(mode="json") for item in request.structures.structures]
        after = [item.model_dump(mode="json") for item in stored.structures]
        rebase["structures"] = {"draft": draft, "before": before["structures"], "after": after}
    if not rebase:
        return
    out = _node(["facility.rebase"], {"rebase": rebase}, "route")
    if out.get("pipes") is not None:
        request.pipes = {**request.pipes, "points": out["pipes"]}
    if out.get("structures") is not None and stored is not None:
        request.structures = type(request.structures).model_validate(
            {"base_revision": stored.revision, "structures": out["structures"]}
        )


async def carry_cross(
    project_id: UUID,
    route_id: int,
    carry: RouteCarryRequest | None,
    before: dict[str, Any],
) -> dict[str, Any]:
    """옛 노선 세션 칸 → 새 노선 횡단 patch(Node `stations.carry`) → {patches, outside}."""
    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points
    from B06_Section.B06_Section_Router import get_section_detail

    detail = await get_section_detail(project_id, route_id)
    if not hasattr(detail, "model_dump"):
        raise StepFailed("cross", "새 노선 횡단 상세를 못 읽었습니다")
    inputs = {
        "detail": detail.model_dump(mode="json"),
        "pipes_before": before["pipes"],
        "pipes_after": _pipes(await get_pipe_points(project_id)),
        **(carry or RouteCarryRequest()).model_dump(mode="json"),
    }
    return _node(["stations.carry"], {"carry": inputs}, "cross")


def overlay_edits(
    carried: dict[str, Any], touch: dict[str, Any], cleared: AlignmentCleared
) -> dict[str, dict[str, float]]:
    """서버 옮김 편집 위에 사용자 손댐만(498-2) — 지운 · 고친 열쇠 자리(0.006 m 안)의 옮김 값을 걷고
    손댐 값을 얹음. 손대지 않은 옮김 열쇠는 서버 측점 그대로(브라우저 열쇠로 안 덮음)."""
    from B05_Profile.B05_Profile_Engine_Grade_Alignment import SAME_STATION_SNAP_M

    out: dict[str, dict[str, float]] = {}
    for field in ("station_offsets", "curve_radii"):
        merged = {str(key): float(value) for key, value in (carried.get(field) or {}).items()}
        mine = {str(key): float(value) for key, value in (touch.get(field) or {}).items()}
        for key in [*getattr(cleared, field), *mine]:
            for near in [k for k in merged if abs(float(k) - float(key)) < SAME_STATION_SNAP_M]:
                del merged[near]
        out[field] = {**merged, **mine}
    return out


async def touched_alignment(
    project_id: UUID, route_id: int, touch: Any, cleared: AlignmentCleared
) -> Any:
    """노선 바뀐 [저장] 계획고 본문(498-2) = 새 종단 편집(노선 차례 서버 옮김) ⊕ 사용자 손댐."""
    import json
    from pathlib import Path

    from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
    from B05_Profile.B05_Profile_Schema import ProfileAlignmentSaveRequest
    from B06_Section.B06_Section_Repository import get_longitudinal_section
    from common_util.common_util_storage import resolve_stored_project_path
    from config.config_db import get_db_pool

    async with get_db_pool().acquire() as connection:
        row = await get_longitudinal_section(connection, project_id, route_id)
        stored = await get_project_storage_relative_path(connection, project_id)
    if not row:
        raise StepFailed("alignment", "새 노선 종단을 못 읽었습니다")
    path = Path(resolve_stored_project_path(stored)) / str(row["longitudinal_file_path"])
    longitudinal = json.loads(path.read_text(encoding="utf-8"))
    carried = (longitudinal.get("profile_alignment") or {}).get("edits") or {}
    edits = overlay_edits(carried, touch.edits() if touch is not None else {}, cleared)
    return ProfileAlignmentSaveRequest(route_id=route_id, **edits)