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
import math
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Callable

import numpy as np
import shapely
from shapely.geometry import LineString, MultiLineString

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 (
    _POOLS,
    _TERRAINS,
    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 config.config_system_terrain import PLAN_ANALYZE_WORKERS
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
# 계류 거르기 칸 · 지형 상자 여유(m) — `near_streams`
STREAM_TILE_M = 200.0
STREAM_PAD_M = 100.0

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 _engine_inputs(criteria: dict[str, Any]) -> dict[str, Any]:
    return {
        "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"],
    }


def segment_fingerprint(
    seg: dict[str, Any],
    terrain: Any,
    streams: list,
    criteria: dict[str, Any],
    corridor_m: float = PLAN_CORRIDOR_M,
) -> str:
    """구간 입력 지문 — 갈래 넷 B 저장 키 · 복도 · 판."""
    expected = [_point(seg["start"]), _point(seg["end"])]
    inputs = _engine_inputs(criteria)
    keys = [
        request_fingerprint(expected, terrain, streams, mode=mode, **inputs)[0]
        for mode in PLAN_MODES
    ]
    return hashlib.blake2b(
        repr((keys, float(corridor_m), ALGO)).encode(), digest_size=16
    ).hexdigest()


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 = _engine_inputs(criteria)
    fingerprint = segment_fingerprint(seg, terrain, streams, criteria, corridor_m)
    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 near_streams(terrain: Any, streams: list) -> list:
    """지형 상자 안 어느 점에서든 가장 가까운 계류일 수 있는 선만 — 나머지는 엔진 값에 안 닿음.

    칸마다 닿는 거리 = 칸 가운데 계류 거리 + 칸 반대각(거리는 1-립시츠) — 그 안에 드는 선만 남기면
    상자 안 점의 계류 거리 · 건너기 · 격자 거리가 통째와 같음(통째 거리 = 추천 시간의 큰 몫).
    """
    if not streams:
        return streams
    x0, x1 = float(np.min(terrain.x)) - STREAM_PAD_M, float(np.max(terrain.x)) + STREAM_PAD_M
    y0, y1 = float(np.min(terrain.y)) - STREAM_PAD_M, float(np.max(terrain.y)) + STREAM_PAD_M
    xs = np.linspace(x0, x1, max(1, math.ceil((x1 - x0) / STREAM_TILE_M)) + 1)
    ys = np.linspace(y0, y1, max(1, math.ceil((y1 - y0) / STREAM_TILE_M)) + 1)
    lo_x, lo_y = np.meshgrid(xs[:-1], ys[:-1])
    hi_x, hi_y = np.meshgrid(xs[1:], ys[1:])
    tiles = shapely.box(lo_x.ravel(), lo_y.ravel(), hi_x.ravel(), hi_y.ravel())
    reach = shapely.distance(shapely.centroid(tiles), MultiLineString(streams)) + 0.5 * math.hypot(
        xs[1] - xs[0], ys[1] - ys[0]
    )
    lines = np.array([LineString(line) for line in streams], dtype=object)
    keep = (shapely.distance(lines[:, None], tiles[None, :]) <= reach[None, :]).any(axis=1)
    return [line for line, kept in zip(streams, keep) if kept]


def _load(project_root: Path, year: int, loader: Loader) -> tuple[Any, list]:
    terrain, streams = loader(
        project_root,
        SELECTION,
        models_dir_of(project_root, year),
        derived_dir(project_root) / f"cost_{year}",
    )
    return terrain, near_streams(terrain, streams)


def _segment_job(
    project_root: Path,
    seg: dict[str, Any],
    criteria: dict[str, Any],
    corridor_m: float,
    loader: Loader,
) -> dict[str, Any]:
    """연도 구간 하나 — 지형 읽기부터(제 프로세스면 끝날 때 비용면 · 묶음 메모리를 놓음)."""
    terrain, streams = _load(project_root, int(seg["year"]), loader)
    return analyze_segment(project_root, seg, terrain, streams, criteria, corridor_m)


def _worker_job(*args: Any) -> dict[str, Any]:
    """나란히 셈 프로세스 — 구간 하나 뒤 B 기억(비용면 · 묶음)을 비워 다음 구간 메모리를 놓음."""
    try:
        return _segment_job(*args)
    finally:
        _POOLS.clear()
        _TERRAINS.clear()


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,
    workers: int = PLAN_ANALYZE_WORKERS,
) -> list[dict[str, Any]]:
    """연도 구간마다(연도 DEM · 연도 비용면 캐시) `analyze_segment`.

    다시 셀 구간이 둘 이상이면 프로세스 `workers` 개로 나란히 — 구간끼리 입력 · 저장본이 따로라
    결과 같음. `workers` 1 = 이 프로세스(가짜 지형 시험)."""

    def report(percent: int, message: str) -> None:
        if on_progress is not None:
            on_progress(percent, message)

    results: list[dict[str, Any] | None] = [None] * len(segments)
    todo = []
    for i, seg in enumerate(segments):
        stored = read_segment(project_root, seg)
        if stored is not None:  # 저장본이 있을 때만 이 프로세스에서 지형을 읽어 지문을 봄
            terrain, streams = _load(project_root, int(seg["year"]), loader)
            if stored.get("fingerprint") == segment_fingerprint(
                seg, terrain, streams, criteria, corridor_m
            ):
                results[i] = {**stored, "stored": True}
                continue
        todo.append(i)

    count = min(workers, len(todo))
    if count <= 1:
        for n, i in enumerate(todo):
            report(100 * n // len(segments), f"{int(segments[i]['year'])}년 구간 노선 추천 중")
            results[i] = _segment_job(project_root, segments[i], criteria, corridor_m, loader)
    else:
        years = "·".join(str(int(segments[i]["year"])) for i in todo)
        report(0, f"{years}년 구간 노선 추천 중(나란히 {count})")
        context = multiprocessing.get_context("spawn")
        with ProcessPoolExecutor(count, mp_context=context) as pool:
            futures = {
                pool.submit(_worker_job, project_root, segments[i], criteria, corridor_m, loader): i
                for i in todo
            }
            for n, future in enumerate(as_completed(futures), 1):
                i = futures[future]
                results[i] = future.result()
                report(
                    100 * n // len(todo), f"{int(segments[i]['year'])}년 구간 끝 ({n}/{len(todo)})"
                )
    report(100, "분석 끝")
    return results  # type: ignore[return-value]


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