common.py

#!/usr/bin/env python3
"""Shared helpers for standalone ScanNet self-localization data generation.

This module intentionally does not import the existing VSI-590K construction
code. Several helpers are copied and lightly adapted from the current QA
construction pipeline so this directory can travel as an independent project.
"""

from __future__ import annotations

import base64
import html
import json
import math
import random
from dataclasses import dataclass
from pathlib import Path
from typing import Any

try:
    import numpy as np
except ImportError as exc:  # pragma: no cover - surfaced as a clear runtime error.
    raise RuntimeError("numpy is required for whereami data generation") from exc


Point2D = tuple[float, float]
Point3D = tuple[float, float, float]
Matrix = list[list[float]]
VisibilityKey = tuple[str, str | None]

DISPLAY_FACING_CANDIDATES: tuple[tuple[int, int], ...] = (
    (0, -1),
    (1, -1),
    (1, 0),
    (1, 1),
    (0, 1),
    (-1, 1),
    (-1, 0),
    (-1, -1),
)

DEFAULT_MIN_MULTI_CAMERA_DISTANCE_M = 0.5
DEFAULT_MIN_MULTI_YAW_CHANGE_DEG = 20.0
DEFAULT_COHERENT_WINDOW_SIZE = 180


@dataclass(frozen=True)
class SceneManifest:
    scene_id: str
    dataset: str
    video_path: Path
    pose_dir: Path
    frame_count: int


@dataclass(frozen=True)
class CoherentClip:
    keyframe_indices: list[int]
    dense_frame_indices: list[int]
    source_span: tuple[int, int]


def load_json(path: Path) -> Any:
    with path.open() as handle:
        return json.load(handle)


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(json_safe(payload), indent=2, sort_keys=True) + "\n")


def json_safe(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(key): json_safe(item) for key, item in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [json_safe(item) for item in value]
    if isinstance(value, Path):
        return str(value)
    if hasattr(value, "item"):
        return json_safe(value.item())
    if isinstance(value, float):
        return value if math.isfinite(value) else None
    return value


def load_cv2() -> Any:
    try:
        import cv2  # type: ignore
    except ImportError as exc:  # pragma: no cover - depends on environment.
        raise RuntimeError("OpenCV is required for frame sharpness and extraction") from exc
    return cv2


def resolve_scene_path(candidate: Any, fallback: Path) -> Path:
    if candidate:
        path = Path(candidate)
        if path.exists():
            return path
    return fallback


def load_manifest(
    manifest_path: Path | None,
    *,
    data_root: Path,
    video_root: Path,
    scene_ids: list[str] | None = None,
    limit: int | None = None,
    frame_category_info: dict[str, Any] | None = None,
) -> list[SceneManifest]:
    selected_ids = list(dict.fromkeys(scene_ids or []))
    rows: list[dict[str, Any]] = []
    if manifest_path is not None and manifest_path.exists():
        payload = load_json(manifest_path)
        scenes = payload.get("scenes", payload if isinstance(payload, list) else None)
        if not isinstance(scenes, list):
            raise ValueError(f"Expected manifest scenes list at {manifest_path}")
        for scene in scenes:
            if selected_ids and str(scene.get("scene_id")) not in selected_ids:
                continue
            rows.append(dict(scene))
    elif selected_ids:
        rows = [{"scene_id": scene_id, "dataset": "scannet"} for scene_id in selected_ids]
    else:
        rows = [
            {"scene_id": path.name, "dataset": "scannet"}
            for path in sorted(data_root.glob("scene*_*"))
            if path.is_dir() and (path / "pose").exists()
        ]

    manifests: list[SceneManifest] = []
    for row in rows:
        scene_id = str(row["scene_id"])
        video_path = resolve_scene_path(row.get("video_path"), video_root / f"{scene_id}.mp4")
        pose_dir = resolve_scene_path(row.get("pose_dir"), data_root / scene_id / "pose")
        frame_count = int(
            row.get("video_frame_count")
            or row.get("frame_count")
            or (len(frame_category_info[scene_id]) if frame_category_info and scene_id in frame_category_info else 0)
        )
        if frame_count <= 0 and video_path.exists():
            frame_count = probe_video_frame_count(video_path)
        manifests.append(
            SceneManifest(
                scene_id=scene_id,
                dataset=str(row.get("dataset", "scannet")),
                video_path=video_path,
                pose_dir=pose_dir,
                frame_count=frame_count,
            )
        )

    if limit is not None:
        manifests = manifests[:limit]
    return manifests


def probe_video_frame_count(video_path: Path) -> int:
    cv2 = load_cv2()
    cap = cv2.VideoCapture(str(video_path))
    try:
        if not cap.isOpened():
            return 0
        return int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    finally:
        cap.release()


def load_pose(path: Path) -> Matrix | None:
    try:
        rows = [[float(value) for value in line.split()] for line in path.read_text().splitlines() if line.strip()]
    except OSError:
        return None
    if len(rows) != 4 or any(len(row) != 4 for row in rows):
        return None
    if any(not math.isfinite(value) for row in rows for value in row):
        return None
    return rows


def camera_center(pose: Matrix) -> Point3D:
    return pose[0][3], pose[1][3], pose[2][3]


def camera_forward_xy(pose: Matrix) -> Point2D | None:
    return normalize_2d((pose[0][2], pose[1][2]))


def point_xy(point: Point3D) -> Point2D:
    return point[0], point[1]


def distance_2d(a: Point2D, b: Point2D) -> float:
    return math.hypot(a[0] - b[0], a[1] - b[1])


def dot_2d(a: Point2D, b: Point2D) -> float:
    return a[0] * b[0] + a[1] * b[1]


def normalize_2d(vector: Point2D) -> Point2D | None:
    norm = math.hypot(vector[0], vector[1])
    if norm <= 1e-9:
        return None
    return vector[0] / norm, vector[1] / norm


def angle_between_2d(a: Point2D, b: Point2D) -> float:
    dot = max(-1.0, min(1.0, dot_2d(a, b)))
    cross = a[0] * b[1] - a[1] * b[0]
    return abs(math.degrees(math.atan2(cross, dot)))


def jaccard(a: set[Any], b: set[Any]) -> float:
    union = a | b
    if not union:
        return 0.0
    return len(a & b) / len(union)


def parse_pixel_value(value: Any) -> int:
    if hasattr(value, "item"):
        value = value.item()
    if isinstance(value, str):
        value = value.strip()
    return int(float(value))


def parse_num_pixels(record: dict[str, Any]) -> int:
    return parse_pixel_value(record.get("num_pixels", 0))


def scalar_list(value: Any) -> list[Any]:
    if value is None:
        return []
    if hasattr(value, "tolist"):
        value = value.tolist()
    if isinstance(value, (list, tuple)):
        return list(value)
    return [value]


def normalize_instance_id(value: Any) -> str:
    if hasattr(value, "item"):
        value = value.item()
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        return str(value)
    if math.isfinite(numeric) and numeric.is_integer():
        return str(int(numeric))
    return str(value)


def instance_pixel_entries(category: str, payload: dict[str, Any]) -> list[tuple[VisibilityKey, int]]:
    inst_ids = scalar_list(payload.get("inst_ids"))
    inst_pixels = scalar_list(payload.get("inst_num_pixels"))
    if inst_ids and inst_pixels and len(inst_ids) == len(inst_pixels):
        return [
            ((category, normalize_instance_id(inst_id)), parse_pixel_value(pixel_count))
            for inst_id, pixel_count in zip(inst_ids, inst_pixels)
        ]
    return [((category, None), parse_num_pixels(payload))]


def build_adaptive_visibility_thresholds(
    frame_category_records: list[dict[str, Any]],
    *,
    min_visible_pixels: int,
    percentile: float,
    min_retained_frames: int,
) -> dict[VisibilityKey, int]:
    samples: dict[VisibilityKey, list[int]] = {}
    for frame_record in frame_category_records:
        for category, payload in frame_record.items():
            if not isinstance(payload, dict):
                continue
            for key, pixel_count in instance_pixel_entries(category, payload):
                if pixel_count > 0:
                    samples.setdefault(key, []).append(pixel_count)

    thresholds: dict[VisibilityKey, int] = {}
    for key, pixel_counts in samples.items():
        percentile_threshold = int(math.ceil(float(np.percentile(pixel_counts, percentile))))
        threshold = max(min_visible_pixels, percentile_threshold)
        if min_retained_frames > 0:
            retained_index = min(min_retained_frames, len(pixel_counts)) - 1
            top_retained_floor = sorted(pixel_counts, reverse=True)[retained_index]
            threshold = min(threshold, max(min_visible_pixels, top_retained_floor))
        thresholds[key] = threshold
    return thresholds


def visible_instances(
    frame_record: dict[str, Any],
    min_visible_pixels: int,
    visibility_thresholds: dict[VisibilityKey, int] | None = None,
) -> set[VisibilityKey]:
    visible: set[VisibilityKey] = set()
    for category, payload in frame_record.items():
        if not isinstance(payload, dict):
            continue
        for key, pixel_count in instance_pixel_entries(category, payload):
            threshold = min_visible_pixels
            if visibility_thresholds is not None:
                threshold = visibility_thresholds.get(key, min_visible_pixels)
            if pixel_count >= threshold:
                visible.add(key)
    return visible


def categories_for_instances(instance_keys: set[VisibilityKey] | list[VisibilityKey]) -> set[str]:
    return {category for category, _instance_id in instance_keys}


def visibility_key_sort_key(key: VisibilityKey) -> tuple[str, str]:
    category, instance_id = key
    return category, "" if instance_id is None else str(instance_id)


def visibility_id_sort_key(value: str) -> tuple[int, float | str]:
    try:
        return (0, float(value))
    except ValueError:
        return (1, value)


def visibility_key_label(key: VisibilityKey) -> str:
    category, instance_id = key
    return category if instance_id is None else f"{category}#{instance_id}"


def visibility_key_records(keys: set[VisibilityKey] | list[VisibilityKey]) -> list[dict[str, str | None]]:
    return [
        {"category": category, "instance_id": instance_id}
        for category, instance_id in sorted(keys, key=visibility_key_sort_key)
    ]


def frame_sharpness_score(image: Any, cv2_module: Any) -> float:
    gray = cv2_module.cvtColor(image, cv2_module.COLOR_BGR2GRAY)
    return float(cv2_module.Laplacian(gray, cv2_module.CV_64F).var())


def compute_selected_frame_sharpness_scores(video_path: Path, frame_indices: list[int]) -> dict[int, float]:
    if not frame_indices:
        return {}
    if not video_path.exists():
        raise FileNotFoundError(f"Missing video file for blur-aware frame selection: {video_path}")

    cv2 = load_cv2()
    requested = set(frame_indices)
    max_frame = max(requested)
    cap = cv2.VideoCapture(str(video_path))
    if not cap.isOpened():
        raise RuntimeError(f"Could not open video for blur-aware frame selection: {video_path}")

    scores: dict[int, float] = {}
    try:
        frame_idx = 0
        while frame_idx <= max_frame:
            ok, image = cap.read()
            if not ok or image is None:
                break
            if frame_idx in requested:
                scores[frame_idx] = frame_sharpness_score(image, cv2)
            frame_idx += 1
    finally:
        cap.release()
    return scores


def export_video_frames(
    video_path: Path,
    frame_indices: list[int],
    output_dir: Path,
    *,
    force: bool = False,
) -> list[str]:
    output_dir.mkdir(parents=True, exist_ok=True)
    paths = [output_dir / f"image_{index:02d}_frame_{frame_idx:06d}.jpg" for index, frame_idx in enumerate(frame_indices)]
    if not force and all(path.exists() for path in paths):
        return [str(path) for path in paths]
    if not frame_indices:
        return []
    if not video_path.exists():
        raise FileNotFoundError(f"Missing video file for frame export: {video_path}")

    cv2 = load_cv2()
    requested = {frame_idx: path for frame_idx, path in zip(frame_indices, paths)}
    cap = cv2.VideoCapture(str(video_path))
    if not cap.isOpened():
        raise RuntimeError(f"Could not open video for frame export: {video_path}")
    try:
        for frame_idx, output_path in requested.items():
            if output_path.exists() and not force:
                continue
            cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
            ok, image = cap.read()
            if not ok or image is None:
                raise RuntimeError(f"Could not read frame {frame_idx} from {video_path}")
            ok = cv2.imwrite(str(output_path), image)
            if not ok:
                raise RuntimeError(f"Could not write frame image: {output_path}")
    finally:
        cap.release()
    return [str(path) for path in paths]


def export_video_frame_cache(
    video_path: Path,
    frame_indices: list[int],
    output_dir: Path,
    *,
    force: bool = False,
) -> list[str]:
    output_dir.mkdir(parents=True, exist_ok=True)
    unique_indices = sorted(dict.fromkeys(int(frame_idx) for frame_idx in frame_indices))
    path_by_frame = {frame_idx: output_dir / f"frame_{frame_idx:06d}.jpg" for frame_idx in unique_indices}
    missing = [frame_idx for frame_idx, path in path_by_frame.items() if force or not path.exists()]
    if missing:
        if not video_path.exists():
            raise FileNotFoundError(f"Missing video file for frame export: {video_path}")
        cv2 = load_cv2()
        cap = cv2.VideoCapture(str(video_path))
        if not cap.isOpened():
            raise RuntimeError(f"Could not open video for frame export: {video_path}")
        try:
            for frame_idx in missing:
                cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
                ok, image = cap.read()
                if not ok or image is None:
                    raise RuntimeError(f"Could not read frame {frame_idx} from {video_path}")
                ok = cv2.imwrite(str(path_by_frame[frame_idx]), image)
                if not ok:
                    raise RuntimeError(f"Could not write frame image: {path_by_frame[frame_idx]}")
        finally:
            cap.release()
    return [str(path_by_frame[int(frame_idx)]) for frame_idx in frame_indices]


def camera_distance_from_poses(pose_a: Matrix, pose_b: Matrix) -> float:
    return distance_2d(point_xy(camera_center(pose_a)), point_xy(camera_center(pose_b)))


def yaw_change_from_poses(pose_a: Matrix, pose_b: Matrix) -> float | None:
    forward_a = camera_forward_xy(pose_a)
    forward_b = camera_forward_xy(pose_b)
    if forward_a is None or forward_b is None:
        return None
    return angle_between_2d(forward_a, forward_b)


def frame_passes_previous_viewpoint_change(
    previous_frame_index: int,
    frame_index: int,
    pose_by_frame: dict[int, Matrix],
    *,
    min_camera_distance_m: float,
    min_yaw_change_deg: float,
) -> bool:
    pose_a = pose_by_frame.get(previous_frame_index)
    pose_b = pose_by_frame.get(frame_index)
    if pose_a is None or pose_b is None:
        return False
    camera_distance_m = camera_distance_from_poses(pose_a, pose_b)
    yaw_change_deg = yaw_change_from_poses(pose_a, pose_b)
    if yaw_change_deg is None:
        return False
    return camera_distance_m >= min_camera_distance_m or yaw_change_deg >= min_yaw_change_deg


def sample_diverse_pose_path_frames(
    frame_indices: list[int],
    count: int,
    pose_by_frame: dict[int, Matrix],
    *,
    min_camera_distance_m: float,
    min_yaw_change_deg: float,
) -> list[int]:
    """Greedily keep the earliest eligible frame with enough change from the prior selection."""
    candidates = sorted(dict.fromkeys(frame_indices))
    if count <= 0 or not candidates:
        return []
    selected = [candidates[0]]
    for frame_index in candidates[1:]:
        if frame_passes_previous_viewpoint_change(
            selected[-1],
            frame_index,
            pose_by_frame,
            min_camera_distance_m=min_camera_distance_m,
            min_yaw_change_deg=min_yaw_change_deg,
        ):
            selected.append(frame_index)
        if len(selected) >= count:
            break
    return selected


def build_coherent_clips(
    *,
    frame_count: int,
    usable_frame_indices: list[int],
    pose_by_frame: dict[int, Matrix],
    max_multi_images: int,
    coherent_window_sizes: list[int],
    coherent_window_stride: int,
    min_multi_camera_distance_m: float = DEFAULT_MIN_MULTI_CAMERA_DISTANCE_M,
    min_multi_yaw_change_deg: float = DEFAULT_MIN_MULTI_YAW_CHANGE_DEG,
) -> list[CoherentClip]:
    usable_set = set(usable_frame_indices)
    keyframe_count = min(max_multi_images, 5)
    min_keyframes = 2
    clips: list[CoherentClip] = []

    for window_size in sorted(set(coherent_window_sizes)):
        for start in range(0, frame_count, coherent_window_stride):
            end = min(frame_count - 1, start + window_size)
            if end <= start:
                continue
            dense_indices = [idx for idx in range(start, end + 1) if idx in usable_set]
            if len(dense_indices) < min_keyframes:
                continue

            keyframes = sample_diverse_pose_path_frames(
                dense_indices,
                keyframe_count,
                pose_by_frame,
                min_camera_distance_m=min_multi_camera_distance_m,
                min_yaw_change_deg=min_multi_yaw_change_deg,
            )
            if len(keyframes) < min_keyframes:
                continue

            clips.append(
                CoherentClip(
                    keyframe_indices=keyframes,
                    dense_frame_indices=dense_indices,
                    source_span=(keyframes[0], keyframes[-1]),
                )
            )

    clips.sort(key=lambda clip: clip.source_span[0])
    return clips


def frame_spans_overlap(span_a: tuple[int, int], span_b: tuple[int, int]) -> bool:
    """Return whether two inclusive video-frame intervals share any frame."""
    return max(span_a[0], span_b[0]) <= min(span_a[1], span_b[1])


def grid_cell_to_indices(cell: str) -> tuple[int, int]:
    token = cell.strip()
    col_token = ""
    row_token = ""
    for char in token:
        if char.isalpha():
            col_token += char.upper()
        elif char.isdigit():
            row_token += char
    if not col_token or not row_token:
        raise ValueError(f"Invalid grid cell label: {cell!r}")
    col_index = 0
    for char in col_token:
        col_index = col_index * 26 + (ord(char) - ord("A") + 1)
    return col_index - 1, int(row_token) - 1


def object_grid_tokens_by_cell(object_grid: dict[str, Any]) -> dict[tuple[int, int], set[str]]:
    tokens: dict[tuple[int, int], set[str]] = {}
    for cell, labels in (object_grid.get("cell_labels") or {}).items():
        col_index, row_index = grid_cell_to_indices(str(cell))
        tokens.setdefault((col_index, row_index), set()).update(str(label) for label in labels)
    return tokens


def object_grid_categories(object_grid: dict[str, Any]) -> set[str]:
    categories: set[str] = set()
    for obj in object_grid.get("objects") or []:
        if obj.get("centroid_cell") is not None:
            categories.add(str(obj.get("category")))
    return categories


def object_grid_instance_id(obj: dict[str, Any]) -> str | None:
    category = obj.get("category")
    instance_index = obj.get("instance_index")
    if category is None or instance_index is None:
        return None
    return f"{category}#{normalize_instance_id(instance_index)}"


def object_grid_instance_ids(object_grid: dict[str, Any], *, require_centroid_cell: bool = False) -> set[str]:
    instance_ids: set[str] = set()
    for obj in object_grid.get("objects") or []:
        if require_centroid_cell and obj.get("centroid_cell") is None:
            continue
        instance_id = object_grid_instance_id(obj)
        if instance_id is not None:
            instance_ids.add(instance_id)
    return instance_ids


def collect_visibility_ids_by_category(frame_category_records: list[dict[str, Any]]) -> dict[str, set[str]]:
    ids_by_category: dict[str, set[str]] = {}
    for frame_record in frame_category_records:
        if not isinstance(frame_record, dict):
            continue
        for category, payload in frame_record.items():
            if not isinstance(payload, dict):
                continue
            for raw_id in scalar_list(payload.get("inst_ids")):
                ids_by_category.setdefault(str(category), set()).add(normalize_instance_id(raw_id))
    return ids_by_category


def build_visibility_instance_mapping(
    object_grid: dict[str, Any],
    frame_category_records: list[dict[str, Any]],
) -> tuple[dict[VisibilityKey, str], dict[str, str]]:
    objects_by_category: dict[str, list[dict[str, Any]]] = {}
    for obj in object_grid.get("objects") or []:
        category = obj.get("category")
        if category is None or obj.get("instance_index") is None:
            continue
        objects_by_category.setdefault(str(category), []).append(obj)
    for rows in objects_by_category.values():
        rows.sort(key=lambda obj: visibility_id_sort_key(normalize_instance_id(obj.get("instance_index"))))

    visibility_ids = collect_visibility_ids_by_category(frame_category_records)
    mapping: dict[VisibilityKey, str] = {}
    status: dict[str, str] = {}
    for category, rows in sorted(objects_by_category.items()):
        ids = sorted(visibility_ids.get(category, set()), key=visibility_id_sort_key)
        if not ids:
            status[category] = "no_visibility_ids"
            continue
        if len(ids) != len(rows):
            status[category] = f"count_mismatch_grid_{len(rows)}_visibility_{len(ids)}"
            continue
        for visibility_id, obj in zip(ids, rows):
            instance_id = object_grid_instance_id(obj)
            if instance_id is not None:
                mapping[(category, visibility_id)] = instance_id
        status[category] = "rank_order_mapped"
    for category in sorted(set(visibility_ids) - set(objects_by_category)):
        status[category] = "no_grid_objects"
    return mapping, status


def categories_for_grid_instance_ids(instance_ids: set[str] | list[str]) -> set[str]:
    return {str(instance_id).split("#", 1)[0] for instance_id in instance_ids}


def object_cells_for_categories(object_grid: dict[str, Any], categories: set[str]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for obj in object_grid.get("objects") or []:
        category = str(obj.get("category"))
        cell = obj.get("centroid_cell")
        if category not in categories or cell is None:
            continue
        col_index, row_index = grid_cell_to_indices(str(cell))
        rows.append(
            {
                "category": category,
                "label": obj.get("label"),
                "cell": cell,
                "display_coord": [col_index + 1, row_index + 1],
                "object_id": obj.get("object_id"),
                "instance_index": obj.get("instance_index"),
                "raw_label": obj.get("raw_label"),
            }
        )
    rows.sort(key=lambda row: (row["display_coord"][1], row["display_coord"][0], row["category"]))
    return rows


def object_cells_for_instance_ids(object_grid: dict[str, Any], instance_ids: set[str]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for obj in object_grid.get("objects") or []:
        instance_id = object_grid_instance_id(obj)
        cell = obj.get("centroid_cell")
        if instance_id not in instance_ids or cell is None:
            continue
        col_index, row_index = grid_cell_to_indices(str(cell))
        category = str(obj.get("category"))
        rows.append(
            {
                "category": category,
                "label": obj.get("label"),
                "cell": cell,
                "display_coord": [col_index + 1, row_index + 1],
                "object_id": obj.get("object_id"),
                "instance_index": obj.get("instance_index"),
                "instance_id": instance_id,
                "raw_label": obj.get("raw_label"),
            }
        )
    rows.sort(key=lambda row: (row["display_coord"][1], row["display_coord"][0], row["category"], row["instance_id"]))
    return rows


def displayed_object_records(object_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    for obj in object_rows:
        records.append(
            {
                "label": obj.get("label"),
                "category": obj.get("category"),
                "instance_id": obj.get("instance_id"),
                "instance_index": obj.get("instance_index"),
                "object_id": obj.get("object_id"),
                "cell": obj.get("cell"),
                "display_coord": obj.get("display_coord"),
            }
        )
    return records


def format_object_coordinate_block(displayed_objects: list[dict[str, Any]] | None) -> str:
    if not displayed_objects:
        return "Object Coordinates:\n- None"
    rows = []
    for obj in displayed_objects:
        coord = obj.get("display_coord")
        if not isinstance(coord, list) or len(coord) != 2:
            continue
        label = str(obj.get("label") or obj.get("category") or "OBJECT")
        x, y = int(coord[0]), int(coord[1])
        rows.append((y, x, label, f"- {label}: ({x}, {y})"))
    if not rows:
        return "Object Coordinates:\n- None"
    return "Object Coordinates:\n" + "\n".join(row[-1] for row in sorted(rows))


def format_object_grid_crop(
    object_grid: dict[str, Any],
    *,
    required_display_coords: list[list[int]],
    required_categories: set[str],
    required_instance_ids: set[str] | None = None,
    grid_size: int,
    margin: int,
) -> tuple[str, dict[str, Any]]:
    object_rows: list[dict[str, Any]]
    if required_instance_ids is None:
        token_by_cell = object_grid_tokens_by_cell(object_grid)
        object_rows = object_cells_for_categories(object_grid, required_categories)
    else:
        object_rows = object_cells_for_instance_ids(object_grid, required_instance_ids)
        token_by_cell: dict[tuple[int, int], set[str]] = {}
        for obj in object_rows:
            x, y = obj["display_coord"]
            label = str(obj.get("label") or obj["category"])
            token_by_cell.setdefault((x - 1, y - 1), set()).add(label)

    required_cells: set[tuple[int, int]] = set()
    for coord in required_display_coords:
        if len(coord) != 2:
            continue
        x, y = int(coord[0]), int(coord[1])
        required_cells.add((x - 1, y - 1))
    for obj in object_rows:
        x, y = obj["display_coord"]
        required_cells.add((x - 1, y - 1))
    if not required_cells:
        required_cells = set(token_by_cell)
    if not required_cells:
        required_cells = {(grid_size // 2, grid_size // 2)}

    min_col = max(0, min(col for col, _row in required_cells) - margin)
    max_col = min(grid_size - 1, max(col for col, _row in required_cells) + margin)
    min_row = max(0, min(row for _col, row in required_cells) - margin)
    max_row = min(grid_size - 1, max(row for _col, row in required_cells) + margin)

    def cell_token(col_index: int, row_index: int) -> str:
        object_tokens = sorted(token_by_cell.get((col_index, row_index), []))
        return "+".join(object_tokens) if object_tokens else "."

    columns = [str(col_index + 1) for col_index in range(min_col, max_col + 1)]
    row_labels = [f"{row_index + 1:02d}" for row_index in range(min_row, max_row + 1)]
    tokens = [
        cell_token(col_index, row_index)
        for row_index in range(min_row, max_row + 1)
        for col_index in range(min_col, max_col + 1)
    ]
    cell_width = max(len(token) for token in tokens + columns + ["."])
    row_label_width = max(len(label) for label in row_labels)
    lines = [" " * (row_label_width + 2) + " ".join(column.rjust(cell_width) for column in columns)]
    for row_index, row_label in zip(range(min_row, max_row + 1), row_labels):
        row_tokens = [
            cell_token(col_index, row_index).rjust(cell_width)
            for col_index in range(min_col, max_col + 1)
        ]
        lines.append(f"{row_label.rjust(row_label_width)}  " + " ".join(row_tokens))

    crop = {
        "min_col": min_col,
        "max_col": max_col,
        "min_row": min_row,
        "max_row": max_row,
        "display_columns": [min_col + 1, max_col + 1],
        "display_rows": [min_row + 1, max_row + 1],
        "required_categories": sorted(required_categories),
        "required_display_coords": required_display_coords,
    }
    if required_instance_ids is not None:
        crop["required_instance_ids"] = sorted(required_instance_ids)
        crop["displayed_instance_ids"] = sorted(
            str(obj["instance_id"]) for obj in object_rows if obj.get("instance_id") is not None
        )
    crop["displayed_objects"] = displayed_object_records(object_rows)
    return "\n".join(lines), crop


def valid_facing_vector(value: Any) -> bool:
    return isinstance(value, list) and len(value) == 2 and tuple(value) in DISPLAY_FACING_CANDIDATES


def quantize_display_facing_vector(dx: float, dy: float) -> tuple[tuple[int, int], float, float] | None:
    norm = math.hypot(dx, dy)
    if norm <= 1e-9:
        return None
    unit_x = dx / norm
    unit_y = dy / norm
    best = max(
        DISPLAY_FACING_CANDIDATES,
        key=lambda item: (unit_x * item[0] + unit_y * item[1]) / math.hypot(item[0], item[1]),
    )
    cos_angle = max(-1.0, min(1.0, (unit_x * best[0] + unit_y * best[1]) / math.hypot(best[0], best[1])))
    angle_to_center = abs(math.degrees(math.acos(cos_angle)))
    margin = 22.5 - angle_to_center
    return best, angle_to_center, margin


def facing_margin_from_camera_row(camera_row: dict[str, Any]) -> dict[str, Any] | None:
    forward = camera_row.get("forward_topdown")
    if not isinstance(forward, dict):
        return None
    try:
        forward_x = float(forward["x_m"])
        forward_z = float(forward["z_m"])
    except (KeyError, TypeError, ValueError):
        return None
    result = quantize_display_facing_vector(forward_x, -forward_z)
    if result is None:
        return None
    vector, angle_to_center, margin = result
    return {
        "computed_facing_vector": list(vector),
        "angle_to_bin_center_deg": angle_to_center,
        "facing_bin_margin_deg": margin,
    }


def cell_margin_from_camera_row(camera_row: dict[str, Any], *, grid_size: int, cell_size_m: float) -> dict[str, Any] | None:
    coord = camera_row.get("display_coord")
    center = camera_row.get("center_topdown_m")
    if not (isinstance(coord, list) and len(coord) == 2 and isinstance(center, dict)):
        return None
    try:
        display_x = int(coord[0])
        display_y = int(coord[1])
        x_m = float(center["x_m"])
        z_m = float(center["z_m"])
    except (KeyError, TypeError, ValueError):
        return None
    col_index = display_x - 1
    row_index = display_y - 1
    origin_col = grid_size // 2
    origin_row = grid_size // 2
    cell_center_x = (col_index - origin_col) * cell_size_m
    cell_center_z = (origin_row - row_index) * cell_size_m
    dx = x_m - cell_center_x
    dz = z_m - cell_center_z
    max_axis_offset = max(abs(dx), abs(dz))
    return {
        "cell_center_offset_m": math.hypot(dx, dz),
        "cell_center_axis_offset_m": max_axis_offset,
        "cell_boundary_margin_m": (0.5 * cell_size_m) - max_axis_offset,
        "cell_center_topdown_m": {"x_m": cell_center_x, "z_m": cell_center_z},
        "camera_center_offset_from_cell_center_m": {"x_m": dx, "z_m": dz},
    }


def line_count(path: Path) -> int:
    with path.open() as handle:
        return sum(1 for _ in handle)


def html_escape_text(text: Any) -> str:
    return html.escape("" if text is None else str(text))


def image_to_data_uri(path: Path, *, max_width: int = 360) -> str | None:
    if not path.exists():
        return None
    try:
        from PIL import Image

        image = Image.open(path).convert("RGB")
        if image.width > max_width:
            ratio = max_width / float(image.width)
            image = image.resize((max_width, max(1, int(image.height * ratio))))
        import io

        buffer = io.BytesIO()
        image.save(buffer, format="JPEG", quality=82)
        encoded = base64.b64encode(buffer.getvalue()).decode("ascii")
    except Exception:
        encoded = base64.b64encode(path.read_bytes()).decode("ascii")
    return f"data:image/jpeg;base64,{encoded}"


def rng_sample_stable(items: list[Any], limit: int | None, rng: random.Random) -> list[Any]:
    if limit is None or limit <= 0 or len(items) <= limit:
        return list(items)
    indices = sorted(rng.sample(range(len(items)), limit))
    return [items[index] for index in indices]