eomsangdonandClaude Opus 5.5
e0051bff72
- 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
98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
"""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
|