Aislo/resources/tester/test_349_0s_draw_work.py

"""PLAN 349-0s sub7 — 끌기 한 걸음 = 종단도 한 번 그림 · 묶음 기록에 알림 받는 쪽 · 그리기 몫(ms).

· 꼭짓점 끌기 다섯 걸음 · 되돌리기 — 걸음마다 종단도 그리기 ≤ 2(종단 먼저 · 횡단 · 유토곡선 다음 틀)
  (옛 길 = 갈아끼움 · 정책 · 이어 공사 기준 · 비정규 측점 · 새 상세 알림마다 바로 그림)
· runs[].work — 가게 · 새 상세 · 종단도 · 쉼 이름별 ms(알림 받는 쪽 · 그리기 몫)
· 되돌리기 = 열기 값 규칙 그대로(restored)
짝: `helper_349_0s_draw_work.cjs`(실제 TS 를 Node 로 · 화면 부품 · 서버 창구만 가짜).
"""

from __future__ import annotations

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

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_Structures_Schema import structure_type_map  # noqa: E402
from common_util.common_util_surface_sampler import dtm_grid_bytes  # noqa: E402
from config.config_system_design import STANDARD_CROSS_SECTION  # noqa: E402

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

PROJECT = ROOT / "storage/1/3/6d3b2812-6d5a-4168-a932-40e4ae303d5a"  # 검증_sub7
ROUTE = PROJECT / "B05_Profile/route"
SECTION = PROJECT / "B06_Section"
GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz"
ONES = {g: {"loose": 1.0, "compacted": 1.0} for g in ("soil", "ripping_rock", "blasting_rock")}


def _csv(path: Path) -> list[list[float]]:
    with path.open(encoding="utf-8") as handle:
        return [[float(r["x"]), float(r["y"])] for r in csv.DictReader(handle)]


def _json(path: Path):
    return json.loads(path.read_text(encoding="utf-8"))


def _between(sections: list[dict], index: int) -> dict:
    """두 측점 사이 한 점(0.37) — 누가거리 · 좌표."""
    a, b = sections[index], sections[index + 1]
    k = 0.37
    return {
        "chainage_m": round(a["chainage_m"] + k * (b["chainage_m"] - a["chainage_m"]), 2),
        "x": a["center_x"] + k * (b["center_x"] - a["center_x"]),
        "y": a["center_y"] + k * (b["center_y"] - a["center_y"]),
    }


@pytest.fixture(scope="module")
def r() -> dict:
    if not (GRID.is_file() and (SECTION / "longitudinal/longitudinal.json").is_file()):
        pytest.skip("검증_sub7 없음")
    curves = _json(ROUTE / "planned_route_curves.json")
    sections = sorted(
        (_json(p) for p in (SECTION / "cross_sections").glob("cross_*.json")),
        key=lambda s: s["chainage_m"],
    )
    pipes = _json(PROJECT / "B04_PreProcess/drainage/edits/pipe_points.json")["points"]
    # 세월교 하나 더(측점 사이) — 저장 상세엔 그 측점이 없음 = 묶음 저장 노선 측점과 다름(검증_sub7 세월교 2 + 1)
    pipes = [*pipes, {"source": "manual", "facility": "ford_bridge", **_between(sections, -6)}]
    # 저장 횡단 면적 = 서버가 상세를 낼 때 셈(파일엔 없음) — 정본 대신 정한 값
    for i, section in enumerate(sections):
        section["design"] = {
            "ground_type": "soil",
            "design_elevation_m": section["center_z"],
            "cut_area_m2": 4.0 + (i % 5),
            "fill_area_m2": 1.5 + (i % 3),
        }
    job = {
        "line": _csv(ROUTE / "planned_route.csv"),
        "nodes": _csv(ROUTE / "planned_route_nodes.csv"),
        "curves": [
            {k: c[k] for k in ("start", "apex", "end", "radius_m")} for c in curves["curves"]
        ],
        "stationSnap": bool(curves.get("station_snap")),
        "minRadius": curves["min_radius_m"],
        "interval": 20.0,
        "types": [
            {"type_id": t.type_id, "group": t.group, "placement": t.placement}
            for t in structure_type_map().values()
        ],
        "pipes": pipes,
        "standard": STANDARD_CROSS_SECTION,
        "conversion": ONES,
        # 구조물 하나 — 첫 관 자리 횡단배수(같은 측점을 덮음)
        "structures": [
            {
                "structure_id": "s-349o",
                "type_id": "cross_drain_exposed",
                "chainage_m": pipes[0]["chainage_m"],
                "x": pipes[0]["x"],
                "y": pipes[0]["y"],
            }
        ],
        # 열기 전에 고친 카드 선택 — 끌면 밀리는 측점(뒤쪽) · 단면 유형은 노선이 바뀌면 안 넘기는 칸
        "touched": {
            str(sections[-3]["chainage_m"]): {
                "ground_type": "blasting_rock",
                "section_mode": "right_cut",
            }
        },
        "steps": 5,
        "detail": {
            "longitudinal": _json(SECTION / "longitudinal/longitudinal.json"),
            "cross_sections": sections,
        },
    }
    with tempfile.TemporaryDirectory() as folder:
        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(GRID))
        proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
            ["node", str(HERE / "helper_349_0s_draw_work.cjs"), str(path), str(grid)],
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=300,
        )
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_걸음마다_종단도_한두_번(r):
    for step in r["steps"]:
        assert [x["outcome"] for x in step["runs"]][-1] == "done", step["runs"]
        assert 1 <= step["draws"] <= 2, step


def test_그리기_몫_기록(r):
    done = [x for step in r["steps"] for x in step["runs"] if x["outcome"] == "done"]
    for x in done:
        assert {"가게", "새 상세 · 종단도", "종단도", "쉼"} <= set(x["work"]), x["work"]
        assert x["totalMs"] >= x["stagesMs"] * 0.99


def test_되돌리기_그대로(r):
    u = r["undo"]
    assert [x["why"] for x in u["runs"]] == ["saved-diff"], u["runs"]  # 세월교 하나 더(0o 꼴)
    assert 1 <= u["draws"] <= 2, u
    assert r["refresh"] == 0