Aislo/resources/tester/test_400_profile_carry.py

# -*- coding: utf-8 -*-
"""400 평면 고친 뒤 종단 편집 이어받기 — 같은 자리(옆 1 m · 방향 30° 안)는 옛 계획고를 새 측점 위에.

거울(파이썬 = 서버 [저장] 정본 · TS = 브라우저 즉시 · 같은 값):
  `B05_Profile_Engine_Grade_Carry` line_matcher · carry_base · carry_edits ↔
  `logic/B05_Profile_Grade_Carry.ts` lineMatcher · carryBase · carryEdits.
묶음 follow(`logic/B05_Profile_Pipeline_Grade.ts`) — 걸음 수와 무관(기준 한 벌에서 한 번) ·
되돌린 노선 = 옛 편집.
"""

from __future__ import annotations

import json
import math
import subprocess
import sys
from pathlib import Path

import numpy as np
import pytest

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

from B05_Profile.B05_Profile_Engine_Grade_Alignment import resolve_pvi  # noqa: E402
from B05_Profile.B05_Profile_Engine_Grade_Carry import (  # noqa: E402
    SAME_LINE_TOL_M,
    carry_base,
    carry_edits,
    line_matcher,
)

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

INTERVAL = 20.0
HALF = INTERVAL / 2


# ── 입력 만들기 ───────────────────────────────────────────────────────────────


def _dense(nodes: list[tuple[float, float]], step: float = 2.0) -> list[list[float]]:
    """꺾은선을 `step` m 안팎 점으로(그린 노선처럼 정점이 많게)."""
    out = [list(nodes[0])]
    for (x0, y0), (x1, y1) in zip(nodes, nodes[1:]):
        n = max(1, int(math.ceil(math.dist((x0, y0), (x1, y1)) / step)))
        out += [[x0 + (x1 - x0) * k / n, y0 + (y1 - y0) * k / n] for k in range(1, n + 1)]
    return out


def _length(line) -> float:
    return sum(math.dist(a, b) for a, b in zip(line, line[1:]))


def _stations(line) -> list[float]:
    total = _length(line)
    grid = [round(k * INTERVAL, 3) for k in range(int(total // INTERVAL) + 1)]
    return grid + ([round(total, 3)] if total - grid[-1] > 1e-6 else [])


def _ground(s: float) -> float:
    return 100.0 + 0.05 * s + 2.0 * math.sin(s / 37.0)


def _ground_pvi(line) -> list[dict]:
    """새 노선 지반 첫 계획선 꼴(측점마다 지반 · 옛 계획고와 다르게 +3 m)."""
    return [{"chainage_m": s, "elevation_m": round(_ground(s) + 3.0, 6)} for s in _stations(line)]


OLD_NODES = [(0.0, 0.0), (200.0, 0.0), (400.0, 100.0), (600.0, 100.0), (800.0, 0.0)]
OLD = _dense(OLD_NODES)
OLD_BASE = [{"chainage_m": s, "elevation_m": round(_ground(s), 6)} for s in _stations(OLD)]


def _old_edits() -> dict:
    """직선화(40 ~ 120 사이 측점) · ▲ 1 m(300) · 종단곡선 R(300) · 자유지정 꺾임(520 · 540)."""
    base = {n["chainage_m"]: n["elevation_m"] for n in OLD_BASE}
    z40, z120 = base[40.0], base[120.0]
    offsets = {}
    for s in (60.0, 80.0, 100.0):
        offsets[f"{s:.3f}"] = round(z40 + (z120 - z40) * (s - 40) / 80 - base[s], 6)
    offsets["300.000"] = 1.0
    offsets["520.000"] = -0.8
    offsets["540.000"] = -1.2
    return {"station_offsets": offsets, "curve_radii": {"300.000": 400.0, "520.000": 250.0}}


OLD_EDITS = _old_edits()


def _moved(index: int, to: tuple[float, float]) -> list[list[float]]:
    nodes = list(OLD_NODES)
    nodes[index] = to
    return _dense(nodes)


CASES = {
    # 먼 교점(끝 앞) 하나 옮김 — 앞쪽 편집 그대로
    "far": _moved(3, (620.0, 130.0)),
    # 시점을 뒤로 15 m 늘림 — 같은 선 · 누가거리만 +15(새 측점은 옛 측점 사이)
    "start": _dense([(-15.0, 0.0)] + OLD_NODES[1:]),
    # 끝을 같은 방향으로 늘림 — 옛 끝까지 그대로 · 넘는 자리만 지반
    "end": _dense(OLD_NODES[:-1] + [(840.0, -20.0)]),
    # 측점 스냅처럼 뒤 교점이 옆으로 0.6 m(같은 자리) · 1.6 m(바뀐 자리) 밀림
    "snap": _dense(OLD_NODES[:2] + [(400.0, 100.6), (600.0, 101.6), (800.0, 1.6)]),
    "same": OLD,
}


def _xy(line):
    return [{"x": x, "y": y} for x, y in line]


def _plan(base: list[dict], edits: dict) -> tuple[np.ndarray, np.ndarray]:
    s = np.array([n["chainage_m"] for n in base], dtype=np.float64)
    z = np.array([n["elevation_m"] for n in base], dtype=np.float64)
    pvi_s, pvi_z, _ = resolve_pvi(s, z, edits["station_offsets"])
    return pvi_s, pvi_z


def _py(name: str) -> tuple[list[dict], dict]:
    new = CASES[name]
    base = carry_base(OLD, new, OLD_BASE, _ground_pvi(new))
    edits = carry_edits(OLD, new, OLD_BASE, OLD_EDITS, _stations(new), HALF)
    return base, edits


# ── TS 한 번 ─────────────────────────────────────────────────────────────────

FOLLOW_STEPS = [
    # 기준 → 먼 교점 옮김 → 더 옮김 · 한 번에 옮김과 같아야
    [("far1", _moved(3, (610.0, 115.0))), ("far2", CASES["far"])],
    [("far2", CASES["far"])],
    # 바뀐 자리를 거쳐 되돌림 — 옛 편집 그대로
    [("mid", _moved(4, (830.0, 30.0))), ("back", OLD)],
]


# 측점 스냅 켠 노선 — 앞 교점 하나 옮기면 뒤 교점이 모두 MC 맞춤으로 밀림(StationFit)
SNAP = {
    "nodes": [[0, 0], [150, 30], [300, 0], [450, 40], [600, 10], [750, 50], [900, 20], [1050, 40]],
    "index": 1,
    "to": [150.0, 37.0],
    "limits": {"minRadiusM": 60.0, "limitRadiusM": 0.0, "limitArcM": 0.0, "intervalM": INTERVAL},
}


@pytest.fixture(scope="module")
def ts(tmp_path_factory) -> dict:
    queries = [0.0, 33.3, 210.0, 430.0, 500.0, 700.0, 815.0, 1e6]
    payload = {
        "match": [
            {"old_xy": _xy(OLD), "new_xy": _xy(line), "tol": SAME_LINE_TOL_M, "queries": queries}
            for line in CASES.values()
        ],
        "base": [
            {
                "old_xy": _xy(OLD),
                "new_xy": _xy(line),
                "old_base": OLD_BASE,
                "ground_pvi": _ground_pvi(line),
            }
            for line in CASES.values()
        ],
        "edits": [
            {
                "old_xy": _xy(OLD),
                "new_xy": _xy(line),
                "old_base": OLD_BASE,
                "edits": OLD_EDITS,
                "stations": _stations(line),
                "half": HALF,
            }
            for line in CASES.values()
        ],
        "follow": [
            {
                "ref_line": OLD,
                "policy": {"station_interval_m": INTERVAL},
                "base_pvi": OLD_BASE,
                "edits": OLD_EDITS,
                "steps": [{"line": line, "stations": _stations(line)} for _, line in steps],
            }
            for steps in FOLLOW_STEPS
        ],
        "snap": [SNAP],
    }
    source = tmp_path_factory.mktemp("s400") / "in.json"
    source.write_text(json.dumps(payload), encoding="utf-8")
    proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
        ["node", str(HERE / "helper_400_profile_carry.cjs"), str(source)],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=120,
        check=False,
    )
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


# ── 거울 ─────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize("index", range(len(CASES)))
def test_거울_같은자리_꼭짓점_편집(ts, index):
    name, line = list(CASES.items())[index]
    match = line_matcher(OLD, line)
    for t, got in zip([0.0, 33.3, 210.0, 430.0, 500.0, 700.0, 815.0, 1e6], ts["match"][index]):
        want = match(t)
        assert (want is None) == (got is None), (name, t)
        if want is not None:
            assert want == pytest.approx(got, abs=1e-9)
    base, edits = _py(name)
    assert len(base) == len(ts["base"][index])
    for a, b in zip(base, ts["base"][index]):
        assert a["chainage_m"] == pytest.approx(b["chainage_m"], abs=1e-9)
        assert a["elevation_m"] == pytest.approx(b["elevation_m"], abs=1e-6)
    got = ts["edits"][index]
    for field in ("station_offsets", "curve_radii"):
        assert edits[field].keys() == got[field].keys(), (name, field)
        for key, value in edits[field].items():
            assert value == pytest.approx(got[field][key], abs=1e-6)


# ── 규칙 ─────────────────────────────────────────────────────────────────────


def _same_place_plan(name: str):
    """같은 자리 새 측점마다 (새 계획고, 옛 계획고) · 바뀐 자리 측점."""
    new = CASES[name]
    base, edits = _py(name)
    match = line_matcher(OLD, new)
    old_s, old_z = _plan(OLD_BASE, OLD_EDITS)
    new_s, new_z = _plan(base, edits)
    kept, moved = [], []
    for t in _stations(new):
        s = match(t)
        if s is None:
            moved.append(t)
        else:
            kept.append((t, float(np.interp(t, new_s, new_z)), float(np.interp(s, old_s, old_z))))
    return kept, moved, base, edits


@pytest.mark.parametrize("name", ["far", "start", "end", "snap", "same"])
def test_같은자리_계획고_그대로(name):
    kept, _, _, _ = _same_place_plan(name)
    assert kept
    for t, new_z, old_z in kept:
        # 옛 자동 꼭짓점 0.006 m 안 측점은 그 꼭짓점 위로 봄 — 기울기 차 × 0.006 m 안
        assert new_z == pytest.approx(old_z, abs=1e-3), (name, t)


def test_먼교점_옮김_앞쪽_편집_누가거리_그대로_바뀐자리만_지반():
    kept, moved, base, edits = _same_place_plan("far")
    # 앞쪽(같은 선 · 같은 측점) 편집 글자 그대로
    for key in ("60.000", "80.000", "100.000", "300.000"):
        assert edits["station_offsets"][key] == pytest.approx(OLD_EDITS["station_offsets"][key])
    assert edits["curve_radii"]["300.000"] == 400.0
    # 바뀐 자리 = 옮긴 교점 둘레 · 그 측점은 지반 첫 계획선 · 편집 없음(옛 520 · 540 자유지정 버림)
    assert moved and min(moved) > 423.0
    ground = {n["chainage_m"]: n["elevation_m"] for n in _ground_pvi(CASES["far"])}
    by = {n["chainage_m"]: n["elevation_m"] for n in base}
    assert all(by[t] == ground[t] for t in moved)
    assert not any(float(k) in moved for k in edits["station_offsets"])


def test_시점_늘림_누가거리_투영_새측점_위():
    kept, moved, _, edits = _same_place_plan("start")
    assert not moved or max(moved) < 1.0
    stations = set(_stations(CASES["start"]))
    assert edits["station_offsets"] and all(float(k) in stations for k in edits["station_offsets"])
    assert all(float(k) in stations for k in edits["curve_radii"])
    # ▲(옛 300) 의 R 은 옛 300 = 새 315 에서 가장 가까운 새 측점(320 · 옛 305)
    assert edits["curve_radii"].get("320.000") == 400.0


def test_끝_늘림_옛끝까지_이어받고_넘는자리만_지반():
    kept, moved, base, _ = _same_place_plan("end")
    old_len = _length(OLD)
    assert moved and min(moved) > old_len
    assert max(t for t, _, _ in kept) > old_len - INTERVAL


def test_스냅처럼_옆으로_밀림_1m_안_이어받고_넘으면_다시():
    match = line_matcher(OLD, CASES["snap"])
    assert match(410.0) is not None  # 0.6 m 쪽
    assert match(650.0) is None  # 1.6 m 쪽
    _, _, _, edits = _same_place_plan("snap")
    assert "300.000" in edits["station_offsets"]


def test_종단_안_고친_프로젝트는_편집_없음_꼭짓점은_옛것_또는_지반():
    empty = {"station_offsets": {}, "curve_radii": {}}
    new = CASES["far"]
    assert carry_edits(OLD, new, OLD_BASE, empty, _stations(new), HALF) == empty
    base = carry_base(OLD, new, OLD_BASE, _ground_pvi(new))
    old_s = [n["chainage_m"] for n in OLD_BASE]
    old_z = [n["elevation_m"] for n in OLD_BASE]
    ground = {n["chainage_m"]: n["elevation_m"] for n in _ground_pvi(new)}
    match = line_matcher(OLD, new)
    for node in base:
        t = node["chainage_m"]
        s = match(t)
        want = ground[t] if s is None else round(float(np.interp(s, old_s, old_z)), 6)
        assert node["elevation_m"] == pytest.approx(want, abs=1e-9)


def test_엇갈려_지나는_선은_같은자리_아님():
    crossing = _dense([(100.0, -50.0), (100.0, 50.0)])
    assert line_matcher(OLD, crossing)(50.0) is None
    reverse = OLD[::-1]
    assert line_matcher(OLD, reverse)(100.0) is None


# ── 묶음 follow ──────────────────────────────────────────────────────────────


def test_follow_걸음수와_무관_서버와_같은값(ts):
    stepwise, direct, back = ts["follow"]
    assert stepwise[-1] == direct[-1]
    _, edits = _py("far")
    got = direct[-1]
    assert edits["station_offsets"].keys() == got["station_offsets"].keys()
    for key, value in edits["station_offsets"].items():
        assert value == pytest.approx(got["station_offsets"][key], abs=1e-6)
    assert edits["curve_radii"] == got["curve_radii"]


def test_follow_바뀐자리_거쳐_되돌리면_옛편집(ts):
    _, _, back = ts["follow"]
    assert "520.000" in back[0]["station_offsets"]  # 끝 교점 옮김 — 앞쪽(먼) 편집은 남음
    assert back[-1] == OLD_EDITS


def test_측점스냅_켠_노선_뒤_교점_밀려도_같은자리_편집_이어받음(ts):
    got = ts["snap"][0]
    old_line, new_line = got["old_line"], got["new_line"]
    shifts = [math.dist(a, b) for a, b in zip(got["old_nodes"][2:], got["new_nodes"][2:])]
    assert max(shifts) > 0.5  # 뒤 교점이 정말 밀림 — 1 mm 정점 비교로는 뒤쪽 전부 「바뀐 구간」
    stations = _stations(old_line)
    base = [{"chainage_m": t, "elevation_m": round(_ground(t), 6)} for t in stations]
    far = [t for t in stations if 600.0 <= t <= 900.0]
    offsets = {f"{t:.3f}": round(0.5 + 0.001 * t, 6) for t in far}  # 뒤쪽 직선화 · ▲ 몫
    edits = {"station_offsets": offsets, "curve_radii": {}}
    new_stations = _stations(new_line)
    new_base = carry_base(old_line, new_line, base, _ground_pvi(new_line))
    new_edits = carry_edits(old_line, new_line, base, edits, new_stations, HALF)
    match = line_matcher(old_line, new_line)
    old_s, old_z = _plan(base, edits)
    new_s, new_z = _plan(new_base, new_edits)
    carried = [t for t in new_stations if f"{t:.3f}" in new_edits["station_offsets"]]
    assert (
        len(carried) >= len(far) // 2
    )  # 뒤쪽 편집 절반 넘게 이어받음(나머지 = 1 m 넘게 밀린 곡선 둘레)
    for t in carried:
        s = match(t)
        assert float(np.interp(t, new_s, new_z)) == pytest.approx(
            float(np.interp(s, old_s, old_z)), abs=1e-3
        )