feat(지형표본): 344-1 격자 통로 · 노선 편집 지반고 브라우저 표본 — GET surface/models/{mid}/grid(원시 이진 · ETag · 첫 요청 때 .bin 저장본 · DTM 아니면 404) · terrain-sources 에 격자 받을 자리(manifest 서명) · IndexedDB 받기 · B05 미리 받기 한 줄 · elevations = 격자 있으면 브라우저 · 없으면 서버

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Se7ytqbohmBQnfYpwYFnYf

A00_Common/b_asset_cache.ts

@@ -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);

B05_Profile/B05_Profile_Api_Replan.ts

@@ -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) }) },

B05_Profile/B05_Profile_Api_TerrainGrid.ts

@@ -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;
+}

B05_Profile/B05_Profile_Router_Terrain.py

@@ -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")

resources/tester/test_133_1_analysis_terrain.py

@@ -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"
 

resources/tester/test_344_1_terrain_grid_route.py

@@ -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