Aislo/resources/tester/test_344_3_pipeline_stations_mirror.py

"""PLAN 344-3 · 346-1 — 계산 묶음 stations 단계 거울(파이썬 ↔ TS).

· 서버 `generate_sections` 측점 목록 = TS `buildStations`(`B05_Profile/logic/B05_Profile_Sections_Stations.ts`)
  — 누가거리 · 종류 · 열쇠 · 자리 · 방위각 · 평면 반경 · 바깥쪽 · 확폭(곡선표 · 스냅 · 곡선표 없음 셋)
· 상수 = config_system_design · Engine_Sections_Core 같은 값
· 노선 한 점 옮김 뒤 앞 구간 측점 열쇠 그대로 · 범위 밖은 앞 객체 그대로(단계 `design_pipeline_stations.ts`)
짝: `helper_344_3_pipeline_stations.cjs`(실제 TS 를 Node 로 돌림).
"""

from __future__ import annotations

import json
import math
import subprocess
import sys
import tempfile
from types import SimpleNamespace
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 import B05_Profile_Engine_Sections_Core as core  # noqa: E402
from B05_Profile.B05_Profile_Structures_Schema import structure_type_map  # noqa: E402
from common_util.common_util_drainage_pipes import parse_pipe_points  # noqa: E402
from common_util.common_util_structure_station import plan_stations  # noqa: E402
from config import config_system_design as cfg  # noqa: E402

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

_TOL = 1.5e-6  # 두 쪽 다 소수 6자리 반올림

PIPES = [
    {"id": "p57", "chainage_m": 57.0, "source": "user"},
    {"id": "p140", "chainage_m": 140.0, "source": "user"},  # 격자 위 — 관 측점이 위
    {
        "id": "p-rev",
        "chainage_m": 250.0,
        "source": "user",
        "facility": "revetment",
        "start_m": 245.0,
        "end_m": 260.0,
    },
    {"id": "p333", "chainage_m": 333.0, "source": "user"},
]
STRUCTURES = [
    {
        "structure_id": "s-wall",
        "type_id": "retaining_wall",
        "placement": "interval",
        "chainage_m": 410.0,
        "start_m": 400.0,
        "end_m": 430.0,
    },
]


class _Flat:
    """평지 표본기 — 측점 부분만 대조(지반은 344-1 거울 몫)."""

    def sample_xy(self, xy):
        xy = np.asarray(xy)
        return np.zeros(len(xy)), np.ones(len(xy), dtype=bool)


def _route(nodes, radii):
    """꺾임점마다 원호(반경) — 그린 노선 · 곡선표(시작 · 꼭짓점 · 끝 · 반경)."""
    line = [list(nodes[0])]
    curves = []
    for i in range(1, len(nodes) - 1):
        p, b, n = (np.asarray(v, dtype=float) for v in (nodes[i - 1], nodes[i], nodes[i + 1]))
        u = (b - p) / np.linalg.norm(b - p)
        v = (n - b) / np.linalg.norm(n - b)
        delta = math.acos(max(-1.0, min(1.0, float(u @ v))))
        r = radii[i - 1]
        t = r * math.tan(delta / 2)
        bc, ec = b - u * t, b + v * t
        turn = 1.0 if u[0] * v[1] - u[1] * v[0] > 0 else -1.0
        center = bc + turn * r * np.array([-u[1], u[0]])
        a0 = math.atan2(*(bc - center)[::-1])
        for k in range(21):
            a = a0 + turn * delta * k / 20
            line.append([float(center[0] + r * math.cos(a)), float(center[1] + r * math.sin(a))])
        curves.append({"start": bc.tolist(), "apex": b.tolist(), "end": ec.tolist(), "radius_m": r})
    line.append(list(nodes[-1]))
    return line, curves


NODES = [(0.0, 0.0), (200.0, 0.0), (300.0, 120.0), (520.0, 130.0), (600.0, 40.0)]
RADII = [25.0, 12.0, 40.0]
LINE, CURVES = _route(NODES, RADII)
MOVED_LINE, MOVED_CURVES = _route([*NODES[:-1], (640.0, 90.0)], RADII)


def _types() -> list[dict]:
    return [
        {
            "type_id": t.type_id,
            "group": t.group,
            "placement": t.placement,
            "managed_by": t.managed_by,
        }
        for t in structure_type_map().values()
    ]


def _server(line, curves, snap):
    extras = tuple(
        (float(s["chainage_m"]), str(s["title"]))
        for s in plan_stations(
            parse_pipe_points(PIPES), [SimpleNamespace(options=None, **item) for item in STRUCTURES]
        )
    )
    data = core.generate_sections(
        line,
        _Flat(),
        core.SectionGenerationOptions(extra_stations=extras),
        design_curves=curves,
        station_snap=snap,
    )
    return data["longitudinal"]["stations"]


def _first_changed_m() -> float:
    """옮긴 꼭짓점의 앞 접선점(BC) 누가거리 — 노선 = 처음 바뀐 접선점부터 끝."""
    _gap, at = core._project_chainage(
        *core._clean_polyline(np.asarray(LINE)), tuple(CURVES[-1]["start"])
    )
    return at


CASES = [
    {"name": "곡선표", "line": LINE, "curves": CURVES, "snap": False},
    {"name": "스냅", "line": LINE, "curves": CURVES, "snap": True},
    {"name": "곡선표없음", "line": LINE, "curves": [], "snap": False},
]


@pytest.fixture(scope="module")
def r() -> dict:
    job = {
        "cases": CASES,
        "pipes": PIPES,
        "structures": STRUCTURES,
        "types": _types(),
        "move": {
            "line": LINE,
            "curves": CURVES,
            "movedLine": MOVED_LINE,
            "movedCurves": MOVED_CURVES,
            "fromM": _first_changed_m(),
        },
    }
    with tempfile.TemporaryDirectory() as folder:
        path = Path(folder) / "job.json"
        path.write_text(json.dumps(job), encoding="utf-8")
        proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
            ["node", str(HERE / "helper_344_3_pipeline_stations.cjs"), str(path)],
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=60,
        )
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_상수_같음(r):
    c = r["constants"]
    assert c["interval"] == cfg.SECTION_STATION_INTERVAL_M
    assert c["endpoint"] == cfg.SECTION_INCLUDE_ENDPOINT
    assert c["taper"] == cfg.CURVE_WIDENING_TAPER_M
    assert [tuple(row) for row in c["table"]] == list(cfg.CURVE_WIDENING_TABLE_M)
    assert c["arm"] == core.PLAN_RADIUS_ARM_M and c["straight"] == core.PLAN_RADIUS_STRAIGHT_M


def _close(a, b) -> bool:
    if a is None or b is None:
        return a is b
    return abs(float(a) - float(b)) <= _TOL


_KIND = {"bp": {"end"}, "ep": {"end"}, "regular": {"grid"}, "irregular": {"pipe", "structure"}}


@pytest.mark.parametrize("index", range(len(CASES)), ids=[c["name"] for c in CASES])
def test_서버_측점_목록과_같음(r, index):
    case = CASES[index]
    server = _server(case["line"], case["curves"], case["snap"])
    ours = r["cases"][index]
    assert [s["chainage_m"] for s in server] == pytest.approx(
        [s["chainageM"] for s in ours], abs=_TOL
    )
    problems = []
    for py, ts in zip(server, ours, strict=True):
        where = f"{py['chainage_m']}"
        if ts["kind"] not in _KIND[py["kind"]]:
            problems.append(f"{where} 종류 {py['kind']} vs {ts['kind']}")
        pairs = {
            "x": (py["center_x"], ts["x"]),
            "y": (py["center_y"], ts["y"]),
            "방위각": (py["azimuth_deg"], ts["azimuthDeg"]),
            "반경": (py["plan_radius_m"], ts["planRadiusM"]),
            "확폭": (py["curve_widening_m"], ts["curveWideningM"]),
            "좌측x": (py["frame"]["left_xy"][0], ts["left"][0]),
            "좌측y": (py["frame"]["left_xy"][1], ts["left"][1]),
        }
        for label, (a, b) in pairs.items():
            if not _close(a, b):
                problems.append(f"{where} {label} {a} vs {b}")
        if py["curve_outer_side"] != ts["curveOuterSide"]:
            problems.append(f"{where} 바깥쪽 {py['curve_outer_side']} vs {ts['curveOuterSide']}")
    assert not problems, problems[:10]
    # 확폭 · 반경이 실제로 섬(시험이 빈 값만 대조하지 않게)
    assert any(s["curveWideningM"] > 0 for s in ours) and any(s["planRadiusM"] for s in ours)


def test_열쇠(r):
    ours = r["cases"][0]
    keys = {s["key"]: s for s in ours}
    assert ours[0]["key"] == "end:BP" and ours[-1]["key"] == "end:EP"
    assert keys["pipe:p57"]["chainageM"] == 57.0 and keys["pipe:p57"]["id"] == "p57"
    assert keys["pipe:p140"]["kind"] == "pipe" and "grid:7" not in keys  # 격자 위 관 = 관 열쇠 하나
    assert {"pipe:p-rev:start", "pipe:p-rev", "pipe:p-rev:end"} <= keys.keys()
    assert {"structure:s-wall:start", "structure:s-wall", "structure:s-wall:end"} <= keys.keys()
    for s in ours:
        if s["kind"] == "grid":
            assert s["key"] == f"grid:{round(s['chainageM'] / 20)}"
    assert len(keys) == len(ours)  # 열쇠 겹침 없음
    assert r["text"] == ["0.00", "12.35", "40.00", "700.00"]


def test_스냅_측점은_원래_격자_번호(r):
    plain = {s["key"]: s["chainageM"] for s in r["cases"][0] if s["kind"] == "grid"}
    snapped = {s["key"]: s["chainageM"] for s in r["cases"][1] if s["kind"] == "grid"}
    assert plain.keys() == snapped.keys()  # 옮겨도 번호 그대로 · 겹침 없음
    moved = {k: (plain[k], snapped[k]) for k in plain if plain[k] != snapped[k]}
    assert moved and all(abs(a - b) <= 10 for a, b in moved.values())


def test_노선_한_점_옮김_앞_구간_열쇠_그대로(r):
    m = r["move"]
    from_m = _first_changed_m()
    before = {k: c for k, c in m["before"]}
    after = dict(m["after"])
    ahead = [k for k, c in m["before"] if c < from_m]
    assert ahead and all(after.get(k) == before[k] for k in ahead)
    # 범위(옮긴 접선점 − 여유 20 m) 밖은 앞 객체 그대로 · 안은 새로 셈
    for (key, chainage), same in zip(m["after"], m["same"], strict=True):
        if chainage < from_m - 20 and key in before:
            assert same, key
    assert not all(m["same"])
    assert m["empty"] == []