Aislo/B06_Section/B06_Section_Engine_SampleExtend.py

"""미교차 측점의 지반 샘플을 **지표면 끝까지** 넓힘 — ㉳ (가) (2026-09-14 브레인 판정).

반폭(기본 20m) 샘플 끝에서 사면이 원지반과 안 만나면 면적이 거기서 잘려 성토가 **작게** 섰다
(936be972 미교차 12곳 중 11곳이 넓히면 닫힘 · 성토 +2,559.9㎥). ⇒ **열린 쪽만** +5m 씩 확정 지표면에서
다시 떠 계산하고, 닫히면 멈춘다. 새 걸음에 지표면 밖(무효) 샘플이 섞이면 그 걸음은 안 붙이고 멈춘다 —
그 측점은 「지표면 끝까지 넓혀도 안 만남」으로 미교차가 남는다.
⚠ 상한 숫자를 두지 않는다 — 지표면이 실제로 있는 끝이라 근거가 필요 없음(인위 기본값 금지).
⚠ 716 판정 「미교차 측점만 반폭 +5m 한 번」을 갈음(그것으론 4곳만 닫혔음).
⚠ 계산식은 안 바뀐다 — 샘플(입력)만 넓어진다. 넓힌 샘플은 횡단 파일에 남겨 화면(TS)·Node·B08 이
  같은 지반을 본다(한 벌).
"""

from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import Any, Callable

import numpy as np

from B05_Profile.B05_Profile_Engine_Sections import find_cross_file
from B06_Section.B06_Section_Engine_Design import (
    _SLOPE_CLOSE_TOLERANCE_M,
    compute_cross_design,
    curve_widening_args,
)
from common_util.common_util_json import atomic_write_json

logger = logging.getLogger(__name__)

#: 한 걸음 — 브레인 판정 「+5m 씩」.
EXTEND_STEP_M = 5.0
#: 지표면 끝까지 넓혀도 안 닫힌 측점의 사유(경고가 그대로 씀).
SURFACE_END_REASON = "지표면 끝까지 넓혀도 성토 비탈이 원지반과 안 만남"


def _valid_sorted(samples: list[dict[str, Any]]) -> list[tuple[float, float]]:
    return sorted(
        (float(s["offset_m"]), float(s["elevation_m"]))
        for s in samples
        if s.get("valid") is not False and s.get("elevation_m") is not None
    )


def _open_ends(samples: list[dict[str, Any]], design: dict[str, Any]) -> tuple[bool, bool]:
    """(우 끝 열림, 좌 끝 열림) — 샘플 끝의 지반과 설계선 높이차가 허용오차를 넘나."""
    ground = _valid_sorted(samples)
    line = sorted(
        (float(p["offset_m"]), float(p["elevation_m"])) for p in design.get("design_line") or []
    )
    if not ground or not line:
        return False, False
    right = abs(ground[0][1] - line[0][1]) > _SLOPE_CLOSE_TOLERANCE_M
    left = abs(ground[-1][1] - line[-1][1]) > _SLOPE_CLOSE_TOLERANCE_M
    return right, left


def extend_unclosed(
    samples: list[dict[str, Any]],
    frame: dict[str, Any],
    compute: Callable[[list[dict[str, Any]]], dict[str, Any]],
    sampler: Any,
    step_m: float = EXTEND_STEP_M,
) -> tuple[list[dict[str, Any]], dict[str, Any]] | None:
    """넓힌 샘플과 `{left_m, right_m, surface_end}` — 이미 닫혔으면 `None`."""
    design = compute(samples)
    if not design.get("slope_unclosed"):
        return None
    offsets = [offset for offset, _z in _valid_sorted(samples)]
    spacing = min(b - a for a, b in zip(offsets, offsets[1:]) if b - a > 1e-9)
    count = max(int(round(step_m / spacing)), 1)
    origin = np.array([float(frame["origin"]["x"]), float(frame["origin"]["y"])])
    left_axis = np.array([float(v) for v in frame["left_xy"]])
    info: dict[str, Any] = {"left_m": 0.0, "right_m": 0.0, "surface_end": False}
    samples = list(samples)
    while design.get("slope_unclosed"):
        right_open, left_open = _open_ends(samples, design)
        added: list[dict[str, Any]] = []
        for side, is_open, sign in (("right", right_open, -1.0), ("left", left_open, 1.0)):
            if not is_open:
                continue
            edge = min(offsets) if sign < 0 else max(offsets)
            new_offsets = np.array([edge + sign * spacing * k for k in range(1, count + 1)])
            xy = origin[None, :] + left_axis[None, :] * new_offsets[:, None]
            z, valid = sampler.sample_xy(xy)
            if not np.all(valid):
                info["surface_end"] = True
                continue
            added.extend(
                {
                    "offset_m": round(float(o), 6),
                    "x": round(float(p[0]), 6),
                    "y": round(float(p[1]), 6),
                    "z": round(float(e), 6),
                    "elevation_m": round(float(e), 6),
                    "valid": True,
                }
                for o, p, e in zip(new_offsets, xy, z)
            )
            info[f"{side}_m"] += step_m
        if not added:
            break
        samples = sorted(samples + added, key=lambda s: float(s["offset_m"]))
        offsets = [offset for offset, _z in _valid_sorted(samples)]
        design = compute(samples)
    return samples, info


def extend_unclosed_sections(
    longitudinal: dict[str, Any],
    cross_sections: list[dict[str, Any]],
    project_root: Path,
    standard: dict[str, Any] | None,
    sampler: Any,
) -> int:
    """저장 설계가 미교차인 측점을 넓혀 샘플·설계를 자리에서 갈고 횡단 파일에 남긴다. 넓힌 수."""
    from B06_Section.B06_Section_Router_Design import (
        USER_TOUCHED_KEYS,
        ford_drop_at,
        ford_surface_drops,
        stored_berm,
        stored_cut_slope,
    )
    from common_util.common_util_route_profile import design_elevation_from_longitudinal
    from config.config_system import STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M

    if sampler is None:
        return 0
    drops = ford_surface_drops(project_root)
    cross_dir = project_root / "B06_Section" / "cross_sections"
    changed = 0
    for section in cross_sections:
        design = section.get("design")
        if not isinstance(design, dict) or not design.get("slope_unclosed"):
            continue
        chainage = float(section.get("chainage_m", 0.0))

        def compute(samples: list[dict[str, Any]], design=design, chainage=chainage) -> dict:
            return compute_cross_design(
                samples,
                design_elevation_from_longitudinal(longitudinal, chainage),
                ground_type=str(design.get("ground_type") or "ripping_rock"),
                section_mode=str(design.get("section_mode") or "left_cut"),
                ditch_side=design.get("ditch_side"),
                ditch_type=str(design.get("ditch_type") or "standard"),
                paved=bool(design.get("paved", False)),
                standard=standard,
                rock_boundary_offset_m=design.get(
                    "rock_boundary_offset_m", STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M
                ),
                two_stage_slope=bool(design.get("two_stage_slope", True)),
                cut_slope_ratio=stored_cut_slope(design),
                ditch_enabled=design.get("ditch_enabled"),
                ditch_choice=design.get("ditch_choice"),
                surface_drop_m=ford_drop_at(chainage, drops),
                berm=stored_berm(design),
                **curve_widening_args(section),
            )

        try:
            result = extend_unclosed(
                section.get("samples") or [], section["frame"], compute, sampler
            )
        except (ValueError, KeyError):
            continue
        if result is None:
            continue
        samples, info = result
        if not info["left_m"] and not info["right_m"]:
            continue  # 첫 걸음부터 지표면 밖 — 샘플이 그대로라 갈 것이 없음
        recomputed = compute(samples)
        for key in ("status", "pavement_suggested", *USER_TOUCHED_KEYS):
            if design.get(key) is not None:
                recomputed[key] = design[key]
        section["samples"] = samples
        section["design"] = recomputed
        cross_path = find_cross_file(cross_dir, chainage)
        if cross_path.is_file():
            stored = json.loads(cross_path.read_text(encoding="utf-8"))
            atomic_write_json(cross_path, {**stored, "samples": samples})
        logger.info(
            "미교차 샘플 넓힘: 측점 %.3f 좌 +%sm 우 +%sm 지표면 끝 %s",
            chainage,
            info["left_m"],
            info["right_m"],
            info["surface_end"],
        )
        changed += 1
    return changed