Aislo/B05_Profile/B05_Profile_Engine_RouteInitial_CrossEarth.py

"""노선 추천 횡단 토공(PLAN 420 · 요구 254).

측점마다 공식 횡단 엔진(B06 `compute_cross_design` — [저장] 정본과 같은 파이썬)으로 절 · 성
단면적을 재고 평균단면법으로 토량을 낸다. 띠 가늠(폭 4 m · 비탈 뺌 · `..._Metrics`)을 갈음하는
추천 토공 — 후보 줄 세움 · 합치기 판정 · 결과 표 절 · 성.

계산 자리 = 서버 단독(③). 측점 횡단 뜨기 · 단면적은 균형 종단(`..._Profile`)과 같은 함수 —
종단 반영 끔 = 계획고 = 지반고([적용] 뒤 1차 종단과 같음).
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

import numpy as np

from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
    STATION_M,
    _station_areas,
    route_sections,
)

if TYPE_CHECKING:
    from B05_Profile.B05_Profile_Engine_RouteInitial_Search import Terrain
    from common_util.common_util_surface_sampler import SurfaceElevationSampler


def _end_area_weights(at: np.ndarray) -> np.ndarray:
    """평균단면법 무게 — 측점마다 앞 · 뒤 구간 반씩(Σ (A_i + A_i+1) / 2 · L = Σ A_i · w_i)."""
    spans = np.diff(at)
    weight = np.zeros(len(at))
    weight[:-1] += spans / 2.0
    weight[1:] += spans / 2.0
    return weight


def _areas(
    line: list[tuple[float, float]],
    chainages: Any,
    design_z: Any,
    terrain: Terrain | SurfaceElevationSampler,
    standard: dict[str, Any] | None,
    design_curves: list[dict[str, Any]] | None,
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """(누가거리, 지반고, 계획고, [절 자연 · 절 다짐 · 성] ㎡ 행렬)."""
    at, ground, sections, modes, widen = route_sections(line, terrain, design_curves, at=chainages)
    plan = ground if design_z is None else np.asarray(design_z, dtype=np.float64)
    if plan.shape != at.shape:
        raise ValueError("계획고 수가 측점 수와 다릅니다.")
    rows = np.asarray(
        [
            _station_areas(sections[i], modes[i], float(plan[i]), widen[i], standard)
            for i in range(len(at))
        ]
    ).reshape(-1, 3)
    return at, ground, plan, rows


def cross_earthwork(
    line: list[tuple[float, float]],
    chainages: Any,
    design_z: Any,
    terrain: Terrain | SurfaceElevationSampler,
    standard: dict[str, Any] | None = None,
    *,
    design_curves: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """노선 하나의 횡단 토공 — `{cut, cut_compacted, fill, per_station}`(㎥ · 절 = 자연 상태).

    `chainages` = 잴 누가거리(None = 20 m 측점 + 종점 · 균형 종단과 같은 틀) · `design_z` =
    그 측점 계획고(None = 지반고 · 종단 반영 끔) · `standard` = 프로젝트 저장 표준 횡단면 설정값
    (None = 설정 기본값) · `design_curves` = 곡선표(확폭 · 없으면 측점별 실측 반경)."""
    at, ground, plan, rows = _areas(line, chainages, design_z, terrain, standard, design_curves)
    weight = _end_area_weights(at)
    cut, cut_c, fill = (float((weight * rows[:, k]).sum()) for k in range(3))
    return {
        "cut": round(cut, 1),
        "cut_compacted": round(cut_c, 1),
        "fill": round(fill, 1),
        "per_station": [
            {
                "chainage_m": round(float(s), 2),
                "ground_m": round(float(g), 3),
                "plan_m": round(float(p), 3),
                "cut_m2": round(float(r[0]), 4),
                "cut_compacted_m2": round(float(r[1]), 4),
                "fill_m2": round(float(r[2]), 4),
            }
            for s, g, p, r in zip(at, ground, plan, rows)
        ],
    }


def cross_earthwork_delta(
    before: list[tuple[float, float]] | dict[str, Any],
    after: list[tuple[float, float]],
    changed_chainages: Any,
    terrain: Terrain | SurfaceElevationSampler,
    standard: dict[str, Any] | None = None,
    *,
    before_chainages: Any = None,
    z_before: Any = None,
    z_after: Any = None,
    design_curves_before: list[dict[str, Any]] | None = None,
    design_curves_after: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """바뀐 측점만 횡단으로 잰 토공 차(새 − 옛 · ㎥) — 합치기 판정(3~6 측점).

    측점 하나 = `STATION_M` 몫(가운데 측점 평균단면법 무게). `before` = 옛 노선 정점 또는 그
    노선의 `cross_earthwork` 결과(그 측점 단면적을 다시 안 셈 · 누가거리 가장 가까운 것) ·
    `before_chainages` = 옛 노선에서 잴 누가거리(없으면 `changed_chainages`) · `z_*` = 그 측점
    계획고(None = 지반고). 결과 `cut` · `fill` · `earth` = 차 · `before` 가 결과면 `*_total` =
    고친 뒤 노선 전체 어림(옛 합 + 차)."""
    at_after = np.asarray(changed_chainages, dtype=np.float64).reshape(-1)
    at_before = at_after if before_chainages is None else np.asarray(before_chainages, float)
    if at_after.size == 0:
        new = old = np.zeros((0, 3))
    else:
        new = _areas(after, at_after, z_after, terrain, standard, design_curves_after)[3]
        if isinstance(before, dict):
            rows = before["per_station"]
            held = np.asarray([row["chainage_m"] for row in rows])
            pick = [int(np.argmin(np.abs(held - s))) for s in at_before]
            old = np.asarray(
                [[rows[i]["cut_m2"], rows[i]["cut_compacted_m2"], rows[i]["fill_m2"]] for i in pick]
            ).reshape(-1, 3)
        else:
            old = _areas(before, at_before, z_before, terrain, standard, design_curves_before)[3]
    cut, cut_c, fill = (float(STATION_M * (new[:, k].sum() - old[:, k].sum())) for k in range(3))
    out = {
        "cut": round(cut, 1),
        "cut_compacted": round(cut_c, 1),
        "fill": round(fill, 1),
        "earth": round(cut + fill, 1),
        "stations": int(at_after.size),
    }
    if isinstance(before, dict):
        out["cut_total"] = round(before["cut"] + cut, 1)
        out["fill_total"] = round(before["fill"] + fill, 1)
    return out


def attach_cross_earth(
    result: dict[str, Any],
    terrain: Terrain | SurfaceElevationSampler,
    standard: dict[str, Any] | None = None,
) -> None:
    """추천 결과 표 절 · 성 = 횡단 값(420) — 종단 켬 = 균형 종단 횡단 ㎥ · 끔 = 계획고 지반고.

    띠 가늠은 `band_cut_m3` · `band_fill_m3` 로 남긴다. 맞추기 전 선(`unsnapped`)도 같이."""
    for item in (result, result.get("unsnapped")):
        if not item or not item.get("planned"):
            continue
        metrics = item["metrics"]
        if "band_cut_m3" in metrics:
            continue
        profile = metrics.get("profile")
        if profile and "error" not in profile:
            cut, fill = float(profile["cut_m3"]), float(profile["fill_m3"])
        else:
            try:
                earth = cross_earthwork(
                    [tuple(p) for p in item["planned"]],
                    None,
                    None,
                    terrain,
                    standard,
                    design_curves=item.get("curves") or None,
                )
            except ValueError:  # 지반을 못 뜬 노선 — 띠 가늠 그대로
                continue
            cut, fill = earth["cut"], earth["fill"]
        metrics["band_cut_m3"] = metrics.get("cut_m3")
        metrics["band_fill_m3"] = metrics.get("fill_m3")
        metrics["cut_m3"] = round(cut, 1)
        metrics["fill_m3"] = round(fill, 1)
        metrics["surplus_m3"] = round(cut - fill, 1)