Aislo/B05_Profile/B05_Profile_Engine.py

"""B05 경로 설계 엔진 오케스트레이터.

경로점(BP/CP/EP/AP/FP)과 옵션을 받아 최적 경로를 계산하고, 폴리라인을
GeoJSON으로 저장하며 DB 기록용 데이터(메타·렌더링 샘플·통계)를 준비한다.
라우터에서 asyncio.to_thread로 호출한다.
"""

import logging
from pathlib import Path
from typing import Any

from B05_Profile.B05_Profile_Debug import log_b05_debug
from B05_Profile.B05_Profile_Engine_RidgeValley import solve_ridge_valley_route
from B05_Profile.B05_Profile_Engine_Sections import ROUTE_DATA_FILE
from B05_Profile.B05_Profile_Engine_Solver import solve_optimal_route
from common_util.common_util_json import atomic_write_json

_ROUTE_SUBDIR = Path("B05_Profile") / "route"
# route_points 테이블에 저장할 렌더링 샘플 최대 개수
_MAX_RENDER_POINTS = 500
logger = logging.getLogger(__name__)


def _route_geojson(polyline: list[list[float]]) -> dict[str, Any]:
    """폴리라인을 3D LineString GeoJSON Feature로 변환한다.

    x · y 4자리 = 계획노선 csv(`write_route_csv`) · 브라우저 route 단계(`draftLine` · 짝 —
    `test_475_1_route_digits.py`) — 같은 선이라야 저장 체인 ↔ 화면 누가토량이 같음(475-1).
    """
    return {
        "type": "Feature",
        "geometry": {
            "type": "LineString",
            "coordinates": [[round(x, 4), round(y, 4), round(z, 3)] for x, y, z in polyline],
        },
        "properties": {},
    }


def _sample_render_points(
    polyline: list[list[float]], chainage_m: list[float], maximum: int
) -> list[dict[str, Any]]:
    """폴리라인을 최대 maximum개로 균등 샘플링해 렌더링용 포인트를 만든다."""
    n = len(polyline)
    if n == 0:
        return []
    if n <= maximum:
        indices = range(n)
    else:
        step = n / maximum
        indices = (int(i * step) for i in range(maximum))

    points: list[dict[str, Any]] = []
    for seq, idx in enumerate(indices):
        idx = min(idx, n - 1)
        x, y, z = polyline[idx]
        # 국소 경사(%) — 직전 정점과의 차이
        slope_pct = 0.0
        if idx > 0:
            x0, y0, z0 = polyline[idx - 1]
            x1, y1, z1 = polyline[idx]
            h = ((x1 - x0) ** 2 + (y1 - y0) ** 2) ** 0.5
            if h > 1e-6:
                slope_pct = abs(z1 - z0) / h * 100.0
        points.append(
            {
                "x": round(x, 3),
                "y": round(y, 3),
                "z": round(z, 3),
                "chainage_m": round(chainage_m[idx], 3) if idx < len(chainage_m) else None,
                "elevation_m": round(z, 3),
                "slope_percent": round(slope_pct, 3),
                "sequence_num": seq,
            }
        )
    return points


def run_route_design(
    project_root: Path,
    filter_key: str,
    method: str,
    smooth: bool,
    points_data: dict[str, Any],
    options: dict[str, Any],
    algorithm: str = "dijkstra",
) -> dict[str, Any]:
    """경로 탐색을 실행하고 GeoJSON 저장 + DB 기록용 데이터를 반환한다.

    algorithm: "dijkstra"(격자 Dijkstra) 또는 "ridge_valley"(능선-계곡 정속경사).

    반환 dict:
      - route_data_path: 저장한 GeoJSON의 프로젝트 상대 경로
      - solver_result: solver 원본 결과 (polyline, metrics, segments 등)
      - render_points: route_points 테이블 저장용 샘플
      - statistics: route_statistics 저장용 요약
    """
    log_b05_debug(
        logger,
        "calculation.start",
        algorithm=algorithm,
        filter_key=filter_key,
        method=method,
        smooth=smooth,
        point_counts={
            "bp": 1 if points_data.get("bp") else 0,
            "ep": 1 if points_data.get("ep") else 0,
            "cp": len(points_data.get("cp", [])),
            "ap": len(points_data.get("ap", [])),
            "fp": len(points_data.get("fp", [])),
        },
        options=options,
    )
    if algorithm == "as_planned":
        # 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀지 않는다(PLAN 0-3 자동탐색 접기).
        # 제어점 목록(bp·cp·ep)이 곧 노선이며, 표고만 지표면에서 뜬다.
        from B05_Profile.B05_Profile_Engine_AsPlanned import solve_as_planned

        sequence = [points_data.get("bp")] + list(points_data.get("cp") or [])
        sequence.append(points_data.get("ep"))
        vertices = [
            (float(point["x"]), float(point["y"])) for point in sequence if isinstance(point, dict)
        ]
        result = solve_as_planned(
            project_root, filter_key, smooth, vertices, options, method=method
        )
    elif algorithm == "ridge_valley":
        result = solve_ridge_valley_route(
            project_root, filter_key, smooth, points_data, options, method=method
        )
    else:
        result = solve_optimal_route(
            project_root, filter_key, smooth, points_data, options, method=method
        )
    polyline = result["polyline"]
    chainage_m = result["chainage_m"]
    log_b05_debug(
        logger,
        "calculation.solver_complete",
        algorithm=algorithm,
        polyline_count=len(polyline),
        segment_count=len(result.get("segments", [])),
        metrics=result.get("metrics", {}),
        required_points_ok=result.get("required_points_ok"),
        warning_count=len(result.get("curve_warning_segments", [])),
    )

    route_dir = project_root / _ROUTE_SUBDIR
    route_dir.mkdir(parents=True, exist_ok=True)
    geojson_path = project_root / ROUTE_DATA_FILE
    atomic_write_json(geojson_path, _route_geojson(polyline))
    log_b05_debug(
        logger,
        "calculation.geojson_saved",
        path=geojson_path.relative_to(project_root).as_posix(),
        coordinate_count=len(polyline),
    )

    # 통계 요약 (solver 메트릭에서 파생)
    metrics = result["metrics"]
    statistics = {
        "min_slope": 0.0,
        "max_slope": metrics.get("max_grade_pct"),
        "mean_slope": metrics.get("avg_grade_pct"),
        "cost_score": None,
    }

    render_points = _sample_render_points(polyline, chainage_m, _MAX_RENDER_POINTS)
    log_b05_debug(
        logger,
        "calculation.output_prepared",
        render_point_count=len(render_points),
        first_render_point=render_points[0] if render_points else None,
        last_render_point=render_points[-1] if render_points else None,
        statistics=statistics,
    )
    return {
        "route_data_path": geojson_path.relative_to(project_root).as_posix(),
        "solver_result": result,
        "render_points": render_points,
        "statistics": statistics,
        "grade_percent": [seg.get("max_grade_pct") for seg in result.get("segments", [])],
        "constraints": result.get("conditions_snapshot", {}),
        "algorithm_params": options.get("weights") or {},
    }