perf(전처리 화면): 343-3 B04 다시 열기 받는 양 — 보관함 저장 실패 고침 · 점군 고정 표본 이진 · 꺼진 층은 켤 때 · 남은 층 ETag · 등고선 주소 한 벌

- b_asset_cache — 쓰기 끝 = 트랜잭션 완료(용량 초과 취소 잡음) · 실패 한 줄 경고(까닭당 한 번) · onblocked · onversionchange · 열기 실패 null 안 기억 · 지우기는 색인 열쇠 · 앞머리 범위(값 안 읽음)
- 점군 미리보기 — 전처리 때 고정 표본 이진(preview_points_{필터}.bin · 원천 지문 같으면 건너뜀) · GET point-cloud/binary(ETag) · 화면은 fetchCachedBytes(IndexedDB) + decodePointPreview(py 짝)
- B04 지도 — 켜진 층만 처음에 받음 · 꺼진 층은 켤 때 · 같은 층 겹쳐 받기 막음
- geojson(지적도 · 행정구역 · 등고선) 파일 그대로 cached_file_response(ETag) · 등고선 단순화본은 원본이 새로우면 다시 만듦
- 등고선 주소 contourAssetUrl 하나(B04 · 미리 받기) — B05 두 곳은 sub5 몫
- 시험 test_343_3_browser_cache 13건 + helper_343_3_asset_cache.cjs

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL
This commit is contained in:
2026-10-16 11:21:51 +09:00
co-authored by Claude Opus 5.5
parent 1fcdf5aa4c
commit e0051bff72
11 changed files with 830 additions and 131 deletions
@@ -0,0 +1,97 @@
"""B04 점군 미리보기 — 고정 표본 이진 파일(전처리 때 한 번 · ETag 로 내보냄).
JSON(27.8 MB · 요청마다 점군 풀고 표본)을 대신한다. 표본 규칙은 JSON 길과 같다
(`sample_points` · 같은 씨앗) — 같은 점 · 같은 float32 값.
파일 = [머리 길이 uint32 LE][머리 JSON utf-8][0 채움 → 4 배수][xyz float32 × 3n][rgb uint16 × 3n].
머리 = {version, source(원천 npz mtime_ns:크기), point_count, sampled_count, bounds, has_rgb}.
TS 짝(읽기) = `B04_PreProcess_Api_Fetch.ts` `decodePointPreview`.
"""
from __future__ import annotations
import json
import struct
from pathlib import Path
from typing import Any
import numpy as np
from common_util.common_util_atomic import atomic_write_bytes
POINT_CLOUD_SAMPLE_LIMIT = 500_000
PREVIEW_VERSION = 1
_SAMPLE_SEED = 20260710
def sample_points(xyz: np.ndarray, rgb: np.ndarray | None) -> tuple[np.ndarray, np.ndarray | None]:
"""화면용 고정 표본 — 상한을 넘으면 같은 씨앗으로 고른다(열 때마다 같은 그림)."""
source_count = int(len(xyz))
if rgb is not None and (rgb.ndim == 0 or len(rgb) != source_count):
rgb = None
if source_count <= POINT_CLOUD_SAMPLE_LIMIT:
return xyz, rgb
indexes = np.random.default_rng(_SAMPLE_SEED).choice(
source_count, POINT_CLOUD_SAMPLE_LIMIT, replace=False
)
return xyz[indexes], (rgb[indexes] if rgb is not None else None)
def preview_path(processed_dir: Path, filter_key: str | None) -> Path:
return processed_dir / f"preview_points_{filter_key or 'raw'}.bin"
def _stamp(path: Path) -> str:
stat = path.stat()
return f"{stat.st_mtime_ns}:{stat.st_size}"
def read_preview_header(path: Path) -> dict[str, Any] | None:
"""머리만 읽는다 — 없거나 깨졌으면 None."""
try:
with path.open("rb") as file:
(length,) = struct.unpack("<I", file.read(4))
return json.loads(file.read(length).decode("utf-8"))
except (OSError, ValueError, struct.error):
return None
def ensure_point_preview(source_path: Path, filter_key: str | None, xyz_source: Any = None) -> Path:
"""원천 npz(structured · ground_points_*) 그대로면 있는 파일 · 아니면 한 번 만든다.
`xyz_source` — 이미 메모리에 있는 xyz(또는 쓸 때 푸는 대리자) · 없으면 npz 에서 푼다.
"""
target = preview_path(source_path.parent, filter_key)
stamp = _stamp(source_path)
header = read_preview_header(target)
if header and header.get("version") == PREVIEW_VERSION and header.get("source") == stamp:
return target
with np.load(source_path) as data:
xyz = np.asarray(data["xyz"] if xyz_source is None else xyz_source, dtype=np.float32)
bounds = np.asarray(data["bounds"], dtype=np.float64)
point_count = int(data["point_count"]) if "point_count" in data else int(len(xyz))
rgb = np.asarray(data["rgb"]) if "rgb" in data else None
sample, rgb_sample = sample_points(xyz, rgb)
header = {
"version": PREVIEW_VERSION,
"source": stamp,
"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]),
},
"has_rgb": rgb_sample is not None,
}
head = json.dumps(header).encode("utf-8")
pad = (-(4 + len(head))) % 4
parts = [struct.pack("<I", len(head)), head, b"\0" * pad]
parts.append(np.ascontiguousarray(sample, dtype="<f4").tobytes())
if rgb_sample is not None:
parts.append(np.ascontiguousarray(rgb_sample, dtype="<u2").tobytes())
atomic_write_bytes(target, b"".join(parts))
return target