Aislo/resources/tester/test_349_0j_draft_vs_saved.py

"""PLAN 349-0j — 노선 초안 미리보기 누가토량 대 [저장] 정본.

갈린 자리(검증_sub2 · 2번째 꺾임점 서쪽 20 m · 서버 체인을 실제로 돌리고 체인 스스로 되살림):
  미리보기 = 서버 이월 직후(보정 전)와 측점 · 계획고 · 면적이 같음(-11,378.5 대 -11,391.3 ㎥)
  · [확정] 보정 중 미교차 넓힘(표본 20 m → 지표면 끝) -9,531 ㎥ · 관 구조물 면적 -334 ㎥(넓힌 뒤 기준 1,971 ㎥).
· cross 단계에 설계 지형 격자를 넘기면 미교차 측점을 서버와 같은 규칙(`extendUnclosed` ↔
  `extend_unclosed`)으로 넓힘 — 넓힌 끝 · 면적 · 누가토량이 서버 엔진과 같음
· 격자를 안 넘기면(지금 묶음) 넓힘 없음 — 서버와 갈림
· 초안 상세의 설계 = cross 단계 결과의 사본(구조물 면적 얹기가 단계 기억을 안 고침)
짝: `helper_349_0j_draft_vs_saved.cjs` · 지형 · 노선 · 관 틀 = `test_346_2_pipeline_cross_page.py`.
"""

from __future__ import annotations

import json
import subprocess
import sys
import tempfile
from pathlib import Path

import pytest

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
for path in (ROOT, HERE):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options  # noqa: E402
from B05_Profile.B05_Profile_Engine_Grade_Profile import (  # noqa: E402
    design_ground_following_profile,
)
from B06_Section.B06_Section_Engine_Design import compute_cross_design  # noqa: E402
from B06_Section.B06_Section_Engine_SampleExtend import extend_unclosed  # noqa: E402
from common_util.common_util_surface_sampler import DtmGridSampler, dtm_grid_bytes  # noqa: E402
from config.config_system_design import STANDARD_CROSS_SECTION  # noqa: E402
from test_346_2_pipeline_cross_page import (  # noqa: E402
    INTERVAL_M,
    MIN_RADIUS_M,
    ONES,
    _fake,
    _line,
    _pipes,
    _real,
    _saved_detail,
    _sections,
    _types,
)

pytestmark = pytest.mark.skipif(
    not (ROOT / "config" / "node_modules" / "typescript").is_dir(),
    reason="프론트엔드 의존성 미설치",
)


def _server(ts: dict, sampler) -> list[dict]:
    """[저장] 정본 길 — 같은 노선 서버 측점 · 지반 → 엔진 설계 → 미교차 넓힘(확정 보정)."""
    pipes = [
        {"id": s["key"].split(":", 1)[1], "chainage_m": s["chainage"], "source": "user"}
        for s in ts["stations"]
        if s["key"].startswith("pipe:")
    ]
    data = _sections(ts["line"], sampler, pipes, ts["curves"])
    out = []
    for a, b in zip(ts["stations"], data["cross_sections"]):
        assert abs(a["chainage"] - b["chainage_m"]) < 1e-3

        def compute(samples, a=a, b=b):
            return compute_cross_design(
                samples,
                a["elevation"],
                ground_type="ripping_rock",
                section_mode=f"{b.get('uphill_side') or 'left'}_cut",
                plan_radius_m=b.get("plan_radius_m"),
                curve_outer_side=b.get("curve_outer_side"),
                curve_widening_m=b.get("curve_widening_m"),
            )

        design = compute(b["samples"])
        wide = extend_unclosed(b["samples"], b["frame"], compute, sampler)
        if wide is not None and (wide[1]["left_m"] or wide[1]["right_m"]):
            design = compute(wide[0])
        line = design["design_line"]
        out.append(
            {
                "chainage": b["chainage_m"],
                "cut": design["cut_area_m2"],
                "fill": design["fill_area_m2"],
                "lineEnds": [line[0]["offset_m"], line[-1]["offset_m"]],
            }
        )
    return out


def _final(rows) -> float:
    nets = [(r["chainage"], r["cut"] - r["fill"]) for r in rows]
    return sum((n0 + n1) / 2 * (c1 - c0) for (c0, n0), (c1, n1) in zip(nets, nets[1:]))


@pytest.fixture(scope="module", params=[_fake, _real], ids=["가짜", "검증_sub7"])
def r(request) -> dict:
    with tempfile.TemporaryDirectory() as folder:
        npz, nodes = request.param(Path(folder))
        line = _line(nodes)
        sampler = DtmGridSampler.from_npz(npz)
        pipes = _pipes(line)
        saved = _sections(line, sampler, pipes)
        options = resolve_grade_options("trunk", terrain_type="normal", paved=False, stored=None)
        policy = design_ground_following_profile(saved["longitudinal"], options)[0]["policy"]
        job = {
            "nodes": nodes,
            "line": line,
            "minRadius": MIN_RADIUS_M,
            "policy": policy,
            "types": _types(),
            "interval": INTERVAL_M,
            "pipes": pipes,
            "standard": STANDARD_CROSS_SECTION,
            "conversion": ONES,
            "detail": _saved_detail(saved),
            "haulDelayMs": 20,
            "drags": [{"index": 1, "dx": -20.0, "dy": 0.0}],
        }
        path = Path(folder) / "job.json"
        path.write_text(json.dumps(job), encoding="utf-8")
        grid = Path(folder) / "grid.bin"
        grid.write_bytes(dtm_grid_bytes(npz))
        proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
            ["node", str(HERE / "helper_349_0j_draft_vs_saved.cjs"), str(path), str(grid)],
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=180,
        )
        assert proc.returncode == 0, proc.stderr
        out = json.loads(proc.stdout)
        out["server"] = _server(out["wide"], sampler)
        out["kind"] = request.param.__name__
    return out


def test_격자를_넘기면_서버와_같은_넓힘_면적(r):
    wide, server = r["wide"], r["server"]
    assert wide["outcome"] == "done"
    assert len(wide["stations"]) == len(server)
    for a, b in zip(wide["stations"], server):
        assert a["lineEnds"] == pytest.approx(b["lineEnds"]), a["key"]
        assert abs(a["cut"] - b["cut"]) < 0.05 + 0.01 * b["cut"], a["key"]
        assert abs(a["fill"] - b["fill"]) < 0.05 + 0.01 * b["fill"], a["key"]


def test_미리보기_누가토량_서버_정본과_같음(r):
    want = _final(r["server"])
    got = r["wide"]["final"]
    assert got is not None
    assert abs(got - want) <= max(1.0, abs(want) * 0.005), (got, want)


def test_격자_없으면_넓힘_없음(r):
    plain, wide = r["plain"], r["wide"]
    widened = [
        s["key"] for s in wide["stations"] if s["lineEnds"] and max(map(abs, s["lineEnds"])) > 20.01
    ]
    assert all(
        s["lineEnds"] is None or max(map(abs, s["lineEnds"])) <= 20.01 for s in plain["stations"]
    )
    if r["kind"] == "_real":  # 실제 지형 — 미교차가 있어 넓힘이 누가토량을 바꿈
        assert widened
        assert abs(plain["final"] - wide["final"]) > 1.0
    else:
        assert plain["final"] == pytest.approx(wide["final"]) or widened


def test_초안_상세_표본은_넓힌_표본(r):
    widened = [
        s for s in r["wide"]["stations"] if s["lineEnds"] and max(map(abs, s["lineEnds"])) > 20.01
    ]
    assert r["wide"]["shownWide"] == len(widened)  # 349-0n — 서버도 넓힌 표본을 상세에 남김
    assert r["plain"]["shownWide"] == 0


def test_초안_상세_설계는_사본(r):
    assert r["wide"]["copied"] is True