"""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