#!/usr/bin/env python3
"""Measure a UV/unroll map — and compare it with the shipped one and the built package.

This is the tool that answers "is the ~40 % of the finger mesh that renders grey *supposed* to be
non-sensing?", with numbers rather than by eye:

* the island's coverage and its bounding box in UV pixels;
* the island in **sensor units** (from ``world_xyz``), so it can be compared with the mesh extent
  the app reports;
* how much of the compiled package's surface has no mapping (that is the grey the operator sees);
* the map's ``field_domain_id`` against the one recorded in the package — a mismatch means the pane
  would refuse live frames (``coordinator.py`` drops a frame whose domain differs from the scene);
* with ``--against``, a second map (normally the shipped one): coverage/bbox deltas and the share of
  pixels whose validity differs.

Offline only; it never runs in the application. Needs only numpy plus the app's own shared package
(the same ``load_legacy_domain`` the compiler uses, so the id is the compiler's id):

    python tools/scene_assets/compare_uv_map.py --map <map.npz> \
        --against <sdk>/sensors/finger/assets/30fps-0703/unroll_map_uv_s0.4.npz \
        --package apps/studio/config/sensors/finger/scene/finger-display-1.0.0.oryscene
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
import zipfile
from pathlib import Path

import numpy as np

_MM_PER_SOURCE_UNIT = 1.0  # the finger export states 1 source unit = 1 mm


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def load_map(path: Path) -> dict:
    with np.load(path, allow_pickle=False) as data:
        arrays = {key: data[key] for key in data.files}
    return arrays


def _scalar(arrays: dict, key: str):
    value = arrays.get(key)
    if value is None:
        return None
    return np.asarray(value).reshape(-1)[0].item() if np.asarray(value).size else None


def describe_map(path: Path, arrays: dict) -> dict:
    """Everything worth knowing about one map, as plain JSON-able values."""
    mask = np.asarray(arrays["valid_mask"]).astype(bool)
    height, width = mask.shape
    rows, cols = np.nonzero(mask)
    report: dict = {
        "path": str(path),
        "sha256": _sha256(path),
        "shape_hw": [int(height), int(width)],
        "valid_pixels": int(mask.sum()),
        "coverage_pct": round(float(mask.mean()) * 100.0, 3),
        "recorded": {
            key: _scalar(arrays, key)
            for key in ("out_width", "out_height", "source_width", "source_height",
                        "flip_v", "edge_band", "edge_expand", "source_scale", "method")
        },
    }
    if rows.size:
        report["island_bbox_px"] = {
            "rows": [int(rows.min()), int(rows.max())],
            "cols": [int(cols.min()), int(cols.max())],
        }
    else:
        report["island_bbox_px"] = None

    world = arrays.get("world_xyz")
    if world is not None and rows.size:
        points = np.asarray(world)[mask]
        finite = np.isfinite(points).all(axis=1)
        if finite.any():
            points = points[finite] * _MM_PER_SOURCE_UNIT
            low, high = points.min(axis=0), points.max(axis=0)
            report["island_sensor_units"] = {
                "low": [round(float(v), 4) for v in low],
                "high": [round(float(v), 4) for v in high],
                "size": [round(float(v), 4) for v in (high - low)],
                "note": "source units; the finger export states 1 unit = 1 mm",
            }
    return report


def field_domain_id(map_path: Path) -> str:
    """The compiler's own domain identity for this map (same code path as the compile)."""
    from orisys_app_shared.scene3d.legacy import load_legacy_domain

    return str(load_legacy_domain(map_path).domain_id)


def package_facts(package: Path) -> dict:
    """What the built package says about the same surface."""
    facts: dict = {"path": str(package)}
    with zipfile.ZipFile(package) as archive:
        names = set(archive.namelist())
        manifest = json.loads(archive.read("manifest.json").decode("utf-8"))
        facts["manifest_keys"] = sorted(manifest.keys())
        for key, value in manifest.items():
            if isinstance(value, dict) and "id" in value:
                facts[f"manifest.{key}.id"] = value["id"]
            elif isinstance(value, str) and "domain" in key:
                facts[f"manifest.{key}"] = value
        for member in ("sensing/surface_valid.npy", "sensing/surface_positions.npy"):
            if member not in names:
                continue
            with archive.open(member) as handle:
                array = np.load(handle, allow_pickle=False)
            facts[member] = list(array.shape)
            if member.endswith("surface_valid.npy"):
                valid = np.asarray(array).astype(bool)
                facts["surface_valid_true"] = int(valid.sum())
                facts["surface_valid_false"] = int((~valid).sum())
                facts["surface_unmapped_pct"] = round(
                    float((~valid).mean()) * 100.0, 3
                )
    return facts


def compare(left: dict, right: dict) -> dict:
    """How two maps differ: coverage, island movement, and which pixels changed."""
    left_mask = np.asarray(left["valid_mask"]).astype(bool)
    right_mask = np.asarray(right["valid_mask"]).astype(bool)
    result: dict = {}
    if left_mask.shape != right_mask.shape:
        result["shape_mismatch"] = [list(left_mask.shape), list(right_mask.shape)]
        return result
    changed = left_mask ^ right_mask
    result["coverage_pct"] = {
        "left": round(float(left_mask.mean()) * 100.0, 3),
        "right": round(float(right_mask.mean()) * 100.0, 3),
        "delta": round(float((left_mask.mean() - right_mask.mean()) * 100.0), 3),
    }
    result["pixels_only_in_left"] = int((left_mask & ~right_mask).sum())
    result["pixels_only_in_right"] = int((right_mask & ~left_mask).sum())
    result["pixels_differing_pct"] = round(float(changed.mean()) * 100.0, 4)
    for name, mask in (("left", left_mask), ("right", right_mask)):
        rows, cols = np.nonzero(mask)
        result[f"{name}_bbox_px"] = (
            [int(rows.min()), int(rows.max()), int(cols.min()), int(cols.max())]
            if rows.size
            else None
        )
    result["identical"] = bool(not changed.any())
    return result


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--map", type=Path, required=True, help="the map to measure")
    parser.add_argument("--against", type=Path, default=None, help="a second map to diff against")
    parser.add_argument("--package", type=Path, default=None, help="the built .oryscene to compare")
    parser.add_argument("--json", action="store_true", help="machine-readable output")
    args = parser.parse_args(argv)

    if not args.map.is_file():
        print(f"no such map: {args.map}", file=sys.stderr)
        return 2

    report: dict = {}
    arrays = load_map(args.map)
    report["map"] = describe_map(args.map, arrays)
    print(f"map            {args.map}")
    info = report["map"]
    print(f"  shape        {info['shape_hw'][1]} x {info['shape_hw'][0]} (W x H)")
    print(f"  coverage     {info['coverage_pct']} % of the domain "
          f"({info['valid_pixels']} pixels)")
    if info["island_bbox_px"]:
        rows, cols = info["island_bbox_px"]["rows"], info["island_bbox_px"]["cols"]
        print(f"  island bbox  rows {rows[0]}..{rows[1]}, cols {cols[0]}..{cols[1]}")
    if "island_sensor_units" in info:
        print(f"  island size  {info['island_sensor_units']['size']} (source units)")
        print(f"  island span  {info['island_sensor_units']['low']} .. "
              f"{info['island_sensor_units']['high']}")
    recorded = {k: v for k, v in info["recorded"].items() if v is not None}
    print(f"  recorded     {recorded}")

    try:
        report["field_domain_id"] = field_domain_id(args.map)
        print(f"  domain id    {report['field_domain_id']}")
    except Exception as exc:  # noqa: BLE001 - the tool reports, it does not crash
        report["field_domain_id_error"] = f"{type(exc).__name__}: {exc}"
        print(f"  domain id    (could not compute: {type(exc).__name__}: {exc})")

    if args.against is not None:
        if not args.against.is_file():
            print(f"no such map to compare against: {args.against}", file=sys.stderr)
            return 2
        other = load_map(args.against)
        report["against"] = describe_map(args.against, other)
        report["comparison"] = compare(arrays, other)
        print(f"\nagainst        {args.against}")
        print(f"  coverage     {report['against']['coverage_pct']} %")
        diff = report["comparison"]
        if "shape_mismatch" in diff:
            print(f"  SHAPE MISMATCH {diff['shape_mismatch']}")
        else:
            print(f"  delta        {diff['coverage_pct']['delta']:+.3f} pp, "
                  f"{diff['pixels_differing_pct']} % of pixels differ")
            print(f"  only here    {diff['pixels_only_in_left']} px; "
                  f"only there {diff['pixels_only_in_right']} px")
            print(f"  verdict      {'IDENTICAL island' if diff['identical'] else 'different island'}")

    if args.package is not None:
        if not args.package.is_file():
            print(f"no such package: {args.package}", file=sys.stderr)
            return 2
        facts = package_facts(args.package)
        report["package"] = facts
        print(f"\npackage        {args.package}")
        if "surface_unmapped_pct" in facts:
            print(f"  surface      {facts['surface_valid_false']} of "
                  f"{facts['surface_valid_true'] + facts['surface_valid_false']} render vertices "
                  f"have no mapping ({facts['surface_unmapped_pct']} %)")
        recorded_ids = {k: v for k, v in facts.items() if k.endswith(".id")}
        if recorded_ids:
            print(f"  recorded ids {recorded_ids}")
        if report.get("field_domain_id"):
            matches = [
                key for key, value in recorded_ids.items()
                if str(value) == report["field_domain_id"]
            ]
            report["package_domain_matches_map"] = bool(matches)
            print(f"  domain match {'yes' if matches else 'NO — the pane would refuse live frames'}")

    if args.json:
        print(json.dumps(report, indent=2, default=str))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())