Aislo/D03_PlanAnalysis/D03_PlanAnalysis_Analyze.py

"""D 예정노선 2단계 분석 — 연도 구간마다 B05 노선 추천 엔진으로 지점을 이음 (요구 345 · PLAN 530-2).

- 구간 = [P(i-1), P(i)] 한 번씩 `generate_initial_route` — 시종점을 옮기지 않아 연도 경계가 맞물림
- 갈래 넷(`PLAN_MODES` · follow 뺌 — 두 점이면 직선) · 연도마다 고름
- 추천 기준 = 프로젝트 임도 종류 · 설계속도 → B 법정 함수 · 지형 구분은 B 폴백(`TERRAIN_TYPE`)
- 계류 이격 = 도엽 하천선(B 로더 그대로) · 복도 반폭 = `PLAN_CORRIDOR_M`(D 만)
- 연도 경계는 꺾인 채(경계점 = 측점) · 누가거리 = 등록 시작 누가부터 고른 갈래 길이로 이어 셈
- 파생 저장본 = `D03_PlanAnalysis/_derived/route_<연도>.json`(입력 지문 같으면 다시 안 셈)

계산 자리 = ③ 서버 단독(B 추천과 같음).

⚠ 파이썬·TS 짝 — `pick_mode` · `chain` · `joined_line` ↔ `logic/D03_PlanAnalysis_Plan.ts`
`pickMode` · `chainRows` · `joinedLine`(화면이 고르는 즉시) · 거울 시험 `test_530_3_plan_logic.py`.
"""

from __future__ import annotations

import hashlib
import json
import logging
from pathlib import Path
from typing import Any, Callable

from B05_Profile.B05_Profile_Engine_Grade import (
    legal_grade_limit_pct,
    legal_plan_radius_min_m,
    resolve_design_speed,
)
from B05_Profile.B05_Profile_Engine_RouteInitial import (
    generate_initial_route,
    load_terrain_and_streams,
)
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint
from common_util.common_util_json import atomic_write_json
from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER
from D03_PlanAnalysis.D03_PlanAnalysis_Terrain import PLAN_CORRIDOR_M, models_dir_of

logger = logging.getLogger(__name__)

PLAN_MODES = ("gradeline", "earthwork", "gentle", "weighted")
# 도엽 DEM(스무딩) — B 「수치지형도」 분석 지형과 같은 고르기
SELECTION = {"source_filter": SHEET_ANALYSIS_FILTER, "method": "dtm", "smooth": True}
# D 등록에 지형 구분 칸 없음 — B `_road_settings` 못 읽을 때 폴백 그대로(가장 완화)
TERRAIN_TYPE = "special"
ALGO = 1

Progress = Callable[[int, str], None]
Loader = Callable[..., tuple[Any, list]]


def derived_dir(project_root: Path) -> Path:
    return project_root / "D03_PlanAnalysis" / "_derived"


def plan_criteria(road_type: str | None, design_speed_kph: int | None) -> dict[str, Any]:
    """추천 기준 — 종단 상한(비율) · 최소 R · 설계속도(B 법정 함수)."""
    grade_class = str(road_type or "work")
    speed = resolve_design_speed(grade_class, design_speed_kph)
    return {
        "max_grade": legal_grade_limit_pct(grade_class, TERRAIN_TYPE, False, speed) / 100.0,
        "min_radius_m": legal_plan_radius_min_m(speed, TERRAIN_TYPE),
        "design_speed": speed,
    }


def _point(value: Any) -> tuple[float, float]:
    """엔진 출력 자리수(0.1 mm)로 — 노선 끝점 = 지점 그대로 · 연도 경계가 정확히 맞물림."""
    return (round(float(value[0]), 4), round(float(value[1]), 4))


def _same_point(a: Any, b: Any) -> bool:
    return abs(float(a[0]) - float(b[0])) < 1e-3 and abs(float(a[1]) - float(b[1])) < 1e-3


def read_segment(project_root: Path, seg: dict[str, Any]) -> dict[str, Any] | None:
    """저장본 — 지점이 지금과 같을 때만(입력 지문은 분석 때 다시 봄)."""
    try:
        stored = json.loads(
            (derived_dir(project_root) / f"route_{int(seg['year'])}.json").read_text("utf-8")
        )
    except (OSError, ValueError):
        return None
    if _same_point(stored["start"], seg["start"]) and _same_point(stored["end"], seg["end"]):
        return stored
    return None


def _route_view(result: dict[str, Any]) -> dict[str, Any]:
    keep = ("planned", "nodes", "curves", "metrics", "relaxed", "terrain_bound_pct")
    return {key: result.get(key) for key in keep}


def analyze_segment(
    project_root: Path,
    seg: dict[str, Any],
    terrain: Any,
    streams: list,
    criteria: dict[str, Any],
    corridor_m: float = PLAN_CORRIDOR_M,
) -> dict[str, Any]:
    """연도 구간 하나 · 갈래 넷 — 입력 지문이 저장본과 같으면 그대로(`stored` True)."""
    expected = [_point(seg["start"]), _point(seg["end"])]
    inputs = {
        "weights": None,
        "max_grade": criteria["max_grade"],
        "min_radius_m": criteria["min_radius_m"],
        "stream_offset_m": None,
        "grade_smooth_m": None,
        "target_grade": None,
        "with_profile": False,
        "design_speed": criteria["design_speed"],
    }
    keys = [
        request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0]
        for mode in PLAN_MODES
    ]
    fingerprint = hashlib.blake2b(
        repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16
    ).hexdigest()
    stored = read_segment(project_root, seg)
    if stored is not None and stored.get("fingerprint") == fingerprint:
        return {**stored, "stored": True}
    routes: dict[str, Any] = {}
    for mode in PLAN_MODES:
        try:
            routes[mode] = _route_view(
                generate_initial_route(
                    expected, terrain, streams, mode=mode, corridor_m=corridor_m, **inputs
                )
            )
        except ValueError as error:  # NoRoute 포함 — 그 갈래만 빔
            routes[mode] = {"error": str(error)}
    result = {
        "year": int(seg["year"]),
        "start": list(expected[0]),
        "end": list(expected[1]),
        "fingerprint": fingerprint,
        "corridor_m": float(corridor_m),
        "criteria": {
            "max_grade_pct": round(criteria["max_grade"] * 100.0, 2),
            "min_radius_m": criteria["min_radius_m"],
            "design_speed_kph": criteria["design_speed"],
            "terrain_type": TERRAIN_TYPE,
        },
        "routes": routes,
    }
    atomic_write_json(derived_dir(project_root) / f"route_{int(seg['year'])}.json", result)
    return {**result, "stored": False}


def analyze_plan(
    project_root: Path,
    segments: list[dict[str, Any]],
    criteria: dict[str, Any],
    *,
    loader: Loader = load_terrain_and_streams,
    corridor_m: float = PLAN_CORRIDOR_M,
    on_progress: Progress | None = None,
) -> list[dict[str, Any]]:
    """연도 구간마다 지형(연도 DEM · 연도 비용면 캐시)을 읽어 `analyze_segment`."""
    results = []
    for i, seg in enumerate(segments):
        year = int(seg["year"])
        if on_progress is not None:
            on_progress(100 * i // len(segments), f"{year}년 구간 노선 추천 중")
        terrain, streams = loader(
            project_root,
            SELECTION,
            models_dir_of(project_root, year),
            derived_dir(project_root) / f"cost_{year}",
        )
        results.append(analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m))
    if on_progress is not None:
        on_progress(100, "분석 끝")
    return results


def pick_mode(result: dict[str, Any], wanted: str | None) -> str | None:
    """고른 갈래 — 없거나 빈 갈래면 차례대로 첫 성공 갈래."""
    routes = result.get("routes") or {}
    if wanted in routes and "error" not in routes[wanted]:
        return wanted
    return next((m for m in PLAN_MODES if m in routes and "error" not in routes[m]), None)


def chain(
    results: list[dict[str, Any]], choices: dict[int, str | None], start_chainage_m: float
) -> list[dict[str, Any]]:
    """연도별 고른 갈래 · 시작 · 끝 누가(이어 셈) — 앞 연도 끝 누가 = 다음 연도 시작 누가."""
    at = float(start_chainage_m or 0.0)
    rows = []
    for result in results:
        mode = pick_mode(result, choices.get(int(result["year"])))
        length = float(result["routes"][mode]["metrics"]["length_m"]) if mode else 0.0
        rows.append(
            {
                "year": int(result["year"]),
                "mode": mode,
                "length_m": round(length, 3),
                "start_chainage_m": round(at, 3),
                "end_chainage_m": round(at + length, 3),
            }
        )
        at += length
    return rows


def joined_line(results: list[dict[str, Any]], rows: list[dict[str, Any]]) -> list[list[float]]:
    """고른 갈래를 이은 중심선 — 경계점 한 번만(꺾인 채)."""
    line: list[list[float]] = []
    for result, row in zip(results, rows):
        if row["mode"] is None:
            continue
        planned = result["routes"][row["mode"]]["planned"]
        line.extend(planned[1:] if line and _same_point(line[-1], planned[0]) else planned)
    return line