refactor(767-1): 안 쓰는 지면 필터 · 서피스 지움 — 필터 넷(CSF · 격자 최저 · PMF · RANSAC) · 표현 셋(TIN + Taubin 평활 · NURBS · meshfree) · 수치지형도 비교 방식 다섯 · 안 부르는 길 정리

- 필터 = 업체 분류 · 점진 TIN · 둘의 다듬기 판만 · 분석 허용 목록 = Ground 등록부(available_filters) 한 벌
- 표현 = DTM · implicit 만 · 평활 = DTM 만 · config 기본 목록 · PRECOMPUTE · 설정 사전 맞춤(설정 지문 바뀜 — 768 에서 다시 지음)
- 등고선 = 격자 · 굳힌 격자만(삼각망 굽기 · 외곽 맞춤 · implicit RBF 폴백 걷음) · 표본기 · 경로 찾기 tin · nurbs · meshfree 갈래 걷음
- 수치지형도 = 라플라스만(TIN 둘 · TPS · ANUDEM형 · 다중해상도 걷음) · B04 단추 하나
- 안 부르는 길 — B04 JSON 점군 GET · ground-stats GET + TS · B07 옛 계획평면도(라이다) 짓기 · lidar_source · LIDAR_ID · plan_lidar 종류 이름 · SOURCE_FILTER_LABELS · MODEL_METHODS(py) · MESH_* 설정
- 옛 manifest 의 지운 갈래 줄은 DB 로 되살리지 않음 · 옛 csf 행은 미리보기 · 등고선 그대로 열림 · 지운 이름 요청 = 400
- 낡은 글 — Ground · Pipeline 머리 · WF1 「없으면 csf」 · B07 음영 기본 「csf」 → 빈 값 · 등고선 폴백 「csf」 → 404 · B04 3D 보기 처음값 · B05 Schema · B06 Prebuild · 씨닝 설명
- 시험 지움 2 — test_b04_contour_raster · test_b04_contour_equivalence(지운 삼각망 굽기 · 없는 프로젝트를 봐 건너뛰던 것)
- 시험 덜어냄 — test_b04_ground_filters CSF · 격자 최저 네 경우(지운 필터)
- 시험 고침 — 허용 목록 · 형 검사에 걸린 csf · tin · grid_min_z · meshfree · tin_sheet 견본을 ptd · implicit · laplace 로(54_22 · 238_1 · 443 · 342_2 · 346_3 · 351_1 · 442_2 · 271 + helper · 344_1 · 133_1 · 343_2 · 637 · 754_1 · 763_1 깨진 tin → 깨진 implicit · 763_3 · b05 둘 · b06 · sheet_surface TIN(도엽선) 경우 지움) · 이름만 바꿈(273_2 · 343_1 · 343_3 · 624_1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-11-03 16:35:12 +09:00
co-authored by Claude Opus 5.5
parent 462a2f50fa
commit 11c75f022a
62 changed files with 232 additions and 2600 deletions
@@ -97,12 +97,6 @@ export interface SurfacePointCloudSampleResponse {
rgb?: [number, number, number][];
}
export interface SurfaceGroundStatsResponse {
status: string;
project_id: string;
filters: Record<string, Record<string, unknown>>;
}
export interface SurfaceStatusResponse {
project_id: string;
status: "pending" | "in_progress" | "completed" | "failed";
@@ -298,14 +292,6 @@ export async function fetchConfirmedSurface(projectId: string): Promise<SurfaceC
});
}
export async function fetchSurfaceGroundStats(
projectId: string,
): Promise<SurfaceGroundStatsResponse> {
return requestJson<SurfaceGroundStatsResponse>(`/projects/${projectId}/surface/ground-stats`, {
method: "GET",
});
}
export async function fetchSurfaceStatus(projectId: string): Promise<SurfaceStatusResponse> {
return requestJson<SurfaceStatusResponse>(`/projects/${projectId}/surface/status`, {
method: "GET",
+7 -3
View File
@@ -14,12 +14,14 @@ from typing import Any
import numpy as np
from B04_PreProcess.B04_PreProcess_Engine_Ground import (
available_filters,
base_filter,
build_ground_masks,
resolve_auto_source_filters,
run_ground_filter,
summarize_masks,
)
from B04_PreProcess.B04_PreProcess_Engine_ModelBuild import BUILDERS
from B04_PreProcess.B04_PreProcess_Engine_Pipeline import build_all_terrain_models
from B04_PreProcess.B04_PreProcess_Engine_Structurize import structurize_las
from common_util.common_util_atomic import atomic_write_npz
@@ -608,8 +610,10 @@ def _collect_analysis_result(
# manifest에서 저장된 모델별 정보 추출 (필터별 대표 모델)
models: list[dict[str, Any]] = []
for filter_key, filter_entry in manifest.get("source_filters", {}).items():
if filter_key not in available_filters():
continue # 지운 갈래(608 · csf 등) 옛 줄은 DB 로 되살리지 않음
for method, meta in filter_entry.get("methods", {}).items():
if meta.get("status") != "completed":
if meta.get("status") != "completed" or method not in BUILDERS:
continue
model_file = meta.get("model_file")
model_path = (models_dir / model_file) if model_file else None
@@ -618,11 +622,11 @@ def _collect_analysis_result(
layers.append(
{
"layer_name": f"{method}_{filter_key}_preview",
"geometry_type": "MESH" if method != "meshfree" else "POINTCLOUD",
"geometry_type": "MESH",
"file_path": _relative_to_project(
project_root, models_dir / meta["preview_file"]
),
"file_format": "glb" if method != "meshfree" else "ply",
"file_format": "glb",
}
)
models.append(
+14 -327
View File
@@ -1,16 +1,13 @@
"""B04 등고선 추출 엔진.
5종 표현(regular_grid/triangular_mesh/bspline_surface/local_rbf_height_field/
meshfree_surfels)의 npz 모델에서 표고 격자를 환원하고, marching squares로
지정 간격 등고선 라인을 추출한다. DTM valid_mask를 footprint로 사용해
경계 누출을 차단한다.
표현(regular_grid · local_rbf_height_field 굳힌 격자)의 표고 격자에서 marching
squares로 지정 간격 등고선 라인을 추출한다. 격자 valid_mask가 경계 누출을 막는다.
"""
from pathlib import Path
from typing import Any
import numpy as np
from scipy.interpolate import RBFInterpolator
from skimage import measure
# 등고선 캐시 형식/추출 규칙이 바뀔 때 증가시킨다.
@@ -139,233 +136,6 @@ def contour_cache_is_stale(contour_path: Path, model_path: Path) -> bool:
return any(path.is_file() and path.stat().st_mtime_ns > built for path in sources)
def _load_footprint_mask(
model_npz_path: Path, x_coords: np.ndarray, y_coords: np.ndarray
) -> np.ndarray | None:
"""같은 source filter의 DTM valid_mask를 현재 격자에 최근접 리샘플한다."""
stem = Path(model_npz_path).stem
if stem.endswith("_smooth"):
stem = stem[:-7]
parts = stem.split("_", 1)
if len(parts) < 2:
return None
filter_key = parts[1]
dtm_path = Path(model_npz_path).parent / f"dtm_{filter_key}.npz"
if not dtm_path.exists():
return None
try:
with np.load(dtm_path) as d:
dtm_x = np.asarray(d["x"]).ravel()
dtm_y = np.asarray(d["y"]).ravel()
dtm_mask = np.asarray(d["valid_mask"], dtype=bool)
except Exception:
return None
if len(dtm_x) < 2 or len(dtm_y) < 2:
return None
def _nearest_idx(axis: np.ndarray, coords: np.ndarray) -> np.ndarray:
ascending = bool(axis[0] <= axis[-1])
a = axis if ascending else axis[::-1]
idx = np.clip(np.searchsorted(a, coords), 1, len(a) - 1)
idx = np.where(np.abs(a[idx - 1] - coords) <= np.abs(a[idx] - coords), idx - 1, idx)
return idx if ascending else (len(axis) - 1 - idx)
xi = _nearest_idx(dtm_x, np.asarray(x_coords, dtype=np.float64))
yi = _nearest_idx(dtm_y, np.asarray(y_coords, dtype=np.float64))
return dtm_mask[np.ix_(yi, xi)]
def _apply_footprint(
model_npz_path: Path, x_coords: np.ndarray, y_coords: np.ndarray, valid_mask: np.ndarray
) -> np.ndarray:
"""valid_mask에 DTM footprint를 교집합으로 적용한다 (형상 다르면 최근접 리샘플)."""
fp = _load_footprint_mask(model_npz_path, x_coords, y_coords)
if fp is not None:
if fp.shape == valid_mask.shape:
return valid_mask & fp
from scipy.ndimage import zoom
zoom_y = valid_mask.shape[0] / fp.shape[0]
zoom_x = valid_mask.shape[1] / fp.shape[1]
fp_resized = zoom(fp.astype(float), (zoom_y, zoom_x), order=0) > 0.5
if fp_resized.shape == valid_mask.shape:
return valid_mask & fp_resized
return valid_mask
def _tin_face_coverage_mask(
vertices: np.ndarray, faces: np.ndarray, xx: np.ndarray, yy: np.ndarray
) -> np.ndarray:
"""저장된 TIN 면이 실제로 덮는 XY 영역만 True로 반환한다."""
vertices = np.asarray(vertices)
faces = np.asarray(faces, dtype=np.int64)
if vertices.ndim != 2 or vertices.shape[1] < 2 or not len(faces):
return np.zeros(xx.shape, dtype=bool)
edges = np.vstack((faces[:, [0, 1]], faces[:, [1, 2]], faces[:, [2, 0]]))
edges = np.sort(edges, axis=1)
unique_edges, counts = np.unique(edges, axis=0, return_counts=True)
boundary_edges = unique_edges[counts == 1]
if not len(boundary_edges):
return np.zeros(xx.shape, dtype=bool)
import affine
import rasterio.features
from shapely import get_parts, linestrings, polygonize
boundary_lines = linestrings(vertices[boundary_edges, :2])
polygons = list(get_parts(polygonize(boundary_lines)))
if not polygons:
return np.zeros(xx.shape, dtype=bool)
x_coords = xx[0, :]
y_coords = yy[:, 0]
dx = float(x_coords[1] - x_coords[0]) if len(x_coords) > 1 else 1.0
dy = float(y_coords[1] - y_coords[0]) if len(y_coords) > 1 else 1.0
transform = affine.Affine(dx, 0.0, x_coords[0] - dx / 2.0, 0.0, dy, y_coords[0] - dy / 2.0)
mask = rasterio.features.rasterize(
polygons, out_shape=xx.shape, transform=transform, fill=0, default_value=1, dtype="uint8"
)
return mask.astype(bool)
def _grid_axes(x_min: float, x_max: float, y_min: float, y_max: float, target_grid_m: float):
cols = max(2, int(np.ceil((x_max - x_min) / target_grid_m)) + 1)
rows = max(2, int(np.ceil((y_max - y_min) / target_grid_m)) + 1)
x_coords = np.linspace(x_min, x_max, cols, dtype=np.float32)
y_coords = np.linspace(y_min, y_max, rows, dtype=np.float32)
return x_coords, y_coords
# 후보 셀 배열이 한 번에 커지지 않도록 삼각형을 나눠 처리하는 단위.
_RASTER_CHUNK = 200_000
# 격자점이 변 위에 놓였을 때 양쪽 삼각형 모두에서 탈락하지 않도록 두는 여유.
_BARYCENTRIC_EPS = 1e-9
def rasterize_triangle_mesh(
vertices: np.ndarray,
triangles: np.ndarray,
x_coords: np.ndarray,
y_coords: np.ndarray,
) -> np.ndarray:
"""삼각망을 규칙격자에 직접 래스터화해 표고 격자를 만든다.
삼각형마다 자기가 덮는 격자 셀만 무게중심 선형보간으로 채우므로, 질의점별
위치탐색(scipy `find_simplex`)이 사라진다. 삼각망 밖 셀은 NaN으로 남아
기존 `griddata` 결과와 같은 유효 판정을 유지한다.
"""
rows, cols = len(y_coords), len(x_coords)
z_grid = np.full((rows, cols), np.nan, dtype=np.float64)
vertices = np.asarray(vertices, dtype=np.float64)
triangles = np.asarray(triangles, dtype=np.int64)
if not len(triangles) or vertices.ndim != 2 or vertices.shape[1] < 3:
return z_grid
axis_x = np.asarray(x_coords, dtype=np.float64)
axis_y = np.asarray(y_coords, dtype=np.float64)
def _index_range(axis: np.ndarray, low: np.ndarray, high: np.ndarray, size: int):
"""축 배열을 직접 탐색해 [low, high]를 덮는 인덱스 구간을 낸다.
`(좌표 - 원점) / 간격` 식은 쓰지 않는다. `_grid_axes`가 float32 축을 만드는데
UTM 좌표(18만대)에서 float32 해상도가 0.015625m라 격자 간격이 0.984~1.0으로
흔들리고, 등간격으로 환산하면 최대 0.81셀까지 어긋난다. 삼각형이 평균 0.3m라
그만큼 어긋나면 격자점을 통째로 놓친다 (2026-08-17 실측).
"""
ascending = size < 2 or axis[-1] >= axis[0]
sorted_axis = axis if ascending else axis[::-1]
lo = np.searchsorted(sorted_axis, low, side="left")
hi = np.searchsorted(sorted_axis, high, side="right") - 1
if ascending:
return lo, hi
return size - 1 - hi, size - 1 - lo
for begin in range(0, len(triangles), _RASTER_CHUNK):
chunk = triangles[begin : begin + _RASTER_CHUNK]
corners = vertices[chunk]
ax, ay, az = corners[:, 0, 0], corners[:, 0, 1], corners[:, 0, 2]
bx, by, bz = corners[:, 1, 0], corners[:, 1, 1], corners[:, 1, 2]
cx, cy, cz = corners[:, 2, 0], corners[:, 2, 1], corners[:, 2, 2]
denom = (by - cy) * (ax - cx) + (cx - bx) * (ay - cy)
col_lo, col_hi = _index_range(
axis_x, np.minimum(np.minimum(ax, bx), cx), np.maximum(np.maximum(ax, bx), cx), cols
)
row_lo, row_hi = _index_range(
axis_y, np.minimum(np.minimum(ay, by), cy), np.maximum(np.maximum(ay, by), cy), rows
)
# 클립 전에 격자 밖 삼각형을 걸러야 경계에 눌러붙지 않는다.
selected = (
np.isfinite(denom)
& (np.abs(denom) > 1e-12)
& (col_hi >= col_lo)
& (row_hi >= row_lo)
& (col_hi >= 0)
& (row_hi >= 0)
& (col_lo <= cols - 1)
& (row_lo <= rows - 1)
)
if not selected.any():
continue
col_lo = np.clip(col_lo[selected], 0, cols - 1).astype(np.int64)
col_hi = np.clip(col_hi[selected], 0, cols - 1).astype(np.int64)
row_lo = np.clip(row_lo[selected], 0, rows - 1).astype(np.int64)
row_hi = np.clip(row_hi[selected], 0, rows - 1).astype(np.int64)
width = col_hi - col_lo + 1
counts = width * (row_hi - row_lo + 1)
total = int(counts.sum())
if total <= 0:
continue
# 삼각형별 후보 셀 수가 제각각이라 반복 인덱스를 펼쳐서 한 번에 계산한다.
starts = np.concatenate(([0], np.cumsum(counts)[:-1]))
local = np.arange(total, dtype=np.int64) - np.repeat(starts, counts)
width_rep = np.repeat(width, counts)
offset_row = local // width_rep
cell_row = np.repeat(row_lo, counts) + offset_row
cell_col = np.repeat(col_lo, counts) + (local - offset_row * width_rep)
point_x = axis_x[cell_col]
point_y = axis_y[cell_row]
ax_r, ay_r, az_r = (
np.repeat(ax[selected], counts),
np.repeat(ay[selected], counts),
np.repeat(az[selected], counts),
)
bx_r, by_r, bz_r = (
np.repeat(bx[selected], counts),
np.repeat(by[selected], counts),
np.repeat(bz[selected], counts),
)
cx_r, cy_r, cz_r = (
np.repeat(cx[selected], counts),
np.repeat(cy[selected], counts),
np.repeat(cz[selected], counts),
)
denom_r = np.repeat(denom[selected], counts)
weight_a = ((by_r - cy_r) * (point_x - cx_r) + (cx_r - bx_r) * (point_y - cy_r)) / denom_r
weight_b = ((cy_r - ay_r) * (point_x - cx_r) + (ax_r - cx_r) * (point_y - cy_r)) / denom_r
weight_c = 1.0 - weight_a - weight_b
hit = (
(weight_a >= -_BARYCENTRIC_EPS)
& (weight_b >= -_BARYCENTRIC_EPS)
& (weight_c >= -_BARYCENTRIC_EPS)
)
if not hit.any():
continue
z_grid[cell_row[hit], cell_col[hit]] = (
weight_a[hit] * az_r[hit] + weight_b[hit] * bz_r[hit] + weight_c[hit] * cz_r[hit]
)
return z_grid
def extract_contours(
model_npz_path: Path,
representation: str,
@@ -415,102 +185,19 @@ def _extract_loaded(
x_coords, y_coords, z_grid, valid_mask, interval, scene_center=scene_center
)
if representation == "triangular_mesh":
vertices, faces = data["vertices"], data["faces"]
x_min, x_max = float(np.min(vertices[:, 0])), float(np.max(vertices[:, 0]))
y_min, y_max = float(np.min(vertices[:, 1])), float(np.max(vertices[:, 1]))
x_coords, y_coords = _grid_axes(x_min, x_max, y_min, y_max, target_grid_m)
xx, yy = np.meshgrid(x_coords, y_coords)
# 저장된 faces를 그대로 쓴다 — griddata는 정점을 다시 삼각분할하느라 긴 변
# 제거·외곽 클리핑이 반영되지 않은 면까지 되살렸다 (PLAN 2026-08-17).
z_grid = rasterize_triangle_mesh(vertices, faces, x_coords, y_coords)
face_mask = _tin_face_coverage_mask(vertices, faces, xx, yy)
valid_mask = np.isfinite(z_grid) & face_mask
valid_mask = _apply_footprint(model_npz_path, x_coords, y_coords, valid_mask)
return extract_contours_from_grid(
x_coords, y_coords, z_grid, valid_mask, interval, scene_center=scene_center
)
if representation == "bspline_surface":
# 모델 빌더와 같은 곡면을 다시 만든다 — 평활 계수·발산 방어를 공유한다.
from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
evaluate_nurbs_spline,
fit_nurbs_spline,
)
control_x, control_y, control_z = data["control_x"], data["control_y"], data["control_z"]
degree = int(data["degree"][0])
spline, z_range = fit_nurbs_spline(control_x, control_y, control_z, degree)
x_coords, y_coords = _grid_axes(
float(control_x[0]),
float(control_x[-1]),
float(control_y[0]),
float(control_y[-1]),
target_grid_m,
)
z_grid = evaluate_nurbs_spline(
spline, y_coords, x_coords, z_range, Path(model_npz_path).stem
).astype(np.float32)
valid_mask = _apply_footprint(
model_npz_path, x_coords, y_coords, np.ones_like(z_grid, dtype=bool)
)
return extract_contours_from_grid(
x_coords, y_coords, z_grid, valid_mask, interval, scene_center=scene_center
)
if representation == "local_rbf_height_field":
# 굳힌 격자(641-1)가 있으면 그 면 그대로 — 등고선 = 서버 표본 · 종횡단과 같은 면(643)
# 굳힌 격자(641-1) 그대로 — 등고선 = 서버 표본 · 종횡단과 같은 면(643) · dtm 늘 먼저(754-1)
from common_util.common_util_surface_sampler import implicit_grid_npz
filter_key = model_npz_path.stem.removeprefix("implicit_")
if (model_npz_path.parent / f"dtm_{filter_key}.npz").is_file():
from common_util.common_util_surface_sampler import implicit_grid_npz
with np.load(implicit_grid_npz(model_npz_path.parent, filter_key)) as grid:
return extract_contours_from_grid(
grid["x"],
grid["y"],
grid["z"],
grid["valid_mask"],
interval,
scene_center=scene_center,
)
centers_xy, center_z = data["centers_xy"], data["center_z"]
smoothing = float(data["smoothing"][0])
interpolator = RBFInterpolator(
centers_xy.astype(np.float64),
center_z.astype(np.float64),
neighbors=min(64, len(centers_xy)),
smoothing=smoothing,
kernel="thin_plate_spline",
)
x_min, x_max = float(np.min(centers_xy[:, 0])), float(np.max(centers_xy[:, 0]))
y_min, y_max = float(np.min(centers_xy[:, 1])), float(np.max(centers_xy[:, 1]))
x_coords, y_coords = _grid_axes(x_min, x_max, y_min, y_max, target_grid_m)
xx, yy = np.meshgrid(x_coords, y_coords)
z_values = interpolator(np.column_stack([xx.ravel(), yy.ravel()])).astype(np.float32)
z_grid = z_values.reshape(len(y_coords), len(x_coords))
valid_mask = _apply_footprint(
model_npz_path, x_coords, y_coords, np.ones_like(z_grid, dtype=bool)
)
return extract_contours_from_grid(
x_coords, y_coords, z_grid, valid_mask, interval, scene_center=scene_center
)
if representation == "meshfree_surfels":
from scipy.spatial import Delaunay
points = data["points"]
x_min, x_max = float(np.min(points[:, 0])), float(np.max(points[:, 0]))
y_min, y_max = float(np.min(points[:, 1])), float(np.max(points[:, 1]))
x_coords, y_coords = _grid_axes(x_min, x_max, y_min, y_max, target_grid_m)
# Delaunay 1회로 삼각분할과 hull 판정을 함께 해결한다. 기존에는 griddata가
# 내부에서 한 번, hull 판정용으로 또 한 번 만들었다 (PLAN 2026-08-17).
triangulation = Delaunay(np.asarray(points[:, :2], dtype=np.float64))
z_grid = rasterize_triangle_mesh(points, triangulation.simplices, x_coords, y_coords)
# hull 밖은 어느 삼각형에도 안 닿아 NaN으로 남는다 = 기존 hull_inside와 같다.
valid_mask = np.isfinite(z_grid)
valid_mask = _apply_footprint(model_npz_path, x_coords, y_coords, valid_mask)
return extract_contours_from_grid(
x_coords, y_coords, z_grid, valid_mask, interval, scene_center=scene_center
)
with np.load(implicit_grid_npz(model_npz_path.parent, filter_key)) as grid:
return extract_contours_from_grid(
grid["x"],
grid["y"],
grid["z"],
grid["valid_mask"],
interval,
scene_center=scene_center,
)
raise ValueError(f"지원하지 않는 표현 방식입니다: {representation}")
@@ -1,122 +0,0 @@
"""B04 CSF(Cloth Simulation Filter) 지면 필터.
Pure NumPy 기반 CSF. 지형을 반전시킨 뒤 가상의 천을 중력으로 낙하시켜
원 지면(반전 최하단)에 밀착시키고, 천과의 오차가 임계값 이내인 포인트를
지면으로 분류한다.
"""
import math
from typing import Any
import numpy as np
from config.config_system import (
SURFACE_CSF_CLASS_THRESHOLD_M,
SURFACE_CSF_CLOTH_RESOLUTION_M,
SURFACE_CSF_ITERATIONS,
SURFACE_CSF_MAX_ITERATIONS,
SURFACE_CSF_RIGIDNESS,
SURFACE_CSF_SLOPE_SMOOTH,
SURFACE_CSF_SLOPE_SMOOTH_THRESHOLD_M,
SURFACE_CSF_TIME_STEP,
)
# rigidness 단계별 스프링 완화 계수 (1: 산악 밀착 ~ 3: 부드럽게 덮음)
_RIGIDNESS_SPRING_COEFF = {1: 0.25, 2: 0.45, 3: 0.65}
# 중력 하강 계수 (time_step에 곱해 반복당 낙하량 산출)
_GRAVITY_BASE = 9.8 * 0.05
def filter_csf(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
cloth_resolution: float = SURFACE_CSF_CLOTH_RESOLUTION_M,
rigidness: int = SURFACE_CSF_RIGIDNESS,
time_step: float = SURFACE_CSF_TIME_STEP,
class_threshold: float = SURFACE_CSF_CLASS_THRESHOLD_M,
iterations: int = SURFACE_CSF_ITERATIONS,
slope_smooth: bool = SURFACE_CSF_SLOPE_SMOOTH,
slope_smooth_threshold: float = SURFACE_CSF_SLOPE_SMOOTH_THRESHOLD_M,
max_iterations: int = SURFACE_CSF_MAX_ITERATIONS,
) -> np.ndarray:
"""CSF로 지면 포인트의 불리언 마스크를 반환한다."""
if not math.isfinite(cloth_resolution) or cloth_resolution <= 0:
raise ValueError("천 해상도는 0보다 큰 유한한 값이어야 합니다.")
if rigidness not in _RIGIDNESS_SPRING_COEFF:
raise ValueError("rigidness는 1, 2, 3 중 하나여야 합니다.")
if not math.isfinite(time_step) or time_step <= 0:
raise ValueError("time_step은 0보다 큰 유한한 값이어야 합니다.")
if not math.isfinite(class_threshold) or class_threshold < 0:
raise ValueError("분류 임계값은 0 이상의 유한한 값이어야 합니다.")
if iterations <= 0:
raise ValueError("반복 횟수는 1 이상이어야 합니다.")
if max_iterations < iterations:
raise ValueError("반복 상한은 정착 여유 반복 수보다 작을 수 없습니다.")
xyz = np.asarray(structured_data["xyz"], dtype=np.float64)
if xyz.ndim != 2 or xyz.shape[1] != 3:
raise ValueError("xyz 배열은 (N, 3) 형태여야 합니다.")
if xyz.shape[0] == 0:
return np.zeros(0, dtype=bool)
xs, ys, zs = xyz[:, 0], xyz[:, 1], xyz[:, 2]
# 1. 지형 반전 — 지표면 추출을 위해 높이를 뒤집는다.
z_max = float(np.max(zs))
inverted_zs = (z_max - zs).astype(np.float32)
# 2. 2D 가상 천 격자 설정 (바운더리 밀착 매핑)
x_min, x_max = float(np.min(xs)), float(np.max(xs))
y_min, y_max = float(np.min(ys)), float(np.max(ys))
cols = int(np.ceil((x_max - x_min) / cloth_resolution)) + 1
rows = int(np.ceil((y_max - y_min) / cloth_resolution)) + 1
# 천 노드 초기 높이 — 반전 지형 최고점보다 약간 높은 곳에서 낙하 시작
start_height = float(np.max(inverted_zs)) + 1.0
cloth_z = np.full((rows, cols), start_height, dtype=np.float32)
# 3. 격자 충돌 타겟 구성 (Drape Target)
collision_grid = np.full((rows, cols), -np.inf, dtype=np.float32)
gx = np.clip(((xs - x_min) / cloth_resolution).astype(np.int32), 0, cols - 1)
gy = np.clip(((ys - y_min) / cloth_resolution).astype(np.int32), 0, rows - 1)
np.maximum.at(collision_grid, (gy, gx), inverted_zs)
collision_grid[collision_grid == -np.inf] = 0.0
# 4. 천 시뮬레이션 반복 루프 (물리 하강)
# 반복 수는 현장 기복에서 뽑는다 — 고정 횟수는 곧 고정 하강 예산이라
# (0.3185m × 150회 = 47.8m) 기복이 그보다 큰 산악지에서 천이 지면에 닿지
# 못하고 공중에 멈춘다. 낙하 구간 + 스프링 정착 여유(iterations)로 잡는다.
gravity = _GRAVITY_BASE * time_step
spring_coeff = _RIGIDNESS_SPRING_COEFF[rigidness]
required_drop = start_height - float(np.min(collision_grid))
total_iterations = int(math.ceil(max(required_drop, 0.0) / gravity)) + iterations
total_iterations = min(total_iterations, max_iterations)
for _ in range(total_iterations):
cloth_z -= gravity
cloth_z = np.maximum(cloth_z, collision_grid)
# 노드 간 스프링 제약 완화 (가로/세로 인접 교정)
diff_h = cloth_z[:, 1:] - cloth_z[:, :-1]
correction_h = diff_h * spring_coeff * 0.5
cloth_z[:, :-1] += correction_h
cloth_z[:, 1:] -= correction_h
diff_v = cloth_z[1:, :] - cloth_z[:-1, :]
correction_v = diff_v * spring_coeff * 0.5
cloth_z[:-1, :] += correction_v
cloth_z[1:, :] -= correction_v
cloth_z = np.maximum(cloth_z, collision_grid)
# 5. 시뮬레이션 천 높이와 원본 대조 → 오차 이내면 지면
simulated_inverted_z = cloth_z[gy, gx]
height_diff = np.abs(inverted_zs - simulated_inverted_z)
mask = height_diff <= class_threshold
# 6. 수목 노이즈 2차 필터 보정 — 셀 지면 후보(반전고 최대 = 원표고 최저)보다
# 얼마나 위에 떠 있는지를 본다. 반전 공간에서 높이는 (지면 후보 - 자기 반전고)다.
# 반대로 빼면 항상 0 이하가 나와 조건이 늘 참이 된다(= 필터가 무력화).
if slope_smooth:
local_ground_z = collision_grid[gy, gx]
mask = mask & ((local_ground_z - inverted_zs) < slope_smooth_threshold)
return np.asarray(mask, dtype=bool)
@@ -1,50 +0,0 @@
"""B04 grid minimum-Z 지면 필터."""
import math
from typing import Any
import numpy as np
from config.config_system import SURFACE_GRID_CELL_SIZE_M, SURFACE_GRID_HEIGHT_THRESHOLD_M
def filter_grid_min_z(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
cell_size: float = SURFACE_GRID_CELL_SIZE_M,
height_threshold: float = SURFACE_GRID_HEIGHT_THRESHOLD_M,
) -> np.ndarray:
"""격자 최저 표면에서 높이 임계값 이내인 포인트의 불리언 마스크를 반환한다."""
if not math.isfinite(cell_size) or cell_size <= 0:
raise ValueError("격자 크기는 0보다 큰 유한한 값이어야 합니다.")
if not math.isfinite(height_threshold) or height_threshold < 0:
raise ValueError("높이 임계값은 0 이상의 유한한 값이어야 합니다.")
xyz = np.asarray(structured_data["xyz"], dtype=np.float64)
bounds = np.asarray(structured_data["bounds"], dtype=np.float64)
if xyz.ndim != 2 or xyz.shape[1] != 3:
raise ValueError("xyz 배열은 (N, 3) 형태여야 합니다.")
if bounds.shape != (3, 2):
raise ValueError("bounds 배열은 (3, 2) 형태여야 합니다.")
if xyz.shape[0] == 0:
return np.zeros(0, dtype=bool)
x_min, y_min = bounds[0, 0], bounds[1, 0]
x_max, y_max = bounds[0, 1], bounds[1, 1]
grid_width = int(np.ceil((x_max - x_min) / cell_size)) + 2
grid_height = int(np.ceil((y_max - y_min) / cell_size)) + 2
# float32로 담으면 셀 최저점(가장 확실한 지면점)이 반올림 때문에 자기 기준면보다
# 낮아져 height_above < 0 으로 탈락한다 — 원본 좌표와 같은 float64로 둔다.
minimum_z = np.full((grid_height, grid_width), np.inf, dtype=np.float64)
grid_x = np.clip(((xyz[:, 0] - x_min) / cell_size).astype(np.int64), 0, grid_width - 1)
grid_y = np.clip(((xyz[:, 1] - y_min) / cell_size).astype(np.int64), 0, grid_height - 1)
np.minimum.at(minimum_z, (grid_y, grid_x), xyz[:, 2])
# 기준면은 점이 실제로 들어 있는 셀에서만 읽는다 — 3x3 최소필터로 이웃 셀 값을
# 끌어오면 급경사에서 기준면이 경사만큼 파고들어(중앙 1.88m > 임계 1.5m) 지면점이
# 통째로 탈락한다. 빈 셀은 아래 인덱싱에서 참조되지 않으므로 채울 필요가 없다.
height_above = xyz[:, 2] - minimum_z[grid_y, grid_x]
return np.asarray(
(height_above >= 0.0) & (height_above <= height_threshold),
dtype=bool,
)
@@ -1,91 +0,0 @@
"""B04 PMF(Progressive Morphological Filter) 지면 필터.
XY 평면을 격자로 투영해 Z-min 지형 맵을 만든 뒤, 윈도우 폭을 단계적으로
키워가며 형태학적 열림(Opening) 연산으로 수목·구조물을 제거하고, 최종 지면
대비 높이차가 임계값 이내인 포인트를 지면으로 분류한다. Pure NumPy 구현.
"""
import math
from typing import Any
import numpy as np
from numpy.lib.stride_tricks import sliding_window_view
from config.config_system import (
SURFACE_PMF_CELL_SIZE_M,
SURFACE_PMF_INITIAL_WINDOW_SIZE,
SURFACE_PMF_MAX_DISTANCE_M,
SURFACE_PMF_MAX_WINDOW_SIZE,
SURFACE_PMF_SLOPE,
)
def _min_max_window_filter(grid: np.ndarray, w_size: int, mode: str = "min") -> np.ndarray:
"""순수 NumPy 2D 이동 윈도우 최솟값/최댓값 필터 (경계는 edge 패딩)."""
pad_val = w_size // 2
padded = np.pad(grid, pad_val, mode="edge")
windows = sliding_window_view(padded, (w_size, w_size))
if mode == "min":
return np.min(windows, axis=(2, 3))
return np.max(windows, axis=(2, 3))
def filter_pmf(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
cell_size: float = SURFACE_PMF_CELL_SIZE_M,
max_window_size: int = SURFACE_PMF_MAX_WINDOW_SIZE,
slope: float = SURFACE_PMF_SLOPE,
initial_window_size: int = SURFACE_PMF_INITIAL_WINDOW_SIZE,
max_distance: float = SURFACE_PMF_MAX_DISTANCE_M,
) -> np.ndarray:
"""PMF로 지면 포인트의 불리언 마스크를 반환한다."""
if not math.isfinite(cell_size) or cell_size <= 0:
raise ValueError("격자 크기는 0보다 큰 유한한 값이어야 합니다.")
if initial_window_size < 1 or max_window_size < initial_window_size:
raise ValueError("윈도우 크기는 1 이상이고 최대가 초기값 이상이어야 합니다.")
if not math.isfinite(max_distance) or max_distance < 0:
raise ValueError("최대 거리 임계값은 0 이상의 유한한 값이어야 합니다.")
xyz = np.asarray(structured_data["xyz"], dtype=np.float64)
bounds = np.asarray(structured_data["bounds"], dtype=np.float64)
if xyz.ndim != 2 or xyz.shape[1] != 3:
raise ValueError("xyz 배열은 (N, 3) 형태여야 합니다.")
if bounds.shape != (3, 2):
raise ValueError("bounds 배열은 (3, 2) 형태여야 합니다.")
if xyz.shape[0] == 0:
return np.zeros(0, dtype=bool)
xs, ys, zs = xyz[:, 0], xyz[:, 1], xyz[:, 2]
x_min, y_min, z_min = bounds[0, 0], bounds[1, 0], bounds[2, 0]
x_max, y_max = bounds[0, 1], bounds[1, 1]
grid_w = int(np.ceil((x_max - x_min) / cell_size)) + 2
grid_h = int(np.ceil((y_max - y_min) / cell_size)) + 2
z_grid = np.full((grid_h, grid_w), np.inf, dtype=np.float32)
gx = np.clip(((xs - x_min) / cell_size).astype(np.int32), 0, grid_w - 1)
gy = np.clip(((ys - y_min) / cell_size).astype(np.int32), 0, grid_h - 1)
# 1. Z-min 지형 구성
np.minimum.at(z_grid, (gy, gx), zs.astype(np.float32))
z_grid[z_grid == np.inf] = z_min
# 2. 점진적 형태학 필터 (Opening = Dilation of Erosion)
current_grid = z_grid.copy()
w_sizes = []
w = initial_window_size
while w <= max_window_size:
w_sizes.append(w)
w = w * 2 + 1
for w_size in w_sizes:
eroded = _min_max_window_filter(current_grid, w_size, mode="min")
opened = _min_max_window_filter(eroded, w_size, mode="max")
t_dist = min(slope * w_size * cell_size * 0.15 + 0.5, max_distance)
mask_elev = (current_grid - opened) > t_dist
current_grid[mask_elev] = opened[mask_elev]
# 3. 마스크 매핑 및 원본 비교
simulated_z = current_grid[gy, gx]
mask = (zs >= simulated_z - 0.4) & (zs <= simulated_z + max_distance)
return np.asarray(mask, dtype=bool)
@@ -1,116 +0,0 @@
"""B04 RANSAC 지면 필터 (Local plane fitting).
공간을 로컬 격자로 분할하고, 각 격자에서 RANSAC 평면 피팅을 수행해
평면과의 거리가 임계값 이내인 인라이어(지면)를 취합한다. Pure NumPy 구현.
"""
import math
from collections.abc import Callable
from typing import Any
import numpy as np
from config.config_system import (
SURFACE_RANSAC_DISTANCE_THRESHOLD_M,
SURFACE_RANSAC_ITERATIONS,
SURFACE_RANSAC_LOCAL_GRID_SIZE_M,
SURFACE_RANSAC_N,
SURFACE_RANSAC_SEED,
)
def fit_plane_ransac(
points: np.ndarray,
distance_threshold: float = SURFACE_RANSAC_DISTANCE_THRESHOLD_M,
ransac_n: int = SURFACE_RANSAC_N,
num_iterations: int = SURFACE_RANSAC_ITERATIONS,
seed: int = SURFACE_RANSAC_SEED,
) -> np.ndarray:
"""평면 ax+by+cz+d=0을 RANSAC으로 피팅하고 인라이어 마스크를 반환한다."""
n_points = len(points)
if n_points < ransac_n:
return np.ones(n_points, dtype=bool)
best_inliers = np.zeros(n_points, dtype=bool)
max_inlier_count = -1
rng = np.random.default_rng(seed)
for _ in range(num_iterations):
idx = rng.choice(n_points, ransac_n, replace=False)
p0, p1, p2 = points[idx]
normal = np.cross(p1 - p0, p2 - p0)
norm = np.linalg.norm(normal)
if norm < 1e-6:
continue # 세 점이 일직선상 → 스킵
normal = normal / norm
d = -np.dot(normal, p0)
distances = np.abs(np.dot(points, normal) + d)
inliers = distances < distance_threshold
inlier_count = int(np.sum(inliers))
if inlier_count > max_inlier_count:
max_inlier_count = inlier_count
best_inliers = inliers
if max_inlier_count > 0:
return best_inliers
return np.ones(n_points, dtype=bool)
def filter_ransac(
structured_data: dict[str, Any] | np.lib.npyio.NpzFile,
distance_threshold: float = SURFACE_RANSAC_DISTANCE_THRESHOLD_M,
ransac_n: int = SURFACE_RANSAC_N,
num_iterations: int = SURFACE_RANSAC_ITERATIONS,
local_grid_size: float = SURFACE_RANSAC_LOCAL_GRID_SIZE_M,
seed: int = SURFACE_RANSAC_SEED,
progress_callback: Callable[[int], None] | None = None,
) -> np.ndarray:
"""로컬 격자별 RANSAC 평면 분할로 지면 포인트 마스크를 반환한다."""
if not math.isfinite(local_grid_size) or local_grid_size <= 0:
raise ValueError("로컬 격자 크기는 0보다 큰 유한한 값이어야 합니다.")
if ransac_n < 3:
raise ValueError("RANSAC 샘플 수는 3 이상이어야 합니다.")
xyz = np.asarray(structured_data["xyz"], dtype=np.float64)
bounds = np.asarray(structured_data["bounds"], dtype=np.float64)
if xyz.ndim != 2 or xyz.shape[1] != 3:
raise ValueError("xyz 배열은 (N, 3) 형태여야 합니다.")
if bounds.shape != (3, 2):
raise ValueError("bounds 배열은 (3, 2) 형태여야 합니다.")
n_points = len(xyz)
mask = np.zeros(n_points, dtype=bool)
if n_points == 0:
return mask
x_min, y_min = bounds[0, 0], bounds[1, 0]
x_max, y_max = bounds[0, 1], bounds[1, 1]
grid_w = int(np.ceil((x_max - x_min) / local_grid_size)) + 1
grid_h = int(np.ceil((y_max - y_min) / local_grid_size)) + 1
xs, ys = xyz[:, 0], xyz[:, 1]
gx = np.clip(((xs - x_min) / local_grid_size).astype(np.int32), 0, grid_w - 1)
gy = np.clip(((ys - y_min) / local_grid_size).astype(np.int32), 0, grid_h - 1)
grid_indices = gy * grid_w + gx
unique_grids = np.unique(grid_indices)
total_grids = len(unique_grids)
for i, grid_id in enumerate(unique_grids):
cell_points_idx = np.where(grid_indices == grid_id)[0]
if len(cell_points_idx) < ransac_n:
mask[cell_points_idx] = True
continue
cell_inliers = fit_plane_ransac(
xyz[cell_points_idx],
distance_threshold=distance_threshold,
ransac_n=ransac_n,
num_iterations=num_iterations,
seed=seed,
)
mask[cell_points_idx[cell_inliers]] = True
if progress_callback:
progress_callback(int(((i + 1) / total_grids) * 100))
if progress_callback:
progress_callback(100) # 모든 격자 처리 완료 보장
return mask
@@ -1,6 +1,6 @@
"""B04 지면 필터 오케스트레이션.
구조화된 포인트클라우드(structured.npz)에 대해 grid_min_z/csf/pmf/ransac/ptd/classification
구조화된 포인트클라우드(structured.npz)에 대해 ptd(점진 TIN)/classification(업체 분류)
(+ `_refine` 다듬기 판)
필터를 실행하여 지면 마스크 딕셔너리를 만든다. 필터 선택은 config의
source_filters를 따른다.
@@ -16,11 +16,7 @@ from B04_PreProcess.B04_PreProcess_Engine_Filter_Classification import (
has_classified_coverage,
has_classified_ground,
)
from B04_PreProcess.B04_PreProcess_Engine_Filter_CSF import filter_csf
from B04_PreProcess.B04_PreProcess_Engine_Filter_Grid import filter_grid_min_z
from B04_PreProcess.B04_PreProcess_Engine_Filter_PMF import filter_pmf
from B04_PreProcess.B04_PreProcess_Engine_Filter_PTD import filter_ptd
from B04_PreProcess.B04_PreProcess_Engine_Filter_RANSAC import filter_ransac
from B04_PreProcess.B04_PreProcess_Engine_Filter_Refine import refine_ground_mask
from config.config_system import SURFACE_AUTO_FALLBACK_FILTER
@@ -30,10 +26,6 @@ _REFINABLE = ("classification", "ptd")
# 필터 키 → 함수 매핑
_FILTERS = {
"grid_min_z": filter_grid_min_z,
"csf": filter_csf,
"pmf": filter_pmf,
"ransac": filter_ransac,
"ptd": filter_ptd,
"classification": filter_classification,
"classification_refine": lambda data: refine_ground_mask(
@@ -1,7 +1,7 @@
"""B04 지표면 5종 표현 빌더 (TIN/DTM/NURBS/implicit/meshfree).
"""B04 지표면 표현 빌더 (DTM/implicit).
TerrainContext에서 파생 격자·샘플을 받아 각 표현의 모델(npz)과 프리뷰
(GLB/PLY)를 저장하고 메타데이터를 반환한다.
(GLB)를 저장하고 메타데이터를 반환한다.
"""
from contextlib import suppress
@@ -10,7 +10,6 @@ from typing import Any
import numpy as np
from scipy.interpolate import RBFInterpolator
from scipy.spatial import Delaunay
from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
ProgressCallback,
@@ -18,63 +17,14 @@ from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
artifact_size,
atomic_npz,
clip_and_compact_mesh,
evaluate_nurbs_spline,
fit_nurbs_spline,
grid_faces,
grid_vertices,
with_footprint,
write_binary_ply,
write_glb,
)
from common_util.common_util_surface_sampler import implicit_grid_npz
def build_tin(
context: TerrainContext, output_dir: Path, stem: str, progress: ProgressCallback
) -> dict[str, Any]:
progress(5)
points = context.sample(int(context.config["tin_max_input_points"]))
unique_xy, unique_indices = np.unique(points[:, :2], axis=0, return_index=True)
points = points[unique_indices]
if len(points) < 3:
raise ValueError("TIN 생성에 필요한 포인트가 부족합니다.")
progress(25)
faces = np.asarray(Delaunay(unique_xy).simplices, dtype=np.uint32)
if len(faces):
triangle_xy = points[faces, :2]
edges = np.stack(
[
np.linalg.norm(triangle_xy[:, 0] - triangle_xy[:, 1], axis=1),
np.linalg.norm(triangle_xy[:, 1] - triangle_xy[:, 2], axis=1),
np.linalg.norm(triangle_xy[:, 2] - triangle_xy[:, 0], axis=1),
],
axis=1,
)
faces = faces[np.max(edges, axis=1) <= float(context.config["tile_size_meters"]) * 2]
valid_vertices = context.contains_xy(points[:, 0], points[:, 1])
points, faces = clip_and_compact_mesh(points, faces, valid_vertices)
if not len(faces):
raise ValueError("외곽 안쪽 기준 적용 후 TIN 면이 남지 않았습니다.")
progress(65)
model_path = output_dir / f"{stem}.npz"
preview_path = output_dir / f"{stem}_preview.glb"
atomic_npz(model_path, vertices=points, faces=faces)
write_glb(preview_path, points, faces, context.bounds)
progress(100)
return with_footprint(
context,
{
"representation": "triangular_mesh",
"model_file": model_path.name,
"preview_file": preview_path.name,
"preview_media_type": "model/gltf-binary",
"vertex_count": int(len(points)),
"face_count": int(len(faces)),
"artifact_bytes": artifact_size(model_path, preview_path),
},
)
def build_dtm(
context: TerrainContext, output_dir: Path, stem: str, progress: ProgressCallback
) -> dict[str, Any]:
@@ -121,56 +71,6 @@ def build_dtm(
)
def build_nurbs(
context: TerrainContext, output_dir: Path, stem: str, progress: ProgressCallback
) -> dict[str, Any]:
degree = max(1, min(5, int(context.config["nurbs_degree"])))
patch_size = float(context.config["nurbs_patch_size_meters"])
controls = max(degree + 1, int(context.config["nurbs_control_points_per_axis"]))
control_resolution = max(patch_size / max(controls - 1, 1), 0.25)
x_control, y_control, z_control = context.grid(control_resolution)
progress(30)
spline, z_range = fit_nurbs_spline(x_control, y_control, z_control, degree)
x_preview, y_preview, _ = context.preview_grid(
float(context.config["dtm_grid_resolution_meters"])
)
z_preview = evaluate_nurbs_spline(spline, y_preview, x_preview, z_range, stem).astype(
np.float32
)
progress(65)
vertices = grid_vertices(x_preview, y_preview, z_preview)
faces = grid_faces(len(y_preview), len(x_preview))
valid_preview = context.contains_xy(vertices[:, 0], vertices[:, 1])
vertices, faces = clip_and_compact_mesh(vertices, faces, valid_preview)
model_path = output_dir / f"{stem}.npz"
preview_path = output_dir / f"{stem}_preview.glb"
atomic_npz(
model_path,
control_x=x_control,
control_y=y_control,
control_z=z_control,
degree=np.array([degree], np.int16),
patch_size_meters=np.array([patch_size], np.float32),
)
write_glb(preview_path, vertices, faces, context.bounds)
progress(100)
return with_footprint(
context,
{
"representation": "bspline_surface",
"model_file": model_path.name,
"preview_file": preview_path.name,
"preview_media_type": "model/gltf-binary",
"degree": degree,
"control_rows": int(len(y_control)),
"control_columns": int(len(x_control)),
"vertex_count": int(len(vertices)),
"face_count": int(len(faces)),
"artifact_bytes": artifact_size(model_path, preview_path),
},
)
def implicit_point_count(ground_count: int, config: dict[str, Any]) -> int:
"""implicit 중심점 수 = clip(지면점 × 비율, 하한, 상한) · 지면점이 하한보다 적으면 전부."""
minimum = int(config["implicit_min_points"])
@@ -240,57 +140,7 @@ def build_implicit(
)
def build_meshfree(
context: TerrainContext, output_dir: Path, stem: str, progress: ProgressCallback
) -> dict[str, Any]:
points = context.sample(int(context.config["meshfree_max_model_points"]))
points = points[context.contains_xy(points[:, 0], points[:, 1])]
if not len(points):
raise ValueError("외곽 안쪽 기준 적용 후 Meshfree 포인트가 남지 않았습니다.")
resolution = float(context.config["dtm_grid_resolution_meters"])
x_grid, y_grid, z_grid = context.grid(resolution)
dz_dy, dz_dx = np.gradient(z_grid, resolution, resolution)
gx = np.clip(np.searchsorted(x_grid, points[:, 0]), 0, len(x_grid) - 1)
gy = np.clip(np.searchsorted(y_grid, points[:, 1]), 0, len(y_grid) - 1)
normals = np.column_stack([-dz_dx[gy, gx], -dz_dy[gy, gx], np.ones(len(points), np.float32)])
normals /= np.maximum(np.linalg.norm(normals, axis=1, keepdims=True), 1e-9)
progress(55)
preview_max = int(context.config["max_preview_vertices"])
if len(points) > preview_max:
selection = np.linspace(0, len(points) - 1, preview_max, dtype=np.int64)
preview_points, preview_normals = points[selection], normals[selection]
else:
preview_points, preview_normals = points, normals
model_path = output_dir / f"{stem}.npz"
preview_path = output_dir / f"{stem}_preview.ply"
radius = float(context.config["meshfree_point_radius_meters"])
atomic_npz(
model_path,
points=points,
normals=normals.astype(np.float32),
radius=np.array([radius], np.float32),
)
write_binary_ply(preview_path, preview_points, preview_normals, context.bounds)
progress(100)
return with_footprint(
context,
{
"representation": "meshfree_surfels",
"model_file": model_path.name,
"preview_file": preview_path.name,
"preview_media_type": "application/octet-stream",
"point_count": int(len(points)),
"preview_point_count": int(len(preview_points)),
"point_radius_meters": radius,
"artifact_bytes": artifact_size(model_path, preview_path),
},
)
BUILDERS = {
"tin": build_tin,
"dtm": build_dtm,
"nurbs": build_nurbs,
"implicit": build_implicit,
"meshfree": build_meshfree,
}
@@ -1,8 +1,8 @@
"""B04 지표면 모델 공통 컨텍스트 및 메시 유틸리티.
지면 마스크가 적용된 포인트에서 footprint(외곽), 격자, 프리뷰 격자를 만들고,
GLB/PLY 프리뷰 및 npz 모델을 원자적으로 저장하는 공통 기능을 제공한다.
5개 표현(TIN/DTM/NURBS/implicit/meshfree) 빌더가 이 컨텍스트를 공유한다.
GLB 프리뷰 및 npz 모델을 원자적으로 저장하는 공통 기능을 제공한다.
표현(DTM/implicit) 빌더가 이 컨텍스트를 공유한다.
"""
import hashlib
@@ -16,7 +16,6 @@ from typing import Any, Callable
import numpy as np
import trimesh
from scipy import ndimage
from scipy.interpolate import RectBivariateSpline
from skimage.restoration import inpaint_biharmonic
from common_util.common_util_atomic import atomic_write_bytes, atomic_write_npz
@@ -24,8 +23,6 @@ from common_util.common_util_atomic import atomic_write_bytes, atomic_write_npz
logger = logging.getLogger(__name__)
MODEL_VERSION = 1
MODEL_METHODS = ("tin", "dtm", "nurbs", "implicit", "meshfree")
SOURCE_FILTER_LABELS = {"grid_min_z": "Grid Min-Z", "csf": "CSF", "pmf": "PMF"}
ProgressCallback = Callable[[int], None]
# 대용량 포인트 배치 처리 크기
@@ -33,63 +30,6 @@ _BATCH_SIZE = 500_000
# implicit 격자 칸 뽑기 씨앗 — 고정이라 다시 지어도 같은 중심점
IMPLICIT_SAMPLE_SEED = 558
# NURBS 평활 계수 — 제어점당 허용 잔차제곱(m²).
# 0.01(RMS 0.1m)은 요구가 너무 빡빡해 FITPACK이 제어점 수(122x113)보다 많은
# knot(126x117)을 밀어넣고 계수가 발산했다 — csf 표고 1e105, grid_min_z 1e10
# (2026-08-17 실측). 1.0(RMS 1m)이면 knot이 17~35로 떨어지고 세 지면 필터 모두
# 데이터 표고 범위 안에 머문다. NURBS는 평활 곡면 표현이라 이 정도 완화가 맞다.
NURBS_RESIDUAL_PER_CONTROL_POINT = 1.0
def fit_nurbs_spline(
x_control: np.ndarray, y_control: np.ndarray, z_control: np.ndarray, degree: int
) -> tuple[RectBivariateSpline, tuple[float, float]]:
"""제어 격자에 B-spline 곡면을 맞추고, 발산 판정용 표고 허용 범위를 함께 준다.
모델 빌더와 등고선 엔진이 같은 곡면을 각각 만들므로 여기서 한 번에 정의한다.
"""
spline = RectBivariateSpline(
y_control,
x_control,
z_control,
kx=min(degree, len(y_control) - 1),
ky=min(degree, len(x_control) - 1),
s=float(len(x_control) * len(y_control)) * NURBS_RESIDUAL_PER_CONTROL_POINT,
)
z_min = float(np.min(z_control))
z_max = float(np.max(z_control))
margin = max((z_max - z_min) * 0.5, 1.0)
return spline, (z_min - margin, z_max + margin)
def evaluate_nurbs_spline(
spline: RectBivariateSpline,
y_coords: np.ndarray,
x_coords: np.ndarray,
z_range: tuple[float, float],
label: str = "",
) -> np.ndarray:
"""스플라인을 평가하고 표고 허용 범위를 벗어난 값을 잘라낸다.
평활 스플라인은 제어점이 성긴 구석에서 발산할 수 있다. 그대로 두면 float32
캐스팅에서 inf가 되고 프리뷰 색상(nan)과 등고선 레벨 산출(`np.arange`가
"Maximum allowed size exceeded"로 실패)까지 번진다.
"""
z_values = np.asarray(spline(y_coords, x_coords), dtype=np.float64)
low, high = z_range
outside = np.count_nonzero(~np.isfinite(z_values) | (z_values < low) | (z_values > high))
if outside:
logger.warning(
"NURBS 곡면이 표고 범위를 벗어나 잘라냈습니다: %s%d개 셀 (허용 %.1f~%.1fm)",
f"{label} " if label else "",
int(outside),
low,
high,
)
z_values = np.nan_to_num(z_values, nan=low, posinf=high, neginf=low)
return np.clip(z_values, low, high)
# 서명에서 빼는 키 — 지오메트리를 만드는 방법이 아니라 "무엇을 만들지"를 고르는 값들.
# source_filters·precompute를 넣으면 조합을 바꿀 때마다 manifest가 통째로 폐기돼
# 이전에 만들어 둔 조합이 사라진다. manifest는 이미 필터·방식별로 키가 갈린다.
@@ -155,44 +95,6 @@ def write_glb(path: Path, vertices: np.ndarray, faces: np.ndarray, bounds: np.nd
atomic_write_bytes(path, payload)
def write_binary_ply(
path: Path, vertices: np.ndarray, normals: np.ndarray, bounds: np.ndarray
) -> None:
verts = scene_vertices(vertices, bounds)
scene_normals = np.empty_like(normals, dtype=np.float32)
scene_normals[:, 0] = normals[:, 0]
scene_normals[:, 1] = normals[:, 2]
scene_normals[:, 2] = -normals[:, 1]
colors = height_colors(vertices)
dtype = np.dtype(
[
("x", "<f4"),
("y", "<f4"),
("z", "<f4"),
("nx", "<f4"),
("ny", "<f4"),
("nz", "<f4"),
("red", "u1"),
("green", "u1"),
("blue", "u1"),
("alpha", "u1"),
]
)
records = np.empty(len(vertices), dtype=dtype)
records["x"], records["y"], records["z"] = verts.T
records["nx"], records["ny"], records["nz"] = scene_normals.T
records["red"], records["green"], records["blue"], records["alpha"] = colors.T
header = (
"ply\nformat binary_little_endian 1.0\n"
f"element vertex {len(vertices)}\n"
"property float x\nproperty float y\nproperty float z\n"
"property float nx\nproperty float ny\nproperty float nz\n"
"property uchar red\nproperty uchar green\nproperty uchar blue\nproperty uchar alpha\n"
"end_header\n"
).encode("ascii")
atomic_write_bytes(path, header + records.tobytes())
def grid_faces(rows: int, cols: int) -> np.ndarray:
"""정규 격자의 삼각형 면 인덱스를 만든다."""
if rows < 2 or cols < 2:
@@ -1,6 +1,6 @@
"""B04 지표면 모델 파이프라인 오케스트레이터.
세 지면 필터(grid_min_z/csf/pmf)와 다섯 표현(TIN/DTM/NURBS/implicit/meshfree)의
지면 필터(업체 분류 · 점진 TIN · 둘의 다듬기 판)와 표현(DTM/implicit)의
캐시를 만들고 manifest.json을 관리한다. 캐시 유효성 검증, 스무딩/등고선 연동,
동일 출력 폴더의 중복 실행 취소를 포함한다.
"""
@@ -88,10 +88,9 @@ def _cache_contours(
smooth_model_path = output_dir / f"{stem}_smooth.npz"
smooth_meta = metadata.get("smooth", {})
if smooth_model_path.exists() and smooth_meta.get("status") == "completed":
smooth_rep = "regular_grid" if method == "dtm" else "triangular_mesh"
smooth_contours = extract_contours(
smooth_model_path,
representation=smooth_rep,
representation="regular_grid",
interval=interval,
target_grid_m=target_grid_m,
scene_center=None,
@@ -143,10 +142,7 @@ def _cache_contours_safely(
_REPRESENTATIONS = {
"meshfree": "meshfree_surfels",
"dtm": "regular_grid",
"tin": "triangular_mesh",
"nurbs": "bspline_surface",
"implicit": "local_rbf_height_field",
}
@@ -157,17 +153,16 @@ def _cache_is_valid(
"""디스크의 결과 파일과 스무딩 메타데이터가 유효한지 검사한다.
manifest 항목이 완료여야 함 — 설정 지문이 바뀌면 manifest 가 비므로 파일만 남은
옛 결과(평활 없는 implicit · nurbs · meshfree)를 다시 짓는다(요구 558).
옛 결과(평활 없는 implicit)를 다시 짓는다(요구 558).
"""
if entry.get("status") != "completed":
return False
ext = "ply" if method == "meshfree" else "glb"
files_exist = (output_dir / f"{stem}_preview.{ext}").exists() and (
files_exist = (output_dir / f"{stem}_preview.glb").exists() and (
output_dir / f"{stem}.npz"
).exists()
if not files_exist:
return False
if method in config.get("smoothing_methods", ("dtm", "tin")):
if method in config.get("smoothing_methods", ("dtm",)):
smooth_entry = entry.get("smooth", {})
smooth_exist = (output_dir / f"{stem}_smooth.npz").exists() and (
output_dir / f"{stem}_smooth_preview.glb"
@@ -190,14 +185,14 @@ def _build_all_terrain_models(
force: bool = False,
progress: ProgressReporter | None = None,
) -> dict[str, Any]:
"""세 지면 필터와 다섯 표현의 캐시를 만들고 manifest를 반환한다."""
"""지면 필터와 표현의 캐시를 만들고 manifest를 반환한다."""
output_dir.mkdir(parents=True, exist_ok=True)
manifest_path = output_dir / "manifest.json"
filters = tuple(key for key in config["source_filters"] if key in ground_masks)
# dtm 은 늘 맨 앞 — 같은 필터 외곽 · implicit 굳힌 격자 · 등고선 자르기의 바탕이라
# 어느 표현을 지어도 낡았으면(manifest 항목 없음 · 지문 다름) 같이 다시 지음(754-1).
methods = ("dtm", *(key for key in config["precompute"] if key in BUILDERS and key != "dtm"))
# 무거운 implicit 은 기본 필터(첫 필터)에만 — 분류 LAS 의 csf 까지 두 번 짓지 않음(요구 558).
# 무거운 implicit 은 기본 필터(첫 필터)에만 — 둘째 필터엔 안 지음(요구 558).
later_methods = tuple(method for method in methods if method != "implicit")
signature = config_signature(config)
bounds = np.asarray(structured_data["bounds"], dtype=np.float64)
@@ -248,7 +243,7 @@ def _build_all_terrain_models(
interval = float(config.get("contour_interval_meters", 1.0))
contour_path = output_dir / f"contour_{filter_key}_{method}_{interval}m.json"
needs_contours = contour_cache_is_stale(contour_path, output_dir / f"{stem}.npz")
if method in config.get("smoothing_methods", ("dtm", "tin")):
if method in config.get("smoothing_methods", ("dtm",)):
smooth_contour_path = (
output_dir / f"contour_{filter_key}_{method}_smooth_{interval}m.json"
)
@@ -302,7 +297,7 @@ def _build_all_terrain_models(
"error": None,
}
)
if method in config.get("smoothing_methods", ("dtm", "tin")):
if method in config.get("smoothing_methods", ("dtm",)):
original_model_path = output_dir / f"{stem}.npz"
if original_model_path.exists():
try:
@@ -1,57 +1,23 @@
"""도엽등고선 → 표고 격자 보간 방식 모음 (비교용).
"""도엽등고선 → 표고 격자 보간 — 라플라스(거리 비례 초기값 + 이웃 평균 다듬기).
문헌(Hutchinson 1988/89 ANUDEM; Chaplot 2006; Arun 2013 등)은 지형 복잡도·자료 밀도에
따라 우열이 갈리며 단일 최적해가 없다고 본다. 그래서 방식을 하나로 고르지 않고 여기
모아 두고 B04 화면에서 바꿔 가며 보게 한다(2026-08-30 사용자 지시).
비교용 다섯 방식(TIN 둘 · TPS · ANUDEM형 · 다중해상도)은 걷음(608) — 확정 · 자동 = 라플라스.
각 builder는 `(spec, burned, features, cell_m) -> (R, C) float32` 격자를 돌려준다.
builder는 `(spec, burned, features, cell_m) -> (R, C) float32` 격자를 돌려준다.
`burned`는 등고 라인이 구워진 격자(라인 셀 = 표고, 그 외 NaN)다. 폐합 링 안쪽 처리와
프리뷰·저장은 호출측(`_SheetSurface`)이 방식과 무관하게 똑같이 해 준다.
"""
import logging
import warnings
from typing import Any, Callable
import numpy as np
logger = logging.getLogger(__name__)
# TIN(도엽선)이 물어 오는 격자 밖 여유 — 테두리가 삼각망 밖으로 나가지 않을 만큼만.
SHEET_TIN_CLIP_MARGIN_M = 100.0
# 화면 버튼에 쓰는 이름 — 키는 surface_models.generation_params.source_filter 접미사다.
SHEET_METHOD_LABELS: dict[str, str] = {
"tin_sheet": "TIN(도엽선)",
"tin": "TIN 격자",
"biharmonic": "TPS(박판)",
"anudem": "ANUDEM형",
"multires": "다중해상도",
"laplace": "라플라스",
}
def _laplacian(values: np.ndarray) -> np.ndarray:
padded = np.pad(values, 1, mode="edge")
return (
padded[:-2, 1:-1] + padded[2:, 1:-1] + padded[1:-1, :-2] + padded[1:-1, 2:] - 4.0 * values
)
def _contour_vertices(burned: np.ndarray, spec: Any) -> tuple[np.ndarray, np.ndarray]:
"""등고 라인 셀을 (N,2) 세계좌표와 표고로 바꾼다."""
rows, cols = np.nonzero(np.isfinite(burned))
xs = spec.cell_centers_x()[cols]
ys = spec.cell_centers_y()[rows]
return np.column_stack([xs, ys]), burned[rows, cols].astype(np.float64)
def _grid_points(spec: Any) -> tuple[np.ndarray, np.ndarray]:
grid_x, grid_y = np.meshgrid(spec.cell_centers_x(), spec.cell_centers_y())
return grid_x, grid_y
# ── ① 거리 비례 (버튼에서는 뺐지만 다른 방식의 초기추정으로 계속 쓴다) ───────
# ── 거리 비례 (라플라스 초기값) ─────────────────────────────────────────────
def build_distance(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""가장 가까운 서로 다른 표고 두 라인 사이를 거리 비례로 나눈다.
@@ -120,394 +86,17 @@ def relax_laplace(surface: np.ndarray, fixed: np.ndarray, iterations: int) -> No
surface[colour] = neighbours[colour]
# ── ② 라플라스(조화) ─────────────────────────────────────────────────────────
# ── 라플라스(조화) ─────────────────────────────────────────────────────────────
def build_laplace(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""등고선을 경계값으로 두고 Δz=0을 푼다.
면이 매끈해지지만 z=r(원뿔·능선)은 조화함수가 아니라 마루가 눌린다. 비교 기준으로
남겨 둔다 — ANUDEM이 라플라스 대신 박판 스플라인을 쓰는 이유를 눈으로 보기 위함.
면이 매끈해지지만 z=r(원뿔·능선)은 조화함수가 아니라 마루가 조금 눌린다.
"""
surface = build_distance(spec, burned, features, cell_m)
relax_laplace(surface, np.isfinite(burned), 400)
return surface
# ── ③ 박판 스플라인(중조화) ──────────────────────────────────────────────────
def solve_min_curvature(constrained: np.ndarray, guess: np.ndarray) -> np.ndarray:
"""제약 셀을 고정하고 Δ²z=0(박판 스플라인)을 최소곡률 최소제곱으로 푼다.
ANUDEM/Topo to Raster가 쓰는 박판 스플라인과 같은 연산자다. 라플라스와 달리 z=r을
그대로 통과시켜 능선·마루가 눌리지 않고, 경사가 등고선 너머로 자연스럽게 이어진다.
`constrained`: 값이 고정된 셀(등고 라인, 필요하면 구조선 앵커) — 그 외는 NaN.
`guess`: 시작값이자 감쇠 기준(보통 거리 보간 결과).
"""
from scipy.sparse.linalg import LinearOperator, lsmr
guess = guess.astype(np.float64)
burned = constrained
# 제약 셀 + **격자 테두리**를 고정한다. 최외곽 등고선 바깥이 통째로 자유면
# 1차함수가 Δ²의 영공간에 남아 해가 하나로 정해지지 않고 켤레기울기가 발산한다
# (2026-08-30 실측: |Δz| 2092m). 테두리는 거리 보간값으로 묶는다.
fixed = np.isfinite(burned)
fixed[0, :] = fixed[-1, :] = True
fixed[:, 0] = fixed[:, -1] = True
free = ~fixed & np.isfinite(guess)
if not free.any():
return guess.astype(np.float32)
index = np.flatnonzero(free.ravel())
values = np.where(np.isfinite(burned), burned, guess)
base = np.where(fixed, np.nan_to_num(values), 0.0)
# Δ²z=0을 정규방정식(CG)으로 풀면 조건수가 격자변 4제곱이라 발산한다(실측).
# 대신 **최소곡률** 최소제곱으로 세운다 — 자유 셀에 대해 ‖Δz‖를 최소화하며,
# 그 정상해가 곧 Δ²z=0이다. 조건수가 제곱으로 줄어 LSMR이 안정적으로 푼다
# (Briggs 1974의 최소곡률 격자화와 같은 목적함수).
# 최소곡률만으로는 제약(등고선)에서 먼 영역이 정해지지 않아 해가 폭주한다.
# ANUDEM의 거칠기 벌점과 같은 취지로 감쇠항을 붙여 거리 보간값에 묶어 둔다:
# minimize ‖Δz‖² + λ‖z − 거리보간‖²
# λ가 작을수록 더 매끈하고 클수록 거리 보간에 가깝다.
total_cells = base.size
free_count = len(index)
damping = np.float64(np.sqrt(0.02))
anchor = guess.ravel()[index]
def forward(vector: np.ndarray) -> np.ndarray:
# 반드시 **선형**이어야 한다 — 고정 셀 기여(base)를 여기서 더하면 아핀이 되어
# LSMR의 전제가 깨지고 해가 폭주한다. base 몫은 우변으로만 넘긴다.
scattered = np.zeros_like(base)
scattered.ravel()[index] = vector
return np.concatenate([_laplacian(scattered).ravel(), damping * vector])
def adjoint(vector: np.ndarray) -> np.ndarray:
curvature = _laplacian(vector[:total_cells].reshape(base.shape)).ravel()[index]
return curvature + damping * vector[total_cells:]
rhs = np.concatenate([-_laplacian(base).ravel(), damping * anchor])
linear = LinearOperator(
(total_cells + free_count, free_count),
matvec=forward,
rmatvec=adjoint,
dtype=np.float64,
)
result = lsmr(linear, rhs, x0=anchor, maxiter=400, atol=1e-8, btol=1e-8)
solution, info = result[0], result[1]
surface = base.copy()
surface.ravel()[index] = solution
surface[fixed] = values[fixed]
# 안전장치 — 발산하면 조용히 틀린 지형을 넘기지 말고 거리 보간으로 되돌린다.
drift = float(np.nanmax(np.abs(surface - guess)))
span = float(np.nanmax(guess) - np.nanmin(guess))
if not np.isfinite(drift) or drift > max(span, 1.0):
logger.warning(
"도엽 서피스(TPS): 해가 발산해(최대 %.1fm) 거리 보간으로 되돌립니다 (info=%s).",
drift,
info,
)
return guess.astype(np.float32)
logger.info("도엽 서피스(TPS): 최소제곱 info=%s, 최대 변화 %.2fm", info, drift)
return surface.astype(np.float32)
def build_biharmonic(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""등고선만 제약으로 둔 박판 스플라인."""
return solve_min_curvature(burned, build_distance(spec, burned, features, cell_m))
# ── ④ TIN ───────────────────────────────────────────────────────────────────
def _triangulate(
spec: Any, shape_: tuple[int, int], points: np.ndarray, values: np.ndarray
) -> np.ndarray:
"""정점 구름을 Delaunay 삼각망 선형 보간해 격자로 편다. 삼각망 밖은 NaN."""
from scipy.interpolate import LinearNDInterpolator
if len(points) < 3:
return np.full(shape_, np.nan, dtype=np.float32)
interpolator = LinearNDInterpolator(points, values)
grid_x, grid_y = _grid_points(spec)
surface = np.empty(shape_, dtype=np.float32)
chunk = max(1, int(4_000_000 // max(spec.n_cols, 1)))
for start in range(0, spec.n_rows, chunk):
stop = min(start + chunk, spec.n_rows)
surface[start:stop] = interpolator(grid_x[start:stop], grid_y[start:stop]).astype(
np.float32
)
return surface
def build_tin(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""격자에 구운 등고 라인 셀을 Delaunay 삼각망 선형 보간한다.
정점이 셀 중심에 맞춰져 있어 1m 계단이 삼각망에 그대로 실린다. 같은 표고 정점
3개로 이루어진 평탄 삼각형이 굴곡부·마루에 계단을 만든다. 비교 기준으로 남긴다.
"""
points, values = _contour_vertices(burned, spec)
if len(points) > 120_000: # 삼각망 비용은 정점 수에 비례한다
step = int(np.ceil(len(points) / 120_000))
points, values = points[::step], values[::step]
return _triangulate(spec, burned.shape, points, values)
def build_tin_sheet(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""**원본 5m 도엽 등고선 정점**을 그대로 이은 고전 TIN (2026-08-30 사용자 지시).
`tin`은 격자에 구운 라인 셀(=1m 계단으로 뭉개진 정점)을 쓰지만, 이쪽은 벡터
등고선의 정점을 좌표 그대로 쓴다 — 도면 등고선을 삼각망으로 잇는 측량 관행 그대로다.
격자 밖 등고선은 물지 않는다(삼각망 비용만 커지고 결과는 같다). 다만 격자 테두리가
삼각망 밖으로 나가지 않도록 여유를 두고 자른다.
길이 필터는 두지 않는다 — 짧은 봉우리 폐합 등고선을 버리면 마루가 통째로 평평해진다.
"""
from shapely.geometry import shape as to_shape
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import (
ELEVATION_KEYS,
iter_linestrings,
)
xs, ys = spec.cell_centers_x(), spec.cell_centers_y()
margin = max(SHEET_TIN_CLIP_MARGIN_M, cell_m * 2.0)
x_lo, x_hi = xs[0] - margin, xs[-1] + margin
y_lo, y_hi = ys[-1] - margin, ys[0] + margin
coords: list[np.ndarray] = []
levels: list[np.ndarray] = []
for feature in features or []:
properties = feature.get("properties") or {}
elevation = next(
(float(properties[key]) for key in ELEVATION_KEYS if properties.get(key) is not None),
None,
)
geometry = feature.get("geometry")
if elevation is None or not geometry:
continue
try:
parsed = to_shape(geometry)
except Exception: # noqa: BLE001 — 손상된 피처는 건너뛴다
continue
for line in iter_linestrings(parsed):
point = np.asarray(line.coords, dtype=np.float64)[:, :2]
inside = (
(point[:, 0] >= x_lo)
& (point[:, 0] <= x_hi)
& (point[:, 1] >= y_lo)
& (point[:, 1] <= y_hi)
)
if not inside.any():
continue
coords.append(point[inside])
levels.append(np.full(int(inside.sum()), elevation, dtype=np.float64))
if not coords:
logger.warning("도엽 서피스: TIN(도엽선)에 쓸 등고선 정점이 없습니다.")
return np.full(burned.shape, np.nan, dtype=np.float32)
points = np.vstack(coords)
values = np.concatenate(levels)
# 도엽 이음매에서 같은 정점이 겹쳐 들어온다 — Qhull 비용만 늘어 미리 접는다.
_, unique = np.unique(np.round(points, 3), axis=0, return_index=True)
points, values = points[unique], values[unique]
logger.info("도엽 서피스: TIN(도엽선) 정점 %d개", len(points))
return _triangulate(spec, burned.shape, points, values)
# ── ⑦ ANUDEM형 (구조선 + 배수 강제) ─────────────────────────────────────────
def _contour_corner_anchors(
spec: Any, burned: np.ndarray, guess: np.ndarray, cell_m: float
) -> tuple[np.ndarray, np.ndarray] | None:
"""등고선의 국소 최대 곡률점(코너)에서 능선·계곡 구조선 앵커를 만든다.
ANUDEM은 등고선 자체의 곡률에서 능선·계곡망을 먼저 뽑아 흐름 구조를 세운다
(Hutchinson 1988/89). 여기서도 같은 순서를 따른다.
① 라인마다 정점 곡률을 재 국소 최대점(V자 꼭짓점)을 고른다
② 굽은 안쪽이 더 높으면 **계곡**(등고선 V가 상류를 가리킴), 낮으면 **능선**
③ 같은 종류의 코너를 이웃 표고끼리 이어 그 사이를 선형 보간해 앵커로 심는다
앵커는 박판 해의 제약으로 들어가 계곡 바닥이 이어져 내려가고 능선 마루가 선다.
"""
from scipy.ndimage import label
from scipy.spatial import cKDTree
levels = np.unique(burned[np.isfinite(burned)])
if len(levels) < 2:
return None
interval = float(np.diff(levels).min())
xs = spec.cell_centers_x()
ys = spec.cell_centers_y()
corners: list[tuple[float, float, float, int]] = [] # x, y, level, +1 계곡 / -1 능선
span = 6 # 곡률을 재는 정점 간격(px) — 짧으면 노이즈, 길면 꼭짓점을 놓친다
for level in levels:
labelled, count = label(burned == level)
for component_id in range(1, count + 1):
line_rows, line_cols = np.nonzero(labelled == component_id)
if len(line_rows) < 3 * span:
continue
# 라인 셀을 한 줄로 세운다 — 좌표 정렬로 근사한다(정밀 추적은 과하다).
order = np.argsort(line_cols + line_rows * 1e-3)
path = np.column_stack([line_cols[order], line_rows[order]]).astype(np.float64)
before = np.roll(path, span, axis=0)
after = np.roll(path, -span, axis=0)
first = path - before
second = after - path
first_len = np.hypot(first[:, 0], first[:, 1])
second_len = np.hypot(second[:, 0], second[:, 1])
valid = (first_len > 1e-6) & (second_len > 1e-6)
cosine = np.ones(len(path))
cosine[valid] = (first[valid] * second[valid]).sum(axis=1) / (
first_len[valid] * second_len[valid]
)
sharp = np.flatnonzero(valid & (cosine < 0.3)) # 70도 이상 꺾인 자리
for i in sharp[:: max(1, span)]:
# 굽은 안쪽 방향 = 두 변 단위벡터 합의 반대
inward = -(first[i] / first_len[i] + second[i] / second_len[i])
norm = float(np.hypot(inward[0], inward[1]))
if norm < 1e-6:
continue
probe = path[i] + inward / norm * 6.0
probe_col = int(round(probe[0]))
probe_row = int(round(probe[1]))
if not (0 <= probe_row < guess.shape[0] and 0 <= probe_col < guess.shape[1]):
continue
inside = float(guess[probe_row, probe_col])
if not np.isfinite(inside) or abs(inside - level) < interval * 0.15:
continue
corners.append(
(
float(xs[int(path[i, 0])]),
float(ys[int(path[i, 1])]),
float(level),
1 if inside > level else -1,
)
)
if len(corners) < 4:
logger.info("도엽 서피스(ANUDEM형): 등고선 코너가 부족해 구조선을 건너뜁니다.")
return None
array = np.asarray(corners, dtype=np.float64)
anchor_xy: list[np.ndarray] = []
anchor_z: list[np.ndarray] = []
reach = interval * 20.0 # 이보다 먼 코너는 같은 구조선으로 보지 않는다
for level in levels[:-1]:
upper = level + interval
lower_set = array[np.abs(array[:, 2] - level) < 1e-6]
upper_set = array[np.abs(array[:, 2] - upper) < 1e-6]
if not len(lower_set) or not len(upper_set):
continue
tree = cKDTree(upper_set[:, :2])
distance, index = tree.query(lower_set[:, :2], k=1)
for i in range(len(lower_set)):
j = int(index[i])
if distance[i] > reach or lower_set[i, 3] != upper_set[j, 3]:
continue
start = lower_set[i, :2]
end = upper_set[j, :2]
steps = max(2, int(distance[i] / max(cell_m, 1e-6) / 4))
fraction = np.linspace(0.0, 1.0, steps + 1)[1:-1]
if not len(fraction):
continue
anchor_xy.append(start + (end - start) * fraction[:, None])
anchor_z.append(level + interval * fraction)
if not anchor_xy:
return None
return np.vstack(anchor_xy), np.concatenate(anchor_z)
def _enforce_drainage(surface: np.ndarray, epsilon: float = 0.01) -> int:
"""가짜 웅덩이를 메운다 — ANUDEM의 배수 강제와 같은 목적.
등고선만으로 만든 면에는 흐름이 끊기는 웅덩이가 남는다. 형태학적 재구성(erosion)
으로 채우되 완전 평탄해지지 않게 아주 작은 값을 얹는다. 채운 셀 수를 반환한다.
"""
from skimage.morphology import reconstruction
if not np.isfinite(surface).all():
return 0
seed = np.full(surface.shape, float(surface.max()), dtype=np.float64)
seed[0, :] = surface[0, :]
seed[-1, :] = surface[-1, :]
seed[:, 0] = surface[:, 0]
seed[:, -1] = surface[:, -1]
filled = reconstruction(seed, surface.astype(np.float64), method="erosion")
raised = filled > surface + 1e-6
if not raised.any():
return 0
# 그냥 채우면 웅덩이가 통째로 평탄해져 흐름 방향이 없어진다. 채운 영역 안쪽으로
# 갈수록 아주 조금 높아지게 해서 물이 가장자리(넘침점)로 빠져나가게 둔다
# (Garbrecht·Martz의 평탄면 해소를 간단히 옮긴 것 — 표고 변화는 cm 단위다).
from scipy.ndimage import distance_transform_edt
inner = distance_transform_edt(raised)
surface[raised] = (filled[raised] + epsilon * inner[raised]).astype(surface.dtype)
return int(raised.sum())
def build_anudem(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""ANUDEM형 — 등고선 곡률에서 능선·계곡 구조선을 뽑아 제약에 더하고, 박판으로 풀고,
가짜 웅덩이를 메운다. Topo to Raster가 밟는 세 단계를 그대로 옮긴 것이다."""
guess = build_distance(spec, burned, features, cell_m).astype(np.float64)
constrained = burned.astype(np.float64).copy()
anchors = _contour_corner_anchors(spec, burned, guess, cell_m)
if anchors is not None:
xy, z = anchors
row, col = spec.world_to_rc(xy[:, 0].copy(), xy[:, 1].copy())
inside = (row >= 0) & (col >= 0)
row, col, z = row[inside], col[inside], z[inside]
free = ~np.isfinite(constrained[row, col])
constrained[row[free], col[free]] = z[free]
logger.info("도엽 서피스(ANUDEM형): 구조선 앵커 %d셀", int(free.sum()))
surface = solve_min_curvature(constrained, guess).astype(np.float32)
logger.info("도엽 서피스(ANUDEM형): 가짜 웅덩이 %d셀 메움", _enforce_drainage(surface))
return surface
# ── ⑧ 다중해상도 (coarse → fine) ────────────────────────────────────────────
def build_multires(spec: Any, burned: np.ndarray, features: Any, cell_m: float) -> np.ndarray:
"""성긴 격자에서 풀고 점차 세밀화한다 — ANUDEM의 다중해상도 전략.
전체 형상은 성긴 격자에서 싸게 잡고, 세밀한 격자에서는 등고선 근처만 다듬는다.
한 해상도에서만 풀 때보다 넓은 밴드가 고르게 퍼지고 값싸게 수렴한다.
"""
from scipy.ndimage import zoom
surface: np.ndarray | None = None
for factor in (8, 4, 2, 1):
if factor == 1:
coarse = burned
else:
# 성긴 격자의 제약 — 블록 안 등고 라인의 평균 표고를 대표로 쓴다.
rows = burned.shape[0] // factor * factor
cols = burned.shape[1] // factor * factor
blocks = burned[:rows, :cols].reshape(rows // factor, factor, cols // factor, factor)
# 라인이 하나도 없는 블록은 NaN이 정상이라 경고를 삼킨다.
with warnings.catch_warnings():
warnings.simplefilter("ignore", RuntimeWarning)
coarse = np.nanmean(blocks, axis=(1, 3)).astype(np.float32)
level = build_distance(spec, coarse, features, cell_m * factor)
if surface is not None:
# 앞 단계 해를 지금 해상도로 올려 절반씩 섞는다 — 성긴 단계의 넓은 추세를
# 이어받되 이번 해상도의 등고선 정보를 덮지 않는다.
scale = (level.shape[0] / surface.shape[0], level.shape[1] / surface.shape[1])
upscaled = zoom(surface, scale, order=1)
free = ~np.isfinite(coarse)
level[free] = (level[free] + upscaled[free]) * 0.5
relax_laplace(level, np.isfinite(coarse), 8)
surface = level
assert surface is not None
if surface.shape != burned.shape: # 블록 자르기로 남은 가장자리 보정
scale = (burned.shape[0] / surface.shape[0], burned.shape[1] / surface.shape[1])
surface = zoom(surface, scale, order=1)
line = np.isfinite(burned)
surface[line] = burned[line]
return surface.astype(np.float32)
SHEET_METHOD_BUILDERS: dict[str, Callable[[Any, np.ndarray, Any, float], np.ndarray]] = {
"tin_sheet": build_tin_sheet,
"tin": build_tin,
"biharmonic": build_biharmonic,
"anudem": build_anudem,
"multires": build_multires,
"laplace": build_laplace,
}
+2 -41
View File
@@ -1,7 +1,7 @@
"""B04 지표면 스무딩 엔진 (DTM 가우시안+B-Spline, TIN Taubin).
"""B04 지표면 스무딩 엔진 (DTM 가우시안+B-Spline).
원본 모델(npz)을 로드해 표고 Z만 평활화한 스무딩 모델과 프리뷰(GLB)를 만든다.
XY 위치와 삼각형 연결은 원본을 유지한다.
XY 격자는 원본을 유지한다.
"""
import hashlib
@@ -10,7 +10,6 @@ from pathlib import Path
from typing import Any
import numpy as np
import trimesh
from scipy import ndimage
from scipy.interpolate import RectBivariateSpline
@@ -104,48 +103,10 @@ def smooth_dtm(
}
def smooth_tin(
context: TerrainContext, output_dir: Path, stem: str, original_model_path: Path
) -> dict[str, Any]:
"""TIN 삼각망에 Taubin 저수축 스무딩을 적용한다 (Z만 평활화)."""
data = np.load(original_model_path)
vertices, faces = data["vertices"], data["faces"]
if len(vertices) < 3 or not len(faces):
raise ValueError("TIN 스무딩을 수행할 삼각망 메쉬 데이터가 올바르지 않습니다.")
iterations = int(context.config.get("smoothing_tin_taubin_iterations", 10))
lamb = float(context.config.get("smoothing_tin_taubin_lambda", 0.5))
mu = float(context.config.get("smoothing_tin_taubin_mu", -0.53))
mesh = trimesh.Trimesh(vertices=vertices, faces=faces, process=False)
if iterations > 0:
trimesh.smoothing.filter_taubin(mesh, lamb=lamb, nu=mu, iterations=iterations)
vertices_smooth = np.asarray(mesh.vertices, dtype=np.float32)
# 지표면 스무딩은 리토폴로지가 아니다: XY·연결은 원본, Z만 평활화.
vertices_smooth[:, :2] = np.asarray(vertices[:, :2], dtype=np.float32)
faces_smooth = np.asarray(mesh.faces, dtype=np.uint32)
model_path = output_dir / f"{stem}_smooth.npz"
preview_path = output_dir / f"{stem}_smooth_preview.glb"
atomic_npz(model_path, vertices=vertices_smooth, faces=faces_smooth)
write_glb(preview_path, vertices_smooth, faces_smooth, context.bounds)
return {
"model_file": model_path.name,
"preview_file": preview_path.name,
"preview_media_type": "model/gltf-binary",
"vertex_count": int(len(vertices_smooth)),
"face_count": int(len(faces_smooth)),
"artifact_bytes": artifact_size(model_path, preview_path),
"smoothing_signature": compute_smoothing_signature(context.config),
}
def run_smoothing(
method: str, context: TerrainContext, output_dir: Path, stem: str, original_model_path: Path
) -> dict[str, Any]:
"""방식에 따라 적절한 스무딩 엔진을 수행한다."""
if method == "dtm":
return smooth_dtm(context, output_dir, stem, original_model_path)
if method == "tin":
return smooth_tin(context, output_dir, stem, original_model_path)
raise ValueError(f"스무딩이 불가능한 지형 표현 방식입니다: {method}")
@@ -4,7 +4,7 @@
만들고, 뒤 단계(지면필터·모델·등고선·배수)는 받는 형식이 그대로라 손대지 않는다.
점이 임계를 넘으면 **칸(기본 0.5m)마다 최저점 하나만** 남긴다(씨닝). 설계가 쓰는 격자가
1m(지면필터 2m·CSF 천 1.5m)라 0.5m 는 설계보다 촘촘해 결과 표고가 사실상 같고, 30GB 두
1m(다듬기 칸 1m)라 0.5m 는 설계보다 촘촘해 결과 표고가 사실상 같고, 30GB 두
장이 메모리 29GB → 1.4GB 로 내려간다. 임계 아래면 원본 점을 그대로 쓴다 — 작은 자료의
결과는 바뀌지 않는다.
"""
@@ -157,8 +157,8 @@ class _ThinGrid:
버리지 않으므로 **지면 결과는 씨닝 전과 완전히 같다**(2026-09-06 용화 실측: 1m 지면
격자 141,969칸 전부 표고 차이 0). 지면점은 원본의 2~3%뿐이라 남겨도 가볍다.
나머지(수목·구조물·잡음)는 칸마다 최저점만 남긴다 — 분류가 없는 자료에서 CSF·PMF가
지면을 찾을 밑그림으로 충분하다(CSF 천 간격 1.5m > 칸 0.5m).
나머지(수목·구조물·잡음)는 칸마다 최저점만 남긴다 — 분류가 없는 자료에서 점진 TIN이
지면을 찾을 밑그림으로 충분하다(씨앗 간격 10m · 다듬기 칸 1m > 칸 0.5m).
한 청크 안에서 같은 칸이 여러 번 나오면 뒤에 쓴 값이 이겨 최저점이 아니게 된다.
그래서 청크를 (칸, 표고)로 정렬해 **칸마다 첫 점**만 골라 낸 뒤 격자와 견준다.
+1 -80
View File
@@ -1,7 +1,6 @@
"""B04 지표면 분석 FastAPI 라우터."""
import asyncio
import json
import logging
from pathlib import Path
from typing import Any
@@ -22,7 +21,7 @@ from B04_PreProcess.B04_PreProcess_Engine_Extent import (
project_epsg_from_prj,
)
from B04_PreProcess.B04_PreProcess_Engine_Ground import available_filters
from B04_PreProcess.B04_PreProcess_Engine_PointPreview import ensure_point_preview, sample_points
from B04_PreProcess.B04_PreProcess_Engine_PointPreview import ensure_point_preview
from B04_PreProcess.B04_PreProcess_Repository import (
get_input_file,
list_project_point_cloud_inputs,
@@ -54,12 +53,10 @@ from B04_PreProcess.B04_PreProcess_Schema import (
SurfaceConfirmedResponse,
SurfaceConfirmRequest,
SurfaceConfirmResponse,
SurfaceGroundStatsResponse,
SurfaceInputFileListResponse,
SurfaceInputFileSummary,
SurfaceModelListResponse,
SurfaceModelSummary,
SurfacePointCloudSampleResponse,
)
from B04_PreProcess.B04_PreProcess_Service import confirm_surface_selection
from common_util.common_util_auth import require_system_admin
@@ -356,51 +353,6 @@ async def get_surface_point_cloud_binary(
)
@router.get("/{project_id}/surface/point-cloud", response_model=SurfacePointCloudSampleResponse)
async def get_surface_point_cloud(
project_id: UUID,
filter: str | None = None,
) -> SurfacePointCloudSampleResponse | JSONResponse:
"""원본 또는 필터링된 B04 3D 미리보기 포인트를 반환한다(JSON · 화면은 이진 길을 씀)."""
try:
source_path = await _point_cloud_source(project_id, filter)
if isinstance(source_path, JSONResponse):
return source_path
with np.load(source_path) as structured:
xyz = np.asarray(structured["xyz"], dtype=np.float32)
bounds = np.asarray(structured["bounds"], dtype=np.float64)
point_count = (
int(structured["point_count"]) if "point_count" in structured else int(len(xyz))
)
rgb = np.asarray(structured["rgb"]) if "rgb" in structured else None
sample, rgb_sample = sample_points(xyz, rgb)
return SurfacePointCloudSampleResponse(
project_id=str(project_id),
point_count=point_count,
sampled_count=int(len(sample)),
bounds={
"x_min": float(bounds[0, 0]),
"x_max": float(bounds[0, 1]),
"y_min": float(bounds[1, 0]),
"y_max": float(bounds[1, 1]),
"z_min": float(bounds[2, 0]),
"z_max": float(bounds[2, 1]),
},
points=sample.astype(float).tolist(),
rgb=rgb_sample.astype(int).tolist() if rgb_sample is not None else None,
)
except LookupError as exc:
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
except Exception:
logger.exception("B04 포인트클라우드 샘플 조회 실패: project_id=%s", project_id)
return JSONResponse(
status_code=500,
content={"status": "error", "message": "포인트클라우드 조회 중 오류가 발생했습니다."},
)
@router.get("/{project_id}/surface/confirmed", response_model=SurfaceConfirmedResponse)
async def get_confirmed_surface(project_id: UUID) -> SurfaceConfirmedResponse | JSONResponse:
"""확정 지표면 구성과 지형 가장자리만 반환한다(포인트 배열 없음).
@@ -496,34 +448,3 @@ async def get_confirmed_surface(project_id: UUID) -> SurfaceConfirmedResponse |
status_code=500,
content={"status": "error", "message": "확정 지표면 정보를 불러오지 못했습니다."},
)
@router.get("/{project_id}/surface/ground-stats", response_model=SurfaceGroundStatsResponse)
async def get_surface_ground_stats(project_id: UUID) -> SurfaceGroundStatsResponse | JSONResponse:
"""manifest에서 필터별 지면 포인트 통계를 반환한다."""
pool = get_db_pool()
try:
async with pool.acquire() as connection:
stored_path = await get_project_storage_relative_path(connection, project_id)
project_root = Path(resolve_stored_project_path(stored_path))
manifest_path = project_root / "B04_PreProcess" / "models" / "manifest.json"
if not manifest_path.is_file():
return SurfaceGroundStatsResponse(project_id=str(project_id), filters={})
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
filters = {
key: {
"source_point_count": value.get("source_point_count"),
"methods": value.get("methods", {}),
}
for key, value in manifest.get("source_filters", {}).items()
if isinstance(value, dict)
}
return SurfaceGroundStatsResponse(project_id=str(project_id), filters=filters)
except LookupError as exc:
return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
except Exception:
logger.exception("B04 지면 통계 조회 실패: project_id=%s", project_id)
return JSONResponse(
status_code=500,
content={"status": "error", "message": "지면 통계 조회 중 오류가 발생했습니다."},
)
@@ -29,10 +29,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B04 Surface Contour"])
MODEL_REPRESENTATIONS = {
"meshfree": "meshfree_surfels",
"dtm": "regular_grid",
"tin": "triangular_mesh",
"nurbs": "bspline_surface",
"implicit": "local_rbf_height_field",
}
@@ -79,7 +76,7 @@ async def get_surface_model_contour(
status_code=404,
content={"status": "error", "message": "해당 모델을 찾을 수 없습니다."},
)
model_type, model_file_path = row[0], row[1]
model_file_path = row[1]
project_root = Path(resolve_stored_project_path(stored_path))
if not model_file_path:
@@ -92,21 +89,16 @@ async def get_surface_model_contour(
models_dir = model_path.parent
structured_path = project_root / "B04_PreProcess" / "processed" / "structured.npz"
stem = model_path.stem
parts = stem.split("_")
if len(parts) >= 2:
method = parts[0]
filter_key = "_".join(parts[1:])
else:
method = model_type
filter_key = "csf"
if smooth and method in ("dtm", "tin"):
# 파일 이름 = `{표현}_{필터}` · 지운 표현(608 · tin 등)이면 404
method, _, filter_key = stem.partition("_")
if smooth and method == "dtm":
contour_filename = f"contour_{filter_key}_{method}_smooth_{interval}m.json"
contour_model_path = models_dir / f"{stem}_smooth.npz"
representation = "regular_grid" if method == "dtm" else "triangular_mesh"
representation = "regular_grid"
else:
contour_filename = f"contour_{filter_key}_{method}_{interval}m.json"
contour_model_path = model_path
representation = MODEL_REPRESENTATIONS.get(method)
representation = MODEL_REPRESENTATIONS.get(method) if filter_key else None
contour_path = models_dir / contour_filename
if recalculate or not _is_contour_cache_current(contour_path, contour_model_path):
if not math.isfinite(interval) or interval < 0.5:
@@ -188,11 +188,10 @@ async def get_surface_model_preview(
models_dir = model_path.parent
stem = model_path.stem
ext = "ply" if model_type == "meshfree" else "glb"
if smooth and model_type in ("dtm", "tin"):
if smooth and model_type == "dtm":
preview_filename = f"{stem}_smooth_preview.glb"
else:
preview_filename = f"{stem}_preview.{ext}"
preview_filename = f"{stem}_preview.glb"
preview_path = models_dir / preview_filename
if not preview_path.is_file():
@@ -204,11 +203,7 @@ async def get_surface_model_preview(
},
)
media_type = "application/octet-stream"
if ext == "glb":
media_type = "model/gltf-binary"
elif ext == "ply":
media_type = "application/ply"
media_type = "model/gltf-binary"
# 브라우저가 이미 같은 파일을 갖고 있으면 304만 돌려준다(새로고침이 빨라진다).
return cached_file_response(request, preview_path, media_type, preview_filename)
+3 -28
View File
@@ -4,19 +4,14 @@ from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from B04_PreProcess.B04_PreProcess_Engine_Ground import available_filters
from config.config_system import (
SURFACE_MODEL_PRECOMPUTE,
SURFACE_MODEL_SOURCE_FILTERS,
)
# classification은 LAS가 지면분류를 싣고 있을 때 엔진이 자동으로 붙이지만,
# 재계산 요청에서 이름으로 지정하는 것도 허용한다.
_ALLOWED_FILTERS = set(SURFACE_MODEL_SOURCE_FILTERS) | {
"ransac",
"classification",
"classification_refine",
"ptd_refine",
}
# 허용 필터 = 지면 필터 등록부 하나(Ground `_FILTERS`)
_ALLOWED_FILTERS = set(available_filters())
_ALLOWED_METHODS = set(SURFACE_MODEL_PRECOMPUTE)
@@ -105,18 +100,6 @@ class SurfaceInputFileListResponse(BaseModel):
files: list[SurfaceInputFileSummary]
class SurfacePointCloudSampleResponse(BaseModel):
"""B04 3D 미리보기용 포인트클라우드 샘플."""
status: str = "success"
project_id: str
point_count: int
sampled_count: int
bounds: dict[str, float]
points: list[list[float]]
rgb: list[list[int]] | None = None
class SurfaceConfirmedResponse(BaseModel):
"""확정 지표면 요약 — 화면 진입 판정·준비화면·B05가 공통으로 쓰는 단일 출처.
@@ -138,14 +121,6 @@ class SurfaceConfirmedResponse(BaseModel):
route_bounds: dict[str, float] | None = None
class SurfaceGroundStatsResponse(BaseModel):
"""필터별 지면 포인트 통계 응답."""
status: str = "success"
project_id: str
filters: dict[str, dict[str, Any]]
class SurfaceAnalyzeResponse(BaseModel):
"""지표면 분석 실행 결과."""
@@ -18,9 +18,6 @@ export const SOURCE_FILTERS = [
"classification_refine",
"ptd",
"ptd_refine",
"grid_min_z",
"csf",
"pmf",
] as const;
const FILTER_LABELS: Record<string, string> = {
+2 -2
View File
@@ -50,12 +50,12 @@ import "./B04_PreProcess_UI_Style.css";
import "./B04_PreProcess_UI_Style_Map.css";
import { currentProjectId, guardProjectLeave } from "../A00_Common/b_project_switch";
const MODEL_METHODS = ["tin", "dtm", "nurbs", "implicit", "meshfree"] as const;
const MODEL_METHODS = ["dtm", "implicit"] as const;
const DEFAULT_FILTER = "ptd";
const DEFAULT_METHOD = "implicit"; // 확정 기본 표현(요구 558) — 확정본이 있으면 그것이 먼저
const ROUTE_STAGE = ROUTES.B05_PROFILE;
// 도엽 서피스 보간 방식 버튼 순서 — 백엔드 SHEET_SURFACE_METHODS와 같은 차례로 둔다.
const SHEET_METHOD_ORDER = ["tin_sheet", "tin", "biharmonic", "anudem", "multires", "laplace"];
const SHEET_METHOD_ORDER = ["laplace"];
function L(key: keyof typeof ui_locales): string {
return ui_locales[key][currentLanguageIndex];
@@ -1,7 +1,6 @@
import * as THREE from "three";
import { OrbitControls } from "three/examples/jsm/controls/OrbitControls.js";
import { GLTFLoader } from "three/examples/jsm/loaders/GLTFLoader.js";
import { PLYLoader } from "three/examples/jsm/loaders/PLYLoader.js";
import { API_BASE_URL } from "@config/config_frontend";
import { fetchCachedBytes } from "../A00_Common/b_asset_cache";
import { currentLanguageIndex, ui_locales } from "@ui/ui_template_locale";
@@ -75,7 +74,7 @@ export function createSurfaceTerrainViewer(): SurfaceTerrainViewer {
minValSpan,
progress,
} = buildTerrainViewerChrome();
let activeFilter = "csf";
let activeFilter = "";
let activeMethod = "dtm";
/** 진행률(0~1, 모르면 null)과 문구를 표시한다. label이 null이면 서클을 감춘다. */
@@ -348,7 +347,7 @@ export function createSurfaceTerrainViewer(): SurfaceTerrainViewer {
}
function syncSmoothingSupport(): void {
const supported = activeMethod === "tin" || activeMethod === "dtm";
const supported = activeMethod === "dtm";
smoothSelect.disabled = !supported;
smoothSelect.value = supported && smoothPreferred ? "on" : "off";
smoothLabel.classList.toggle("is-disabled", !supported);
@@ -372,8 +371,8 @@ export function createSurfaceTerrainViewer(): SurfaceTerrainViewer {
showProgress(null, "모델 조회 중…");
// 1. Find matching model in list
// model_type is TIN / DTM / NURBS / Implicit / Meshfree (we match activeMethod)
// model_file_path contains the activeFilter (e.g. csf, pmf, grid_min_z)
// model_type is DTM / Implicit (we match activeMethod)
// model_file_path contains the activeFilter (e.g. ptd_refine, classification)
const match = currentModelsList.find((m) => {
const typeMatches = m.model_type.toLowerCase() === activeMethod.toLowerCase();
const configuredFilter = m.generation_params?.source_filter;
@@ -391,7 +390,7 @@ export function createSurfaceTerrainViewer(): SurfaceTerrainViewer {
}
const modelId = match.id;
const isSmooth = (activeMethod === "tin" || activeMethod === "dtm") && smoothingOn();
const isSmooth = activeMethod === "dtm" && smoothingOn();
currentModelId = modelId;
currentModelSmooth = isSmooth;
const generation = ++loadGeneration;
@@ -410,52 +409,34 @@ export function createSurfaceTerrainViewer(): SurfaceTerrainViewer {
});
if (generation !== loadGeneration) return;
if (activeMethod === "meshfree") {
const geometry = new PLYLoader().parse(buffer);
geometry.computeBoundingSphere();
const material = new THREE.PointsMaterial({
size: 0.35,
vertexColors: geometry.hasAttribute("color"),
sizeAttenuation: true,
});
const points = new THREE.Points(geometry, material);
points.visible = surfCheck.checked;
terrainMesh = points;
scene.add(points);
fitCamera(points);
showProgress(1, "등고선을 그리는 중…");
await loadSelectedContours(modelId, isSmooth);
showProgress(null, null);
} else {
new GLTFLoader().parse(
buffer,
"",
async (gltf) => {
if (generation !== loadGeneration) {
disposeObject(gltf.scene);
return;
new GLTFLoader().parse(
buffer,
"",
async (gltf) => {
if (generation !== loadGeneration) {
disposeObject(gltf.scene);
return;
}
gltf.scene.traverse((child) => {
if (child instanceof THREE.Mesh) {
child.material.side = THREE.DoubleSide;
child.material.vertexColors = child.geometry.hasAttribute("color");
}
gltf.scene.traverse((child) => {
if (child instanceof THREE.Mesh) {
child.material.side = THREE.DoubleSide;
child.material.vertexColors = child.geometry.hasAttribute("color");
}
});
gltf.scene.visible = surfCheck.checked;
terrainMesh = gltf.scene;
scene.add(gltf.scene);
fitCamera(gltf.scene);
showProgress(1, "등고선을 그리는 중…");
await loadSelectedContours(modelId, isSmooth);
showProgress(null, null);
},
() => {
if (generation !== loadGeneration) return;
statusSpan.textContent = "3D 메쉬 파일이 없거나 로드할 수 없습니다.";
showProgress(null, null);
},
);
}
});
gltf.scene.visible = surfCheck.checked;
terrainMesh = gltf.scene;
scene.add(gltf.scene);
fitCamera(gltf.scene);
showProgress(1, "등고선을 그리는 중…");
await loadSelectedContours(modelId, isSmooth);
showProgress(null, null);
},
() => {
if (generation !== loadGeneration) return;
statusSpan.textContent = "3D 메쉬 파일이 없거나 로드할 수 없습니다.";
showProgress(null, null);
},
);
} catch (e) {
statusSpan.textContent = "3D 파일 로드에 실패했습니다.";
showProgress(null, null);