Aislo/B05_wf2_Route/B05_wf2_Route_Router_Confirm.py

"""B05 경로 확정 시 종단 정본 파일에 비정규 측점·상단측 변경을 병합하는 헬퍼.

라우터(`B05_wf2_Route_Router.py`)의 confirm 엔드포인트가 호출한다. 파일 기반 병합이라
재확정해도 중복이 생기지 않으며, 모든 호출은 비치명적(실패해도 경로 확정은 진행)이다.
"""

import asyncio
import json
from pathlib import Path
from typing import Any
from uuid import UUID

import aiomysql

from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
from B05_wf2_Route.B05_wf2_Route_Engine_Sections import generate_irregular_sections
from B05_wf2_Route.B05_wf2_Route_Engine_Sections_Core import SectionGenerationOptions
from B05_wf2_Route.B05_wf2_Route_Repository import get_surface_crs_epsg
from B05_wf2_Route.B05_wf2_Route_Schema import RouteConfirmRequest
from B06_wf3_ProfileCross.B06_wf3_ProfileCross_Repository import (
    get_latest_section_options,
    get_longitudinal_section,
)
from common_util.common_util_json import atomic_write_json
from common_util.common_util_storage import resolve_stored_project_path


def _section_options_from_stored(stored: dict[str, Any] | None) -> SectionGenerationOptions:
    """저장된 종횡단 옵션(단일 소스)만으로 옵션을 재구성한다(확정 시 비정규 측점 샘플링용)."""
    defaults = SectionGenerationOptions()
    stored = stored or {}
    return SectionGenerationOptions(
        station_interval_m=stored.get("station_interval_m") or defaults.station_interval_m,
        cross_half_width_m=stored.get("cross_half_width_m") or defaults.cross_half_width_m,
        cross_sample_interval_m=stored.get("cross_sample_interval_m")
        or defaults.cross_sample_interval_m,
        long_sample_interval_m=stored.get("long_sample_interval_m")
        or defaults.long_sample_interval_m,
        include_endpoint=defaults.include_endpoint,
    )


def _merge_uphill_overrides_into_longitudinal(
    project_root: Path, longitudinal_file_path: str, overrides: list[dict[str, Any]]
) -> None:
    """종단 정본 파일 stations의 uphill_side를 사용자 변경값으로 덮어쓴다.

    solve가 자동 판정한 상단측(=측구 방향)을 3D 램프 클릭으로 바꾼 경우, 확정 시점에
    정본에 반영해 B06이 값만 읽으면 되게 한다. 사용자 지정임을 소스 필드로 남긴다.
    """
    if not overrides:
        return
    root = project_root.resolve()
    path = (root / longitudinal_file_path).resolve()
    if root not in path.parents or not path.is_file():
        return
    data = json.loads(path.read_text(encoding="utf-8"))
    stations = data.get("stations")
    if not isinstance(stations, list):
        return
    by_chainage = {round(float(item["chainage_m"]), 3): str(item["side"]) for item in overrides}
    changed = False
    for station in stations:
        side = by_chainage.get(round(float(station.get("chainage_m", -1.0)), 3))
        if side is None:
            continue
        station["uphill_side"] = side
        station["uphill_side_source"] = "user"
        changed = True
    if changed:
        data["stations"] = stations
        atomic_write_json(path, data)


def _merge_irregular_into_longitudinal(
    project_root: Path, longitudinal_file_path: str, irregular_stations: list[dict[str, Any]]
) -> None:
    """종단 정본 파일의 stations에 비정규 측점을 병합한다(기존 비정규는 교체·정렬).

    B06 상세는 종단 파일의 stations를 읽고 그에 맞는 cross 파일만 노출하므로, 이 병합과
    (엔진이 이미 쓴) cross 파일만으로 다음 페이지에 횡단이 나타난다. 계획선·규칙 측점·표고
    샘플은 손대지 않는다. 파일 기반이라 재확정해도 중복이 생기지 않는다.
    """
    root = project_root.resolve()
    path = (root / longitudinal_file_path).resolve()
    if root not in path.parents or not path.is_file():
        return
    data = json.loads(path.read_text(encoding="utf-8"))
    stations = data.get("stations")
    if not isinstance(stations, list):
        return
    regular = [station for station in stations if station.get("kind") != "irregular"]
    merged = regular + list(irregular_stations)
    merged.sort(key=lambda station: float(station.get("chainage_m", 0.0)))
    data["stations"] = merged
    atomic_write_json(path, data)


async def _append_irregular_cross_sections(
    connection: aiomysql.Connection,
    project_id: UUID,
    route: dict[str, Any],
    request: RouteConfirmRequest,
) -> None:
    """확정 시 비정규 측점의 횡단을 생성해 종단 파일에 병합한다(파일 기반, 비치명적 호출용)."""
    stored_path = await get_project_storage_relative_path(connection, project_id)
    project_root = Path(resolve_stored_project_path(stored_path))
    stored_options = await get_latest_section_options(connection, project_id)
    crs_epsg = await get_surface_crs_epsg(connection, project_id, request.surface_model_id)
    irregular_stations = await asyncio.to_thread(
        generate_irregular_sections,
        project_root,
        str(route["route_data_path"]),
        request.filter_key,
        request.method,
        request.smooth,
        extra_stations=request.extra_stations(),
        options=_section_options_from_stored(stored_options),
        crs=f"EPSG:{crs_epsg}" if crs_epsg is not None else None,
    )
    if not irregular_stations:
        return
    longitudinal = await get_longitudinal_section(connection, project_id, route["id"])
    if longitudinal:
        await asyncio.to_thread(
            _merge_irregular_into_longitudinal,
            project_root,
            str(longitudinal["longitudinal_file_path"]),
            irregular_stations,
        )