Aislo/B05_Profile/B05_Profile_Engine_RouteInitial.py

"""초기 계획노선 갈래 — 예상노선 추종 · 토공 최소 · 기울기 순한 노선 · 비중(PLAN 20 · 23장).

계산 자리 = 서버 단독(③). 결과는 화면 캐시용 — 이 모듈은 파일을 쓰지 않는다. 적용은 화면이
받은 노드로 기존 `POST /route/replan`([확인])을 부른다.

갈래(`mode`):
  · follow    — 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv`)과 같은 규칙 · 같은 선.
                제약 탐색은 하지 않고, 넘은 곳은 `metrics.violations` 에 뜬다.
  · earthwork — 후보 중 절 · 성토 합(`cut_m3 + fill_m3`) 최소.
  · gentle    — 후보 중 최대 종단기울기 최소(같으면 평균 최소).
  · weighted  — 후보 중 예상노선 이격 · 절성토 · 기울기(최대, 같으면 평균) 세 값을 후보
                범위로 0~1 정규화해 비중(합 100)으로 더한 값 최소. 끝값은 그 갈래와 같은 선 —
                토공 100 = 토공 최소 · 기울기 100 = 기울기 순한 · 예상 100 = 추적 선 그대로(23-3e).

후보 = 목표 비중(토공 · 기울기 · 이격 · 기본 비중)을 바꿔 가며 격자 탐색(`..._Search`)한 노선들.
갈래는 **비교값(화면에 보이는 값)으로 고른다** — 탐색 목표(4 m 링크 기울기 · 옆경사)는 비교값
(측점 20 m 기울기 · 절성토 가늠)과 어긋나 탐색 한 벌로는 갈래 순서가 안 맞는다(PLAN 23-1).

공통 제약(follow 밖 모든 갈래 · **넘지 못함**) — 종단기울기 상한 · 최소 곡선반지름 · 계류 이격:
  · 종단기울기 — 지반 기울기가 상한을 넘는 링크는 탐색에서 뺌. 곡선 맞춤 뒤 측점(20 m) 사이
    지반이 상한을 넘으면 탐색 상한을 조금씩 낮춰 다시 찾음(`GRADE_CAP_STEPS`).
  · 최소 곡선반지름 — 한 번에 이웃 방향(약 22.5°)으로만 꺾고, 꺾은 뒤 원호 두 개의 접선 길이만큼
    곧게 가야 다시 꺾음 → 노드마다 최소 반지름 원호가 들어감.
  · 계류 이격 — 버퍼 안에서는 계류 쪽 · 반대쪽 ±45° 로만 움직임(건너기 · 빠져나가기).
  · 그래도 길이 없을 때만(지형상 불가능) 벌점 탐색으로 넘긴 길을 내고, 넘긴 곳을
    `metrics.violations`(측점 · 값 · 사유)에 적음. `over_limit` 수 = 그 목록 수. 이때 후보는
    벌점 크기를 바꿔 가며 만들고(`RELAXED_GRADE_COSTS`), 가장 센 벌점(`SOFT_GRADE_COST`)의
    기준 길보다 못 지킨 곳이 많은 후보는 뺀다 — 센 벌점 하나로만 찾으면 벌점이 목표를 덮어
    네 갈래가 한 길로 모인다(PLAN 23-1 원인).

기준값 확정(2026-09-29 사용자 결정, PLAN 20장) — 아래 임시값이 그대로 **기본값**이 됐다
(지식DB `resources/knowledge/technical_info/01_임도` 후보 표 그대로). 종단기울기 상한 ·
최소 곡선반지름은 여전히 호출부(`_road_settings`·`_plan_criteria`)가 프로젝트별로 넘기고,
계류 이격 · 기울기 잴 지반 폭은 이 머리의 상수가 기본값이다 — 넷 모두 화면(`RouteInitialCriteria`)
에서 프로젝트마다 겹쳐 쓸 수 있다(`generate_initial_route` 의 `stream_offset_m`·`grade_smooth_m`
인자, 라우터의 `max_grade`·`min_radius_m` 대입). 응답 `criteria.pending` 은 이제 늘 빈 목록이다.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

import numpy as np

from B05_Profile.B05_Profile_Engine_RouteInitial_Metrics import GRADE_SMOOTH_M, route_metrics
from B05_Profile.B05_Profile_Engine_RouteInitial_Search import (
    CORRIDOR_M,
    ROAD_WIDTH_M,
    NoRoute,
    RouteGraph,
    SearchGrid,
    Terrain,
)
from common_util.common_util_route_polyline import build_planned_polyline

__all__ = ["NoRoute", "Terrain", "generate_initial_route"]

# ── 사용자 확정 대기(지식DB 후보 · PLAN 20장) ─────────────────────────────────────
# 계류 이격(m) — 법정 이격 수치 없음. 타당성평가 「산지계류 100 m 이내 구간 비율」(별표1 비고,
# 1/25,000 주요 수계)을 후보로 씀.
STREAM_OFFSET_M = 100.0
# 이격을 따질 계류 — 도엽 하천중심선 `구분`. 「주요 수계」 가늠으로 세류 · 연결선은 뺌.
STREAM_CLASSES = ("소하천", "지방하천", "국가하천")
# 기준값 확정(2026-09-29 사용자 결정) — 예전엔 여기 여섯 칸이 「사용자 확정 대기」였다.
# 이제 후보 표 값 그대로 기본값으로 굳었고(화면에서 프로젝트마다 바꿀 수 있음), 빈 목록이 곧
# 「확정됨」이다(`criteria.pending`).
PENDING: tuple[str, ...] = ()

# ── 탐색 · 측정 값 ───────────────────────────────────────────────────────────────
MODES = ("follow", "earthwork", "gentle", "weighted")
DEFAULT_WEIGHTS = {"expected": 50.0, "earthwork": 25.0, "grade": 25.0}
MODE_WEIGHTS = {
    "follow": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
    "earthwork": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
    "gentle": {"expected": 0.0, "earthwork": 0.0, "grade": 100.0},
}
# 후보를 만들 탐색 비중 — 목표 하나씩
CANDIDATE_WEIGHTS = (MODE_WEIGHTS["earthwork"], MODE_WEIGHTS["gentle"], MODE_WEIGHTS["follow"])
# 탐색 상한 = 상한 × 이 값들 차례로 — 곡선 맞춤 뒤 측점 사이 지반이 넘으면 다음 값으로 다시 찾음
GRADE_CAP_STEPS = (1.0, 0.93, 0.85)
# 모든 제약을 지키는 길이 없을 때만 쓰는 벌점(1 m 당 × 넘은 기울기 / 상한) — 기준 길(가장 적게
# 넘긴 길)은 센 값, 갈래 후보는 목표와 겨룰 만한 값들(목표는 1 m 당 평균 1 로 정규화됨)
SOFT_GRADE_COST = 500.0
RELAXED_GRADE_COSTS = (50.0, 5.0)
# 후보 묶음은 갈래와 무관 — 같은 입력이면 다시 짜지 않음(최근 몇 벌만 · 프로세스 메모리)
POOL_CACHE_SIZE = 4
_POOLS: dict[bytes, tuple[list[tuple[Any, dict[str, Any]]], float | None]] = {}


def load_terrain_and_streams(
    project_root: Path, selection: dict[str, Any]
) -> tuple[Terrain, list[list[tuple[float, float]]]]:
    """지반 격자(탐색 비용면 캐시 그대로)와 이격을 따질 계류(사업지 좌표계)를 읽는다."""
    from pyproj import Transformer

    from B04_PreProcess.B04_PreProcess_Router_Watershed import (
        STREAM_FILE,
        _load_features,
        _reproject_features,
    )
    from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR, _load_or_build_cost_surface
    from common_util.common_util_crs import resolve_project_crs

    x, y, z, valid, *_ = _load_or_build_cost_surface(
        project_root,
        project_root / _MODELS_SUBDIR,
        str(selection.get("source_filter")),
        str(selection.get("method") or "dtm"),
        bool(selection.get("smooth")),
    )
    to_metric = Transformer.from_crs("EPSG:4326", resolve_project_crs(project_root), always_xy=True)
    features = _load_features(project_root / "B04_PreProcess" / "processed", STREAM_FILE)
    streams = stream_lines(_reproject_features(features, to_metric))
    return Terrain(np.asarray(x), np.asarray(y), np.asarray(z), np.asarray(valid)), streams


def _choice_path(project_root: Path) -> Path:
    return project_root / "B05_Profile" / "route" / "initial_choice.json"


def read_initial_choice(project_root: Path) -> dict[str, Any]:
    """현재 초기값(갈래 · 비중 · 제약 겹침) — [확인] 때 적어 둔 것. 없거나 못 읽으면 추종.

    `criteria` 는 **사용자가 겹쳐 쓴 값만**(비운 칸은 안 담김) — 나머지는 프로젝트 기본값이라
    화면이 다시 열 때 겹친 칸만 채우고 나머지는 placeholder(기본값)로 비워 둔다.
    """
    try:
        data = json.loads(_choice_path(project_root).read_text(encoding="utf-8"))
        logic = data["logic"] if data.get("logic") in MODES else "follow"
        criteria = data.get("criteria")
        return {
            "logic": logic,
            "weights": resolve_weights(logic, data.get("weights")),
            "criteria": criteria if isinstance(criteria, dict) and criteria else None,
        }
    except (OSError, ValueError, KeyError, TypeError, AttributeError):
        return {"logic": "follow", "weights": resolve_weights("follow", None), "criteria": None}


def write_initial_choice(
    project_root: Path,
    logic: str,
    weights: dict[str, float] | None,
    criteria: dict[str, float] | None = None,
) -> dict[str, Any]:
    """[확인] 때만 부름 — 작업본 한 칸. 다시 열기 · 다른 PC 에서도 같은 초기값.

    `criteria` 는 사용자가 화면 칸에 직접 넣은 겹침 값만(빈 칸은 담지 않는다) — 프로젝트
    기본값이 나중에 바뀌어도 사용자가 안 만진 칸은 그 새 기본값을 그대로 받는다.
    """
    choice = {
        "logic": logic,
        "weights": resolve_weights(logic, weights),
        "criteria": criteria or None,
    }
    path = _choice_path(project_root)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(choice, ensure_ascii=False), encoding="utf-8")
    return choice


def resolve_weights(mode: str, weights: dict[str, float] | None) -> dict[str, float]:
    """갈래의 비중(합 100). weighted 가 아니면 갈래가 정한 값."""
    if mode != "weighted":
        return dict(MODE_WEIGHTS[mode])
    merged = {**DEFAULT_WEIGHTS, **(weights or {})}
    return {key: float(merged[key]) for key in DEFAULT_WEIGHTS}


def stream_lines(features: list[dict[str, Any]]) -> list[list[tuple[float, float]]]:
    """이격을 따질 계류만 좌표 목록으로(사업지 좌표계로 이미 옮긴 피처)."""
    lines: list[list[tuple[float, float]]] = []
    for feature in features:
        if (feature.get("properties") or {}).get("구분") not in STREAM_CLASSES:
            continue
        geometry = feature.get("geometry") or {}
        parts = geometry.get("coordinates") or []
        if geometry.get("type") == "LineString":
            parts = [parts]
        elif geometry.get("type") != "MultiLineString":
            continue
        for part in parts:
            points = [(float(p[0]), float(p[1])) for p in part if len(p) >= 2]
            if len(points) >= 2:
                lines.append(points)
    return lines


def generate_initial_route(
    expected: list[tuple[float, float]],
    terrain: Terrain,
    streams: list[list[tuple[float, float]]],
    *,
    mode: str,
    weights: dict[str, float] | None,
    max_grade: float,
    min_radius_m: float,
    stream_offset_m: float | None = None,
    grade_smooth_m: float | None = None,
) -> dict[str, Any]:
    """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).

    follow 밖 갈래는 후보(`_candidates`) 중 제 목표 비교값이 가장 좋은 것을 고른다.
    `stream_offset_m`·`grade_smooth_m` 을 안 주면(None) 이 머리의 기본값을 쓴다 — 화면에서
    프로젝트마다 겹쳐 쓸 수 있다(PLAN 20장 확정).
    """
    if mode not in MODES:
        raise ValueError(f"모르는 갈래: {mode}")
    if len(expected) < 2:
        raise ValueError("예상노선 정점이 2개 미만입니다.")
    used = resolve_weights(mode, weights)
    stream_offset = STREAM_OFFSET_M if stream_offset_m is None else float(stream_offset_m)
    grade_smooth = GRADE_SMOOTH_M if grade_smooth_m is None else float(grade_smooth_m)

    def measure(outline: Any, bound: float | None, follow: bool = False) -> dict[str, Any]:
        return route_metrics(
            outline.vertices,
            expected,
            terrain,
            streams,
            outline.curves,
            max_grade=max_grade,
            min_radius_m=min_radius_m,
            stream_offset_m=stream_offset,
            road_width_m=ROAD_WIDTH_M,
            bound_pct=None if bound is None else bound * 100.0,
            grade_smooth_m=grade_smooth,
            follow=follow,
        )

    def follow_line() -> tuple[Any, dict[str, Any]]:
        # 예상노선 추적 = 전처리 초기값(`planned_route_initial.csv`)과 같은 규칙 · 같은 입력 →
        # 같은 선(`_write_planned_polyline` 기본값). 제약은 따지지 않고 넘은 곳만 목록에 뜸.
        outline = build_planned_polyline(expected, min_radius_m=min_radius_m)
        return outline, measure(outline, None, follow=True)

    bound = None
    # 비중 예상 100 = 예상노선 추적 그 선(PLAN 23-3e) — 사이 비중은 제약 지킨 후보에서만 고름
    if mode == "follow" or (mode == "weighted" and used["earthwork"] == used["grade"] == 0):
        outline, metrics = follow_line()
    else:
        key = _pool_key(
            expected, terrain, streams, max_grade, min_radius_m, stream_offset, grade_smooth
        )
        if key not in _POOLS:
            grid = SearchGrid(
                expected,
                terrain,
                streams,
                stream_offset_m=stream_offset,
                grade_smooth_m=grade_smooth,
            )
            graph = RouteGraph(grid, expected, limit=max_grade, min_radius_m=min_radius_m)
            while len(_POOLS) >= POOL_CACHE_SIZE:
                _POOLS.pop(next(iter(_POOLS)))
            _POOLS[key] = _candidates(graph, max_grade, min_radius_m, measure)
        candidates, bound = _POOLS[key]
        outline, metrics = _pick(mode, used, candidates)
    return {
        "mode": mode,
        "logic": mode,
        "weights": used,
        "relaxed": bound is not None,
        # 지형상 불가능할 때만 — 복도 안 어떤 길도 넘어야 하는 기울기(%) · 그 위는 막았음
        "terrain_bound_pct": None if bound is None else round(bound * 100.0, 2),
        "planned": [[round(x, 4), round(y, 4)] for x, y in outline.vertices],
        "nodes": [node.as_dict() for node in outline.nodes],
        "curves": [curve.as_dict() for curve in outline.curves],
        "metrics": metrics,
        "criteria": {
            "max_grade_pct": round(max_grade * 100.0, 2),
            "min_radius_m": round(min_radius_m, 2),
            "stream_offset_m": round(stream_offset, 2),
            "road_width_m": ROAD_WIDTH_M,
            "corridor_m": CORRIDOR_M,
            "grade_smooth_m": round(grade_smooth, 2),
            # 기준값 확정(PLAN 20장, 2026-09-29) — 더는 후보가 아니다. 항상 빈 목록.
            "pending": list(PENDING),
        },
    }


def _candidates(
    graph: RouteGraph,
    max_grade: float,
    min_radius_m: float,
    measure: Any,
) -> tuple[list[tuple[Any, dict[str, Any]]], float | None]:
    """갈래가 고를 후보 노선들(폴리라인 · 비교값)과 지형상 불가능 때 막은 기울기(아니면 None).

    제약을 지키는 길이 있으면 비중마다 탐색 상한을 `GRADE_CAP_STEPS` 차례로 낮춰 찾은 노선들 ·
    없으면 막은 기울기 안에서 벌점 크기 · 비중을 바꿔 찾은 노선들. 못 지킨 곳 수가 기준
    (엄격: 후보 최소 · 풂: 센 벌점 기준 길)보다 많은 후보는 뺀다.
    """

    def trace(nodes: list[tuple[float, float]], bound: float | None) -> tuple[Any, dict]:
        # 노드마다 곡선(묶지 않음) — [확인] 재구성(`/route/replan`)과 같은 규칙 → 적용 뒤도 같은 선
        outline = build_planned_polyline(
            nodes, min_radius_m=min_radius_m, simplify=False, curve_flags=[True] * len(nodes)
        )
        return outline, measure(outline, bound)

    found: list[tuple[Any, dict[str, Any]]] = []
    for weights in CANDIDATE_WEIGHTS:
        for step in GRADE_CAP_STEPS:
            try:
                nodes = graph.solve(weights, cap=max_grade * step, strict_stream=True)
            except NoRoute:
                break
            found.append(trace(nodes, None))
            if found[-1][1]["over_limit"]["total"] == 0:
                break
        if not found:  # 제약을 지키는 길 없음 — 비중과 무관하니 더 볼 것 없음
            break
    if found:
        least = min(item[1]["over_limit"]["total"] for item in found)
        return [item for item in found if item[1]["over_limit"]["total"] == least], None

    # 지형상 불가능 — 반지름 · 계류 규칙을 지키는 어떤 길도 넘어야 하는 가장 작은 기울기를
    # 구해(`terrain_bound`) 그 위는 막고, 상한과 그 사이는 벌점으로 넘긴다.
    bound = graph.terrain_bound()
    cap = bound + 1e-9
    seed = trace(
        graph.solve(DEFAULT_WEIGHTS, cap=cap, strict_stream=False, grade_cost=SOFT_GRADE_COST),
        bound,
    )
    found = [seed]
    for weights in CANDIDATE_WEIGHTS:
        for cost in RELAXED_GRADE_COSTS:
            nodes = graph.solve(weights, cap=cap, strict_stream=False, grade_cost=cost)
            found.append(trace(nodes, bound))
    most = seed[1]["over_limit"]["total"]
    return [item for item in found if item[1]["over_limit"]["total"] <= most], bound


def _pool_key(
    expected: list[tuple[float, float]],
    terrain: Terrain,
    streams: list[list[tuple[float, float]]],
    *numbers: float,
) -> bytes:
    """후보 묶음을 가를 입력 지문 — 지형 · 예상노선 · 계류 · 제약 값."""
    digest = hashlib.blake2b(digest_size=16)
    for array in (terrain.x, terrain.y, terrain.z, terrain.valid, np.asarray(expected)):
        digest.update(np.ascontiguousarray(array).tobytes())
    digest.update(repr((streams, numbers)).encode())
    return digest.digest()


def _pick(
    mode: str, weights: dict[str, float], candidates: list[tuple[Any, dict[str, Any]]]
) -> tuple[Any, dict[str, Any]]:
    """갈래 목표 비교값이 가장 좋은 후보. 같으면 앞 후보."""

    def earth(metrics: dict[str, Any]) -> float:
        return metrics["cut_m3"] + metrics["fill_m3"]

    def grade(metrics: dict[str, Any]) -> float:
        # 최대 먼저 · 평균은 같을 때만(기울기 순한과 같은 순서 — 값은 0.01 단위라 평균 몫 < 0.01)
        return metrics["max_grade_pct"] + 1e-4 * metrics["avg_grade_pct"]

    def offset(metrics: dict[str, Any]) -> float:
        return metrics["offset_avg_m"]

    if mode == "earthwork":
        return min(candidates, key=lambda item: earth(item[1]))
    if mode == "gentle":
        return min(
            candidates, key=lambda item: (item[1]["max_grade_pct"], item[1]["avg_grade_pct"])
        )
    # 비중 — 세 값을 후보 범위로 0~1 로 맞춘 뒤 비중으로 더함(단위가 다른 값을 겨루게)
    parts = []
    for key, value in (("expected", offset), ("earthwork", earth), ("grade", grade)):
        values = [value(item[1]) for item in candidates]
        low, span = min(values), max(values) - min(values)
        parts.append([weights[key] * ((v - low) / span if span > 1e-9 else 0.0) for v in values])
    scores = [sum(column) for column in zip(*parts)]
    return candidates[scores.index(min(scores))]