Aislo/B06_Section/B06_Section_Engine_Design_Node.py

"""횡단 설계선 · 면적 · 소단 구간을 서버 Node 입구로(PLAN 472-4 ② · 등록부 `cross.design` ·
`cross.berm_spans`) — 브라우저와 같은 TS(`common_util_cross_design` · `common_util_berm_spans`).

파이썬은 입력 조립만 — 측점마다 `compute_cross_design` 키를 모아 Node 한 번에 보냄.
Node 를 못 부르면 파이썬 짝으로(거울 시험 `test_b06_cross_design_mirror` · `test_272_berm_source` ·
알려진 차 = 설계선 점 반올림 동점 1e-4 m — 파이썬 round 짝수 쪽 · TS 위쪽).
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

from B06_Section.B06_Section_Engine_Design import _resolve_group, compute_cross_design
from common_util.common_util_calc_runner import run_calc

logger = logging.getLogger(__name__)

# 한 측점 입력 = (지반 표본, 계획고, `compute_cross_design` 키 — standard 뺌)
DesignJob = tuple[list[dict[str, Any]], float | None, dict[str, Any]]


def full_standard(standard: dict[str, Any] | None) -> dict[str, Any]:
    """config 기본값 위에 저장분을 덮은 표준 횡단 한 벌 — TS 는 config 를 모름(입력으로 받음)."""
    out: dict[str, Any] = {}
    for key in ("soil", "rock", "paved"):
        g = _resolve_group(key, standard)
        out[key] = {
            "road_width_m": g["road_width_m"],
            "shoulder_left_m": g["shoulder_left_m"],
            "shoulder_right_m": g["shoulder_right_m"],
            "ditch": {
                "top_width_m": g["ditch_top_width_m"],
                "bottom_width_m": g["ditch_bottom_width_m"],
                "depth_m": g["ditch_depth_m"],
            },
            "ditch_l_type": {"width_m": g["l_ditch_width_m"], "depth_m": g["l_ditch_depth_m"]},
            "cross_slope_pct": {"min": g["cross_slope_pct"]},
            "fill_slope_ratio": g["fill_slope_ratio"],
            "cut_slope_ratio": g["cut_slope_ratio"],
            "pavement_thickness_m": g["pavement_thickness_m"],
        }
    return out


def _python(job: DesignJob, standard: dict[str, Any] | None) -> dict[str, Any] | None:
    samples, elevation, options = job
    try:
        return compute_cross_design(samples, elevation, standard=standard, **options)
    except (ValueError, KeyError):
        return None


# 시설 구간(포장 · 기울임 · 세월교 하강) = TS `facilityRanges` — (프로젝트 뿌리, 측점 누가거리)
Facilities = tuple[Path, list[float]]


def facility_inputs(project_root: Path) -> dict[str, list[dict[str, Any]] | None]:
    """TS `facilityRanges` 입력 — 관 = 지금 노선에 얹은 관 지점(`route_pipe_points`) · 구조물 정본.

    못 읽은 관 = 빈 목록 · 구조물 = None(포장 · 맹암거 구간 없음 · 소단 = 저장분 그대로 · 파이썬
    `pavement_ranges` · `berm_specs_at` 의 비치명 태도 그대로)."""
    from dataclasses import asdict

    from B05_Profile.B05_Profile_Engine_Sections import route_pipe_points
    from B05_Profile.B05_Profile_Structures_Repository import load_structures

    try:
        pipes = [asdict(point) for point in route_pipe_points(project_root)]
    except Exception:  # noqa: BLE001 — 정본을 못 읽어도 설계 계산은 이어 간다
        logger.exception("B06 관 지점을 읽지 못했습니다 (시설 구간 없음으로 본다)")
        pipes = []
    structures: list[dict[str, Any]] | None
    try:
        structures = [s.model_dump(mode="json") for s in load_structures(str(project_root))[1]]
    except Exception:  # noqa: BLE001
        logger.exception("B06 구조물 정본을 읽지 못했습니다 (포장 · 맹암거 구간 없음으로 본다)")
        structures = None
    return {"pipes": pipes, "structures": structures}


def _python_facility(root: Path) -> Any:
    """Node 를 못 부를 때 — 파이썬 구간 짝(`Router_Design_Ranges` · `Engine_Tilt`)."""
    from B06_Section.B06_Section_Engine_Tilt import tilt_args, tilt_ranges
    from B06_Section.B06_Section_Router_Design_Ranges import (
        ford_drop_at,
        ford_surface_drops,
        paved_at,
        pavement_ranges,
    )

    paved, tilts, drops = pavement_ranges(root), tilt_ranges(root), ford_surface_drops(root)
    return lambda c: {
        "paved": paved_at(c, paved),
        "surface_drop_m": ford_drop_at(c, drops),
        **tilt_args(c, tilts),
    }


def compute_cross_designs(
    jobs: list[DesignJob],
    standard: dict[str, Any] | None,
    facilities: Facilities | None = None,
) -> list[dict[str, Any] | None]:
    """측점 묶음 횡단 설계 — 차례 그대로 결과 · None = 못 세움(파이썬 ValueError 자리).

    `facilities` = (뿌리, 누가거리) 면 포장 · 기울임 · 세월교 하강을 TS 구간이 측점마다 얹음
    (511-B B-4 · C3)."""
    if not jobs:
        return []
    stations = [
        {
            "samples": samples,
            "design_elevation_m": elevation,
            "options": {
                key: (value._asdict() if key == "berm" and value is not None else value)
                for key, value in options.items()
            },
        }
        for samples, elevation, options in jobs
    ]
    batch: dict[str, Any] = {"standard": full_standard(standard), "stations": stations}
    if facilities is not None:
        root, chainages = facilities
        batch["facilities"] = facility_inputs(root)
        for station, chainage in zip(stations, chainages, strict=True):
            station["chainage_m"] = chainage
    results = run_calc(["cross.design"], {"cross_design": batch})
    if results is None:
        logger.warning("Node 횡단 설계 실패 — 파이썬 짝으로 셈(%d 측점)", len(jobs))
        if facilities is not None:
            at = _python_facility(facilities[0])
            jobs = [(s, z, {**o, **at(c)}) for (s, z, o), c in zip(jobs, facilities[1])]
        return [_python(job, standard) for job in jobs]
    return [None if "error" in row else row for row in results["cross.design"]]


def berm_specs_at(
    project_root: Path, chainages: list[float]
) -> list[dict[str, float] | None] | None:
    """측점마다 소단 제원(구조물 정본 구간) · None = 정본을 못 읽음(부르는 쪽이 저장분 그대로)."""
    from B05_Profile.B05_Profile_Structures_Repository import load_structures
    from B06_Section.B06_Section_Engine_Structures_Wall import berm_span_at, load_berm_spans

    try:
        structures = [
            item.model_dump(include={"type_id", "start_m", "end_m", "chainage_m", "options"})
            for item in load_structures(str(project_root))[1]
        ]
    except Exception:  # noqa: BLE001 — 정본을 못 읽으면 저장분으로 물러섬
        logger.exception("B06 소단 구간을 읽지 못했습니다 (저장분 소단을 씀)")
        return None
    results = run_calc(
        ["cross.berm_spans"], {"berm": {"structures": structures, "chainages": chainages}}
    )
    if results is not None:
        return results["cross.berm_spans"]["at"]
    logger.warning("Node 소단 구간 실패 — 파이썬 짝으로 셈")
    spans = load_berm_spans(project_root)
    return None if spans is None else [berm_span_at(spans, c) for c in chainages]