Aislo/resources/tester/helper_472_1_paths.py

"""472-1 기준값 — 고정 자료(`fixtures/golden_472`) 하나로 길 넷의 값을 뜸(시험 · 값 뜨기 함께 씀).

  stored        GET 상세(정본 그대로) → 누가토량(서버 Node 묶음 `computeMassHaul`)
  server_save   [저장] 서버 재계산 — 정본 상세 → 종단 경고 다시 셈(`refresh_alignment` · 472 덧) →
                `enforce_design_lines` → `_enforce_stored_designs` → server-calc
  server_route  노선 저장 체인(같은 노선 · DB 없이) — `run_section_generation`(고정 geojson · npz)
                → `_carry_profile` 같은 노선 갈래(옛 base_pvi · 편집) → `_apply_alignment_edits` →
                측점 열쇠 이월(`carry_cross_inputs`) → `compute_default_designs` → 상세 읽기 →
                server_save 와 같은 끝
  browser       진짜 단계 함수 묶음(`helper_472_1_browser.mjs` · vite ssr) — 격자 = 같은 npz 의
                `dtm_grid_bytes`

값 꼴(`values_of`) — 측점마다 열쇠 · 누가거리(0.01) · x · y(0.001) · 계획고(0.001) · 면적(0.0001) ·
측구 · 층따기 밑수 · 단면 모양 · 누가토량(0.01) + 끝값 · 관 길이 · 종단 경고.
DB · 서버 · 사용자 storage 안 씀(고정 자료 사본만).
"""

from __future__ import annotations

import copy
import json
import shutil
import subprocess
from pathlib import Path
from typing import Any

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
FIXTURE = HERE / "fixtures" / "golden_472"
VALUES = FIXTURE / "golden_472_values.json"
ROUTE = "B05_Profile/route/route_main.geojson"
LONG = "B06_Section/longitudinal/longitudinal.json"
NPZ = "B04_PreProcess/models/dtm_classification_smooth.npz"
SURFACE = ("classification", "dtm", True)


def read(name: str) -> Any:
    return json.loads((FIXTURE / name).read_text(encoding="utf-8"))


def stage_project(root: Path) -> Path:
    """고정 storage 사본을 임시 프로젝트 뿌리로 — 체인이 파일을 쓰므로 늘 새 사본."""
    shutil.copytree(FIXTURE / "project", root, dirs_exist_ok=True)
    return root


# ── 값 꼴 ────────────────────────────────────────────────────────────────────


def _r(value: Any, digits: int) -> float | None:
    return round(float(value), digits) if isinstance(value, (int, float)) else None


def values_of(
    sections: list[dict[str, Any]],
    keys: list[str | None],
    mass_haul: dict[str, Any] | None,
    warnings: list[str],
    violations: list[dict[str, Any]],
) -> dict[str, Any]:
    """상세 측점(누가거리 차례) · 열쇠 · 누가토량 → 비교 값. 누가토량은 같은 측점 id 로 짝."""
    rows = sorted(zip(sections, keys, strict=True), key=lambda pair: float(pair[0]["chainage_m"]))
    points = {p["station_id"]: p for p in (mass_haul or {}).get("points") or []}
    stations = []
    for section, key in rows:
        design = section.get("design") or {}
        point = points.get(section.get("station_id")) or {}
        stations.append(
            {
                "key": key,
                "chainage_m": _r(section["chainage_m"], 2),
                "x": _r(section.get("center_x"), 3),
                "y": _r(section.get("center_y"), 3),
                "plan_z": _r(design.get("design_elevation_m"), 3),
                "mode": design.get("section_mode"),
                "cut_m2": _r(design.get("cut_area_m2"), 4),
                "fill_m2": _r(design.get("fill_area_m2"), 4),
                "cut_soil_m2": _r(design.get("cut_soil_area_m2"), 4),
                "cut_rock_m2": _r(design.get("cut_rock_area_m2"), 4),
                "ditch_m2": _r(design.get("ditch_area_m2"), 4),
                "bench_m": _r(design.get("bench_cut_length_m"), 4),
                "pipe_length_m": _r(design.get("pipe_length_m"), 2),
                "cumulative_m3": _r(point.get("cumulative_volume_m3"), 2),
            }
        )
    return {
        "station_count": len(stations),
        "stations": stations,
        "final_cumulative_m3": _r((mass_haul or {}).get("final_cumulative_m3"), 2),
        "pipe_lengths": {
            s["key"]: s["pipe_length_m"] for s in stations if s["pipe_length_m"] is not None
        },
        "warnings": sorted(warnings),
        "violations": len(violations),
    }


def _keys(longitudinal: dict[str, Any], sections: list[dict[str, Any]], root: Path) -> list:
    """서버 측점 열쇠(`station_keys` · 노선 저장 이월이 쓰는 한 벌) — 상세 측점 차례."""
    from B05_Profile.B05_Profile_Route_Save_Parts import station_keys
    from B05_Profile.B05_Profile_Structures_Repository import load_structures
    from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file

    pipes = read_pipe_points_file(pipe_points_path_in(root))
    _, structures = load_structures(str(root))
    interval = float(
        (longitudinal.get("profile_alignment") or {}).get("policy", {}).get("station_interval_m")
        or 20.0
    )
    found = station_keys(longitudinal.get("stations") or [], pipes, structures, interval)
    by_m = {round(at, 3): key for key, at in found.items()}
    return [by_m.get(round(float(s["chainage_m"]), 3)) for s in sections]


# ── 서버 끝(server_save · server_route 공통) ──────────────────────────────────


def _context() -> dict[str, Any]:
    """유토곡선 문맥 — 서버 `_mass_haul_context` + 고정 자료 계수 · 장비 경계(화면이 받는 값)."""
    from B06_Section.B06_Section_Server_Calc_Prebuild import _mass_haul_context

    context = read("context.json")
    limits = [(row["key"], row["max_distance_m"]) for row in context["haul_equipment_limits"]]
    return _mass_haul_context(None, context["earthwork_conversion"], limits)


def _save_tail(detail: dict[str, Any], root: Path) -> tuple[list, dict | None]:
    """[저장] 재계산(`_recompute`) — 종단 경고 · 설계선 · 넓힘 · 보정 → Node 면적 · 누가토량."""
    from B06_Section.B06_Section_Router_Design import enforce_design_lines
    from B06_Section.B06_Section_Server_Calc_Prebuild import (
        _AREA_KEYS,
        _NPM_SCRIPT,
        BUNDLE,
        _enforce_stored_designs,
        refresh_alignment,
    )
    from common_util.common_util_node_bundle import run_bundle_json
    from common_util.common_util_surface_sampler import build_surface_sampler

    longitudinal = detail["longitudinal"]
    sections = detail["cross_sections"]
    refresh_alignment(longitudinal)
    enforce_design_lines(longitudinal, sections, root, None)
    unclosed = any((s.get("design") or {}).get("slope_unclosed") for s in sections)
    sampler = (
        build_surface_sampler(root / "B04_PreProcess" / "models", *SURFACE) if unclosed else None
    )
    _enforce_stored_designs(longitudinal, sections, root, None, sampler)
    detail = json.loads(
        json.dumps(detail, default=str)
    )  # 응답 직렬화와 같은 꼴(`model_dump(json)`)
    output = run_bundle_json(BUNDLE, _NPM_SCRIPT, {"detail": detail, "context": _context()})
    if not isinstance(output, dict):
        raise RuntimeError("server-calc 묶음 실패")
    sections = detail["cross_sections"]
    for row in output.get("areas") or []:
        for section in sections:
            if abs(float(section["chainage_m"]) - float(row["chainage_m"])) < 1e-6:
                design = section.setdefault("design", {})
                design.update(
                    {k: row[k] for k in _AREA_KEYS if isinstance(row.get(k), (int, float))}
                )
    return sections, output.get("mass_haul")


def server_save(root: Path) -> dict[str, Any]:
    detail = copy.deepcopy(read("detail.json"))
    keys = _keys(detail["longitudinal"], detail["cross_sections"], root)
    sections, haul = _save_tail(detail, root)
    alignment = detail["longitudinal"].get("profile_alignment") or {}
    return values_of(
        sections, keys, haul, alignment.get("warnings") or [], alignment.get("violations") or []
    )


def server_route(root: Path) -> dict[str, Any]:
    """노선 저장 체인 — 같은 노선(바뀐 구간 없음) · DB 행 대신 고정 자료(요약 옵션 · 정본 설계)."""
    from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions
    from B05_Profile.B05_Profile_Engine_Grade_Carry import carry_policy
    from B05_Profile.B05_Profile_Engine_Sections import run_section_generation
    from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
    from B05_Profile.B05_Profile_Route_Save_Parts import carry_cross_inputs, station_keys
    from B05_Profile.B05_Profile_Router import _apply_alignment_edits
    from B05_Profile.B05_Profile_Structures_Repository import load_structures
    from B06_Section.B06_Section_Engine_Design import ROUTE_CARRY_KEYS
    from B06_Section.B06_Section_Router import _read_section_detail
    from B06_Section.B06_Section_Router_Design import (
        CARRY_INPUT_KEYS,
        STRUCTURE_ADJUST_KEYS,
        attach_default_designs,
        compute_default_designs,
    )
    from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
    from common_util.common_util_json import atomic_write_json

    saved = read("detail.json")
    summary = read("summary.json")["longitudinal"]["data"]
    options = {k: v for k, v in summary["options"].items() if k != "extra_stations"}
    run_section_generation(
        root,
        ROUTE,
        *SURFACE,
        options=SectionGenerationOptions(**options),
        grade_options=GradeDesignOptions(**summary["grade_options"]),
        crs=f"EPSG:{read('context.json')['crs_epsg']}",
    )
    # 계획고 옮김(`_carry_profile` · 바뀐 구간 없음 = 옛 base_pvi · 편집 통째)
    old = saved["longitudinal"]["profile_alignment"]
    longitudinal = json.loads((root / LONG).read_text(encoding="utf-8"))
    alignment = longitudinal.get("profile_alignment") or {}
    line = json.loads((root / ROUTE).read_text(encoding="utf-8"))["geometry"]["coordinates"]
    xy = [[float(p[0]), float(p[1])] for p in line]
    alignment["policy"] = carry_policy(xy, xy, old, alignment)
    alignment["base_pvi"] = old["base_pvi"]
    longitudinal["profile_alignment"] = alignment
    atomic_write_json(root / LONG, longitudinal)
    alignment, _summary, _dropped = _apply_alignment_edits(root, LONG, old.get("edits") or {})
    # 횡단 이월(`_carry_cross`) — 측점 열쇠로 정본 사용자 입력 → 새 측점 · 전 측점 새 계획고로 셈
    new_long = json.loads((root / LONG).read_text(encoding="utf-8"))
    pipes = read_pipe_points_file(pipe_points_path_in(root))
    _, structures = load_structures(str(root))
    interval = float(options["station_interval_m"])
    old_keys = station_keys(saved["longitudinal"]["stations"], pipes, structures, interval)
    new_keys = station_keys(new_long.get("stations") or [], pipes, structures, interval)
    designs = {
        round(float(s["chainage_m"]), 3): s["design"]
        for s in saved["cross_sections"]
        if isinstance(s.get("design"), dict)
    }
    inputs, adjusts, _ = carry_cross_inputs(
        old_keys, new_keys, designs, (*CARRY_INPUT_KEYS, *ROUTE_CARRY_KEYS), STRUCTURE_ADJUST_KEYS
    )
    chainages = [float(s["chainage_m"]) for s in new_long.get("stations") or []]
    entries = compute_default_designs(root, LONG, chainages, None, inputs)
    by_m = {round(at, 3): {**design, **adjusts.get(round(at, 3), {})} for at, design in entries}
    # [확정] 상세 읽기(`get_section_detail`) — 행 설계(0.01 m) 얹기 · 없는 측점 기본값
    detail = _read_section_detail(
        root, LONG, ROUTE
    )  # 관 = 이 노선에 얹은 것(get_section_detail 과 같음)
    for section in detail["cross_sections"]:
        hit = next(
            (d for at, d in by_m.items() if abs(at - float(section["chainage_m"])) < 0.01), None
        )
        if hit is not None:
            section["design"] = hit
    attach_default_designs(detail["longitudinal"], detail["cross_sections"], root, None)
    keys = _keys(detail["longitudinal"], detail["cross_sections"], root)
    sections, haul = _save_tail(detail, root)
    return values_of(
        sections, keys, haul, alignment.get("warnings") or [], alignment.get("violations") or []
    )


def stored(root: Path) -> dict[str, Any]:
    """정본 그대로 — GET 상세 면적 · 서버 Node 묶음과 같은 누가토량(화면 배지 값)."""
    from B06_Section.B06_Section_Server_Calc_Prebuild import _NPM_SCRIPT, BUNDLE
    from common_util.common_util_node_bundle import run_bundle_json

    detail = read("detail.json")
    alignment = detail["longitudinal"].get("profile_alignment") or {}
    keys = _keys(detail["longitudinal"], detail["cross_sections"], root)
    output = run_bundle_json(BUNDLE, _NPM_SCRIPT, {"detail": detail, "context": _context()})
    haul = (output or {}).get("mass_haul")
    return values_of(
        detail["cross_sections"],
        keys,
        haul,
        alignment.get("warnings") or [],
        alignment.get("violations") or [],
    )


# ── 브라우저 길 ──────────────────────────────────────────────────────────────


def browser(work: Path) -> dict[str, Any]:
    """vite ssr 로 진짜 모듈을 묶어 헬퍼를 돌림(`helper_272_detail_memory` 꼴) · 격자 = 고정 npz."""
    from common_util.common_util_surface_sampler import dtm_grid_bytes

    grid = work / "grid.bin"
    grid.write_bytes(dtm_grid_bytes(FIXTURE / "project" / NPZ))
    build = subprocess.run(  # noqa: S603 — 고정 실행 파일
        [
            "node",
            str(ROOT / "config" / "node_modules" / "vite" / "bin" / "vite.js"),
            "build",
            "--configLoader",
            "runner",
            "--ssr",
            "../resources/tester/helper_472_1_browser_entry.ts",
            "--outDir",
            str(work / "b"),
            "--logLevel",
            "error",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=300,
    )
    if build.returncode:
        raise RuntimeError(build.stderr)
    bundle = work / "b" / "bundle.mjs"  # 조각(assets/)과 같은 폴더
    shutil.copy(work / "b" / "helper_472_1_browser_entry.js", bundle)
    proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
        ["node", str(HERE / "helper_472_1_browser.mjs"), str(bundle), str(FIXTURE), str(grid)],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=120,
    )
    if proc.returncode:
        raise RuntimeError(proc.stderr)
    out = json.loads(proc.stdout)
    return values_of(
        out["sections"], out["keys"], out["mass_haul"], out["warnings"], out["violations"]
    )


PATHS = ("stored", "server_save", "server_route", "browser")


def run_path(name: str, work: Path) -> dict[str, Any]:
    if name == "browser":
        return browser(work)
    return globals()[name](stage_project(work / "project"))