Aislo/B05_Profile/B05_Profile_Router_Confirm.py

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

라우터(`B05_Profile_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_Profile.B05_Profile_Engine_Sections import generate_irregular_sections
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
from B05_Profile.B05_Profile_Schema import RouteConfirmRequest
from B06_Section.B06_Section_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)


async def sync_uphill_overrides_into_designs(
    connection: aiomysql.Connection,
    project_id: UUID,
    route_id: int,
    project_root: Path,
    longitudinal_file_path: str,
    overrides: list[dict[str, Any]],
) -> None:
    """상단측 변경 측점의 **저장된 횡단 설계**를 새 방향으로 재계산해 저장한다.

    종단 정본(stations.uphill_side)만 바꾸면 저장 설계(design.section_mode/ditch_side)가
    옛 방향으로 남아 B06이 그것을 계속 표시하고, 종횡단 확정의 역반영
    (ditch_side→uphill_side)이 사용자 지정을 다시 옛 방향으로 되돌린다
    (2026-08-06 13측점 보고: uphill=right vs design=left_cut 순환 덮어쓰기).
    편절은 절토측을, 양절은 측구측을 새 방향으로 맞추고 기하를 즉시 재계산한다.
    양성(both_fill)은 측구가 없으므로 건드리지 않는다.
    """
    # 지연 import — B06 라우터 모듈 로드는 이 함수가 실제 불릴 때만 필요하다.
    from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args
    from B06_Section.B06_Section_Repository import get_cross_section_designs
    from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
    from B06_Section.B06_Section_Router_Design import (
        ford_drop_at,
        ford_surface_drops,
        pavement_suggestions,
        read_cross_design_inputs,
        resolve_longitudinal_path,
    )

    if not overrides:
        return
    by_chainage = {round(float(item["chainage_m"]), 3): str(item["side"]) for item in overrides}
    designs = await get_cross_section_designs(connection, route_id)
    # 표준단면 수치는 종단 정본 options에 병합 저장돼 있다(종횡단 확정 시) — 없으면 기본값.
    longitudinal = await get_longitudinal_section(connection, project_id, route_id)
    stored_standard = None
    if longitudinal and isinstance(longitudinal.get("data"), dict):
        options = longitudinal["data"].get("options")
        if isinstance(options, dict):
            stored_standard = options.get("standard_cross_section")
    # 다시 계산할 측점만 먼저 고른다 — 파일·계산은 아래에서 **스레드 한 번**에 몰아 한다.
    jobs: list[tuple[float, dict[str, Any], str, str]] = []
    for record in designs:
        chainage = round(float(record["chainage_m"]), 3)
        side = by_chainage.get(chainage)
        design = record.get("design")
        if side is None or not isinstance(design, dict):
            continue
        mode = design.get("section_mode")
        if mode == "both_fill":
            continue
        next_mode = f"{side}_cut" if mode in ("left_cut", "right_cut") else mode
        if next_mode == mode and design.get("ditch_side") == side:
            continue
        jobs.append((chainage, design, str(next_mode), side))
    if not jobs:
        return

    def _recompute() -> list[tuple[float, dict[str, Any]]]:
        """측점마다 종단 정본을 다시 열던 것을 한 번으로 줄인다(측점당 13.8ms 였다)."""
        root = Path(project_root)
        longitudinal_path = resolve_longitudinal_path(root, longitudinal_file_path)
        longitudinal_json = json.loads(longitudinal_path.read_text(encoding="utf-8"))
        preloaded = (longitudinal_path, longitudinal_json, pavement_suggestions(longitudinal_json))
        ford_drops = ford_surface_drops(root)
        out: list[tuple[float, dict[str, Any]]] = []
        for chainage, design, next_mode, side in jobs:
            samples, design_elevation, pavement_suggested, cross_record = read_cross_design_inputs(
                root, longitudinal_file_path, float(chainage), preloaded
            )
            next_design = compute_cross_design(
                samples,
                design_elevation,
                ground_type=str(design.get("ground_type", "soil")),
                section_mode=next_mode,
                ditch_side=side,
                ditch_type=str(design.get("ditch_type", "standard")),
                paved=bool(design.get("paved", False)),
                standard=stored_standard,
                rock_boundary_offset_m=design.get("rock_boundary_offset_m"),
                two_stage_slope=bool(design.get("two_stage_slope", True)),
                ditch_enabled=design.get("ditch_enabled"),
                surface_drop_m=ford_drop_at(float(chainage), ford_drops),
                **curve_widening_args(cross_record),
            )
            next_design["status"] = design.get("status", "provisional")
            next_design["pavement_suggested"] = design.get("pavement_suggested", pavement_suggested)
            # 개별 표시 반폭 등 계산과 무관한 표시 설정은 그대로 이월한다.
            if design.get("display_half_width_m") is not None:
                next_design["display_half_width_m"] = design["display_half_width_m"]
            out.append((float(chainage), next_design))
        return out

    # 쓰기도 한 문장으로 — 행마다 내면 원격 DB 왕복이 측점 수만큼 난다(건당 24.5ms).
    await merge_cross_section_designs(
        connection,
        route_id=route_id,
        entries=await asyncio.to_thread(_recompute),
        replace=True,
        project_id=project_id,
    )


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,
        )