eomsangdonandClaude Opus 5.5
51cebefd40
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Se7ytqbohmBQnfYpwYFnYf
277 lines
12 KiB
Python
277 lines
12 KiB
Python
"""횡단 지반선 생성 거울 — 파이썬 ↔ TS(`B05_Profile/logic/B05_Profile_Sections_Ground.ts`).
|
|
|
|
PLAN 344-1 — 노선을 고치는 동안 브라우저가 지은 지반선이 서버 정본과 같아야 함.
|
|
짝: `generate_sections` 지반 부분 ↔ `sectionsGround` · `ground_profile` + 측점 `np.interp` ↔
|
|
`groundProfile` · `groundAt` · `extend_unclosed` ↔ `extendUnclosed`. 저장된 종 · 횡단 파일과도 대조.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
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_Engine_Grade import ground_profile # noqa: E402
|
|
from B06_Section import B06_Section_Engine_SampleExtend as extend_py # noqa: E402
|
|
from common_util.common_util_surface_sampler import DtmGridSampler, dtm_grid_bytes # noqa: E402
|
|
from config import config_system # noqa: E402
|
|
|
|
TS_FILE = ROOT / "B05_Profile" / "logic" / "B05_Profile_Sections_Ground.ts"
|
|
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
|
|
PROJECT = ROOT / "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a"
|
|
REAL_GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz"
|
|
REAL_ROUTE = PROJECT / "B05_Profile/route/planned_route_initial.csv"
|
|
_TOL = 1.5e-6 # 두 쪽 다 소수 6자리 반올림
|
|
#: 저장값 대조 — 저장된 x · y 가 6자리로 잘려 그 자리 표고가 기울기만큼(수 μm) 움직임.
|
|
_STORED_TOL = 1e-5
|
|
|
|
#: 덧붙임 시험용 설계 — 쪽마다 닫히는 반폭(우 27 · 좌 31 m)에 닿기 전엔 끝이 1 m 떠 있음.
|
|
_COMPUTE_PY_RADII = {"right": 27.0, "left": 31.0}
|
|
|
|
_RUNNER = """
|
|
import { readFileSync, writeFileSync } from "node:fs";
|
|
import { parseDtmGrid, heightAt } from "./common_util/common_util_surface_sampler.js";
|
|
import {
|
|
sectionsGround, groundProfile, groundAt, extendUnclosed,
|
|
SECTION_CROSS_HALF_WIDTH_M, SECTION_CROSS_SAMPLE_INTERVAL_M, SECTION_LONG_SAMPLE_INTERVAL_M,
|
|
EXTEND_STEP_M,
|
|
} from "./B05_Profile/logic/B05_Profile_Sections_Ground.js";
|
|
const bytes = readFileSync(process.argv[2]);
|
|
const grid = parseDtmGrid(bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.length));
|
|
const job = JSON.parse(readFileSync(process.argv[3], "utf8"));
|
|
const compute = (samples) => {
|
|
const ground = samples.filter((s) => s.valid !== false && s.elevation_m != null)
|
|
.sort((a, b) => a.offset_m - b.offset_m);
|
|
const right = ground[0], left = ground[ground.length - 1];
|
|
const rightOpen = Math.abs(right.offset_m) < job.radii.right;
|
|
const leftOpen = Math.abs(left.offset_m) < job.radii.left;
|
|
return {
|
|
slope_unclosed: rightOpen || leftOpen,
|
|
design_line: [
|
|
{ offset_m: right.offset_m, elevation_m: right.elevation_m + (rightOpen ? 1 : 0) },
|
|
{ offset_m: left.offset_m, elevation_m: left.elevation_m + (leftOpen ? 1 : 0) },
|
|
],
|
|
};
|
|
};
|
|
const ground = sectionsGround(job.line, job.stations, grid);
|
|
const profile = groundProfile(ground.samples);
|
|
const out = {
|
|
constants: [SECTION_CROSS_HALF_WIDTH_M, SECTION_CROSS_SAMPLE_INTERVAL_M,
|
|
SECTION_LONG_SAMPLE_INTERVAL_M, EXTEND_STEP_M],
|
|
ground,
|
|
stationGround: job.stations.map((c) => groundAt(profile, c)),
|
|
extended: ground.stations.map((s) =>
|
|
extendUnclosed(s.samples, [s.center_x, s.center_y], s.left_xy, compute, grid)),
|
|
stored: job.stored.map(([x, y]) => {
|
|
const z = heightAt(grid, x, y);
|
|
return Number.isNaN(z) ? null : z;
|
|
}),
|
|
};
|
|
writeFileSync(process.argv[4], JSON.stringify(out));
|
|
"""
|
|
|
|
|
|
def _compute_py(samples: list[dict]) -> dict:
|
|
ground = sorted(
|
|
(s for s in samples if s.get("valid") is not False and s.get("elevation_m") is not None),
|
|
key=lambda s: s["offset_m"],
|
|
)
|
|
right, left = ground[0], ground[-1]
|
|
right_open = abs(right["offset_m"]) < _COMPUTE_PY_RADII["right"]
|
|
left_open = abs(left["offset_m"]) < _COMPUTE_PY_RADII["left"]
|
|
return {
|
|
"slope_unclosed": right_open or left_open,
|
|
"design_line": [
|
|
{"offset_m": right["offset_m"], "elevation_m": right["elevation_m"] + right_open},
|
|
{"offset_m": left["offset_m"], "elevation_m": left["elevation_m"] + left_open},
|
|
],
|
|
}
|
|
|
|
|
|
def _run_ts(tmp_path: Path, npz: Path, job: dict) -> dict:
|
|
out = tmp_path / "js"
|
|
subprocess.run( # noqa: S603 — 고정 실행 파일
|
|
[
|
|
"node",
|
|
str(TSC),
|
|
str(TS_FILE),
|
|
"--outDir",
|
|
str(out),
|
|
"--rootDir",
|
|
str(ROOT),
|
|
"--module",
|
|
"esnext",
|
|
"--target",
|
|
"es2022",
|
|
"--moduleResolution",
|
|
"bundler",
|
|
"--ignoreConfig",
|
|
],
|
|
cwd=str(ROOT),
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
for emitted in out.rglob("*.js"):
|
|
emitted.write_text(
|
|
re.sub(r'(from\s+"\.\.?/[^"]+)"', r'\1.js"', emitted.read_text(encoding="utf-8")),
|
|
encoding="utf-8",
|
|
)
|
|
(out / "runner.mjs").write_text(_RUNNER, encoding="utf-8")
|
|
grid = tmp_path / "grid.bin"
|
|
grid.write_bytes(dtm_grid_bytes(npz))
|
|
payload = tmp_path / "job.json"
|
|
result = tmp_path / "out.json"
|
|
payload.write_text(json.dumps(job), encoding="utf-8")
|
|
subprocess.run( # noqa: S603
|
|
["node", str(out / "runner.mjs"), str(grid), str(payload), str(result)],
|
|
cwd=str(ROOT),
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
return json.loads(result.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _same(a, b, where: str, problems: list[str]) -> None:
|
|
if a is None or b is None:
|
|
if a is not b:
|
|
problems.append(f"{where}: {a!r} vs {b!r}")
|
|
elif abs(float(a) - float(b)) > _TOL:
|
|
problems.append(f"{where}: {a!r} vs {b!r}")
|
|
|
|
|
|
def _check(npz: Path, line: list, tmp_path: Path, stored: list | None = None) -> dict:
|
|
sampler = DtmGridSampler.from_npz(npz)
|
|
expected = core.generate_sections(np.asarray(line, dtype=np.float64), sampler)
|
|
stations = [s["chainage_m"] for s in expected["longitudinal"]["stations"]]
|
|
job = {"line": line, "stations": stations, "radii": _COMPUTE_PY_RADII, "stored": stored or []}
|
|
actual = _run_ts(tmp_path, npz, job)
|
|
problems: list[str] = []
|
|
ground = actual["ground"]
|
|
_same(ground["length_m"], expected["longitudinal"]["length_m"], "길이", problems)
|
|
long_py = expected["longitudinal"]["samples"]
|
|
assert len(ground["samples"]) == len(long_py)
|
|
for i, (ts, py) in enumerate(zip(ground["samples"], long_py, strict=True)):
|
|
assert ts["valid"] == py["valid"], f"종단 {i} 유효 어긋남"
|
|
for key in ("chainage_m", "x", "y", "z", "elevation_m"):
|
|
_same(ts[key], py[key], f"종단 {i} {key}", problems)
|
|
cross_py = expected["cross_sections"]
|
|
assert len(ground["stations"]) == len(cross_py)
|
|
for ts, py in zip(ground["stations"], cross_py, strict=True):
|
|
where = f"측점 {py['chainage_m']}"
|
|
assert ts["uphill_side"] == py["uphill_side"], f"{where} 상단측"
|
|
for key in ("chainage_m", "center_x", "center_y", "center_z"):
|
|
_same(ts[key], py[key], f"{where} {key}", problems)
|
|
for key in ("left_xy", "tangent_xy"):
|
|
for axis in (0, 1):
|
|
_same(ts[key][axis], py["frame"][key][axis], f"{where} {key}", problems)
|
|
assert len(ts["samples"]) == len(py["samples"])
|
|
for a, b in zip(ts["samples"], py["samples"], strict=True):
|
|
assert a["valid"] == b["valid"], f"{where} offset {b['offset_m']} 유효 어긋남"
|
|
for key in ("offset_m", "x", "y", "z"):
|
|
_same(a[key], b[key], f"{where} offset {b['offset_m']} {key}", problems)
|
|
|
|
profile_c, profile_z = ground_profile(expected["longitudinal"])
|
|
for c, z in zip(stations, actual["stationGround"], strict=True):
|
|
_same(z, float(np.interp(c, profile_c, profile_z)), f"측점 지반고 {c}", problems)
|
|
|
|
extended_count = 0
|
|
for ts, py in zip(actual["extended"], cross_py, strict=True):
|
|
want = extend_py.extend_unclosed(py["samples"], py["frame"], _compute_py, sampler)
|
|
if want is None:
|
|
assert ts is None
|
|
continue
|
|
extended_count += 1
|
|
assert ts["info"] == want[1], f"측점 {py['chainage_m']} 넓힘 {ts['info']} vs {want[1]}"
|
|
assert len(ts["samples"]) == len(want[0])
|
|
for a, b in zip(ts["samples"], want[0], strict=True):
|
|
for key in ("offset_m", "x", "y", "z"):
|
|
_same(a[key], b[key], f"넓힘 {py['chainage_m']} {b['offset_m']} {key}", problems)
|
|
assert not problems, "짝이 갈림:\n" + "\n".join(problems[:40])
|
|
return {"actual": actual, "extended": extended_count, "stations": len(stations)}
|
|
|
|
|
|
@pytest.fixture
|
|
def holed_npz(tmp_path: Path) -> Path:
|
|
"""칸 폭 고르지 않음 · 빈칸 띠 · 격자 끝 — 노선이 빈칸 · 끝을 지나감."""
|
|
rng = np.random.default_rng(3441)
|
|
x = (np.cumsum(rng.uniform(0.95, 1.1, 160)) + 1000.0).astype(np.float32)
|
|
y = (np.cumsum(rng.uniform(0.95, 1.1, 90)) + 2000.0).astype(np.float32)
|
|
xx, yy = np.meshgrid(x.astype(np.float64), y.astype(np.float64))
|
|
z = (500 + 0.08 * (xx - 1000) + 0.3 * (yy - 2000) + rng.normal(0, 0.2, xx.shape)).astype(
|
|
np.float32
|
|
)
|
|
mask = np.ones(z.shape, dtype=bool)
|
|
mask[40:46, 60:70] = False
|
|
path = tmp_path / "dtm_holed.npz"
|
|
np.savez(path, x=x, y=y, z=z, valid_mask=mask)
|
|
return path
|
|
|
|
|
|
@pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치")
|
|
def test_가짜_지형_빈칸_끝_넓힘이_서버와_같다(tmp_path: Path, holed_npz: Path) -> None:
|
|
arc = [[1080 + 30 * np.cos(a), 2040 + 30 * np.sin(a)] for a in np.linspace(-1.2, 0.9, 25)]
|
|
line = [[1003.0, 2010.0], [1040.0, 2016.0], *arc, [1150.0, 2083.0]]
|
|
result = _check(holed_npz, [[float(a), float(b)] for a, b in line], tmp_path)
|
|
assert result["actual"]["constants"] == [
|
|
config_system.SECTION_CROSS_HALF_WIDTH_M,
|
|
config_system.SECTION_CROSS_SAMPLE_INTERVAL_M,
|
|
config_system.SECTION_LONG_SAMPLE_INTERVAL_M,
|
|
extend_py.EXTEND_STEP_M,
|
|
]
|
|
assert result["extended"] >= 3
|
|
infos = [e["info"] for e in result["actual"]["extended"] if e]
|
|
assert any(info["surface_end"] for info in infos), "격자 끝 멈춤 사례 없음"
|
|
assert any(not info["surface_end"] for info in infos), "끝까지 닫힌 사례 없음"
|
|
|
|
|
|
def _stored_points() -> list[list[float]]:
|
|
longitudinal = json.loads(
|
|
(PROJECT / "B06_Section/longitudinal/longitudinal.json").read_text(encoding="utf-8")
|
|
)
|
|
points = [[s["x"], s["y"], s["z"]] for s in longitudinal["samples"]]
|
|
for path in sorted((PROJECT / "B06_Section/cross_sections").glob("cross_*.json")):
|
|
section = json.loads(path.read_text(encoding="utf-8"))
|
|
points += [[s["x"], s["y"], s.get("z")] for s in section.get("samples") or []]
|
|
return points
|
|
|
|
|
|
@pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치")
|
|
@pytest.mark.skipif(not (REAL_GRID.is_file() and REAL_ROUTE.is_file()), reason="검증_sub8 없음")
|
|
def test_검증_프로젝트_노선과_저장된_종횡단이_같다(tmp_path: Path) -> None:
|
|
with REAL_ROUTE.open(encoding="utf-8") as handle:
|
|
line = [[float(row["x"]), float(row["y"])] for row in csv.DictReader(handle)]
|
|
stored = _stored_points()
|
|
result = _check(REAL_GRID, line, tmp_path, [[x, y] for x, y, _z in stored])
|
|
assert result["stations"] > 50
|
|
# 저장된 종단 · 횡단(넓힌 표본 포함)의 자리에서 브라우저 표본 = 저장값
|
|
wrong = [
|
|
(x, y, z, got)
|
|
for (x, y, z), got in zip(stored, result["actual"]["stored"], strict=True)
|
|
if (z is None) != (got is None) or (z is not None and abs(z - got) > _STORED_TOL)
|
|
]
|
|
assert len(stored) > 5000
|
|
assert not wrong, f"저장값과 다름 {len(wrong)}점 — 처음 {wrong[:3]}"
|
|
|
|
|
|
def test_짝_머리글이_서로를_가리킨다() -> None:
|
|
text = TS_FILE.read_text(encoding="utf-8")
|
|
for name in (
|
|
"B05_Profile_Engine_Sections_Core.py",
|
|
"B05_Profile_Engine_Grade.py",
|
|
"B06_Section_Engine_SampleExtend.py",
|
|
):
|
|
assert name in text
|
|
assert "B05_Profile_Sections_Ground.ts" in (core.__doc__ or "")
|
|
assert "B05_Profile_Sections_Ground.ts" in (extend_py.__doc__ or "")
|