Aislo/resources/tester/test_428_pipe_move_mirror.py

"""PLAN 428 — 평면 · 관 옮김 뒤 횡단 형상 · 측점 한 값.

요구 255-1 · 257-1 · 261 · 분석 424 1장 · 7장.

못박는 것
  - 0.5 m 넘게 옮긴 관 측점 = 기본 설계(새 상단측 깎기 · 리핑암 · 표준 측구)
    + 사용자가 만진 칸 — 서버가 그 행을 지우고 기본 설계를 세우는 것과 같음 ·
    0.5 m 안이면 앞 설계 그대로.
  - 저장 노선 다시 셈(`refreshCrossDesigns`)이 미교차 측점을 서버 `extend_unclosed`
    와 같이 넓힘 → 브라우저 관 길이 = 서버 관 길이(같은 기하 함수 · 표본 · 설계).
  - 관 초안 얹기 — 따라가기만 한 관(땅 위 자리 = 정본 x · y) = 서버 누가거리 ·
    사용자가 옮긴 관 = 그대로.
  - 관 PUT 끝 서버가 측점을 다시 세움 · GET 관 지점에 x · y.
짝: `helper_428_pipe_move.mjs` · 진입점 `helper_428_pipe_move_entry.ts` ·
지형 · 노선 틀 = `test_346_2_pipeline_cross_page.py`.
"""

from __future__ import annotations

import inspect
import json
import shutil
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 test_346_2_pipeline_cross_page import _fake, _line, _sections  # noqa: E402

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 import STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M  # noqa: E402
from config.config_system_design import STANDARD_CROSS_SECTION  # noqa: E402

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

CULVERT = {
    "type": "pipe",
    "pipe_kind": "흄관",
    "diameter_m": 1.0,
    "min_cover_m": 0.5,
    "inlet": {"role": "inlet", "structure": "기슭막이", "revet_form": "돌쌓기(찰)"},
    "outlet": {"role": "outlet", "structure": "기슭막이", "revet_form": "돌쌓기(찰)"},
}


def _pipe_section(line, sampler, chainage: float) -> dict:
    data = _sections(line, sampler, [{"id": "p-a", "chainage_m": chainage, "source": "user"}])
    return next(s for s in data["cross_sections"] if abs(s["chainage_m"] - chainage) < 1e-6)


def _server(section: dict, elevation: float, sampler):
    """서버 길 — 옮긴 관 측점 = 기본 설계(상단측 깎기 · `compute_default_designs`)
    → [저장] 보정 맨 앞 미교차 넓힘(`extend_unclosed_sections` — 저장 설계의
    단면 유형 · 측구 칸을 입력으로 다시 셈)."""
    base = {
        "ground_type": "ripping_rock",
        "ditch_type": "standard",
        "standard": STANDARD_CROSS_SECTION,
        "rock_boundary_offset_m": STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
    }
    stored = compute_cross_design(
        section["samples"],
        elevation,
        section_mode=f"{section.get('uphill_side') or 'left'}_cut",
        **base,
    )

    def compute(samples):
        return compute_cross_design(
            samples,
            elevation,
            section_mode=stored["section_mode"],
            ditch_side=stored.get("ditch_side"),
            ditch_enabled=stored.get("ditch_enabled"),
            ditch_choice=stored.get("ditch_choice"),
            two_stage_slope=bool(stored.get("two_stage_slope", False)),
            **base,
        )

    wide = extend_unclosed(section["samples"], section["frame"], compute, sampler)
    if not (wide and (wide[1]["left_m"] or wide[1]["right_m"])):
        return section["samples"], stored, wide
    return wide[0], compute(wide[0]), wide


@pytest.fixture(scope="module")
def r(tmp_path_factory) -> dict:
    work = tmp_path_factory.mktemp("b428")
    build = subprocess.run(  # noqa: S603 — 고정 실행 파일
        [
            "node",
            str(ROOT / "config" / "node_modules" / "vite" / "bin" / "vite.js"),
            "build",
            "--configLoader",
            "runner",
            "--ssr",
            "../resources/tester/helper_428_pipe_move_entry.ts",
            "--outDir",
            str(work),
            "--logLevel",
            "error",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=300,
    )
    assert build.returncode == 0, build.stderr
    bundle = work / "bundle.mjs"
    shutil.copy(work / "helper_428_pipe_move_entry.js", bundle)
    with tempfile.TemporaryDirectory() as folder:
        npz, nodes = _fake(Path(folder))
        line = _line(nodes)
        sampler = DtmGridSampler.from_npz(npz)
        old = _pipe_section(line, sampler, 90.0)
        moved = _pipe_section(line, sampler, 96.0)  # 6 m 옮김 — 서버가 옛 행을 지우고 새로 세움
        # 표본 20 m 밖까지 닿는 성토 높이 — 넓힘이 실제로 도는 자리
        for rise in (10.0, 14.0, 18.0, 24.0):
            elevation = float(moved["center_z"]) + rise
            samples, design, wide = _server(moved, elevation, sampler)
            if samples is not moved["samples"]:
                break
        else:
            pytest.fail("넓힘이 도는 계획고를 못 찾음")
        uphill = moved.get("uphill_side") or "left"
        other = "right" if uphill == "left" else "left"
        old["design"] = {
            "design_elevation_m": elevation,
            "ground_type": "soil",
            "section_mode": f"{other}_cut",
            "ditch_side": other,
            "two_stage_slope": True,
            "rock_boundary_offset_m": 9.0,
            "inlet_structure": "revet",
        }
        job = {
            "old": old,
            "moved": moved,
            "elevation": elevation,
            "standard": STANDARD_CROSS_SECTION,
            "rockDefault": STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
            "culvert": CULVERT,
            "serverSamples": samples,
            "serverDesign": design,
        }
        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_428_pipe_move.mjs"), str(bundle), str(path), str(grid)],
            cwd=ROOT,
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=120,
        )
        assert proc.returncode == 0, proc.stderr
        out = json.loads(proc.stdout)
        out["job"] = {"uphill": uphill, "other": other, "wide": wide[1], "samples": len(samples)}
        out["serverDesign"] = design
    return out


def test_옮긴_관_측점_기본_설계와_만진_칸(r):
    s = r["swapped"]
    assert s["uphill"] == r["job"]["uphill"]  # 상단측 = 새 지반
    assert s["section_mode"] == f"{r['job']['uphill']}_cut"
    assert s["ground_type"] == "ripping_rock"
    assert s["ditch_side"] is None and s["two_stage_slope"] is None
    assert s["rock_boundary_offset_m"] is None
    assert s["inlet_structure"] == "revet"  # 사용자가 만진 칸은 따라감


def test_0_5m_안_옮김_앞_설계_그대로(r):
    n = r["near"]
    assert n["section_mode"] == f"{r['job']['other']}_cut"
    assert n["ground_type"] == "soil" and n["ditch_side"] == r["job"]["other"]
    assert n["rock_boundary_offset_m"] == 9.0


def test_저장_노선_다시_셈이_서버처럼_넓힘(r):
    b = r["browser"]
    s = r["server"]  # 구조물 폐회로 면적까지(서버 Node 와 같은 함수)
    assert r["job"]["wide"]["left_m"] or r["job"]["wide"]["right_m"]
    assert r["firstSamples"] < b["samples"] == r["job"]["samples"]  # 격자 받은 뒤 넓힘
    assert abs(b["cut"] - s["cut"]) < 1e-3 and abs(b["fill"] - s["fill"]) < 1e-3
    d = r["serverDesign"]
    ends = [d["design_line"][0]["offset_m"], d["design_line"][-1]["offset_m"]]
    assert all(abs(a - e) < 1e-3 for a, e in zip(b["ends"], ends))


def test_옮긴_관_관_길이_브라우저_대_서버(r):
    got = r["browser"]["pipeLength"]
    want = r["server"]["pipeLength"]
    assert got is not None and want is not None and got > 0
    assert abs(got - want) < 1e-3, (got, want)


def test_관_초안_얹기_따라가기만_한_관은_서버_누가거리(r):
    assert r["rebase"] == [["f", 104], ["u", 210], ["s", 301]]


def test_관_PUT_끝_측점_다시_세움_GET_좌표():
    from B04_PreProcess import B04_PreProcess_Router_Basins as basins

    put = inspect.getsource(basins.put_pipe_points)
    assert (
        put.index("save_pipe_points") < put.index("_stand_pipe_stations(") < put.index("_payload(")
    )
    stand = inspect.getsource(basins._stand_pipe_stations)
    assert "ensure_structure_stations(project_id)" in stand and "except Exception" in stand
    raw = inspect.getsource(basins._raw_payload)
    assert '"x": round(float(pipe.x), 3)' in raw and '"y": round(float(pipe.y), 3)' in raw