Aislo/B05_Profile/B05_Profile_Engine_Sections_Node.py

"""서버 측점 생성 = 화면과 같은 TS(PLAN 511-B B-2 · 요구 320-4) — 늘 떠 있는 Node
`stations.sections`.

`build_sections`(B03 체인 · 노선 저장 · 재생성 · 넓힘 · 추천 초기 종단)가 부름 — 측점 자리 ·
종류 · 열쇠 · 반경 · 확폭 · 지반 · 틀 · 상단측 =
`B05_Profile/logic/B05_Profile_Sections_Generate.ts`
(화면 묶음 stations · ground 단계와 같은 함수). 파이썬은 머리(schema_version · source ·
coordinate_reference · cad_exchange · options)만 붙임.

DTM · implicit(굳힌 격자 641-1) 지표면만(격자 이진 = `dtm_grid_file`) — 그 밖 ·
Node 못 돌림 = None → 부르는 쪽이 파이썬 `generate_sections`(짝 · 로그 남김).
측점 열쇠(`key`) = 노선 저장 이월 열쇠(`B05_Profile_Route_Save._carry_cross`).
"""

from __future__ import annotations

import logging
from typing import Any

from B05_Profile.B05_Profile_Engine_Sections_Core import (
    SECTION_SCHEMA_VERSION,
    SectionGenerationOptions,
)

logger = logging.getLogger(__name__)


def plan_pipe(pipe: Any) -> dict[str, Any]:
    """관 정본 → TS `PlanPipe`(누가거리 그대로 · 파이썬 `plan_stations` 가 읽는 칸)."""
    return {
        "id": getattr(pipe, "id", None),
        "chainage_m": float(pipe.chainage_m),
        "facility": getattr(pipe, "facility", None) or "pipe",
        "start_m": getattr(pipe, "start_m", None),
        "end_m": getattr(pipe, "end_m", None),
        "options": getattr(pipe, "options", None) or {},
    }


def plan_structure(item: Any) -> dict[str, Any]:
    """구조물 정본 → TS `PlanStructure`(JSON 꼴)."""
    if hasattr(item, "model_dump"):
        return item.model_dump(mode="json")
    return dict(item) if isinstance(item, dict) else dict(vars(item))


def structure_table() -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """구조물 종류(`/projects/structure-types` 꼴) · 정의(화면 `useCatalogDefinition` 꼴)."""
    from B05_Profile.B05_Profile_Structures_Schema import _read_definition, _stamp
    from M02_MasterTemplete.StructOptions.M02_SO_Options import with_visible

    types = [item.model_dump(mode="json") for item in with_visible().types]
    return types, {"types": types, **_read_definition(*_stamp())}


def stations_input(
    xy: list[list[float]],
    pipes: list[Any],
    structures: list[Any],
    plus: list[float],
    options: SectionGenerationOptions,
    design_curves: list[dict[str, Any]],
    station_snap: bool,
    station_snap_step: float | None,
) -> dict[str, Any]:
    """Node `stations.list` · `stations.sections` 입력(격자 빼고)."""
    types, definition = structure_table()
    return {
        "line": xy,
        "curves": design_curves,
        "pipes": [plan_pipe(pipe) for pipe in pipes],
        "structures": [plan_structure(item) for item in structures],
        "types": types,
        "definition": definition,
        "plus": [float(at) for at in plus],
        "interval_m": options.station_interval_m,
        "include_endpoint": options.include_endpoint,
        "station_snap": station_snap,
        "station_snap_step": station_snap_step,
        "cross_half_width_m": options.cross_half_width_m,
        "cross_sample_interval_m": options.cross_sample_interval_m,
        "long_sample_interval_m": options.long_sample_interval_m,
    }


def node_stations(stations: dict[str, Any]) -> list[dict[str, Any]] | None:
    """측점 목록(`buildStations` · 열쇠 · 누가거리 · 종류) — 못 돌림 None."""
    from common_util.common_util_calc_runner import run_calc_served

    answer = run_calc_served(["stations.list"], {"stations": stations})
    if not answer or "error" in answer:
        logger.warning("Node 측점 목록 실패: %s", (answer or {}).get("error"))
        return None
    return answer["results"]["stations.list"]


def node_plan_stations(
    groups: list[tuple[list[Any], list[Any], list[float]]],
) -> list[list[dict[str, Any]]] | None:
    """(관 · 구조물 · 추가 측점) 묶음마다 측점 계획(`planStations` · 파이썬 `plan_stations` 짝) —
    한 Node 부름 · 못 돌림 None(부르는 쪽이 파이썬 짝)."""
    from common_util.common_util_calc_runner import run_calc

    if not groups:
        return []
    types, definition = structure_table()
    plan = [
        {
            "pipes": [plan_pipe(pipe) for pipe in pipes],
            "structures": [plan_structure(item) for item in structures],
            "plus": [float(at) for at in plus],
        }
        for pipes, structures, plus in groups
    ]
    try:
        results = run_calc(
            ["stations.plan"],
            {"stations": {"types": types, "definition": definition, "plan": plan}},
        )
    except Exception:  # noqa: BLE001 — 파이썬 짝으로 이어 감
        logger.exception("Node 측점 계획 실패 — 파이썬 길")
        return None
    if results is None:
        logger.warning("Node 측점 계획 못 돌림 — 파이썬 길")
        return None
    return results["stations.plan"]


def node_sections(
    polyline: list[list[float]],
    sampler: Any,
    facilities: tuple[list[Any], list[Any], list[float]],
    options: SectionGenerationOptions,
    *,
    source_snapshot: dict[str, Any],
    crs: str | None,
    design_curves: list[dict[str, Any]],
    station_snap: bool,
    station_snap_step: float | None,
) -> dict[str, Any] | None:
    """`generate_sections` 와 같은 꼴(+ 측점 `key`) — DTM 아님 · Node 못 돌림 = None."""
    from common_util.common_util_calc_runner import run_calc_served
    from common_util.common_util_surface_sampler import dtm_grid_file

    source = getattr(sampler, "source", None)
    if not source:
        return None
    options.validate()
    pipes, structures, plus = facilities
    stations = stations_input(
        [[float(p[0]), float(p[1])] for p in polyline],
        pipes,
        structures,
        plus,
        options,
        design_curves,
        station_snap,
        station_snap_step,
    )
    stations["grid"] = str(dtm_grid_file(source))
    answer = run_calc_served(["stations.sections"], {"stations": stations})
    if not answer or "error" in answer:
        logger.warning("Node 측점 생성 실패 — 파이썬 길: %s", (answer or {}).get("error"))
        return None
    out = answer["results"]["stations.sections"]
    return {
        "schema_version": SECTION_SCHEMA_VERSION,
        "status": "completed",
        "source": source_snapshot or {},
        "coordinate_reference": {
            "crs": crs,
            "world_axes": {"x": "project_easting", "y": "project_northing", "z": "elevation"},
            "units": {"horizontal": "m", "vertical": "m", "angle": "degree"},
        },
        "cad_exchange": {
            "station_origin": "BP",
            "chainage_direction": "BP_to_EP",
            "cross_offset_sign": {"negative": "right", "positive": "left"},
            "cross_local_axes": {"x": "offset_m", "y": "elevation_m"},
            "recommended_drawing_datum_m": out["datum"],
        },
        "options": {
            "station_interval_m": options.station_interval_m,
            "cross_half_width_m": options.cross_half_width_m,
            "cross_sample_interval_m": options.cross_sample_interval_m,
            "long_sample_interval_m": options.long_sample_interval_m,
            "include_endpoint": options.include_endpoint,
            "extra_stations": [list(entry) for entry in options.extra_stations],
        },
        "longitudinal": out["longitudinal"],
        "cross_sections": out["cross_sections"],
        "summary": out["summary"],
    }