Merge remote-tracking branch 'origin/sub_laptop_8' into main_laptop_1

This commit is contained in:
2026-10-16 11:16:26 +09:00
13 changed files with 1216 additions and 5 deletions
+4
View File
@@ -11,6 +11,7 @@
import { API_BASE_URL } from "@config/config_frontend";
import { fetchConfirmedSurface } from "../B04_PreProcess/B04_PreProcess_Api_Fetch";
import { loadDesignTerrainGrid } from "../B05_Profile/B05_Profile_Api_TerrainGrid";
const DB_NAME = "aislo-asset-cache";
// 담아 둔 자료의 형식이나 내용이 바뀌면 이 번호를 올린다 — 올리면 기존 보관분을 통째로 버린다.
@@ -380,6 +381,9 @@ export async function preloadSurfaceAssets(
{ onProgress: (ratio) => report("등고선을 준비하는 중…", ratio) },
);
// 노선 편집 지반고를 브라우저가 읽는 설계 지형 격자(PLAN 344-1) — 못 받아도 멈추지 않음(null).
await loadDesignTerrainGrid(projectId);
// 배수유역도 배경(도엽 표시본·위성사진)도 함께 담는다 — 없으면 B05가 진입할 때마다 받는다.
// 이 자료가 없어도 화면은 뜨므로 실패해도 준비를 멈추지 않는다.
report("배경 지도를 준비하는 중…", null);
+12 -1
View File
@@ -17,6 +17,8 @@
import { API_BASE_URL } from "@config/config_frontend";
import { readState, writeState } from "../A00_Common/b_page_state";
import { heightAt } from "../common_util/common_util_surface_sampler";
import { loadDesignTerrainGrid } from "./B05_Profile_Api_TerrainGrid";
/** 편집 창 분석 지형 — las = 확정 지표면(라이다) · sheet = 수치지형도 서피스. */
export type AnalysisTerrain = "las" | "sheet";
@@ -181,11 +183,20 @@ export async function replanRoute(
/** 점 묶음의 **지반고**를 묻는다(계획서 0-9 ⑤·⑧).
*
* 새 계산이 아니라 확정된 지표면을 **읽기만** 하므로 편집 중에 불러도 된다 — 다만 끄는 동안
* 프레임마다 부르지는 않는다(찍는 순간에만). 지표면 밖은 `null` 로 온다. */
* 프레임마다 부르지는 않는다(찍는 순간에만). 지표면 밖은 `null` 로 온다.
* 설계 지형 격자를 받아 두었고 분석 지형이 그것과 같으면 **브라우저 표본**(서버와 같은 값 ·
* PLAN 344-1) — 아니면 서버에 물음. 자릿수는 서버 응답과 같게 mm. */
export async function fetchRouteElevations(
projectId: string,
points: Array<[number, number]>,
): Promise<Array<number | null>> {
const local = await loadDesignTerrainGrid(projectId);
if (local && analysisTerrain(projectId) === local.design) {
return points.map(([x, y]) => {
const z = heightAt(local.grid, x, y);
return Number.isNaN(z) ? null : Math.round(z * 1000) / 1000;
});
}
const payload = await requestJson<{ z: Array<number | null> }>(
`/projects/${projectId}/route/elevations`,
{ method: "POST", body: JSON.stringify({ points, terrain: analysisTerrain(projectId) }) },
@@ -0,0 +1,55 @@
/* =============================================================================
* B05_Profile_Api_TerrainGrid.ts
* 설계 지형 격자를 받아 둔다 — 노선 편집이 지반고를 브라우저에서 바로 읽게(PLAN 344-1).
*
* GET /projects/{id}/route/terrain-sources → 설계 지형 · 격자 받을 자리(`grid`)
* GET /projects/{id}/surface/models/{mid}/grid → DTM 격자 원시 이진(ETag · IndexedDB)
*
* 주소에 manifest 서명을 붙임 — 보관함은 옛 저장본을 먼저 내므로(`fetchCachedBytes`) 지형이
* 바뀐 뒤 옛 격자로 한 번 그리는 일을 막음. DTM 아닌 표현 · 못 받음 = null → 서버 표본 길.
* ========================================================================== */
import { API_BASE_URL } from "@config/config_frontend";
import { fetchCachedBytes } from "../A00_Common/b_asset_cache";
import { type DtmGrid, parseDtmGrid } from "../common_util/common_util_surface_sampler";
export interface DesignTerrainGrid {
/** 이 격자의 지형 — 편집 창 분석 지형이 이것과 같을 때만 씀. */
design: "las" | "sheet";
grid: DtmGrid;
}
interface TerrainSourcesPayload {
design?: "las" | "sheet";
grid?: { model_id: number; smooth: boolean; signature: string } | null;
}
const loaded = new Map<string, Promise<DesignTerrainGrid | null>>();
async function download(projectId: string): Promise<DesignTerrainGrid | null> {
const response = await fetch(`${API_BASE_URL}/projects/${projectId}/route/terrain-sources`, {
credentials: "include",
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const payload = (await response.json()) as TerrainSourcesPayload;
const info = payload.grid;
if (!info) return null;
const url =
`${API_BASE_URL}/projects/${projectId}/surface/models/${info.model_id}/grid` +
`?smooth=${info.smooth}&sig=${encodeURIComponent(info.signature)}`;
const grid = parseDtmGrid(await fetchCachedBytes(projectId, url));
return { design: payload.design === "sheet" ? "sheet" : "las", grid };
}
/** 설계 지형 격자 — 페이지 수명 동안 한 번만 받음 · 받다 실패하면 null(다음 부름에 다시 해 봄). */
export function loadDesignTerrainGrid(projectId: string): Promise<DesignTerrainGrid | null> {
let pending = loaded.get(projectId);
if (!pending) {
pending = download(projectId).catch(() => {
loaded.delete(projectId);
return null;
});
loaded.set(projectId, pending);
}
return pending;
}
@@ -9,6 +9,7 @@
⚠ 파이썬·TS 짝 — 측점 스냅 `_chainages` · `_design_curve_spans` · `snap_curve_stations` ↔
`B05_Profile_UI_RouteEdit_Frame.ts` `stationsAlong` · `curveSpans` · `snapCurveStations`. 거울 시험
`resources/tester/test_224_2_station_snap.py`.
⚠ 파이썬·TS 짝 — `generate_sections` 지반 부분 ↔ `B05_Profile/logic/B05_Profile_Sections_Ground.ts`.
"""
import math
+87 -3
View File
@@ -11,19 +11,22 @@
POST /api/projects/{id}/route/elevations → 점 묶음의 지반고
POST /api/projects/{id}/route/cross-preview → 고치던 노선의 한 측점 횡단 미리보기
GET /api/projects/{id}/route/terrain-sources → 고를 수 있는 분석 지형(라이다 · 수치지형도)
GET /api/projects/{id}/surface/models/{mid}/grid → DTM 격자 원시 이진(브라우저 표본기 · ETag)
분석 지형(`terrain`, PLAN 133-1) — las = 라이다 · sheet = 수치지형도 · None = 지금 설계 지형.
모달 분석만 고른 지형을 쓴다 — 설계 지형(확정)은 [확인] 때만 바뀜(133-2 · `/route/replan`).
"""
import asyncio
import json
import logging
from pathlib import Path
from uuid import UUID
import aiomysql
import numpy as np
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, Response
from pydantic import BaseModel, Field
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
@@ -32,6 +35,7 @@ from B05_Profile.B05_Profile_Engine_Sections_Core import (
generate_sections,
)
from B06_Section.B06_Section_Engine_Design import compute_cross_design, curve_widening_args
from common_util.common_util_http_cache import cached_file_response
from common_util.common_util_route_polyline import build_planned_polyline
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_surface_confirmation import (
@@ -41,7 +45,7 @@ from common_util.common_util_surface_confirmation import (
design_terrain,
get_surface_confirmation_params,
)
from common_util.common_util_surface_sampler import build_surface_sampler
from common_util.common_util_surface_sampler import build_surface_sampler, dtm_grid_file
from config.config_db import get_db_pool
logger = logging.getLogger(__name__)
@@ -262,6 +266,7 @@ async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse:
content={"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."},
)
params = await get_surface_confirmation_params(connection, str(project_id))
confirmed = await _confirmed_model(connection, project_id)
models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR
sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir)
return {
@@ -269,4 +274,83 @@ async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse:
"project_id": str(project_id),
"sources": sources,
"design": design_terrain(params), # 지금 설계 지형(PLAN 133-2)
"grid": await asyncio.to_thread(_design_grid, params, models_dir, confirmed),
}
async def _confirmed_model(
connection: aiomysql.Connection, project_id: UUID
) -> tuple[int, str, str] | None:
"""확정 지표면 모델 (id, 표현, 파일 경로) — 없으면 None."""
async with connection.cursor() as cursor:
await cursor.execute(
"SELECT id, model_type, model_file_path FROM surface_models"
" WHERE project_id = %s AND status = 'CONFIRMED' ORDER BY created_at DESC LIMIT 1",
(str(project_id),),
)
row = await cursor.fetchone()
return (int(row[0]), str(row[1]), str(row[2] or "")) if row else None
def _design_grid(
params: dict, models_dir: Path, confirmed: tuple[int, str, str] | None
) -> dict | None:
"""설계 지형 격자 받을 자리(PLAN 344-1) — 서버 표본기(`build_surface_sampler`)와 같은 파일만.
DTM 아닌 표현(TIN 등)은 None → 화면은 서버 표본 길 그대로. `signature` = manifest 구성 ·
평활 서명 — 화면이 주소에 붙여 지형이 바뀐 뒤 옛 격자를 먼저 내지 않음.
"""
source_filter = str(params.get("source_filter") or "")
if not confirmed or params.get("method") != "dtm" or confirmed[1] != "dtm":
return None
if Path(confirmed[2]).name != f"dtm_{source_filter}.npz":
return None
smooth = bool(params.get("smooth"))
try:
manifest = json.loads((models_dir / "manifest.json").read_text(encoding="utf-8"))
except (OSError, ValueError):
manifest = {}
dtm = (
manifest.get("source_filters", {}).get(source_filter, {}).get("methods", {}).get("dtm", {})
)
smoothing = (dtm.get("smooth") or {}).get("smoothing_signature") if smooth else None
return {
"model_id": confirmed[0],
"smooth": smooth,
"signature": f"{manifest.get('config_signature') or ''}.{smoothing or ''}",
}
@router.get("/{project_id}/surface/models/{model_id}/grid", response_model=None)
async def read_surface_grid(
request: Request, project_id: UUID, model_id: int, smooth: bool = False
) -> Response | JSONResponse:
"""DTM 격자 원시 이진(`dtm_grid_bytes` 꼴 · ETag) — 첫 요청 때 npz 곁에 `.bin` 저장본을 씀.
평활 고르기는 서버 표본기와 같음(평활 파일 없으면 원본). DTM 아닌 표현은 404.
"""
pool = get_db_pool()
async with pool.acquire() as connection:
stored = await get_project_storage_relative_path(connection, project_id)
async with connection.cursor() as cursor:
await cursor.execute(
"SELECT model_type, model_file_path FROM surface_models"
" WHERE id = %s AND project_id = %s",
(model_id, str(project_id)),
)
row = await cursor.fetchone()
if not stored or not row or row[0] != "dtm" or not row[1]:
return JSONResponse(
status_code=404,
content={"status": "error", "message": "DTM 격자가 없는 지표면 모델입니다."},
)
model_path = Path(resolve_stored_project_path(stored)) / str(row[1])
smooth_path = model_path.with_name(f"{model_path.stem}_smooth.npz")
npz_path = smooth_path if smooth and smooth_path.is_file() else model_path
if not npz_path.is_file():
return JSONResponse(
status_code=404,
content={"status": "error", "message": "DTM 격자 파일이 없습니다."},
)
grid_path = await asyncio.to_thread(dtm_grid_file, npz_path)
return cached_file_response(request, grid_path, "application/octet-stream")
@@ -0,0 +1,311 @@
/* =============================================================================
* B05_Profile_Sections_Ground.ts
* 노선 한 벌의 **지반선** — 종단 1 m 표본 · 측점 지반고 · 횡단 ±20 m 0.5 m · 미교차 덧붙임.
* 노선을 고치는 동안 브라우저가 지형 격자(`common_util_surface_sampler.ts`)로 바로 셈한다(PLAN 344-1).
*
* ⚠ 파이썬·TS 짝 — `B05_Profile_Engine_Sections_Core.py` `generate_sections` 의 지반 부분
* (종단 표본 · 측점 중심 · 횡단 표본 · `uphill_side`) ↔ `sectionsGround`.
* ⚠ 파이썬·TS 짝 — `B05_Profile_Engine_Grade.py` `ground_profile` + 측점 `np.interp` ↔
* `groundProfile` · `groundAt`.
* ⚠ 파이썬·TS 짝 — `B06_Section_Engine_SampleExtend.py` `extend_unclosed` ↔ `extendUnclosed`.
* 한쪽을 고치면 다른 쪽도 고치고 거울 시험 `resources/tester/test_344_1_sections_ground_mirror.py`.
*
* 측점 자리(규칙 · 스냅 · 구조물 측점 병합)는 부르는 쪽이 정해 넘김 — 틀은
* `B05_Profile_UI_RouteEdit_Frame.ts`(이미 짝). 자릿수는 서버 출력과 같게 소수 6자리.
* ========================================================================== */
import { type DtmGrid, heightAt } from "../../common_util/common_util_surface_sampler";
import type { CrossDesignEdge } from "../../common_util/common_util_cross_design";
import { stationFrameAt, stationsAlong } from "../B05_Profile_UI_RouteEdit_Frame";
type Vertex = [number, number];
/** 짝: `config_system_design.py` `SECTION_CROSS_HALF_WIDTH_M` · `_SAMPLE_INTERVAL_M` · `SECTION_LONG_…`. */
export const SECTION_CROSS_HALF_WIDTH_M = 20.0;
export const SECTION_CROSS_SAMPLE_INTERVAL_M = 0.5;
export const SECTION_LONG_SAMPLE_INTERVAL_M = 1.0;
/** 짝: `B06_Section_Engine_SampleExtend.EXTEND_STEP_M`. */
export const EXTEND_STEP_M = 5.0;
/** 짝: `B06_Section_Engine_Design._SLOPE_CLOSE_TOLERANCE_M`. */
const SLOPE_CLOSE_TOLERANCE_M = 0.01;
export interface GroundSample {
x: number;
y: number;
z: number | null;
elevation_m: number | null;
valid: boolean;
}
export interface LongGroundSample extends GroundSample {
chainage_m: number;
}
export interface CrossGroundPoint extends GroundSample {
offset_m: number;
}
export interface StationGround {
chainage_m: number;
center_x: number;
center_y: number;
center_z: number | null;
tangent_xy: Vertex;
/** 좌측(+offset) 단위 벡터. */
left_xy: Vertex;
/** 횡단 기준 등고가 높은 쪽 — 좌우 유효 표본 평균 비교 · 모르면 null. */
uphill_side: "left" | "right" | null;
samples: CrossGroundPoint[];
}
export interface SectionsGround {
length_m: number;
/** 종단 지반 표본(1 m · 끝점 포함). */
samples: LongGroundSample[];
stations: StationGround[];
}
export interface GroundOptions {
crossHalfWidthM?: number;
crossSampleIntervalM?: number;
longSampleIntervalM?: number;
}
/** 파이썬 `round(x, 6)` 자리. */
const round6 = (value: number): number => Math.round(value * 1e6) / 1e6;
const round9 = (value: number): number => Math.round(value * 1e9) / 1e9;
function sample(grid: DtmGrid, x: number, y: number): GroundSample {
const z = heightAt(grid, x, y);
const valid = !Number.isNaN(z);
const value = valid ? round6(z) : null;
return { x: round6(x), y: round6(y), z: value, elevation_m: value, valid };
}
/** `_clean_polyline` — 앞 점과 길이 0(≤ 1e-8)인 점을 버리고 누가거리. */
function cleanLine(line: ReadonlyArray<Vertex>): {
xs: number[];
ys: number[];
chainage: number[];
} {
const kept = line.filter(
(vertex, i) =>
i === 0 || Math.hypot(vertex[0] - line[i - 1][0], vertex[1] - line[i - 1][1]) > 1e-8,
);
const chainage = [0];
for (let i = 1; i < kept.length; i += 1) {
chainage.push(
chainage[i - 1] + Math.hypot(kept[i][0] - kept[i - 1][0], kept[i][1] - kept[i - 1][1]),
);
}
return { xs: kept.map((v) => v[0]), ys: kept.map((v) => v[1]), chainage };
}
/** 횡단 offset 열 — `np.arange(-hw, hw + si/2, si)` · hw 넘는 것 버림 · 0 없으면 끼움. */
function crossOffsets(halfWidth: number, interval: number): number[] {
const count = Math.ceil((halfWidth + interval * 0.5 + halfWidth) / interval);
const offsets: number[] = [];
for (let i = 0; i < count; i += 1) {
const value = -halfWidth + i * interval;
if (value <= halfWidth + 1e-9) offsets.push(value);
}
if (!offsets.some((value) => Math.abs(value) <= 1e-9)) {
offsets.push(0);
offsets.sort((a, b) => a - b);
}
return offsets;
}
/** 노선 지반선 — 종단 표본 · 측점마다 중심 · 좌측 축 · 횡단 표본. `stationsM` = 측점 누가거리. */
export function sectionsGround(
line: ReadonlyArray<Vertex>,
stationsM: ReadonlyArray<number>,
grid: DtmGrid,
options: GroundOptions = {},
): SectionsGround {
const halfWidth = options.crossHalfWidthM ?? SECTION_CROSS_HALF_WIDTH_M;
const interval = options.crossSampleIntervalM ?? SECTION_CROSS_SAMPLE_INTERVAL_M;
const longInterval = options.longSampleIntervalM ?? SECTION_LONG_SAMPLE_INTERVAL_M;
const probe = Math.max(longInterval, 0.5);
// 자리 = `_interpolate_xy`(축마다 `np.interp`) · 좌측 축만 측점 틀 짝(`stationFrameAt`).
const { xs, ys, chainage: along } = cleanLine(line);
const at = (c: number): Vertex => [npInterp(c, along, xs), npInterp(c, along, ys)];
const samples: LongGroundSample[] = stationsAlong(line, longInterval, true).map((chainage) => {
const [x, y] = at(chainage);
return { chainage_m: round6(chainage), ...sample(grid, x, y) };
});
const offsets = crossOffsets(halfWidth, interval);
let center = 0;
offsets.forEach((value, index) => {
if (Math.abs(value) < Math.abs(offsets[center])) center = index;
});
const stations: StationGround[] = [];
for (const chainage of stationsM) {
const frame = stationFrameAt(line, chainage, probe);
if (!frame) continue;
const [ox, oy] = at(chainage);
const [lx, ly] = frame.left;
const cross: CrossGroundPoint[] = [];
const raw: number[] = [];
for (const offset of offsets) {
const x = ox + lx * offset;
const y = oy + ly * offset;
raw.push(heightAt(grid, x, y));
cross.push({ offset_m: round6(offset), ...sample(grid, x, y) });
}
const mean = (pick: (offset: number) => boolean): number | null => {
const values = raw.filter((z, i) => pick(offsets[i]) && !Number.isNaN(z));
return values.length ? values.reduce((a, b) => a + b, 0) / values.length : null;
};
const leftMean = mean((offset) => offset > 1e-9);
const rightMean = mean((offset) => offset < -1e-9);
let uphill: StationGround["uphill_side"] = null;
if (leftMean !== null && rightMean !== null && Math.abs(leftMean - rightMean) > 1e-6) {
uphill = leftMean > rightMean ? "left" : "right";
}
const centerZ = raw[center];
stations.push({
chainage_m: round6(chainage),
center_x: round6(ox),
center_y: round6(oy),
center_z: Number.isNaN(centerZ) ? null : round6(centerZ),
tangent_xy: [round9(ly), round9(-lx)],
left_xy: [round9(lx), round9(ly)],
uphill_side: uphill,
samples: cross,
});
}
return { length_m: round6(along[along.length - 1] ?? 0), samples, stations };
}
/** `np.interp(x, xp, fp)` — xp 오름차순 · 범위 밖은 끝값 · 식 차례도 numpy 와 같음. */
function npInterp(x: number, xp: ReadonlyArray<number>, fp: ReadonlyArray<number>): number {
const last = xp.length - 1;
if (x <= xp[0]) return fp[0];
if (x >= xp[last]) return fp[last];
let low = 0;
let high = last;
while (high - low > 1) {
const mid = (low + high) >> 1;
if (xp[mid] <= x) low = mid;
else high = mid;
}
const slope = (fp[low + 1] - fp[low]) / (xp[low + 1] - xp[low]);
return slope * (x - xp[low]) + fp[low];
}
/** 종단 표본 → (누가거리, 지반고) · 결측은 이웃 유효 값으로 선형 보간. 유효 둘 미만이면 throw. */
export function groundProfile(samples: ReadonlyArray<Partial<LongGroundSample>>): {
chainage: number[];
ground: number[];
} {
const rows = samples
.map((s) => ({
chainage: Number(s.chainage_m ?? 0),
z: s.valid !== false && s.elevation_m != null ? Number(s.elevation_m) : NaN,
}))
.sort((a, b) => a.chainage - b.chainage);
const known = rows.filter((row) => Number.isFinite(row.z));
if (known.length < 2 || rows.length < 2) {
throw new Error("계획선을 만들기에 유효한 종단 표고 샘플이 부족합니다.");
}
const xp = known.map((row) => row.chainage);
const fp = known.map((row) => row.z);
return {
chainage: rows.map((row) => row.chainage),
ground: rows.map((row) => npInterp(row.chainage, xp, fp)),
};
}
/** 측점 지반고 — `groundProfile` 위 선형 보간(파이썬 `station_ground`). */
export function groundAt(profile: { chainage: number[]; ground: number[] }, chainageM: number) {
return npInterp(chainageM, profile.chainage, profile.ground);
}
/** 횡단 설계가 덧붙임에 돌려줘야 할 칸. */
export interface UnclosedDesign {
slope_unclosed: boolean;
design_line?: ReadonlyArray<CrossDesignEdge> | null;
}
export interface ExtendInfo {
left_m: number;
right_m: number;
surface_end: boolean;
}
function validSorted(samples: ReadonlyArray<Partial<CrossGroundPoint>>): Array<[number, number]> {
return samples
.filter((s) => s.valid !== false && s.elevation_m != null)
.map((s): [number, number] => [Number(s.offset_m), Number(s.elevation_m)])
.sort((a, b) => a[0] - b[0] || a[1] - b[1]);
}
/** 미교차 측점 지반 표본을 **열린 쪽만** +`stepM` 씩 지표면 끝까지 넓힘 — 닫혔으면 null.
* 새 걸음에 무효 표본이 섞이면 그 쪽은 안 붙이고 `surface_end`. 상한 없음(파이썬과 같음). */
export function extendUnclosed<T extends Partial<CrossGroundPoint>>(
samples: ReadonlyArray<T>,
origin: Vertex,
leftAxis: Vertex,
compute: (samples: Array<T | CrossGroundPoint>) => UnclosedDesign,
grid: DtmGrid,
stepM: number = EXTEND_STEP_M,
): { samples: Array<T | CrossGroundPoint>; info: ExtendInfo } | null {
let current: Array<T | CrossGroundPoint> = [...samples];
let design = compute(current);
if (!design.slope_unclosed) return null;
let offsets = validSorted(current).map(([offset]) => offset);
let spacing = Infinity;
offsets.slice(1).forEach((value, index) => {
const gap = value - offsets[index];
if (gap > 1e-9 && gap < spacing) spacing = gap;
});
if (!Number.isFinite(spacing)) throw new Error("넓힐 표본 간격을 잴 수 없습니다.");
const count = Math.max(Math.round(stepM / spacing), 1);
const info: ExtendInfo = { left_m: 0, right_m: 0, surface_end: false };
while (design.slope_unclosed) {
const ground = validSorted(current);
const lineEnds = (design.design_line ?? [])
.map((p): [number, number] => [Number(p.offset_m), Number(p.elevation_m)])
.sort((a, b) => a[0] - b[0] || a[1] - b[1]);
const open =
ground.length && lineEnds.length
? {
right: Math.abs(ground[0][1] - lineEnds[0][1]) > SLOPE_CLOSE_TOLERANCE_M,
left:
Math.abs(ground[ground.length - 1][1] - lineEnds[lineEnds.length - 1][1]) >
SLOPE_CLOSE_TOLERANCE_M,
}
: { right: false, left: false };
const added: CrossGroundPoint[] = [];
for (const [side, sign] of [
["right", -1],
["left", 1],
] as const) {
if (!open[side]) continue;
const edge = sign < 0 ? Math.min(...offsets) : Math.max(...offsets);
const step: CrossGroundPoint[] = [];
for (let k = 1; k <= count; k += 1) {
const offset = edge + sign * spacing * k;
const point = sample(
grid,
origin[0] + leftAxis[0] * offset,
origin[1] + leftAxis[1] * offset,
);
step.push({ offset_m: round6(offset), ...point });
}
if (!step.every((point) => point.valid)) {
info.surface_end = true;
continue;
}
added.push(...step);
info[`${side}_m`] += stepM;
}
if (!added.length) break;
current = [...current, ...added].sort((a, b) => Number(a.offset_m) - Number(b.offset_m));
offsets = validSorted(current).map(([offset]) => offset);
design = compute(current);
}
return { samples: current, info };
}
@@ -8,6 +8,7 @@
⚠ 716 판정 「미교차 측점만 반폭 +5m 한 번」을 갈음(그것으론 4곳만 닫혔음).
⚠ 계산식은 안 바뀐다 — 샘플(입력)만 넓어진다. 넓힌 샘플은 횡단 파일에 남겨 화면(TS)·Node·B08 이
같은 지반을 본다(한 벌).
⚠ 파이썬·TS 짝 — `extend_unclosed` ↔ `B05_Profile_Sections_Ground.ts`(B05 logic) `extendUnclosed`.
"""
from __future__ import annotations
+60 -1
View File
@@ -8,10 +8,16 @@ DTM valid_mask를 footprint로 결합해 데이터가 없는 영역을 임의
B05 종횡단 전용이었으나 B04 배수유역 세부 설계도 같은 표고면을 써야 해서
common_util로 옮겼다 — 두 화면의 종단 Z가 갈라지면 세부유역 경계가 달라진다
(2026-08-01 구조 개편).
⚠ 파이썬·TS 짝 — `DtmGridSampler.sample_xy` ↔ `common_util_surface_sampler.ts` `heightAt`
(격자 = `dtm_grid_bytes` 꼴 · 쌍선형 · 네 칸 유효). 한쪽을 고치면 다른 쪽도 고치고 거울 시험
`resources/tester/test_344_1_surface_sampler_mirror.py` 를 돌림.
"""
import os
from collections.abc import Callable
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Literal, Protocol, get_args
@@ -89,6 +95,12 @@ class DtmGridSampler:
with np.load(Path(path), allow_pickle=False) as data:
return cls(data["x"], data["y"], data["z"], data["valid_mask"])
@classmethod
def open_cached(cls, path: Path | str) -> "DtmGridSampler":
"""`from_npz` 를 프로세스 메모리에 기억 — 열쇠 = 경로 · 수정시각 · 크기."""
stat = os.stat(path)
return _open_dtm(str(Path(path).resolve()), stat.st_mtime_ns, stat.st_size)
def sample_xy(self, xy: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
xy = np.asarray(xy, dtype=np.float64)
if xy.ndim != 2 or xy.shape[1] != 2:
@@ -116,6 +128,53 @@ class DtmGridSampler:
return z, valid
# ponytail: 지표면 4벌(sheet 격자 한 벌 약 17 MB) — 서버 프로세스 하나 전제 · 늘면 maxsize 를 셈.
@lru_cache(maxsize=4)
def _open_dtm(path: str, _mtime_ns: int, _size: int) -> DtmGridSampler:
return DtmGridSampler.from_npz(path)
#: 브라우저로 보내는 격자 꼴(`dtm_grid_bytes`) 머리 — TS `parseDtmGrid` 가 같은 값을 봄.
GRID_MAGIC = b"AGRD"
GRID_VERSION = 1
def dtm_grid_bytes(path: Path | str) -> bytes:
"""DTM 격자를 브라우저 표본기(TS 짝)용 원시 이진으로 — 값을 하나도 바꾸지 않음.
꼴(little endian): `AGRD` · u32 판 · u32 nx · u32 ny · u32 z 바이트(4 | 8) · u32 0 ·
f64 x[nx] · f64 y[ny] · z[ny × nx](행 = y) · 유효 비트(행 우선 · 낮은 비트 먼저).
축 · 행렬은 `DtmGridSampler` 가 오름차순으로 맞춘 그대로 — 서버 표본과 같은 칸을 봄.
"""
grid = DtmGridSampler.open_cached(path)
with np.load(Path(path), allow_pickle=False) as data:
z_dtype = np.dtype("<f4") if data["z"].dtype == np.float32 else np.dtype("<f8")
head = np.array(
[GRID_VERSION, len(grid.x), len(grid.y), z_dtype.itemsize, 0], dtype="<u4"
).tobytes()
return b"".join(
[
GRID_MAGIC,
head,
grid.x.astype("<f8").tobytes(),
grid.y.astype("<f8").tobytes(),
grid.z.astype(z_dtype).tobytes(),
np.packbits(grid.valid_mask.reshape(-1), bitorder="little").tobytes(),
]
)
def dtm_grid_file(npz_path: Path | str) -> Path:
"""`dtm_grid_bytes` 저장본(npz 곁 `<이름>_grid.bin`) — 없거나 npz 보다 옛것이면 새로 씀."""
npz_path = Path(npz_path)
target = npz_path.with_name(f"{npz_path.stem}_grid.bin")
if not target.is_file() or target.stat().st_mtime_ns < npz_path.stat().st_mtime_ns:
temporary = target.with_name(f"{target.name}.{os.getpid()}.tmp")
temporary.write_bytes(dtm_grid_bytes(npz_path))
os.replace(temporary, target)
return target
@dataclass
class CallableSurfaceSampler:
"""테스트와 어댑터에 사용할 함수 기반 sampler."""
@@ -160,7 +219,7 @@ def build_surface_sampler(
dtm_path = dtm_smooth if smooth and dtm_smooth.exists() else dtm_original
if not dtm_path.exists():
raise FileNotFoundError(f"기준 DTM이 없습니다: {dtm_path.name}")
footprint = DtmGridSampler.from_npz(dtm_path)
footprint = DtmGridSampler.open_cached(dtm_path)
if method == "dtm":
return footprint
@@ -0,0 +1,82 @@
/* =============================================================================
* common_util_surface_sampler.ts
* 확정 지표면(DTM 격자) 표고를 브라우저에서 바로 읽는다 — 노선을 고치는 동안 서버 왕복 없이.
*
* ⚠ 파이썬·TS 짝 — `common_util/common_util_surface_sampler.py` 의 `DtmGridSampler.sample_xy`
* (scipy `RegularGridInterpolator` linear + 네 칸 유효) · 격자 꼴 = `dtm_grid_bytes`.
* 한쪽을 고치면 다른 쪽도 고치고 거울 시험 `resources/tester/test_344_1_surface_sampler_mirror.py`
* 를 돌림.
*
* 칸 폭이 고르지 않음(축이 float32 로 저장돼 0.97~1.09 m) — 「(x − x0) / 1 m」 셈 금지 · 축 배열을
* 이진 탐색(`searchsorted(side="right") − 1`)해 서버와 같은 칸을 봄.
* ========================================================================== */
/** 오름차순 축 · 행렬 · 유효 비트 — 서버 `dtm_grid_bytes` 를 그대로 푼 것. */
export interface DtmGrid {
x: Float64Array;
y: Float64Array;
/** 행 = y · 열 = x (`z[iy * nx + ix]`). */
z: Float32Array | Float64Array;
/** 칸마다 한 비트(행 우선 · 낮은 비트 먼저). */
validBits: Uint8Array;
}
const GRID_MAGIC = "AGRD";
const GRID_VERSION = 1;
const HEAD_BYTES = 24;
/** 서버 `dtm_grid_bytes` 이진을 푼다 — 꼴이 다르면 throw. */
export function parseDtmGrid(buffer: ArrayBuffer): DtmGrid {
const magic = String.fromCharCode(...new Uint8Array(buffer, 0, 4));
const [version, nx, ny, zBytes] = new Uint32Array(buffer, 4, 4);
if (magic !== GRID_MAGIC || version !== GRID_VERSION || (zBytes !== 4 && zBytes !== 8)) {
throw new Error("지형 격자 꼴이 다름");
}
const x = new Float64Array(buffer, HEAD_BYTES, nx);
const y = new Float64Array(buffer, HEAD_BYTES + nx * 8, ny);
const zStart = HEAD_BYTES + (nx + ny) * 8;
const z =
zBytes === 4
? new Float32Array(buffer, zStart, nx * ny)
: new Float64Array(buffer, zStart, nx * ny);
const validBits = new Uint8Array(buffer, zStart + nx * ny * zBytes, Math.ceil((nx * ny) / 8));
return { x, y, z, validBits };
}
/** `np.searchsorted(axis, value, side="right") - 1` — value 이하인 마지막 칸 번호 · 앞이면 −1. */
function cellIndex(axis: Float64Array, value: number): number {
if (!(value >= axis[0])) return -1; // NaN 도 여기(밖)
let low = 0;
let high = axis.length;
while (high - low > 1) {
const mid = (low + high) >> 1;
if (axis[mid] <= value) low = mid;
else high = mid;
}
return low;
}
function isValid(grid: DtmGrid, index: number): boolean {
return ((grid.validBits[index >> 3] >> (index & 7)) & 1) === 1;
}
/** (x, y) 표고(m) — 네 칸이 모두 유효할 때만 값 · 밖 · 빈칸 · 마지막 선 위는 NaN(임의 표고로 안 메움).
* 식 · 셈 차례 = scipy `evaluate_linear_2d`(같은 IEEE 연산 → 같은 값). */
export function heightAt(grid: DtmGrid, x: number, y: number): number {
const nx = grid.x.length;
const ix = cellIndex(grid.x, x);
const iy = cellIndex(grid.y, y);
if (ix < 0 || iy < 0 || ix >= nx - 1 || iy >= grid.y.length - 1) return NaN;
const i00 = iy * nx + ix;
const i10 = i00 + nx;
if (!isValid(grid, i00) || !isValid(grid, i00 + 1)) return NaN;
if (!isValid(grid, i10) || !isValid(grid, i10 + 1)) return NaN;
const ty = (y - grid.y[iy]) / (grid.y[iy + 1] - grid.y[iy]);
const tx = (x - grid.x[ix]) / (grid.x[ix + 1] - grid.x[ix]);
let value = 0.0;
value = value + grid.z[i00] * (1 - ty) * (1 - tx);
value = value + grid.z[i00 + 1] * (1 - ty) * tx;
value = value + grid.z[i10] * ty * (1 - tx);
value = value + grid.z[i10 + 1] * ty * tx;
return Number.isFinite(value) ? value : NaN;
}
@@ -76,6 +76,11 @@ def _patch_db(monkeypatch, project_root: Path, params: dict) -> None:
monkeypatch.setattr(terrain_router, "get_surface_confirmation_params", confirmed)
monkeypatch.setattr(terrain_router, "resolve_stored_project_path", lambda _s: project_root)
async def no_model(_connection, _project_id): # 격자 받을 자리(344-1)는 따로 시험
return None
monkeypatch.setattr(terrain_router, "_confirmed_model", no_model)
PID = "1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a"
@@ -0,0 +1,276 @@
"""횡단 지반선 생성 거울 — 파이썬 ↔ 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 "")
@@ -0,0 +1,206 @@
"""지형 격자 표본기 거울 — 파이썬 `DtmGridSampler.sample_xy` ↔ TS `common_util_surface_sampler.ts`.
PLAN 344-1 — 노선 편집이 브라우저에서 지반고를 읽어도 서버 값과 같아야 함.
가장자리 · 빈칸 · 마지막 선 · 칸 폭 고르지 않음 · 내림차순 축 · 무작위 2만 점.
"""
from __future__ import annotations
import json
import os
import re
import subprocess
import sys
import time
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 common_util import common_util_surface_sampler as sampler_py # noqa: E402
TS_FILE = ROOT / "common_util" / "common_util_surface_sampler.ts"
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
#: 검증 프로젝트(검증_sub8) 설계 지형 — 없으면 그 시험만 건너뜀.
REAL_GRID = (
ROOT
/ "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a/B04_PreProcess/models"
/ "dtm_classification_smooth.npz"
)
REAL_LONGITUDINAL = REAL_GRID.parents[2] / "B06_Section" / "longitudinal" / "longitudinal.json"
_RUNNER = """
import { readFileSync, writeFileSync } from "node:fs";
import { parseDtmGrid, heightAt } from "./common_util_surface_sampler.js";
const bytes = readFileSync(process.argv[2]);
const grid = parseDtmGrid(bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.length));
const points = JSON.parse(readFileSync(process.argv[3], "utf8"));
const z = points.map(([x, y]) => {
const value = heightAt(grid, x, y);
return Number.isNaN(value) ? null : value;
});
writeFileSync(process.argv[4], JSON.stringify(z));
"""
def _compile(tmp_path: Path) -> Path:
out = tmp_path / "js"
subprocess.run( # noqa: S603 — 고정 실행 파일
[
"node",
str(TSC),
str(TS_FILE),
"--outDir",
str(out),
"--module",
"esnext",
"--target",
"es2022",
"--moduleResolution",
"bundler",
"--ignoreConfig",
],
cwd=str(ROOT),
check=True,
capture_output=True,
)
for emitted in out.glob("*.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")
return out
def _ts_heights(tmp_path: Path, npz: Path, points: np.ndarray) -> list[float | None]:
out = _compile(tmp_path)
grid = tmp_path / "grid.bin"
grid.write_bytes(sampler_py.dtm_grid_bytes(npz))
payload = tmp_path / "points.json"
result = tmp_path / "z.json"
payload.write_text(json.dumps(points.tolist()), 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 _compare(npz: Path, points: np.ndarray, actual: list[float | None]) -> float:
"""유효 어긋남 0 을 묶고 최대 표고 차를 돌려줌."""
expected, valid = sampler_py.DtmGridSampler.from_npz(npz).sample_xy(points)
mismatched = [i for i, ok in enumerate(valid) if bool(ok) != (actual[i] is not None)]
assert not mismatched, (
f"유효 어긋남 {len(mismatched)}점 — 처음 {points[mismatched[:5]].tolist()}"
)
diffs = [abs(actual[i] - float(expected[i])) for i in np.flatnonzero(valid)]
return max(diffs) if diffs else 0.0
def _edge_points(x: np.ndarray, y: np.ndarray) -> np.ndarray:
"""격자 꼭짓점 · 첫 · 마지막 선 · 그 바로 안팎 · 칸 가운데."""
xs = np.r_[x, (x[:-1] + x[1:]) / 2, x[0] - 1e-9, x[-1] - 1e-9, x[-1] + 1e-9]
ys = np.r_[y[[0, 1, -2, -1]], y[-1] - 1e-9, y[0] + 1e-9, (y[0] + y[1]) / 2]
edge = np.array([[a, b] for a in xs for b in ys])
edge2 = np.array([[a, b] for a in x[[0, 1, -2, -1]] for b in y])
return np.vstack([edge, edge2])
@pytest.fixture
def synthetic_npz(tmp_path: Path) -> Path:
"""칸 폭 고르지 않음(float32) · y 내림차순 · 빈칸 · 유효인데 z NaN 한 칸."""
rng = np.random.default_rng(344)
x = np.cumsum(rng.uniform(0.95, 1.1, 60)).astype(np.float32) + np.float32(208300.0)
y = (np.cumsum(rng.uniform(0.95, 1.1, 40)) + 367400.0).astype(np.float32)[::-1]
z = (800 + rng.normal(0, 5, (40, 60))).astype(np.float32)
mask = np.ones((40, 60), dtype=bool)
mask[10:14, 20:25] = False
mask[0, :] = False
mask[:, -1] = False
z[30, 30] = np.nan
path = tmp_path / "dtm_test.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, synthetic_npz: Path) -> None:
data = np.load(synthetic_npz)
x, y = np.sort(data["x"].astype(np.float64)), np.sort(data["y"].astype(np.float64))
rng = np.random.default_rng(1)
random = np.column_stack(
[rng.uniform(x[0] - 5, x[-1] + 5, 20000), rng.uniform(y[0] - 5, y[-1] + 5, 20000)]
)
points = np.vstack([_edge_points(x, y), random])
worst = _compare(synthetic_npz, points, _ts_heights(tmp_path, synthetic_npz, points))
assert worst <= 1e-9, f"표고 차 {worst}"
@pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치")
@pytest.mark.skipif(not REAL_GRID.is_file(), reason="검증_sub8 지형 없음")
def test_검증_프로젝트_격자와_저장된_종단이_서버와_같다(tmp_path: Path) -> None:
grid = sampler_py.DtmGridSampler.from_npz(REAL_GRID)
rng = np.random.default_rng(2)
random = np.column_stack(
[
rng.uniform(grid.x[0] - 5, grid.x[-1] + 5, 20000),
rng.uniform(grid.y[0] - 5, grid.y[-1] + 5, 20000),
]
)
stored = []
if REAL_LONGITUDINAL.is_file():
samples = json.loads(REAL_LONGITUDINAL.read_text(encoding="utf-8"))["samples"]
stored = [[s["x"], s["y"]] for s in samples]
points = np.vstack([_edge_points(grid.x, grid.y), random, np.asarray(stored).reshape(-1, 2)])
actual = _ts_heights(tmp_path, REAL_GRID, points)
worst = _compare(REAL_GRID, points, actual)
assert worst <= 1e-9, f"표고 차 {worst}"
def test_격자_꼴이_축_칸_비트를_그대로_싣는다(synthetic_npz: Path) -> None:
blob = sampler_py.dtm_grid_bytes(synthetic_npz)
assert blob[:4] == b"AGRD"
version, nx, ny, z_bytes, _ = np.frombuffer(blob, "<u4", 5, 4)
assert (version, nx, ny, z_bytes) == (1, 60, 40, 4)
grid = sampler_py.DtmGridSampler.from_npz(synthetic_npz)
x = np.frombuffer(blob, "<f8", nx, 24)
assert np.array_equal(x, grid.x) and np.all(np.diff(x) > 0)
z = np.frombuffer(blob, "<f4", nx * ny, 24 + 8 * (nx + ny)).reshape(ny, nx)
assert np.array_equal(z, grid.z.astype(np.float32), equal_nan=True)
bits = np.frombuffer(blob, np.uint8, offset=24 + 8 * (nx + ny) + 4 * nx * ny)
mask = np.unpackbits(bits, bitorder="little")[: nx * ny].reshape(ny, nx).astype(bool)
assert np.array_equal(mask, grid.valid_mask)
def test_표본기_기억은_파일이_바뀌면_새로_읽는다(synthetic_npz: Path) -> None:
first = sampler_py.DtmGridSampler.open_cached(synthetic_npz)
assert sampler_py.DtmGridSampler.open_cached(synthetic_npz) is first
data = dict(np.load(synthetic_npz))
data["z"] = data["z"] + np.float32(1.0)
np.savez(synthetic_npz, **data)
later = time.time() + 5
os.utime(synthetic_npz, (later, later))
assert sampler_py.DtmGridSampler.open_cached(synthetic_npz) is not first
def test_격자_저장본은_npz_보다_옛것이면_다시_쓴다(synthetic_npz: Path) -> None:
target = sampler_py.dtm_grid_file(synthetic_npz)
assert target.name == "dtm_test_grid.bin"
assert target.read_bytes() == sampler_py.dtm_grid_bytes(synthetic_npz)
stamp = target.stat().st_mtime_ns
assert sampler_py.dtm_grid_file(synthetic_npz).stat().st_mtime_ns == stamp # 그대로 씀
later = time.time() + 10
os.utime(synthetic_npz, (later, later))
assert sampler_py.dtm_grid_file(synthetic_npz).stat().st_mtime_ns != stamp
def test_짝_머리글이_서로를_가리킨다() -> None:
assert "common_util_surface_sampler.py" in TS_FILE.read_text(encoding="utf-8")
assert "common_util_surface_sampler.ts" in (sampler_py.__doc__ or "")
@@ -0,0 +1,116 @@
"""PLAN 344-1 — 격자 통로(`/surface/models/{mid}/grid`) · 받을 자리(`terrain-sources` 의 `grid`).
DTM 만 격자 · 평활 고르기 = 서버 표본기 · 첫 요청 때 `.bin` 저장본 · ETag 304 · DTM 아닌 표현 404 ·
주소 서명 = manifest 구성 · 평활 서명.
"""
from __future__ import annotations
import asyncio
import json
from contextlib import asynccontextmanager
from pathlib import Path
import numpy as np
from starlette.requests import Request
from B05_Profile import B05_Profile_Router_Terrain as terrain_router
from common_util.common_util_surface_sampler import dtm_grid_bytes
PID = "1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a"
PARAMS = {"source_filter": "csf", "method": "dtm", "smooth": True, "contour_interval_m": 1.0}
def _dtm(path: Path, z: float) -> None:
x = np.arange(0.0, 11.0, dtype=np.float32)
np.savez(
path,
x=x,
y=x,
z=np.full((11, 11), z, dtype=np.float32),
valid_mask=np.ones((11, 11), dtype=bool),
)
def _project(tmp_path: Path) -> Path:
models = tmp_path / "B04_PreProcess" / "models"
models.mkdir(parents=True)
_dtm(models / "dtm_csf.npz", 100.0)
_dtm(models / "dtm_csf_smooth.npz", 101.0)
manifest = {
"config_signature": "cfg1",
"source_filters": {"csf": {"methods": {"dtm": {"smooth": {"smoothing_signature": "sm1"}}}}},
}
(models / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
return tmp_path
def _patch(monkeypatch, root: Path, row: tuple | None) -> None:
class _Cursor:
async def execute(self, *_args):
return None
async def fetchone(self):
return row
class _Connection:
@asynccontextmanager
async def cursor(self):
yield _Cursor()
class _Pool:
@asynccontextmanager
async def acquire(self):
yield _Connection()
async def stored(_connection, _project_id):
return "x"
monkeypatch.setattr(terrain_router, "get_db_pool", lambda: _Pool())
monkeypatch.setattr(terrain_router, "get_project_storage_relative_path", stored)
monkeypatch.setattr(terrain_router, "resolve_stored_project_path", lambda _s: root)
def _request(etag: str | None = None) -> Request:
headers = [(b"if-none-match", etag.encode())] if etag else []
return Request({"type": "http", "method": "GET", "path": "/", "headers": headers})
def _grid(smooth: bool, etag: str | None = None):
return asyncio.run(terrain_router.read_surface_grid(_request(etag), PID, 7, smooth))
def test_평활_격자를_저장본으로_내고_같은_ETag_면_304(tmp_path, monkeypatch):
root = _project(tmp_path)
models = root / "B04_PreProcess" / "models"
_patch(monkeypatch, root, ("dtm", "B04_PreProcess/models/dtm_csf.npz"))
response = _grid(True)
assert response.status_code == 200
saved = models / "dtm_csf_smooth_grid.bin"
assert Path(response.path) == saved
assert saved.read_bytes() == dtm_grid_bytes(models / "dtm_csf_smooth.npz")
assert _grid(True, response.headers["etag"]).status_code == 304
assert Path(_grid(False).path).name == "dtm_csf_grid.bin"
def test_DTM_아닌_표현은_404(tmp_path, monkeypatch):
root = _project(tmp_path)
_patch(monkeypatch, root, ("tin", "B04_PreProcess/models/tin_csf.npz"))
assert _grid(True).status_code == 404
_patch(monkeypatch, root, None)
assert _grid(True).status_code == 404
def test_받을_자리_서명은_manifest_구성과_평활(tmp_path):
models = _project(tmp_path) / "B04_PreProcess" / "models"
confirmed = (7, "dtm", "B04_PreProcess/models/dtm_csf.npz")
grid = terrain_router._design_grid(PARAMS, models, confirmed)
assert grid == {"model_id": 7, "smooth": True, "signature": "cfg1.sm1"}
raw = terrain_router._design_grid({**PARAMS, "smooth": False}, models, confirmed)
assert raw["signature"] == "cfg1."
# 서버 표본기와 다른 파일 · DTM 아닌 표현 · 확정 없음 = 격자 안 냄(서버 표본 길)
assert (
terrain_router._design_grid({**PARAMS, "source_filter": "pmf"}, models, confirmed) is None
)
assert terrain_router._design_grid({**PARAMS, "method": "tin"}, models, confirmed) is None
assert terrain_router._design_grid(PARAMS, models, None) is None