feat(762-2): 서피스 품질 지표 넷 — 튐 넓이 % · 1 m 등고선 꺾임 /km · 땅 점 빈 곳 % · 필터 차 % · B04 모델마다 표시

- 새 Engine_Quality — 599 정의 그대로(5 × 5 중간값 0.5 m · 2 m 다시 찍어 30° · 2 m 안 지면점 없음 · 짝 필터 |차| > 0.5 m)
- 파이프라인이 dtm · implicit 을 지을 때 셈해 models/quality_{필터}_{표현}.json 에 저장 · 모델 · 짝 모델 지문이 다를 때만 다시 셈
- GET /surface/models/{id}/quality · B04 지표면 분석 칸에 값 목록(부품 kv) · 문턱 판정 없음(사용자 협의)
- B04 필터 드롭다운 · 이름을 새 UI_Filter 로 옮김(UI_Page 679 → 661 줄) · 업체 분류 + 다듬기 · 점진 TIN + 다듬기
- 지킴 시험 test_762_1_refine — 덤불 혹 지움 · 능선 안 깎음 · 스위치 끔 = 지금 길 · 튐 지표

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B04_PreProcess/B04_PreProcess_Engine_Pipeline.py

@@ -26,6 +26,7 @@ from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
     bounds_dict,
     config_signature,
 )
+from B04_PreProcess.B04_PreProcess_Engine_Quality import quality_after_build
 from B04_PreProcess.B04_PreProcess_Engine_Smooth import (
     compute_smoothing_signature,
     run_smoothing,
@@ -262,6 +263,7 @@ def _build_all_terrain_models(
                         manifest.get("bounds", {}),
                         entry,
                     )
+                quality_after_build(output_dir, filter_key, method, context, fresh=False)
                 done_units += 1
                 _report(f"{filter_key}-{method} 캐시 재사용")
                 continue
@@ -320,6 +322,7 @@ def _build_all_terrain_models(
                     manifest.get("bounds", {}),
                     metadata,
                 )
+                quality_after_build(output_dir, filter_key, method, context, fresh=True)
             except Exception as exc:
                 failures += 1
                 filter_entry["methods"][method] = {

B04_PreProcess/B04_PreProcess_Engine_Quality.py

@@ -0,0 +1,210 @@
+"""B04 서피스 품질 지표 넷 — 기준 없이 재는 값(762-2 · _분석_599 정의 · _분석_601-1 3-3).
+
+① 튐 넓이 % — 5 × 5 칸 중간값 대비 0.5 m 넘는 칸(8 방향 묶음 = 곳 수)
+② 1 m 등고선 꺾임 /km — 2 m 마다 다시 찍어 30° 넘게 꺾인 곳
+③ 땅 점 빈 곳 % — 2 m 안 지면점 없는 칸 · 그중 기울기 > 15° 칸
+④ 필터 차 % — 같은 표현 · 짝 필터 모델과 |차| > 0.5 m 넓이
+저장 = `models/quality_{필터}_{표현}.json`(파생 저장본 — 모델 · 짝 모델 지문이 다를 때만 다시 셈).
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+from scipy import ndimage
+from skimage.measure import find_contours
+
+from common_util.common_util_derived_store import file_stamp
+from common_util.common_util_json import atomic_write_json
+
+logger = logging.getLogger(__name__)
+
+QUALITY_VERSION = 1
+SPIKE_M = 0.5
+SPIKE_WINDOW = 5
+CONTOUR_INTERVAL_M = 1.0
+KINK_STEP_M = 2.0
+KINK_DEG = 30.0
+GAP_M = 2.0
+GAP_STEEP_DEG = 15.0
+DIFF_M = 0.5
+GRID_METHODS = ("dtm", "implicit")
+# 필터 차 짝 — 업체 분류 ↔ 점진 TIN(다듬기 판끼리) · 없으면 다듬기 전 ↔ 뒤
+_PAIRS = {
+    "classification": ("ptd", "classification_refine"),
+    "ptd": ("classification", "ptd_refine"),
+    "classification_refine": ("ptd_refine", "classification"),
+    "ptd_refine": ("classification_refine", "ptd"),
+}
+
+
+def _grid_path(models_dir: Path, filter_key: str, method: str) -> Path:
+    suffix = "_grid" if method == "implicit" else ""
+    return models_dir / f"{method}_{filter_key}{suffix}.npz"
+
+
+def _load_grid(path: Path) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
+    with np.load(path) as data:
+        z = np.asarray(data["z"], dtype=np.float64)
+        valid = np.asarray(data["valid_mask"], dtype=bool) & np.isfinite(z)
+        cell = float(np.asarray(data["resolution"]).ravel()[0]) if "resolution" in data else 1.0
+    return np.where(valid, z, 0.0), valid, np.asarray(cell)
+
+
+def spike_stats(z: np.ndarray, valid: np.ndarray) -> dict[str, Any]:
+    residual = z - ndimage.median_filter(z, size=SPIKE_WINDOW, mode="nearest")
+    inner = ndimage.binary_erosion(valid, iterations=SPIKE_WINDOW // 2)
+    spike = (np.abs(residual) > SPIKE_M) & inner
+    _, count = ndimage.label(spike, structure=np.ones((3, 3), dtype=bool))
+    area = int(inner.sum())
+    return {
+        "spike_pct": round(100.0 * int(spike.sum()) / area, 2) if area else None,
+        "spike_count": int(count),
+    }
+
+
+def contour_kinks(z: np.ndarray, valid: np.ndarray, cell: float) -> dict[str, Any]:
+    if not valid.any():
+        return {"contour_km": 0.0, "kinks_per_km": None, "bubbles": 0}
+    low, high = float(z[valid].min()), float(z[valid].max())
+    total, kinks, bubbles = 0.0, 0, 0
+    step = KINK_STEP_M / cell
+    for level in np.arange(np.ceil(low / CONTOUR_INTERVAL_M) * CONTOUR_INTERVAL_M, high, 1.0):
+        for line in find_contours(z, level, mask=valid):
+            seg = np.hypot(*np.diff(line, axis=0).T)
+            length = float(seg.sum())
+            if length < step * 2:
+                continue
+            total += length
+            if np.allclose(line[0], line[-1]) and length * cell < 30.0:
+                bubbles += 1
+            along = np.r_[0.0, np.cumsum(seg)]
+            t = np.arange(0.0, length, step)
+            rows, cols = np.interp(t, along, line[:, 0]), np.interp(t, along, line[:, 1])
+            angle = np.arctan2(np.diff(rows), np.diff(cols))
+            turn = np.abs((np.diff(angle) + np.pi) % (2 * np.pi) - np.pi)
+            kinks += int((np.degrees(turn) > KINK_DEG).sum())
+    km = total * cell / 1000.0
+    return {
+        "contour_km": round(km, 2),
+        "kinks_per_km": round(kinks / km, 1) if km else None,
+        "bubbles": bubbles,
+    }
+
+
+def gap_stats(
+    z: np.ndarray, valid: np.ndarray, occupied: np.ndarray, cell: float
+) -> dict[str, Any]:
+    gap = (ndimage.distance_transform_edt(~occupied) * cell > GAP_M) & valid
+    grad_y, grad_x = np.gradient(z, cell)
+    steep = np.degrees(np.arctan(np.hypot(grad_x, grad_y))) > GAP_STEEP_DEG
+    area = int(valid.sum())
+    return {
+        "gap_pct": round(100.0 * int(gap.sum()) / area, 2) if area else None,
+        "gap_steep_pct": round(100.0 * int((gap & steep).sum()) / area, 2) if area else None,
+    }
+
+
+def ground_occupancy(xyz: Any, indices: np.ndarray, bounds: np.ndarray, shape: tuple) -> np.ndarray:
+    """지면점이 하나라도 든 칸 — dtm 격자와 같은 칸 나눔(`TerrainContext.grid`)."""
+    rows, cols = shape
+    cell_x = (bounds[0, 1] - bounds[0, 0]) / max(cols - 1, 1)
+    cell_y = (bounds[1, 1] - bounds[1, 0]) / max(rows - 1, 1)
+    occupied = np.zeros(shape, dtype=bool)
+    for start in range(0, len(indices), 2_000_000):
+        points = np.asarray(xyz[indices[start : start + 2_000_000]], dtype=np.float64)
+        gx = np.clip(((points[:, 0] - bounds[0, 0]) / cell_x).astype(np.int64), 0, cols - 1)
+        gy = np.clip(((points[:, 1] - bounds[1, 0]) / cell_y).astype(np.int64), 0, rows - 1)
+        occupied[gy, gx] = True
+    return occupied
+
+
+def diff_pct(z: np.ndarray, valid: np.ndarray, other_path: Path) -> float | None:
+    other_z, other_valid, _ = _load_grid(other_path)
+    if other_z.shape != z.shape:
+        return None
+    both = valid & other_valid
+    area = int(both.sum())
+    if not area:
+        return None
+    return round(100.0 * int(((np.abs(z - other_z) > DIFF_M) & both).sum()) / area, 2)
+
+
+def _pair(models_dir: Path, filter_key: str, method: str) -> Path | None:
+    for other in _PAIRS.get(filter_key, ()):
+        path = _grid_path(models_dir, other, method)
+        if path.is_file():
+            return path
+    return None
+
+
+def ensure_quality(
+    models_dir: Path,
+    filter_key: str,
+    method: str,
+    occupied: np.ndarray | None = None,
+) -> dict[str, Any] | None:
+    """지표 넷을 돌려줌 — 저장본 지문이 맞으면 그대로 · ③ 은 지을 때 넘긴 지면점 칸만."""
+    if method not in GRID_METHODS:
+        return None
+    grid_path = _grid_path(models_dir, filter_key, method)
+    if not grid_path.is_file():
+        return None
+    pair = _pair(models_dir, filter_key, method)
+    key = {
+        "version": QUALITY_VERSION,
+        "model": file_stamp(grid_path),
+        "pair": [pair.name, file_stamp(pair)] if pair else None,
+    }
+    path = models_dir / f"quality_{filter_key}_{method}.json"
+    try:
+        stored = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        stored = {}
+    if stored.get("key") == key and occupied is None:
+        return stored
+    z, valid, cell = _load_grid(grid_path)
+    cell = float(cell)
+    same_model = (stored.get("key") or {}).get("model") == key["model"]
+    result: dict[str, Any] = {"key": key, "filter": filter_key, "method": method}
+    if same_model and occupied is None:
+        result.update({k: stored.get(k) for k in ("spike_pct", "spike_count", "contour_km")})
+        result.update({k: stored.get(k) for k in ("kinks_per_km", "bubbles")})
+    else:
+        result.update(spike_stats(z, valid))
+        result.update(contour_kinks(z, valid, cell))
+    if occupied is not None and occupied.shape == z.shape:
+        result.update(gap_stats(z, valid, occupied, cell))
+    elif same_model:
+        result.update({k: stored.get(k) for k in ("gap_pct", "gap_steep_pct")})
+    result["diff_against"] = pair.name if pair else None
+    result["diff_pct"] = diff_pct(z, valid, pair) if pair else None
+    atomic_write_json(path, result)
+    return result
+
+
+def ensure_quality_safely(models_dir: Path, filter_key: str, method: str, **kwargs: Any) -> None:
+    """지을 때 부름 — 실패가 모델 짓기를 막지 않음."""
+    try:
+        ensure_quality(models_dir, filter_key, method, **kwargs)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("서피스 품질 지표 실패: %s-%s %s", filter_key, method, exc)
+
+
+def quality_after_build(
+    models_dir: Path, filter_key: str, method: str, context: Any, fresh: bool
+) -> None:
+    """파이프라인이 표현 하나를 짓거나 다시 쓴 뒤 부름 — ③ 은 지면점이 손에 있을 때만 셈."""
+    if method not in GRID_METHODS:
+        return
+    occupied = None
+    if fresh or not (models_dir / f"quality_{filter_key}_{method}.json").is_file():
+        try:
+            resolution = float(context.config.get("dtm_grid_resolution_meters", 1.0))
+            shape = context.grid(resolution)[2].shape
+            occupied = ground_occupancy(context.xyz, context.indices(), context.bounds, shape)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("서피스 품질 지면점 칸 실패: %s %s", filter_key, exc)
+    ensure_quality_safely(models_dir, filter_key, method, occupied=occupied)

B04_PreProcess/B04_PreProcess_Router_Contour.py

@@ -190,3 +190,34 @@ async def get_surface_model_contour(
             status_code=500,
             content={"status": "error", "message": "등고선 파일 조회 중 오류가 발생했습니다."},
         )
+
+
+@router.get("/{project_id}/surface/models/{model_id}/quality")
+async def get_surface_model_quality(project_id: UUID, model_id: int) -> JSONResponse:
+    """서피스 품질 지표 넷(762-2) — 저장본 지문이 다르면 다시 셈 · 격자 표현 밖이면 null."""
+    from B04_PreProcess.B04_PreProcess_Engine_Quality import ensure_quality
+
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        stored_path = await get_project_storage_relative_path(connection, project_id)
+        async with connection.cursor() as cursor:
+            await cursor.execute(
+                "SELECT model_file_path FROM surface_models WHERE id = %s AND project_id = %s",
+                (model_id, str(project_id)),
+            )
+            row = await cursor.fetchone()
+    if not row or not row[0]:
+        return JSONResponse(
+            status_code=404,
+            content={"status": "error", "message": "해당 모델을 찾을 수 없습니다."},
+        )
+    model_path = Path(resolve_stored_project_path(stored_path)) / row[0]
+    method, _, filter_key = model_path.stem.partition("_")
+    try:
+        quality = await asyncio.to_thread(ensure_quality, model_path.parent, filter_key, method)
+    except Exception:
+        logger.exception("서피스 품질 지표 실패: project_id=%s, model_id=%s", project_id, model_id)
+        quality = None
+    if quality:
+        quality = {key: value for key, value in quality.items() if key != "key"}
+    return JSONResponse(content={"status": "success", "quality": quality})

B04_PreProcess/B04_PreProcess_UI_Filter.ts

@@ -0,0 +1,118 @@
+/* =============================================================================
+ * B04_PreProcess_UI_Filter.ts
+ * B04 지면 필터 드롭다운(고를 갈래 · 이름) + 고른 모델의 서피스 품질 지표 넷(762-2).
+ *   지표 = GET /api/projects/{id}/surface/models/{model}/quality — 서버가 지을 때 셈해 둔 값
+ *   (튐 넓이 % · 1 m 등고선 꺾임 /km · 땅 점 빈 곳 % · 필터 차 %) · 문턱 판정은 안 함(사용자 협의).
+ * ========================================================================== */
+
+import { API_BASE_URL } from "@config/config_frontend";
+import { createSelectField } from "@ui/ui_template_elements";
+import { createKv } from "@ui/ui_template_kv";
+import type { SurfaceModelSummary } from "./B04_PreProcess_Api_Fetch";
+
+// 고를 수 있는 필터. 자동 전처리는 이 중 기본 하나만 만들고, 나머지는 관리자가
+// 드롭다운에서 고를 때 그 조합만 계산한다 (2026-09-01 사용자 확정).
+// `_refine` = 다듬기 판(762) — 바탕 필터 바로 뒤에 둠(이름 찾기는 가장 긴 것이 이김).
+export const SOURCE_FILTERS = [
+  "classification",
+  "classification_refine",
+  "ptd",
+  "ptd_refine",
+  "grid_min_z",
+  "csf",
+  "pmf",
+] as const;
+
+const FILTER_LABELS: Record<string, string> = {
+  ptd: "점진 TIN",
+  classification_refine: "업체 분류 + 다듬기",
+  ptd_refine: "점진 TIN + 다듬기",
+};
+
+function optionLabel(value: string): string {
+  return FILTER_LABELS[value] ?? value.replaceAll("_", " ").toUpperCase();
+}
+
+/** 라벨 한 줄 + 드롭다운. 지표면 분석 컨테이너 안의 항목 하나를 만든다. */
+export function buildSelectField(
+  label: string,
+  values: readonly string[],
+  defaultValue: string,
+): { root: HTMLElement; select: HTMLSelectElement } {
+  const { root, select } = createSelectField({
+    label,
+    options: values.map((value) => ({ value, text: optionLabel(value) })),
+    value: defaultValue,
+    size: 36,
+  });
+  root.classList.add("b04-surface__field");
+  return { root, select };
+}
+
+interface SurfaceQuality {
+  spike_pct?: number | null;
+  spike_count?: number | null;
+  kinks_per_km?: number | null;
+  gap_pct?: number | null;
+  gap_steep_pct?: number | null;
+  diff_pct?: number | null;
+  diff_against?: string | null;
+}
+
+function fmt(value: number | null | undefined, unit: string): string | null {
+  return value == null ? null : `${value.toLocaleString()} ${unit}`;
+}
+
+function rows(quality: SurfaceQuality | null): ReturnType<typeof createKv> {
+  const against = quality?.diff_against
+    ?.replace(/^(dtm|implicit)_/, "")
+    .replace(/(_grid)?\.npz$/, "");
+  return createKv([
+    {
+      label: "튐 넓이",
+      value:
+        quality?.spike_pct == null
+          ? null
+          : `${quality.spike_pct} % · ${quality.spike_count ?? 0} 곳`,
+    },
+    { label: "등고선 꺾임", value: fmt(quality?.kinks_per_km, "/km") },
+    {
+      label: "땅 점 빈 곳",
+      value:
+        quality?.gap_pct == null
+          ? null
+          : `${quality.gap_pct} % · 급경사 ${quality.gap_steep_pct} %`,
+    },
+    {
+      label: "필터 차",
+      value:
+        quality?.diff_pct == null
+          ? null
+          : `${quality.diff_pct} % · ${optionLabel(against ?? "")} 대비`,
+    },
+  ]);
+}
+
+/** 고른 모델의 품질 지표 넷 — 늦게 온 옛 응답은 버림. */
+export function createQualityPanel(): {
+  root: HTMLElement;
+  show: (projectId: string, model: SurfaceModelSummary | undefined) => void;
+} {
+  const root = document.createElement("div");
+  root.append(rows(null));
+  let ticket = 0;
+  function show(projectId: string, model: SurfaceModelSummary | undefined): void {
+    const mine = ++ticket;
+    root.replaceChildren(rows(null));
+    if (!model) return;
+    void fetch(`${API_BASE_URL}/projects/${projectId}/surface/models/${model.id}/quality`, {
+      credentials: "include",
+    })
+      .then((response) => (response.ok ? response.json() : null))
+      .then((payload: { quality?: SurfaceQuality | null } | null) => {
+        if (mine === ticket) root.replaceChildren(rows(payload?.quality ?? null));
+      })
+      .catch(() => undefined);
+  }
+  return { root, show };
+}

B04_PreProcess/B04_PreProcess_UI_Page.ts

@@ -40,6 +40,7 @@ import {
   type SurfacePointCloudSampleResponse,
   type SurfaceStatusResponse,
 } from "./B04_PreProcess_Api_Fetch";
+import { buildSelectField, createQualityPanel, SOURCE_FILTERS } from "./B04_PreProcess_UI_Filter";
 import { buildSurfaceGroup } from "./B04_PreProcess_UI_Group";
 import { createSurfaceMapViewer } from "./B04_PreProcess_UI_MapViewer";
 import { createSurfaceTerrainViewer } from "./B04_PreProcess_UI_TerrainViewer";
@@ -49,9 +50,6 @@ import "./B04_PreProcess_UI_Style.css";
 import "./B04_PreProcess_UI_Style_Map.css";
 import { currentProjectId, guardProjectLeave } from "../A00_Common/b_project_switch";
 
-// 고를 수 있는 필터. 자동 전처리는 이 중 기본 하나만 만들고, 나머지는 관리자가
-// 드롭다운에서 고를 때 그 조합만 계산한다 (2026-09-01 사용자 확정).
-const SOURCE_FILTERS = ["classification", "ptd", "grid_min_z", "csf", "pmf"] as const;
 const MODEL_METHODS = ["tin", "dtm", "nurbs", "implicit", "meshfree"] as const;
 const DEFAULT_FILTER = "ptd";
 const DEFAULT_METHOD = "implicit"; // 확정 기본 표현(요구 558) — 확정본이 있으면 그것이 먼저
@@ -67,30 +65,11 @@ function infoRow(label: string, value: unknown): KvRow {
   return { label, value: value == null || value === "" ? "-" : String(value) };
 }
 
-/** 라벨 한 줄 + 드롭다운. 지표면 분석 컨테이너 안의 항목 하나를 만든다. */
-function buildSelectField(
-  label: string,
-  values: readonly string[],
-  defaultValue: string,
-): { root: HTMLElement; select: HTMLSelectElement } {
-  const { root, select } = createSelectField({
-    label,
-    options: values.map((value) => ({
-      value,
-      text: value === "ptd" ? "점진 TIN" : value.replaceAll("_", " ").toUpperCase(),
-    })),
-    value: defaultValue,
-    size: 36,
-  });
-  root.classList.add("b04-surface__field");
-  return { root, select };
-}
-
 function getModelFilter(model: SurfaceModelSummary): string {
   const configured = model.generation_params?.source_filter;
   if (typeof configured === "string") return configured.toLowerCase();
   const path = model.model_file_path?.toLowerCase() ?? "";
-  return SOURCE_FILTERS.find((filter) => path.includes(filter)) ?? "";
+  return SOURCE_FILTERS.filter((filter) => path.includes(filter)).pop() ?? "";
 }
 
 export async function renderB04Surface(root: HTMLElement): Promise<void> {
@@ -138,6 +117,7 @@ export async function renderB04Surface(root: HTMLElement): Promise<void> {
   let appliedFilter = DEFAULT_FILTER;
   let appliedMethod = DEFAULT_METHOD;
   const viewer = createSurfacePointCloudViewer();
+  const quality = createQualityPanel(); // 762-2 고른 모델의 서피스 품질 지표 넷
   const terrainViewer = createSurfaceTerrainViewer();
   const mapViewer = createSurfaceMapViewer();
   // 도엽등고 3D 서피스 — 전처리에서 함께 생성되는 참고 서피스(LAS 없는 설계의 지형 원천,
@@ -268,6 +248,7 @@ export async function renderB04Surface(root: HTMLElement): Promise<void> {
     filterGroup.root,
     methodGroup.root,
     terrainViewer.smoothingField,
+    quality.root,
   ]);
   // 표시 옵션은 3D 모델 토글과 포인트 슬라이더를 한 칸에 둔다(슬라이더가 맨 아래).
   const displayGroup = buildSurfaceGroup(L("B04_Surface_Group_Display"), [
@@ -460,6 +441,7 @@ export async function renderB04Surface(root: HTMLElement): Promise<void> {
     if (!projectId) return;
     terrainViewer.setSelection(filterGroup.select.value, methodGroup.select.value);
     terrainViewer.render(projectId, models);
+    quality.show(projectId, findSelectedModel());
     confirmButton.disabled = !findSelectedModel();
   }
 

resources/tester/test_762_1_refine.py

@@ -0,0 +1,80 @@
+"""762-1 지면점 다듬기(SMRF 꼴) · 제안 흐름 스위치 · 762-2 품질 지표 지킴 시험.
+
+합성 비탈(17°) + 능선(높이 6 m · 폭 약 16 m) + 덤불 혹(3 × 3 m · 1.5 m 뜸) — 필터가
+덤불까지 지면으로 고른 판 → 다듬기가 혹은 지우고 능선 · 비탈은 남김.
+"""
+
+import numpy as np
+import pytest
+
+import B04_PreProcess.B04_PreProcess_Engine_Ground as ground
+from B04_PreProcess.B04_PreProcess_Engine_Filter_Refine import refine_ground_mask
+from B04_PreProcess.B04_PreProcess_Engine_Quality import spike_stats
+
+RNG = np.random.default_rng(7621)
+
+
+def _terrain(x: np.ndarray, y: np.ndarray) -> np.ndarray:
+    return 0.3 * x + 6.0 * np.exp(-(((x - 60.0) / 8.0) ** 2)) + 0.02 * y
+
+
+def _scene() -> tuple[np.ndarray, np.ndarray, np.ndarray]:
+    count = 120 * 120 * 4
+    x = RNG.uniform(0, 120, count)
+    y = RNG.uniform(0, 120, count)
+    z = _terrain(x, y) + RNG.normal(0, 0.03, count)
+    bush = np.zeros(count, dtype=bool)
+    for cx, cy in RNG.uniform(10, 110, (15, 2)):
+        if abs(cx - 60) < 12:  # 능선 위 덤불은 따로 가름 — 여기선 비탈 · 골 덤불
+            cx += 25
+        inside = (np.abs(x - cx) < 1.5) & (np.abs(y - cy) < 1.5)
+        z[inside] += 1.5
+        bush |= inside
+    ridge = (np.abs(x - 60) < 4) & ~bush
+    return np.column_stack((x, y, z)), bush, ridge
+
+
+def test_refine_drops_bushes_keeps_ridge() -> None:
+    xyz, bush, ridge = _scene()
+    kept = refine_ground_mask(xyz, np.ones(len(xyz), dtype=bool))
+    print(kept[bush].mean(), kept[ridge].mean(), kept[~bush].mean())
+    assert kept[bush].mean() < 0.1  # 덤불 혹 90 % 넘게 지움
+    assert kept[ridge].mean() > 0.95  # 능선 안 깎음
+    assert kept[~bush].mean() > 0.97  # 땅 점 거의 그대로
+
+
+def test_refine_never_adds_points() -> None:
+    xyz, _, _ = _scene()
+    mask = RNG.random(len(xyz)) < 0.5
+    kept = refine_ground_mask(xyz, mask)
+    assert not (kept & ~mask).any()
+    assert refine_ground_mask(xyz, np.zeros(len(xyz), dtype=bool)).sum() == 0
+
+
+def test_refine_flow_switch(monkeypatch: pytest.MonkeyPatch) -> None:
+    xyz, _, _ = _scene()
+    bounds = np.array([[0, 120], [0, 120], [0, 50]], dtype=np.float64)
+    data = {"xyz": xyz, "bounds": bounds, "classification": np.full(len(xyz), 2, np.uint8)}
+    # 기본 끔 = 지금 길 그대로
+    assert ground.resolve_auto_source_filters(data) == ["classification", "ptd"]
+    monkeypatch.setattr(ground, "SURFACE_REFINE_FLOW", True)
+    assert ground.resolve_auto_source_filters(data) == ["classification_refine", "ptd_refine"]
+    data["classification"][:] = 0
+    assert ground.resolve_auto_source_filters(data) == ["ptd_refine"]
+    assert ground.base_filter("ptd_refine") == "ptd"
+    assert ground.base_filter("ptd") is None
+    assert {"classification_refine", "ptd_refine"} <= set(ground.available_filters())
+    # 바탕 마스크를 넘기면 바탕 필터를 다시 돌리지 않고 다듬기만
+    base = np.ones(len(xyz), dtype=bool)
+    refined = ground.run_ground_filter("ptd_refine", data, base)
+    assert refined.sum() < base.sum()
+
+
+def test_spike_stat_sees_bumps() -> None:
+    yy, xx = np.mgrid[0:100, 0:100].astype(float)
+    z = 0.3 * xx
+    valid = np.ones(z.shape, dtype=bool)
+    assert spike_stats(z, valid)["spike_pct"] == 0.0
+    z[40:42, 40:42] += 1.5
+    stats = spike_stats(z, valid)
+    assert stats["spike_count"] == 1 and stats["spike_pct"] > 0