feat(B03·B04): 지형 라이다 여러 장 입력·병합 전처리

- 지형 파일 개수 제한 해제(정확히 1개 → 1장 이상), 한 카드에 여러 장 담기
  (화면 표시 「용화_서편.las 외 1장」, 업로드는 한 장씩 차례로 전송)
- 구조화 엔진이 여러 파일을 합친 범위로 한 벌 생성 — WF1 자동 전처리·B04 재분석 모두
  프로젝트 지형 파일 전부를 대상으로 실행
- 점이 1억 개를 넘으면 씨닝 — 지면 분류점은 전부 남기고 나머지만 0.5m 칸 최저점으로 축소
  (용화 실측: 4,900만점 → 249만점, 지면점 1,140,716개 그대로, 1m 지면격자 표고차 0.0000m)
- 머리글만 읽어 5km 넘게 떨어진 파일은 업로드 거부 (임도는 길어도 2~3km)
- 화면 조립부 700줄 준수를 위해 terrainCoverage 를 판정 모듈로 이동

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

B03_FileInput/B03_FileInput_Service_WF1.py

@@ -79,11 +79,18 @@ async def trigger_wf1_analysis_and_email(
                 await connection.commit()
             stored_path = await get_project_storage_relative_path(connection, project_id)
             project_info = await _get_project_notification_info(connection, project_id)
-            from B04_PreProcess.B04_PreProcess_Repository import get_input_file
+            from B04_PreProcess.B04_PreProcess_Repository import (
+                get_input_file,
+                list_project_point_cloud_paths,
+            )
 
             input_file = await get_input_file(connection, project_id, input_file_id)
+            project_root = Path(resolve_stored_project_path(stored_path))
+            # 지형 파일이 여러 장이면 합쳐서 한 벌로 전처리한다(2026-09-06 사용자 확정).
+            terrain_paths = await list_project_point_cloud_paths(
+                connection, project_id, project_root
+            )
 
-        project_root = Path(resolve_stored_project_path(stored_path))
         source_path = project_root / Path(str(input_file["raw_file_path"]))
         # LAS 없는 설계(2026-08-30): 입력이 계획노선 CSV면 도엽등고선 서피스 분석으로 간다.
         las_free = str(input_file.get("file_type") or "").lower() not in {"las", "laz"}
@@ -118,7 +125,7 @@ async def trigger_wf1_analysis_and_email(
             analysis_result = await asyncio.to_thread(
                 run_surface_analysis,
                 project_root,
-                source_path,
+                terrain_paths or [source_path],
                 source_filters=None,
                 methods=methods,
                 force=False,

B03_FileInput/B03_FileInput_UI_Page.ts

@@ -24,15 +24,11 @@ import { createUploadFlow } from "./B03_FileInput_UI_Page_Flow";
 import {
   isSlotRequired,
   slotForOverviewFile,
+  terrainCoverage,
   validateFileForSlot,
   validateSlots,
 } from "./B03_FileInput_UI_Page_Rules";
-import {
+import { readCrsLabel, renderSlotPreview } from "./B03_FileInput_UI_Preview";
-  readCrsLabel,
-  readExtent,
-  renderSlotPreview,
-  type PreviewExtent,
-} from "./B03_FileInput_UI_Preview";
 import { confirmReplaceUpload } from "./B03_FileInput_UI_Upload";
 import {
   createFileCardTemplate,
@@ -41,9 +37,11 @@ import {
   initializeSlots,
   makeSessionKey,
   planSlotAssignments,
+  pushExtraFile,
   ROUTE_SLOTS,
   SHAPEFILE_DEPENDENT_SLOTS,
   slotConfigs,
+  slotFileLabel,
   TERRAIN_SLOTS,
   type FileSlot,
   type FileSlotState,
@@ -209,7 +207,7 @@ export async function renderB03FileInput(root: HTMLElement): Promise<void> {
 
     const percent = state.file ? Math.min(100, (state.progressBytes / state.file.size) * 100) : 0;
     // 로컬 파일이 없어도 서버에 업로드된 파일이 있으면 그 정보(정본)를 보여준다.
-    if (fileName) fileName.textContent = state.file?.name ?? state.serverUploaded?.name ?? "";
+    if (fileName) fileName.textContent = slotFileLabel(state);
     if (fileSize) {
       fileSize.textContent = state.file
         ? formatBytes(state.file.size)
@@ -240,7 +238,7 @@ export async function renderB03FileInput(root: HTMLElement): Promise<void> {
 
     const preview = card.querySelector<HTMLDivElement>(".b03-file__preview");
     if (preview) {
-      const terrain = terrainCoverage();
+      const terrain = terrainCoverage(slots);
       renderSlotPreview(preview, {
         metadata: state.serverUploaded?.metadata,
         // 업로드·분석이 끝난 카드에만 보인다 (2026-09-04 사용자 지시).
@@ -258,26 +256,6 @@ export async function renderB03FileInput(root: HTMLElement): Promise<void> {
    * 들어오는지 대조하는 기준. 초기 계산 실패의 주된 원인이 범위 불일치라
    * 카드에서 바로 보이게 한다(2026-09-04).
    */
-  function terrainCoverage(): { extent: PreviewExtent | null; crs: string | null } {
-    let extent: PreviewExtent | null = null;
-    let crs: string | null = null;
-    for (const slot of TERRAIN_SLOTS) {
-      const metadata = slots.get(slot)?.serverUploaded?.metadata;
-      const next = readExtent(metadata);
-      if (!next) continue;
-      crs ??= readCrsLabel(metadata);
-      extent = extent
-        ? {
-            xMin: Math.min(extent.xMin, next.xMin),
-            xMax: Math.max(extent.xMax, next.xMax),
-            yMin: Math.min(extent.yMin, next.yMin),
-            yMax: Math.max(extent.yMax, next.yMax),
-          }
-        : next;
-    }
-    return { extent, crs };
-  }
-
   function showErrorMessage(slot: FileSlot, error: string): void {
     const state = slots.get(slot);
     if (!state) return;
@@ -297,8 +275,13 @@ export async function renderB03FileInput(root: HTMLElement): Promise<void> {
       showErrorMessage(state.slot, `${validation} ${file.name}`);
       return;
     }
-    if (!targetSlot && state.file && state.file.name !== file.name) {
+    // 지형 자료는 도엽별로 여러 장이 온다 — 카드에 더 담고 전처리가 합쳐 쓴다.
-      showErrorMessage(state.slot, `${L("B03_File_Error_DuplicateSlot")} ${file.name}`);
+    if (state.file && state.file.name !== file.name) {
+      if (!pushExtraFile(state, file)) {
+        showErrorMessage(state.slot, `${L("B03_File_Error_DuplicateSlot")} ${file.name}`);
+        return;
+      }
+      renderSlot(state.slot);
       return;
     }
     // 서버에 이미 완료된 슬롯이면 교체 확인을 받는다(2026-08-04 사용자 지시). 이어올리기로
@@ -382,6 +365,7 @@ export async function renderB03FileInput(root: HTMLElement): Promise<void> {
       localStorage.removeItem(makeSessionKey(activeProjectId, state.file));
     }
     state.file = undefined;
+    state.extraFiles = undefined;
     state.uploadSessionId = undefined;
     state.uploadStatus = "pending";
     state.progressBytes = 0;

B03_FileInput/B03_FileInput_UI_Page_Flow.ts

@@ -201,14 +201,24 @@ export function createUploadFlow(ctx: UploadFlowContext): UploadFlowHandle {
     ctx.pageError.textContent = "";
     ctx.setUploading(true);
     try {
-      for (let index = 0; index < targetStates.length; index += 1) {
+      // 지형 자료는 한 카드에 여러 장이 담길 수 있다 — 카드 순서대로 한 장씩 올린다.
-        const state = targetStates[index];
+      const jobs = targetStates.flatMap((state) =>
+        [state.file!, ...(state.extraFiles ?? [])].map((file) => ({ state, file })),
+      );
+      for (let index = 0; index < jobs.length; index += 1) {
+        const { state, file } = jobs[index];
+        if (file !== state.file) {
+          // 앞 파일이 쓰던 전송 세션·진행률을 물려받지 않게 되돌린다.
+          state.uploadSessionId = undefined;
+          state.progressBytes = 0;
+        }
         await uploadOneFile(
           ctx.projectId(),
           state,
-          index === targetStates.length - 1,
+          index === jobs.length - 1,
           () => ctx.renderSlot(state.slot),
           ctx.lasFreeDesign(),
+          file,
         );
       }
       clearDerivedCaches(ctx.projectId());

B03_FileInput/B03_FileInput_UI_Page_Rules.ts

@@ -10,6 +10,7 @@
 import { UPLOAD_MAX_FILES, UPLOAD_MAX_MB } from "@config/config_frontend";
 import { currentLanguageIndex, ui_locales } from "@ui/ui_template_locale";
 import type { UploadOverviewFile } from "./B03_FileInput_Api_Fetch";
+import { readCrsLabel, readExtent, type PreviewExtent } from "./B03_FileInput_UI_Preview";
 import {
   getExtension,
   SHAPEFILE_DEPENDENT_SLOTS,
@@ -105,3 +106,31 @@ export function slotForOverviewFile(
     (candidate) => candidate.slot !== "route_prj" && candidate.extensions.includes(extension),
   )?.slot;
 }
+
+/**
+ * 서버에 올라온 지형 자료가 덮는 범위와 좌표계 — 카드 미리보기가 쓴다.
+ * 여러 카드(포인트클라우드·좌표계·래스터)의 범위를 합친다. 화면 조립부가 700줄을
+ * 넘어 옮겨 온 순수 함수다(2026-09-06).
+ */
+export function terrainCoverage(slots: SlotMap): {
+  extent: PreviewExtent | null;
+  crs: string | null;
+} {
+  let extent: PreviewExtent | null = null;
+  let crs: string | null = null;
+  for (const slot of TERRAIN_SLOTS) {
+    const metadata = slots.get(slot)?.serverUploaded?.metadata;
+    const next = readExtent(metadata);
+    if (!next) continue;
+    crs ??= readCrsLabel(metadata);
+    extent = extent
+      ? {
+          xMin: Math.min(extent.xMin, next.xMin),
+          xMax: Math.max(extent.xMax, next.xMax),
+          yMin: Math.min(extent.yMin, next.yMin),
+          yMax: Math.max(extent.yMax, next.yMax),
+        }
+      : next;
+  }
+  return { extent, crs };
+}

B03_FileInput/B03_FileInput_UI_Support.ts

@@ -28,6 +28,12 @@ export interface SlotConfig {
 
 export interface FileSlotState extends SlotConfig {
   file?: File;
+  /**
+   * 같은 카드에 더 담은 파일 — 지형 자료(포인트클라우드)만 여러 장을 받는다.
+   * 드론 라이다는 사업지가 넓으면 도엽별로 나뉘어 오고, 전처리가 합쳐서 쓴다
+   * (2026-09-06 사용자 확정). 업로드는 이 목록을 한 장씩 차례로 올린다.
+   */
+  extraFiles?: File[];
   uploadSessionId?: string;
   uploadStatus: UploadStatus;
   progressBytes: number;
@@ -123,6 +129,25 @@ const SLOT_CONFIGS: readonly SlotConfig[] = [
   },
 ];
 
+/** 카드에 적을 파일 이름 — 여러 장이면 「첫 장 외 N장」. */
+export function slotFileLabel(state: FileSlotState): string {
+  const name = state.file?.name ?? state.serverUploaded?.name ?? "";
+  const extras = state.extraFiles?.length ?? 0;
+  return extras > 0 ? `${name} 외 ${extras}장` : name;
+}
+
+/**
+ * 같은 카드에 파일을 더 담는다 — 담았으면 true, 이 카드가 한 장짜리면 false.
+ * 지형 자료(포인트클라우드)만 여러 장을 받는다. 같은 이름은 다시 담지 않는다.
+ */
+export function pushExtraFile(state: FileSlotState, file: File): boolean {
+  if (state.slot !== "las_laz") return false;
+  const extras = state.extraFiles ?? [];
+  if (!extras.some((item) => item.name === file.name)) state.extraFiles = [...extras, file];
+  state.error = undefined;
+  return true;
+}
+
 export function getExtension(fileName: string): string {
   const index = fileName.lastIndexOf(".");
   return index >= 0 ? fileName.slice(index).toLowerCase() : "";

B03_FileInput/B03_FileInput_UI_Upload.ts

@@ -83,8 +83,10 @@ export async function uploadOneFile(
   completeUpload: boolean,
   onProgress: () => void,
   lasFree = false,
+  // 지형 자료는 한 카드에 여러 장이 담긴다 — 올릴 파일을 지정받는다(2026-09-06).
+  target?: File,
 ): Promise<UploadedFileResult[]> {
-  const file = state.file;
+  const file = target ?? state.file;
   if (!file) return [];
   state.error = undefined;
   state.uploadStatus = "uploading";

B04_PreProcess/B04_PreProcess_Engine.py

@@ -7,7 +7,7 @@
 import json
 import logging
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from pathlib import Path
 from typing import Any
 
@@ -37,14 +37,22 @@ GROUND_POINT_SAMPLE_LIMIT = 500_000
 GROUND_POINT_CACHE_VERSION = 2
 
 
-def _source_identity(las_path: Path) -> dict[str, Any]:
+def _source_identity(las_paths: list[Path]) -> dict[str, Any]:
-    """입력 LAS의 정체성(이름·크기·수정시각)으로 캐시 세대를 식별한다 (PLAN B-2)."""
+    """입력 지형 파일들의 정체성(이름·크기·수정시각)으로 캐시 세대를 식별한다 (PLAN B-2).
-    stat = las_path.stat()
+
-    return {
+    여러 장을 병합하므로 **한 장이라도 바뀌거나 늘고 줄면** 다시 계산해야 한다
-        "filename": las_path.name,
+    (2026-09-06 다중 입력).
-        "size_bytes": int(stat.st_size),
+    """
-        "mtime": float(stat.st_mtime),
+    files = [
-    }
+        {
+            "filename": path.name,
+            "size_bytes": int(path.stat().st_size),
+            "mtime": float(path.stat().st_mtime),
+        }
+        for path in sorted(las_paths, key=lambda item: item.name)
+    ]
+    # 한 장일 때는 옛 형식과 같은 모양을 유지한다 — 이미 만든 캐시를 헛되이 버리지 않는다.
+    return files[0] if len(files) == 1 else {"files": files}
 
 
 def _relative_to_project(project_root: Path, path: Path) -> str:
@@ -108,7 +116,7 @@ def cache_ground_points(
 
 def run_surface_analysis(
     project_root: Path,
-    las_path: Path,
+    las_path: Path | Sequence[Path],
     *,
     source_filters: list[str] | None,
     methods: list[str],
@@ -117,6 +125,9 @@ def run_surface_analysis(
 ) -> dict[str, Any]:
     """구조화→필터→모델 빌드를 수행하고 산출 메타데이터를 반환한다.
 
+    `las_path`는 지형 파일 한 장 또는 여러 장이다 — 여러 장이면 합친 범위로 한 벌을
+    만든다(2026-09-06 사용자 확정).
+
     `source_filters`가 비면 입력 LAS를 보고 기본 필터를 정한다(자동 전처리 경로).
 
     반환 dict:
@@ -132,6 +143,9 @@ def run_surface_analysis(
             on_progress(percent, stage, message)
 
     total_started = time.monotonic()
+    las_paths = [las_path] if isinstance(las_path, Path) else [Path(item) for item in las_path]
+    if not las_paths:
+        raise ValueError("지형 파일이 없습니다.")
     stage_root = project_root / "B04_PreProcess"
     processed_dir = stage_root / "processed"
     models_dir = stage_root / "models"
@@ -140,7 +154,7 @@ def run_surface_analysis(
 
     # 0. 입력 세대 검증: LAS가 바뀌었으면 모든 캐시를 재계산한다 (PLAN B-2)
     identity_path = processed_dir / "source_identity.json"
-    current_identity = _source_identity(las_path)
+    current_identity = _source_identity(las_paths)
     stored_identity: dict[str, Any] | None = None
     if identity_path.is_file():
         try:
@@ -154,10 +168,12 @@ def run_surface_analysis(
     if rebuild or not structured_path.is_file():
         _report(10, "structurize", "LAS 구조화 중")
         step_started = time.monotonic()
-        structured_path = structurize_las(las_path, processed_dir)
+        structured_path = structurize_las(las_paths, processed_dir)
         atomic_write_json(identity_path, current_identity)
         logger.info(
-            "B04 LAS 구조화 완료: %s (%.1fs)", las_path.name, time.monotonic() - step_started
+            "B04 LAS 구조화 완료: %s (%.1fs)",
+            ", ".join(path.name for path in las_paths),
+            time.monotonic() - step_started,
         )
     else:
         _report(10, "structurize", "구조화 캐시 재사용")
@@ -254,7 +270,7 @@ def run_surface_analysis(
         "z": [float(bounds[2, 0]), float(bounds[2, 1])],
     }
     download_geodata(
-        project_root, processed_dir, las_bounds_dict, las_path.parent, rebuild, report=_report
+        project_root, processed_dir, las_bounds_dict, las_paths[0].parent, rebuild, report=_report
     )
 
     # 3-4. 도엽등고선 3D 서피스 — LAS가 있어도 참고용으로 같이 만들어 영구저장한다

B04_PreProcess/B04_PreProcess_Engine_Structurize.py

@@ -1,82 +1,308 @@
-"""B04 LAS/LAZ 고속 구조화 엔진."""
+"""B04 LAS/LAZ 고속 구조화 엔진 — 여러 장을 한 벌로 병합한다 (2026-09-06 사용자 확정).
 
+드론 라이다는 사업지가 넓으면 도엽별로 여러 장이 온다. 여기서 **합친 범위**로 한 벌을
+만들고, 뒤 단계(지면필터·모델·등고선·배수)는 받는 형식이 그대로라 손대지 않는다.
+
+점이 임계를 넘으면 **칸(기본 0.5m)마다 최저점 하나만** 남긴다(씨닝). 설계가 쓰는 격자가
+1m(지면필터 2m·CSF 천 1.5m)라 0.5m 는 설계보다 촘촘해 결과 표고가 사실상 같고, 30GB 두
+장이 메모리 29GB → 1.4GB 로 내려간다. 임계 아래면 원본 점을 그대로 쓴다 — 작은 자료의
+결과는 바뀌지 않는다.
+"""
+
+import logging
 import os
 import tempfile
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from pathlib import Path
+from typing import Any
 
 import laspy
 import numpy as np
 
 from common_util.common_util_json import replace_with_retry
-from config.config_system import SURFACE_DEFAULT_RGB_VALUE, SURFACE_LAS_CHUNK_SIZE
+from config.config_system import (
+    SURFACE_DEFAULT_RGB_VALUE,
+    SURFACE_LAS_CHUNK_SIZE,
+    SURFACE_MERGE_MAX_GAP_M,
+    SURFACE_THIN_CELL_SIZE_M,
+    SURFACE_THIN_MAX_CELLS,
+    SURFACE_THIN_TRIGGER_POINTS,
+)
+
+logger = logging.getLogger(__name__)
+
+ProgressCallback = Callable[[int], None]
+PathLike = str | Path
+# 청크에서 점과 함께 옮기는 속성들 — 파일에 없으면 기본값이 남는다.
+_ATTRIBUTES = ("intensity", "rgb", "return_number", "number_of_returns", "classification")
+
+
+def _as_list(las_path: PathLike | Sequence[PathLike]) -> list[Path]:
+    if isinstance(las_path, (str, Path)):
+        return [Path(las_path)]
+    return [Path(item) for item in las_path]
+
+
+def point_cloud_extent(path: PathLike) -> tuple[int, tuple[float, float, float, float]]:
+    """머리글만 읽어 점 수와 XY 범위를 돌려준다 — 파일 크기와 무관하게 즉시 끝난다."""
+    with laspy.open(Path(path)) as las_file:
+        header = las_file.header
+        return int(header.point_count), (
+            float(header.mins[0]),
+            float(header.mins[1]),
+            float(header.maxs[0]),
+            float(header.maxs[1]),
+        )
+
+
+def merge_gap_error(
+    paths: Sequence[PathLike], gap_m: float = SURFACE_MERGE_MAX_GAP_M
+) -> str | None:
+    """서로 멀리 떨어진 지형 파일이 섞였는지 — 문제면 안내 문구, 없으면 None.
+
+    다른 사업지 파일이나 좌표계가 다른 파일이 섞이면 합친 범위가 통째로 어긋나 격자가
+    터진다. 도엽으로 나뉜 자료는 경계가 맞닿으므로 여유를 두고 **어느 파일과도 만나지
+    않는 파일**만 걸러 낸다.
+    """
+    sources = _as_list(paths)
+    if len(sources) < 2:
+        return None
+    boxes = [(path, point_cloud_extent(path)[1]) for path in sources]
+    for index, (path, box) in enumerate(boxes):
+        near = any(
+            box[0] - gap_m <= other[2]
+            and other[0] - gap_m <= box[2]
+            and box[1] - gap_m <= other[3]
+            and other[1] - gap_m <= box[3]
+            for other_index, (_, other) in enumerate(boxes)
+            if other_index != index
+        )
+        if not near:
+            return (
+                f"지형 파일 「{path.name}」의 좌표가 다른 파일과 "
+                f"{int(gap_m):,}m 넘게 떨어져 있습니다."
+                " 같은 사업지의 파일인지, 좌표계가 같은지 확인해 주십시오."
+            )
+    return None
+
+
+class _Merged:
+    """합친 점을 담는 그릇 — 원본 유지형과 씨닝형이 같은 모양으로 낸다."""
+
+    def __init__(self, capacity: int) -> None:
+        self.xyz = np.empty((capacity, 3), dtype=np.float64)
+        self.intensity = np.zeros(capacity, dtype=np.uint16)
+        self.rgb = np.full((capacity, 3), SURFACE_DEFAULT_RGB_VALUE, dtype=np.uint8)
+        self.return_number = np.ones(capacity, dtype=np.uint8)
+        self.number_of_returns = np.ones(capacity, dtype=np.uint8)
+        self.classification = np.zeros(capacity, dtype=np.uint8)
+        self.size = 0
+
+    def arrays(self) -> dict[str, np.ndarray]:
+        end = self.size
+        return {
+            "xyz": self.xyz[:end],
+            "intensity": self.intensity[:end],
+            "rgb": self.rgb[:end],
+            "return_number": self.return_number[:end],
+            "number_of_returns": self.number_of_returns[:end],
+            "classification": self.classification[:end],
+        }
+
+    def append(self, columns: dict[str, np.ndarray]) -> None:
+        count = len(columns["x"])
+        section = slice(self.size, self.size + count)
+        self.xyz[section, 0] = columns["x"]
+        self.xyz[section, 1] = columns["y"]
+        self.xyz[section, 2] = columns["z"]
+        for key in _ATTRIBUTES:
+            if key in columns:
+                getattr(self, key)[section] = columns[key]
+        self.size += count
+
+
+def _chunk_columns(chunk: Any, dimensions: set[str]) -> dict[str, np.ndarray]:
+    """청크에서 쓸 값만 꺼낸다. 파일에 없는 항목은 키를 빼서 기본값이 남게 한다."""
+    columns: dict[str, np.ndarray] = {
+        "x": np.asarray(chunk.x, dtype=np.float64),
+        "y": np.asarray(chunk.y, dtype=np.float64),
+        "z": np.asarray(chunk.z, dtype=np.float64),
+    }
+    if "intensity" in dimensions:
+        columns["intensity"] = np.asarray(chunk.intensity, dtype=np.uint16)
+    if {"red", "green", "blue"}.issubset(dimensions):
+        colors = np.stack(
+            [
+                np.asarray(chunk.red, dtype=np.float64),
+                np.asarray(chunk.green, dtype=np.float64),
+                np.asarray(chunk.blue, dtype=np.float64),
+            ],
+            axis=1,
+        )
+        if colors.size and float(colors.max()) > 255.0:
+            colors /= 256.0
+        columns["rgb"] = colors.clip(0, 255).astype(np.uint8)
+    if {"return_number", "number_of_returns"}.issubset(dimensions):
+        columns["return_number"] = np.asarray(chunk.return_number, dtype=np.uint8)
+        columns["number_of_returns"] = np.asarray(chunk.number_of_returns, dtype=np.uint8)
+    if "classification" in dimensions:
+        columns["classification"] = np.asarray(chunk.classification, dtype=np.uint8)
+    return columns
+
+
+class _ThinGrid:
+    """씨닝형 — **지면 분류점은 전부** 남기고, 나머지는 칸마다 최저점 하나만 남긴다.
+
+    설계 지표면을 만드는 것은 지면점이다(업체가 분류해 준 ASPRS class 2). 그 점을 하나도
+    버리지 않으므로 **지면 결과는 씨닝 전과 완전히 같다**(2026-09-06 용화 실측: 1m 지면
+    격자 141,969칸 전부 표고 차이 0). 지면점은 원본의 2~3%뿐이라 남겨도 가볍다.
+
+    나머지(수목·구조물·잡음)는 칸마다 최저점만 남긴다 — 분류가 없는 자료에서 CSF·PMF가
+    지면을 찾을 밑그림으로 충분하다(CSF 천 간격 1.5m > 칸 0.5m).
+
+    한 청크 안에서 같은 칸이 여러 번 나오면 뒤에 쓴 값이 이겨 최저점이 아니게 된다.
+    그래서 청크를 (칸, 표고)로 정렬해 **칸마다 첫 점**만 골라 낸 뒤 격자와 견준다.
+    """
+
+    #: ASPRS 지면 분류 코드.
+    GROUND_CLASS = 2
+
+    def __init__(self, bounds: np.ndarray, cell_size: float) -> None:
+        self.cell_size = cell_size
+        self.x_min = float(bounds[0, 0])
+        self.y_min = float(bounds[1, 0])
+        self.width = int(np.ceil((float(bounds[0, 1]) - self.x_min) / cell_size)) + 1
+        self.height = int(np.ceil((float(bounds[1, 1]) - self.y_min) / cell_size)) + 1
+        cells = self.width * self.height
+        if cells > SURFACE_THIN_MAX_CELLS:
+            raise ValueError(
+                "지형 자료의 합친 범위가 너무 넓습니다."
+                " 같은 사업지의 파일인지, 좌표계가 같은지 확인해 주십시오."
+            )
+        self.best_z = np.full(cells, np.inf, dtype=np.float64)
+        self.x = np.zeros(cells, dtype=np.float64)
+        self.y = np.zeros(cells, dtype=np.float64)
+        self.intensity = np.zeros(cells, dtype=np.uint16)
+        self.rgb = np.full((cells, 3), SURFACE_DEFAULT_RGB_VALUE, dtype=np.uint8)
+        self.return_number = np.ones(cells, dtype=np.uint8)
+        self.number_of_returns = np.ones(cells, dtype=np.uint8)
+        self.classification = np.zeros(cells, dtype=np.uint8)
+        # 그대로 남길 지면점 — 청크마다 모아 두었다가 마지막에 잇는다.
+        self.ground: list[dict[str, np.ndarray]] = []
+
+    def add(self, columns: dict[str, np.ndarray]) -> None:
+        classification = columns.get("classification")
+        if classification is not None:
+            is_ground = classification == self.GROUND_CLASS
+            if is_ground.any():
+                self.ground.append({key: value[is_ground] for key, value in columns.items()})
+                keep = ~is_ground
+                columns = {key: value[keep] for key, value in columns.items()}
+        x, y, z = columns["x"], columns["y"], columns["z"]
+        if not len(x):
+            return
+        grid_x = np.clip(((x - self.x_min) / self.cell_size).astype(np.int64), 0, self.width - 1)
+        grid_y = np.clip(((y - self.y_min) / self.cell_size).astype(np.int64), 0, self.height - 1)
+        cell = grid_y * self.width + grid_x
+        order = np.lexsort((z, cell))
+        sorted_cell = cell[order]
+        first = np.ones(len(order), dtype=bool)
+        first[1:] = sorted_cell[1:] != sorted_cell[:-1]
+        candidate = order[first]
+        candidate_cell = cell[candidate]
+        better = z[candidate] < self.best_z[candidate_cell]
+        chosen = candidate[better]
+        target = candidate_cell[better]
+        self.best_z[target] = z[chosen]
+        self.x[target] = x[chosen]
+        self.y[target] = y[chosen]
+        for key in _ATTRIBUTES:
+            if key in columns:
+                getattr(self, key)[target] = columns[key][chosen]
+
+    def collect(self) -> _Merged:
+        occupied = np.flatnonzero(np.isfinite(self.best_z))
+        ground_count = sum(len(item["x"]) for item in self.ground)
+        merged = _Merged(len(occupied) + ground_count)
+        merged.xyz[: len(occupied), 0] = self.x[occupied]
+        merged.xyz[: len(occupied), 1] = self.y[occupied]
+        merged.xyz[: len(occupied), 2] = self.best_z[occupied]
+        for key in _ATTRIBUTES:
+            getattr(merged, key)[: len(occupied)] = getattr(self, key)[occupied]
+        merged.size = len(occupied)
+        for item in self.ground:
+            merged.append(item)
+        return merged
+
+
+def _headers(sources: list[Path]) -> tuple[int, np.ndarray, bool]:
+    """전체 점 수·합친 범위(3x2)·색 보유 여부를 머리글만 읽어 구한다."""
+    total = 0
+    has_rgb = False
+    mins = np.full(3, np.inf, dtype=np.float64)
+    maxs = np.full(3, -np.inf, dtype=np.float64)
+    for source in sources:
+        with laspy.open(source) as las_file:
+            header = las_file.header
+            total += int(header.point_count)
+            mins = np.minimum(mins, np.asarray(header.mins, dtype=np.float64))
+            maxs = np.maximum(maxs, np.asarray(header.maxs, dtype=np.float64))
+            dimensions = set(header.point_format.dimension_names)
+            has_rgb = has_rgb or {"red", "green", "blue"}.issubset(dimensions)
+    if not np.isfinite(mins).all():
+        mins = np.zeros(3, dtype=np.float64)
+        maxs = np.zeros(3, dtype=np.float64)
+    return total, np.column_stack((mins, maxs)), has_rgb
+
+
+def _merge_sources(
+    sources: list[Path],
+    total_points: int,
+    bounds: np.ndarray,
+    thin: bool,
+    progress_callback: ProgressCallback | None,
+) -> _Merged:
+    grid = _ThinGrid(bounds, SURFACE_THIN_CELL_SIZE_M) if thin else None
+    merged = _Merged(total_points) if grid is None else None
+    done = 0
+    for source in sources:
+        with laspy.open(source) as las_file:
+            dimensions = set(las_file.header.point_format.dimension_names)
+            for chunk in las_file.chunk_iterator(SURFACE_LAS_CHUNK_SIZE):
+                columns = _chunk_columns(chunk, dimensions)
+                if grid is not None:
+                    grid.add(columns)
+                else:
+                    merged.append(columns)
+                done += len(columns["x"])
+                if progress_callback:
+                    progress_callback(int(done / total_points * 100) if total_points else 100)
+    return grid.collect() if grid is not None else merged
 
 
 def structurize_las(
-    las_path: str | Path,
+    las_path: PathLike | Sequence[PathLike],
     output_dir: str | Path,
-    progress_callback: Callable[[int], None] | None = None,
+    progress_callback: ProgressCallback | None = None,
 ) -> Path:
-    """LAS/LAZ 속성을 청크로 읽어 B04 structured.npz로 원자적 저장한다."""
+    """지형 파일 한 장 또는 여러 장을 청크로 읽어 B04 structured.npz로 원자적 저장한다."""
-    source = Path(las_path)
+    sources = _as_list(las_path)
+    if not sources:
+        raise ValueError("구조화할 지형 파일이 없습니다.")
     target_dir = Path(output_dir)
     target_dir.mkdir(parents=True, exist_ok=True)
     target = target_dir / "structured.npz"
 
-    with laspy.open(source) as las_file:
+    total_points, bounds, has_rgb = _headers(sources)
-        header = las_file.header
+    thin = total_points > SURFACE_THIN_TRIGGER_POINTS
-        total_points = int(header.point_count)
+    merged = _merge_sources(sources, total_points, bounds, thin, progress_callback)
-        point_format = header.point_format
+    logger.info(
-        dimensions = set(point_format.dimension_names)
+        "B04 구조화: 파일 %d장 원본 %d점 → 저장 %d점 (씨닝 %s)",
-        has_rgb = {"red", "green", "blue"}.issubset(dimensions)
+        len(sources),
-        has_intensity = "intensity" in dimensions
+        total_points,
-        has_returns = {"return_number", "number_of_returns"}.issubset(dimensions)
+        merged.size,
-        has_classification = "classification" in dimensions
+        f"{SURFACE_THIN_CELL_SIZE_M}m 칸" if thin else "없음",
-        bounds = np.array(
+    )
-            [
-                [float(header.mins[0]), float(header.maxs[0])],
-                [float(header.mins[1]), float(header.maxs[1])],
-                [float(header.mins[2]), float(header.maxs[2])],
-            ],
-            dtype=np.float64,
-        )
-
-        xyz = np.empty((total_points, 3), dtype=np.float64)
-        intensity = np.zeros(total_points, dtype=np.uint16)
-        rgb = np.full((total_points, 3), SURFACE_DEFAULT_RGB_VALUE, dtype=np.uint8)
-        return_number = np.ones(total_points, dtype=np.uint8)
-        number_of_returns = np.ones(total_points, dtype=np.uint8)
-        classification = np.zeros(total_points, dtype=np.uint8)
-
-        offset = 0
-        for chunk in las_file.chunk_iterator(SURFACE_LAS_CHUNK_SIZE):
-            chunk_size = len(chunk)
-            section = slice(offset, offset + chunk_size)
-            xyz[section, 0] = np.asarray(chunk.x, dtype=np.float64)
-            xyz[section, 1] = np.asarray(chunk.y, dtype=np.float64)
-            xyz[section, 2] = np.asarray(chunk.z, dtype=np.float64)
-            if has_intensity:
-                intensity[section] = np.asarray(chunk.intensity, dtype=np.uint16)
-            if has_rgb:
-                colors = np.stack(
-                    [
-                        np.asarray(chunk.red, dtype=np.float64),
-                        np.asarray(chunk.green, dtype=np.float64),
-                        np.asarray(chunk.blue, dtype=np.float64),
-                    ],
-                    axis=1,
-                )
-                if colors.size and float(colors.max()) > 255.0:
-                    colors /= 256.0
-                rgb[section] = colors.clip(0, 255).astype(np.uint8)
-            if has_returns:
-                return_number[section] = np.asarray(chunk.return_number, dtype=np.uint8)
-                number_of_returns[section] = np.asarray(chunk.number_of_returns, dtype=np.uint8)
-            if has_classification:
-                classification[section] = np.asarray(chunk.classification, dtype=np.uint8)
-            offset += chunk_size
-            if progress_callback:
-                progress_callback(int(offset / total_points * 100) if total_points else 100)
 
     temporary_path: Path | None = None
     try:
@@ -90,14 +316,12 @@ def structurize_las(
             temporary_path = Path(temporary.name)
             np.savez_compressed(
                 temporary,
-                xyz=xyz,
+                **merged.arrays(),
-                intensity=intensity,
-                rgb=rgb,
-                return_number=return_number,
-                number_of_returns=number_of_returns,
-                classification=classification,
                 bounds=bounds,
-                total_points=np.array([total_points], dtype=np.int64),
+                total_points=np.array([merged.size], dtype=np.int64),
+                source_point_count=np.array([total_points], dtype=np.int64),
+                source_file_count=np.array([len(sources)], dtype=np.int64),
+                thinned=np.array([int(thin)], dtype=np.int8),
                 has_rgb=np.array([int(has_rgb)], dtype=np.int8),
             )
             temporary.flush()
@@ -109,6 +333,6 @@ def structurize_las(
         if temporary_path is not None:
             temporary_path.unlink(missing_ok=True)
 
-    if progress_callback and total_points == 0:
+    if progress_callback:
         progress_callback(100)
     return target

B04_PreProcess/B04_PreProcess_Repository.py

@@ -6,7 +6,7 @@ terrain_layers(지형 레이어) 테이블에 메타데이터와 상대 경로
 """
 
 import json
-from pathlib import PurePosixPath
+from pathlib import Path, PurePosixPath
 from typing import Any
 from uuid import UUID
 
@@ -230,6 +230,25 @@ async def list_project_point_cloud_inputs(
     ]
 
 
+async def list_project_point_cloud_paths(
+    connection: aiomysql.Connection, project_id: UUID, project_root: Path
+) -> list[Path]:
+    """전처리가 병합할 지형 파일 경로 목록 — 실제로 있는 파일만 (2026-09-06 다중 입력).
+
+    교체된 옛 행(`SUPERSEDED`)은 조회에서 이미 빠지므로, 지금 살아 있는 파일만 남는다.
+    """
+    rows = await list_project_point_cloud_inputs(connection, project_id)
+    paths: list[Path] = []
+    for row in rows:
+        raw = str(row.get("raw_file_path") or "")
+        if not raw:
+            continue
+        path = project_root / Path(raw)
+        if path.is_file() and path not in paths:
+            paths.append(path)
+    return paths
+
+
 async def list_surface_models(
     connection: aiomysql.Connection, project_id: UUID
 ) -> list[dict[str, Any]]:

B04_PreProcess/B04_PreProcess_Router.py

@@ -26,6 +26,7 @@ from B04_PreProcess.B04_PreProcess_Repository import (
     clear_confirmed_surface_models,
     get_input_file,
     list_project_point_cloud_inputs,
+    list_project_point_cloud_paths,
     list_surface_models,
     save_surface_analysis_to_db,
 )
@@ -117,6 +118,10 @@ async def analyze_surface(
                     status_code=404,
                     content={"status": "error", "message": "원본 LAS 파일을 찾을 수 없습니다."},
                 )
+            # 지형 파일이 여러 장이면 합쳐서 다시 만든다 — 자동 전처리와 같은 대상을 쓴다.
+            terrain_paths = await list_project_point_cloud_paths(
+                connection, project_id, project_root
+            )
 
             # 분석 시작 진행률 기록 (별도 스레드의 콜백은 파일에만 원자적 기록).
             write_surface_progress(project_root, 5, "analyzing", "WF1 분석을 시작합니다.")
@@ -128,7 +133,7 @@ async def analyze_surface(
             result = await asyncio.to_thread(
                 run_surface_analysis,
                 project_root,
-                las_path,
+                terrain_paths or [las_path],
                 source_filters=source_filters,
                 methods=methods,
                 force=request.force,

config/config_system_terrain.py

@@ -14,6 +14,20 @@ MESH_SMOOTHING_ITERATIONS = int(os.getenv("MESH_SMOOTHING_ITERATIONS", "0"))
 SURFACE_LAS_CHUNK_SIZE = int(os.getenv("SURFACE_LAS_CHUNK_SIZE", "500000"))
 SURFACE_DEFAULT_RGB_VALUE = int(os.getenv("SURFACE_DEFAULT_RGB_VALUE", "128"))
 SURFACE_GRID_CELL_SIZE_M = float(os.getenv("SURFACE_GRID_CELL_SIZE_M", "2.0"))
+
+# 지형 파일 여러 장 병합 (2026-09-06 사용자 확정)
+# ─────────────────────────────────────────────────────────────────────────
+# 다른 사업지 파일이 섞여 들어오면 합친 범위가 통째로 어긋난다 — 어느 파일과도 이
+# 거리 안에서 만나지 않는 파일은 업로드에서 막는다. 임도는 길어도 2~3km 라 5km 면
+# 넉넉하다(2026-09-06 사용자 확정). 도엽이 나뉜 자료는 경계가 맞닿아 걸리지 않는다.
+SURFACE_MERGE_MAX_GAP_M = float(os.getenv("SURFACE_MERGE_MAX_GAP_M", "5000"))
+# 점이 이 수를 넘으면 아래 칸 크기로 씨닝한다(칸마다 최저점 하나). 넘지 않으면 원본
+# 그대로 쓴다 — 작은 자료의 결과는 바뀌지 않는다. 1억점 = 메모리 약 3.2GB.
+SURFACE_THIN_TRIGGER_POINTS = int(os.getenv("SURFACE_THIN_TRIGGER_POINTS", "100000000"))
+# 씨닝 칸 크기(m). 설계 격자 1m·지면필터 2m·CSF 천 1.5m 보다 촘촘해야 결과가 안 변한다.
+SURFACE_THIN_CELL_SIZE_M = float(os.getenv("SURFACE_THIN_CELL_SIZE_M", "0.5"))
+# 씨닝 격자가 이보다 많아지면 범위가 비정상이다(먼 파일이 섞였거나 좌표계 불일치).
+SURFACE_THIN_MAX_CELLS = int(os.getenv("SURFACE_THIN_MAX_CELLS", "400000000"))
 SURFACE_GRID_HEIGHT_THRESHOLD_M = float(os.getenv("SURFACE_GRID_HEIGHT_THRESHOLD_M", "1.5"))
 
 # CSF (Cloth Simulation Filter) 지면 분류 파라미터