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    — 후보 중 최대 종단기울기 최소(같으면 평균 최소).
  · gradeline — 영선 노선: 목표 기울기(`target_grade` · 없으면 상한)로 등고선을 건너가는
                등경사선(PEGGER 식) · 후보 중 목표 기울기 편차(`grade_dev_pct`) 최소(PLAN 23-5).
  · safe      — 안전 우선: 후보 중 위험 통과 길이(옆경사 35° 이상 · 계류 버퍼 안) 최소.
  · crossings — 계류 횡단 최소: 후보 중 계류 횡단 수 최소 · 같으면 계류에서 먼 쪽.
  · cost      — 공사비 어림 최소: 후보 중 상대 공사비(토공 + 계류 횡단 구조물 + 연장 · 임시 상대
                단가 `..._Metrics.COST_*`) 최소.
  · weighted  — 탐색 비용의 가중합 한 번(후보 고르기 아님 · PLAN 23-4a · `..._Pool.weighted`).
                끝값은 그 갈래와 같은 선 — 예상 100 = 추적 · 토공 100 = 토공 최소 · 기울기 100 =
                기울기 순한. 제약은 (1 − 예상 몫)만큼 벌점으로만 따지고 넘은 곳은 목록에 뜸.

후보 묶음(`..._Pool.build_pool`) = 목표를 바꿔 가며 격자 탐색(`..._Search`)한 노선들 + 토공 ↔ 기울기
비중 쓸기. 갈래는 **비교값(화면에 보이는 값)으로 고른다** — 탐색 목표(4 m 링크 기울기 · 옆경사)는
비교값(측점 20 m 기울기 · 절성토 가늠)과 어긋나 탐색 한 벌로는 갈래 순서가 안 맞는다(PLAN 23-1).
가파른 지형(옆경사 중앙값 67 %)에서는 제약 안의 길이 폭 10~20 m 한 띠뿐이라 갈래끼리 가까울 수
있다 — 응답 `similar`(10 m 안인 다른 갈래) · `band_width_m`(띠 폭 어림)로 알린다(PLAN 23-4).

공통 제약(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_Pool import (
    Pool,
    build_pool,
    pick,
    similar,
    weighted,
)
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", "gradeline", "earthwork", "gentle", "safe", "crossings", "cost", "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},
    # 비중 몫이 없는 갈래(PLAN 23-5) — 제 목표로 찾은 후보에서 비교값으로 고름
    **{
        mode: {"expected": 0.0, "earthwork": 0.0, "grade": 0.0}
        for mode in ("gradeline", "safe", "crossings", "cost")
    },
}
# 후보 묶음은 갈래와 무관 — 같은 입력이면 다시 짜지 않음(최근 몇 벌만 · 프로세스 메모리)
POOL_CACHE_SIZE = 4
_POOLS: dict[bytes, Pool] = {}


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,
    target_grade: float | None = None,
) -> dict[str, Any]:
    """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).

    follow 는 전처리 초기값 규칙 그대로 · weighted 는 비중 탐색 한 번(끝값은 그 갈래) · 나머지는
    후보 묶음(`..._Pool`)에서 제 목표 비교값이 가장 좋은 것. `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)
    # 영선 노선 목표 기울기(비율) — 없으면 상한 · 상한 넘게 넣어도 상한(제약 안에서만)
    goal_grade = max_grade if target_grade is None else min(float(target_grade), max_grade)

    def measure(
        outline: Any, *, bound: float | None, follow: bool = False, weighted: 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,
            weighted=weighted,
            target_grade=goal_grade,
        )

    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, bound=None, follow=True)

    bound = None
    pool: Pool | None = None
    target = _weighted_end(used) if mode == "weighted" else mode
    key = _pool_key(
        expected, terrain, streams, max_grade, min_radius_m, stream_offset, grade_smooth, goal_grade
    )
    if target == "follow":  # 비중 예상 100 = 추적 그 선(PLAN 23-3e) · 묶음은 짜지 않음(빠른 길)
        outline, metrics = follow_line()
        pool = _POOLS.get(key)
    else:
        if key not in _POOLS:
            grid = SearchGrid(
                expected,
                terrain,
                streams,
                stream_offset_m=stream_offset,
                grade_smooth_m=grade_smooth,
                target_grade=goal_grade,
            )
            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] = build_pool(graph, follow_line(), max_grade, min_radius_m, measure)
        pool = _POOLS[key]
        bound = pool.bound
        if target == "weighted":
            outline, metrics = weighted(pool, used, min_radius_m, measure)
        else:
            outline, metrics = pick(target, pool.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,
        # 10 m 안인 다른 갈래(같은 노선 알림) · 제약 안 띠 폭 어림 — 추적은 묶음이 이미 있을 때만
        "similar": None if pool is None else similar(pool, target, outline),
        "band_width_m": None if pool is None else round(pool.band_width_m, 1),
        "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),
            "target_grade_pct": round(goal_grade * 100.0, 2),
            # 기준값 확정(PLAN 20장, 2026-09-29) — 더는 후보가 아니다. 항상 빈 목록.
            "pending": list(PENDING),
        },
    }


def _weighted_end(weights: dict[str, float]) -> str:
    """비중이 한쪽 100 이면 그 갈래(끝값 = 갈래와 같은 선 · PLAN 23-3e) · 아니면 비중 탐색."""
    for name, mode in (("expected", "follow"), ("earthwork", "earthwork"), ("grade", "gentle")):
        if all(value == 0 for key, value in weights.items() if key != name):
            return mode
    return "weighted"


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