perf(전처리): 343-1 같은 지형 파일 다시 올림 = 전처리 결과 살림 · 다시 돌 때 점군 안 풂 · 도엽 병합 · 국가 GIS · 타일 기억 열쇠

- B03 재올림 — 지형 파일이 source_identity.json 과 같으면 B04 폴더 · 지표면 DB 행 남기고 단계 표시만 되돌림
- 확인 창 「처음부터 다시」 따로 고름(fresh_start) · 409 응답에 preprocess_kept
- B04 다시 돎 — 점 수 · 평균 Z 를 structured_stats.json 에 · xyz 는 쓸 때만 풂
- 도엽 병합 열쇠 = 도엽 번호 목록 · 국가 GIS 열쇠 = 범위 · 좌표계
- 지도 타일 메모리 기억 열쇠에 mtime · 크기
- 시험 test_343_1_preprocess_keep 13건

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

B04_PreProcess/B04_PreProcess_Engine.py

@@ -55,6 +55,87 @@ def _source_identity(las_paths: list[Path]) -> dict[str, Any]:
     return files[0] if len(files) == 1 else {"files": files}
 
 
+class _LazyXyz:
+    """structured.npz 의 xyz — 처음 값을 쓸 때만 푼다. 길이는 저장해 둔 점 수로 답한다."""
+
+    def __init__(self, structured: Any, count: int, array: np.ndarray | None = None) -> None:
+        self._structured = structured
+        self._count = count
+        self._array = array
+
+    def _load(self) -> np.ndarray:
+        if self._array is None:
+            self._array = self._structured["xyz"]
+        return self._array
+
+    def __len__(self) -> int:
+        return self._count
+
+    def __getitem__(self, key: Any) -> Any:
+        return self._load()[key]
+
+    def __array__(self, dtype: Any = None, copy: Any = None) -> np.ndarray:
+        return np.asarray(self._load(), dtype=dtype)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._load(), name)
+
+
+class _LazyStructured(dict):
+    """구조화 결과 — 열쇠마다 처음 쓸 때만 푼다(저장본이 다 맞는 다시 돎은 점군을 안 풂)."""
+
+    def __init__(self, structured: Any, xyz: _LazyXyz) -> None:
+        super().__init__()
+        self._structured = structured
+        self["xyz"] = xyz
+
+    # 옛 dict 와 같은 열쇠만 보인다 — 필터가 열쇠 유무로 갈리는 일이 없게.
+    _KEYS = ("classification",)
+
+    def __missing__(self, key: str) -> Any:
+        if key not in self._KEYS:
+            raise KeyError(key)
+        value = self._structured[key]
+        self[key] = value
+        return value
+
+    def __contains__(self, key: object) -> bool:
+        return dict.__contains__(self, key) or key in self._KEYS
+
+
+def _structured_stats(structured: Any, structured_path: Path) -> tuple[dict[str, Any], Any]:
+    """점 수 · 평균 Z 를 `structured_stats.json` 에서 읽는다 — 없거나 낡으면 한 번 풀어 적는다.
+
+    돌려주는 둘째 값 = 이번에 푼 xyz(없으면 None) — 같은 부름 안에서 다시 풀지 않게.
+    """
+    stats_path = structured_path.parent / "structured_stats.json"
+    stat = structured_path.stat()
+    stamp = {"size_bytes": int(stat.st_size), "mtime_ns": int(stat.st_mtime_ns)}
+    try:
+        stored = json.loads(stats_path.read_text(encoding="utf-8"))
+        if stored.get("structured") == stamp:
+            return stored, None
+    except (OSError, ValueError, AttributeError):
+        pass
+    xyz = structured["xyz"]
+    count = int(len(xyz))
+    stats = {
+        "structured": stamp,
+        "point_count": count,
+        "mean_z": float(np.mean(xyz[:, 2])) if count else None,
+    }
+    atomic_write_json(stats_path, stats)
+    return stats, xyz
+
+
+def _key_matches(key_path: Path, key: Any) -> bool:
+    """결과 옆 열쇠 파일이 지금 입력과 같은지."""
+    try:
+        return json.loads(key_path.read_text(encoding="utf-8")) == key
+    except (OSError, ValueError):
+        return False
+
+
 def _relative_to_project(project_root: Path, path: Path) -> str:
     """프로젝트 루트 기준 posix 상대 경로 문자열."""
     return path.relative_to(project_root).as_posix()
@@ -179,13 +260,14 @@ def run_surface_analysis(
         _report(10, "structurize", "구조화 캐시 재사용")
         logger.info("B04 구조화 캐시 재사용: %s", structured_path.name)
     with np.load(structured_path) as structured:
-        xyz = structured["xyz"]
+        # 점 수 · 평균 Z 는 저장값 — 저장본이 다 맞으면 점군(841 MB 압축)을 풀지 않는다.
+        stored_stats, loaded_xyz = _structured_stats(structured, structured_path)
         bounds = structured["bounds"]
-        total_points = int(len(xyz))
+        total_points = int(stored_stats["point_count"])
         stats = {
             "min_z": float(bounds[2, 0]),
             "max_z": float(bounds[2, 1]),
-            "mean_z": float(np.mean(xyz[:, 2])) if total_points else None,
+            "mean_z": stored_stats["mean_z"],
         }
         bounds_dict = {
             "x_min": float(bounds[0, 0]),
@@ -193,7 +275,8 @@ def run_surface_analysis(
             "y_min": float(bounds[1, 0]),
             "y_max": float(bounds[1, 1]),
         }
-        data = {"xyz": xyz, "bounds": bounds, "classification": structured["classification"]}
+        data = _LazyStructured(structured, _LazyXyz(structured, total_points, loaded_xyz))
+        data["bounds"] = bounds
 
         # 2. 지면 필터 실행 — mask_{filter}.npy 영구 캐시 우선 재사용 (PLAN A-1)
         _report(40, "ground_filter", "지면 필터 적용 중")
@@ -386,14 +469,21 @@ def download_geodata(
 
         # 등고선은 국가 gpkg 추가 시점이 전처리보다 늦을 수 있으므로 개별로 존재 여부를 확인한다.
         contour_geojson = processed_dir / "등고선_bounds.geojson"
+        # 열쇠 = 받은 범위 · 좌표계 — 노선이 바뀌어 범위가 달라지면 다시 받는다.
+        gis_key_path = processed_dir / "gis_vectors.key.json"
+        gis_key = json.loads(
+            json.dumps({"epsg": project_epsg, "bounds": bounds_dict_for_download}, default=float)
+        )
         if (
             rebuild
             or not any(processed_dir.glob("*_bounds.geojson"))
             or not contour_geojson.exists()
+            or not _key_matches(gis_key_path, gis_key)
         ):
             try:
                 step_started = time.monotonic()
                 download_all_gis_vectors(prj_path, bounds_dict_for_download, processed_dir)
+                atomic_write_json(gis_key_path, gis_key)
                 logger.info(
                     "B04 국가 GIS 벡터 다운로드 완료 (%.1fs)", time.monotonic() - step_started
                 )
@@ -433,9 +523,13 @@ def download_geodata(
             # 선정에서 빠진 zip은 쓰이지 않으므로 정리한다(다른 지역 잔재 포함).
             prune_sheets(sheet_store, sheet_grid)
 
-            # 확보된 도엽을 레이어별 병합 GeoJSON으로 산출 (기존 산출물 있으면 스킵)
+            # 확보된 도엽을 레이어별 병합 GeoJSON으로 산출 — 열쇠(도엽 번호 목록)가 같으면 스킵
+            merge_key_path = processed_dir / "sheet_merge.key.json"
+            merge_key = {"sheets": sorted(str(item) for item in sheet_result["available"])}
             if sheet_result["available"] and (
-                rebuild or not any(processed_dir.glob("도엽_*.geojson"))
+                rebuild
+                or not any(processed_dir.glob("도엽_*.geojson"))
+                or not _key_matches(merge_key_path, merge_key)
             ):
                 from B04_PreProcess.B04_PreProcess_Engine_SheetParser import (
                     parse_and_merge_sheets,
@@ -450,6 +544,7 @@ def download_geodata(
                     {k: v["count"] for k, v in merged.items()},
                     time.monotonic() - step_started,
                 )
+                atomic_write_json(merge_key_path, merge_key)
         except Exception as exc:
             logger.warning("B04 수치지형도 도엽 확보 실패: %s", exc)
     except Exception as exc: