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 지표면만(격자 이진 = `dtm_grid_file`) — DTM 아닌 지표면(tin · nurbs · implicit · meshfree) ·
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 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": [
            item.model_dump(mode="json") if hasattr(item, "model_dump") else dict(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_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"],
    }