"""Asynchronous RGB-depth benchmark loader with geometry-aware resizing. Evaluation samples follow the processed benchmark directory contract. This module loads RGB, depth, normalized intrinsics, and optional segmentation, applies the benchmark-configured view transformation, and returns aligned PyTorch tensors plus an organized ground-truth point map. """ from typing import Any, Dict, Iterator, List, Optional, Tuple, Union from pathlib import Path import numpy as np import torch from PIL import Image import cv2 import utils3d import pipeline from ..utils.io import read_depth, read_image, read_json, read_segmentation from ..utils.data_augmentation import sample_perspective, warp_perspective, resize_to_cover_center_crop_view def _resolve_image_path(instance_path: Union[str, Path]) -> Path: """Resolve a processed sample's RGB path with PNG precedence. Args: instance_path: Directory containing one processed benchmark sample. Returns: ``image.png`` when present, otherwise ``image.jpg``. """ instance_path = Path(instance_path) image_png = instance_path / 'image.png' if image_png.exists(): return image_png return instance_path / 'image.jpg' def _resize_to_cover_center_crop_mask(mask: np.ndarray, raw_width: int, raw_height: int, tgt_width: int, tgt_height: int) -> np.ndarray: """Apply resize-to-cover and center crop to a discrete label mask. Args: mask: Integer/boolean source mask ``[H_raw,W_raw]``. raw_width: Source image width. raw_height: Source image height. tgt_width: Output width. tgt_height: Output height. Returns: Nearest-resized and center-cropped mask ``[H_tgt,W_tgt]``. """ scale = max(tgt_width / raw_width, tgt_height / raw_height) resized_width = max(tgt_width, int(round(raw_width * scale))) resized_height = max(tgt_height, int(round(raw_height * scale))) resized_mask = cv2.resize( mask.astype(np.uint8), (resized_width, resized_height), interpolation=cv2.INTER_NEAREST, ) x0 = max(0, (resized_width - tgt_width) // 2) y0 = max(0, (resized_height - tgt_height) // 2) return resized_mask[y0:y0 + tgt_height, x0:x0 + tgt_width].copy() def _intrinsics_normalized_to_pixel(intrinsics: np.ndarray, width: int, height: int) -> np.ndarray: """Convert normalized camera intrinsics to pixel coordinates. Args: intrinsics: Floating camera matrix ``[3,3]`` normalized by image size. width: Image width used to scale the first matrix row. height: Image height used to scale the second matrix row. Returns: Pixel-space ``float32 [3,3]`` camera matrix. """ intrinsics_px = intrinsics.astype(np.float32).copy() intrinsics_px[0, :] *= float(width) intrinsics_px[1, :] *= float(height) intrinsics_px[2, 0] = 0.0 intrinsics_px[2, 1] = 0.0 intrinsics_px[2, 2] = 1.0 return intrinsics_px def _intrinsics_pixel_to_normalized(intrinsics_px: np.ndarray, width: int, height: int) -> np.ndarray: """Convert pixel camera intrinsics to normalized coordinates. Args: intrinsics_px: Pixel-space camera matrix ``[3,3]``. width: Image width used to normalize the first matrix row. height: Image height used to normalize the second matrix row. Returns: Normalized ``float32 [3,3]`` camera matrix. """ intrinsics = intrinsics_px.astype(np.float32).copy() intrinsics[0, :] /= float(width) intrinsics[1, :] /= float(height) intrinsics[2, 0] = 0.0 intrinsics[2, 1] = 0.0 intrinsics[2, 2] = 1.0 return intrinsics def _resize_depth_nearest_preserve_nan(depth: np.ndarray, size: Tuple[int, int]) -> np.ndarray: """Nearest-resize positive finite depth while preserving invalid support. Args: depth: Source depth ``float [H,W]`` with NaN/Inf invalid values. size: OpenCV target tuple ``(width,height)``. Returns: ``float32 [height,width]`` depth. Pixels whose nearest source was invalid are represented by NaN. """ width, height = size valid = np.isfinite(depth) & (depth > 0) resized_depth = cv2.resize( np.where(valid, depth, 0.0).astype(np.float32), (width, height), interpolation=cv2.INTER_NEAREST, ) resized_valid = cv2.resize( valid.astype(np.uint8), (width, height), interpolation=cv2.INTER_NEAREST, ).astype(bool) return np.where(resized_valid, resized_depth, np.nan).astype(np.float32) def _mda_boundary_view( image: np.ndarray, depth: np.ndarray, intrinsics: np.ndarray, target_size: Tuple[int, int], segmentation_mask: Optional[np.ndarray] = None, ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, Optional[np.ndarray]]: """Create the principal-point-centered view used by boundary benchmarks. The image is first cropped symmetrically around the principal point, then resized to cover the target and center-cropped. RGB uses Lanczos for downsampling and bicubic for upsampling; depth and segmentation use nearest interpolation. Pixel intrinsics are updated after every crop and resize. Args: image: RGB uint8 array ``[H,W,3]``. depth: Depth array ``[H,W]`` with NaN invalid values. intrinsics: Normalized camera matrix ``[3,3]``. target_size: Output ``(height,width)``. segmentation_mask: Optional integer labels ``[H,W]``. Returns: image: Transformed RGB ``[H_t,W_t,3]``. depth: Transformed depth ``float32 [H_t,W_t]``. intrinsics: Updated normalized matrix ``float32 [3,3]``. segmentation_mask: Transformed labels ``[H_t,W_t]`` or ``None``. """ tgt_height, tgt_width = target_size raw_height, raw_width = image.shape[:2] intrinsics_px = _intrinsics_normalized_to_pixel(intrinsics, raw_width, raw_height) cx = float(intrinsics_px[0, 2]) cy = float(intrinsics_px[1, 2]) margin_x = max(1.0, min(cx, raw_width - cx)) margin_y = max(1.0, min(cy, raw_height - cy)) crop_left = max(0, int(round(cx - margin_x))) crop_right = min(raw_width, int(round(cx + margin_x))) crop_top = max(0, int(round(cy - margin_y))) crop_bottom = min(raw_height, int(round(cy + margin_y))) if crop_right - crop_left < 2 or crop_bottom - crop_top < 2: crop_left, crop_top = 0, 0 crop_right, crop_bottom = raw_width, raw_height image = image[crop_top:crop_bottom, crop_left:crop_right].copy() depth = depth[crop_top:crop_bottom, crop_left:crop_right].copy() if segmentation_mask is not None: segmentation_mask = segmentation_mask[crop_top:crop_bottom, crop_left:crop_right].copy() intrinsics_px[0, 2] -= float(crop_left) intrinsics_px[1, 2] -= float(crop_top) crop_height, crop_width = image.shape[:2] scale = max(tgt_width / crop_width, tgt_height / crop_height) resized_width = max(tgt_width, int(np.floor(crop_width * scale))) resized_height = max(tgt_height, int(np.floor(crop_height * scale))) if resized_width < tgt_width or resized_height < tgt_height: resized_width = max(tgt_width, int(np.ceil(crop_width * scale))) resized_height = max(tgt_height, int(np.ceil(crop_height * scale))) image_resample = Image.Resampling.LANCZOS if scale < 1.0 else Image.Resampling.BICUBIC resized_image = np.array(Image.fromarray(image).resize((resized_width, resized_height), image_resample)) resized_depth = _resize_depth_nearest_preserve_nan(depth, (resized_width, resized_height)) resized_segmentation_mask = None if segmentation_mask is not None: resized_segmentation_mask = cv2.resize( segmentation_mask, (resized_width, resized_height), interpolation=cv2.INTER_NEAREST, ) intrinsics_px[:2, :] *= float(scale) x0 = int(round((resized_width - tgt_width) * 0.5)) y0 = int(round((resized_height - tgt_height) * 0.5)) x0 = min(max(x0, 0), resized_width - tgt_width) y0 = min(max(y0, 0), resized_height - tgt_height) x1, y1 = x0 + tgt_width, y0 + tgt_height tgt_image = resized_image[y0:y1, x0:x1].copy() tgt_depth = resized_depth[y0:y1, x0:x1].copy().astype(np.float32) tgt_segmentation_mask = None if resized_segmentation_mask is not None: tgt_segmentation_mask = resized_segmentation_mask[y0:y1, x0:x1].copy() intrinsics_px[0, 2] -= float(x0) intrinsics_px[1, 2] -= float(y0) tgt_intrinsics = _intrinsics_pixel_to_normalized(intrinsics_px, tgt_width, tgt_height) return tgt_image, tgt_depth, tgt_intrinsics, tgt_segmentation_mask class EvalDataLoaderPipeline: """Asynchronously load and geometrically standardize one benchmark dataset. The pipeline emits one sample at a time. It supports exact resolutions, center-crop sizes, or aspect-preserving token budgets, and can optionally include segmentation or normal annotations for local and boundary metrics. """ def __init__( self, path: str, width: Optional[int] = None, height: Optional[int] = None, center_crop_size: Optional[int] = None, split: int = '.index.txt', drop_max_depth: float = 1000., num_load_workers: int = 4, num_process_workers: int = 8, include_segmentation: bool = False, include_normal: bool = False, depth_to_normal: bool = False, max_segments: int = 100, min_seg_area: int = 1000, depth_unit: str = None, min_depth: Optional[float] = None, max_depth: Optional[float] = None, has_sharp_boundary = False, subset: int = None, filenames: Optional[List[str]] = None, num_tokens: Optional[int] = None, patch_size: Optional[int] = None, disable_augmentations: bool = True, disable_perspective: bool = True, resize_to_cover_center_crop: bool = False, mda_boundary_transform: bool = False, ): """Configure benchmark indexing, transforms, and worker stages. Args: path: Processed benchmark root containing the split index. width: Exact output width when token/crop modes are disabled. height: Exact output height when token/crop modes are disabled. center_crop_size: Optional square output side length. split: Relative index filename under ``path``. drop_max_depth: Relative dynamic-range multiplier used to suppress extreme finite values after transformation. num_load_workers: Number of parallel disk readers. num_process_workers: Number of parallel geometry workers. include_segmentation: Load ``segmentation.png`` and label metadata. include_normal: Derive a normal map from depth. depth_to_normal: Retained benchmark compatibility flag. max_segments: Maximum segmentation labels retained by area. min_seg_area: Minimum number of pixels for a retained label. depth_unit: Optional scalar converting stored depth to metric units. min_depth: Optional lower valid-depth bound in converted units. max_depth: Optional upper valid-depth bound in converted units. has_sharp_boundary: Mark samples for boundary metric computation. subset: Optional number of leading index entries to evaluate. filenames: Optional explicit relative paths replacing the split file. num_tokens: Optional approximate image-token count. patch_size: Patch divisibility used with token sampling. disable_augmentations: Disable random flip/color transforms. disable_perspective: Use identity perspective mapping. resize_to_cover_center_crop: Resize to cover then center crop. mda_boundary_transform: Use the principal-point-centered boundary benchmark transformation. Returns: ``None``. Workers start when the context manager is entered. """ if filenames is None: filenames = Path(path).joinpath(split).read_text(encoding='utf-8').splitlines() else: filenames = list(filenames) if subset is not None: subset = int(subset) if subset > 0: filenames = filenames[:subset] self.width = int(width) if width is not None else None self.height = int(height) if height is not None else None self.center_crop_size = int(center_crop_size) if center_crop_size is not None else None self.drop_max_depth = drop_max_depth self.path = Path(path) self.filenames = filenames self.include_segmentation = include_segmentation self.include_normal = include_normal self.max_segments = max_segments self.min_seg_area = min_seg_area self.depth_to_normal = depth_to_normal self.depth_unit = depth_unit self.min_depth = float(min_depth) if min_depth is not None else None self.max_depth = float(max_depth) if max_depth is not None else None self.has_sharp_boundary = has_sharp_boundary self.num_tokens = int(num_tokens) if num_tokens is not None else None self.patch_size = int(patch_size) if patch_size is not None else None self.disable_augmentations = bool(disable_augmentations) self.disable_perspective = bool(disable_perspective) self.resize_to_cover_center_crop = bool(resize_to_cover_center_crop) self.mda_boundary_transform = bool(mda_boundary_transform) self.rng = np.random.default_rng(seed=0) self.pipeline = pipeline.Sequential([ self._generator, pipeline.Parallel([self._load_instance] * num_load_workers), pipeline.Parallel([self._process_instance] * num_process_workers), pipeline.Buffer(4) ]) def __len__(self): """Return the number of configured benchmark samples. The value reflects explicit filenames and optional subset truncation. Returns: Integer length of the selected filename list. """ return len(self.filenames) def _resolve_target_size(self, raw_width: int, raw_height: int) -> Tuple[int, int]: """Resolve output dimensions from exact, crop, or token settings. Args: raw_width: Source image width. raw_height: Source image height. Returns: Integer tuple ``(target_width,target_height)``, optionally rounded to patch multiples. """ if self.num_tokens is not None: if self.patch_size is None: raise ValueError("patch_size must be set when using num_tokens.") aspect_ratio = raw_width / raw_height target_area = self.num_tokens * (self.patch_size ** 2) tgt_width = int(round((target_area * aspect_ratio) ** 0.5)) tgt_height = int(round(tgt_width / aspect_ratio)) elif self.center_crop_size is not None: tgt_width = self.center_crop_size tgt_height = self.center_crop_size else: if self.width is None or self.height is None: raise ValueError("width/height or center_crop_size must be set when num_tokens is not provided.") tgt_width, tgt_height = self.width, self.height if self.patch_size is not None: tgt_width = max(self.patch_size, int(round(tgt_width / self.patch_size)) * self.patch_size) tgt_height = max(self.patch_size, int(round(tgt_height / self.patch_size)) * self.patch_size) return tgt_width, tgt_height def _generator(self): """Yield sequential sample indices to the asynchronous pipeline. Disk loading and processing are parallelized after this ordered stage. Yields: Integer indices from zero through ``len(self)-1``. """ for idx in range(len(self)): yield idx def _load_instance(self, idx): """Read one indexed RGB-depth sample and optional segmentation. Args: idx: Integer index into ``self.filenames``. Returns: Dictionary containing RGB ``uint8 [H,W,3]``, depth ``float [H,W]``, normalized intrinsics ``float32 [3,3]``, masks, and optional segmentation; ``None`` for an out-of-range index. """ if idx >= len(self.filenames): return None path = self.path.joinpath(self.filenames[idx]) instance = { 'filename': self.filenames[idx], } instance['image'] = read_image(_resolve_image_path(path)) depth = read_depth(Path(path, 'depth.png')) # ignore depth unit from depth file, use config instead instance.update({ 'depth': depth, 'depth_mask': np.isfinite(depth) & (depth > 0), 'depth_mask_inf': np.isinf(depth), }) if self.include_segmentation: segmentation_mask, segmentation_labels = read_segmentation(Path(path,'segmentation.png')) instance.update({ 'segmentation_mask': segmentation_mask, 'segmentation_labels': segmentation_labels, }) meta = read_json(Path(path, 'meta.json')) instance['intrinsics'] = np.array(meta['intrinsics'], dtype=np.float32) return instance def _process_instance(self, instance: dict): """Transform one loaded instance and build its ground-truth point map. Args: instance: Raw dictionary returned by :meth:`_load_instance`, or ``None`` propagated from a failed/out-of-range stage. Returns: Processed dictionary with image ``float32 [3,H,W]``, depth and masks ``[H,W]``, normalized intrinsics ``[3,3]``, point map ``[H,W,3]``, metadata flags, and optional normals/segmentation tensors. Returns ``None`` when the input is ``None``. """ if instance is None: return None image = instance['image'] depth = instance['depth'] intrinsics = instance['intrinsics'] segmentation_mask = instance.get('segmentation_mask', None) segmentation_labels = instance.get('segmentation_labels', None) raw_height, raw_width = image.shape[:2] tgt_width, tgt_height = self._resolve_target_size(raw_width, raw_height) tgt_aspect = tgt_width / tgt_height raw_depth_mask = np.isfinite(depth) & (depth > 0) raw_depth_ratio = raw_depth_mask.mean() if raw_depth_ratio < 0.001: depth = np.ones_like(depth, dtype=np.float32) raw_depth_mask = np.isfinite(depth) else: depth = np.where(raw_depth_mask, depth, np.nan) if self.include_normal: raw_normal, raw_normal_mask = utils3d.np.depth_map_to_normal_map( depth, intrinsics=intrinsics, mask=raw_depth_mask, edge_threshold=88 ) raw_normal = np.where(raw_normal_mask[..., None], raw_normal, np.nan) else: raw_normal = None if self.mda_boundary_transform: tgt_image, tgt_depth, tgt_intrinsics, tgt_segmentation_mask = _mda_boundary_view( image, depth, intrinsics, (tgt_height, tgt_width), segmentation_mask=segmentation_mask, ) if self.include_normal: tgt_normal, tgt_normal_mask = utils3d.np.depth_map_to_normal_map( tgt_depth, intrinsics=tgt_intrinsics, mask=np.isfinite(tgt_depth) & (tgt_depth > 0), edge_threshold=88 ) tgt_normal = np.where(tgt_normal_mask[..., None], tgt_normal, np.nan) else: tgt_normal = None elif self.resize_to_cover_center_crop: tgt_image, tgt_depth, tgt_intrinsics = resize_to_cover_center_crop_view( image, depth, intrinsics, (tgt_height, tgt_width), image_interpolation='lanczos', depth_interpolation='mixed', ) if self.include_normal: tgt_normal, tgt_normal_mask = utils3d.np.depth_map_to_normal_map( tgt_depth, intrinsics=tgt_intrinsics, mask=np.isfinite(tgt_depth) & (tgt_depth > 0), edge_threshold=88 ) tgt_normal = np.where(tgt_normal_mask[..., None], tgt_normal, np.nan) else: tgt_normal = None tgt_segmentation_mask = None if segmentation_mask is not None: tgt_segmentation_mask = _resize_to_cover_center_crop_mask( segmentation_mask, raw_width, raw_height, tgt_width, tgt_height, ) elif self.disable_perspective: tgt_intrinsics = intrinsics.copy() R = np.eye(3, dtype=np.float32) transform = np.eye(3, dtype=np.float32) else: tgt_intrinsics, R = sample_perspective( intrinsics, tgt_aspect=tgt_aspect, center_augmentation=0.0, fov_range_absolute=(1, 179), fov_range_relative=(1.0, 1.0), rng=self.rng, ) transform = tgt_intrinsics @ R @ np.linalg.inv(intrinsics) if not self.resize_to_cover_center_crop and not self.mda_boundary_transform: tgt_image = warp_perspective(image, transform, (tgt_height, tgt_width), interpolation='lanczos') depth_edge_mask = utils3d.np.depth_map_edge(depth, mask=raw_depth_mask, kernel_size=5, ltol=0.01) depth_bilinear_mask = raw_depth_mask & ~depth_edge_mask warped_depth_bilinear_mask = warp_perspective( depth_bilinear_mask.astype(np.float32), transform, (tgt_height, tgt_width), interpolation='bilinear', ) warped_depth_nearest = warp_perspective( depth, transform, (tgt_height, tgt_width), interpolation='nearest', sparse_mask=~np.isnan(depth), ) warped_depth_bilinear = 1 / warp_perspective( 1 / depth, transform, (tgt_height, tgt_width), interpolation='bilinear', ) warped_depth = np.where(warped_depth_bilinear_mask == 1.0, warped_depth_bilinear, warped_depth_nearest) tgt_uvhomo = np.concatenate( [utils3d.np.uv_map((tgt_height, tgt_width)), np.ones((tgt_height, tgt_width, 1), dtype=np.float32)], axis=-1, ) tgt_depth = warped_depth / np.dot(tgt_uvhomo, np.linalg.inv(transform)[2, :]) if raw_normal is not None: warped_normal = warp_perspective(raw_normal, transform, (tgt_height, tgt_width), interpolation='bilinear') tgt_normal = warped_normal @ R.T else: tgt_normal = None if segmentation_mask is not None: tgt_segmentation_mask = warp_perspective( segmentation_mask, transform, (tgt_height, tgt_width), interpolation='nearest' ) else: tgt_segmentation_mask = None if not self.disable_augmentations: if self.rng.choice([True, False]): tgt_image = np.flip(tgt_image, axis=1).copy() tgt_depth = np.flip(tgt_depth, axis=1).copy() if tgt_normal is not None: tgt_normal = np.flip(tgt_normal, axis=1).copy() * [-1, 1, 1] if self.depth_unit is not None: tgt_depth *= self.depth_unit is_metric = True else: is_metric = False depth_range_mask = np.isfinite(tgt_depth) & (tgt_depth > 0) if self.min_depth is not None: depth_range_mask &= tgt_depth >= self.min_depth if self.max_depth is not None: depth_range_mask &= tgt_depth <= self.max_depth tgt_depth = np.where(depth_range_mask, tgt_depth, np.nan) drop_max_depth = np.nanquantile(np.where(np.isfinite(tgt_depth), tgt_depth, np.nan), 0.01) * self.drop_max_depth tgt_depth = np.where(np.isfinite(tgt_depth), np.clip(tgt_depth, 0, drop_max_depth), tgt_depth) tgt_depth_mask_inf = np.isinf(tgt_depth) tgt_depth_mask = np.isfinite(tgt_depth) & (tgt_depth > 0) if not np.any(tgt_depth_mask): tgt_depth_mask = np.ones_like(tgt_depth_mask) tgt_depth = np.ones_like(tgt_depth) tgt_points = utils3d.np.depth_map_to_point_map(tgt_depth, intrinsics=tgt_intrinsics) if self.include_segmentation and tgt_segmentation_mask is not None: for k in ['undefined', 'unannotated', 'background', 'sky']: if k in segmentation_labels: del segmentation_labels[k] seg_id2count = dict(zip(*np.unique(tgt_segmentation_mask, return_counts=True))) sorted_labels = sorted(segmentation_labels.keys(), key=lambda x: seg_id2count.get(segmentation_labels[x], 0), reverse=True) segmentation_labels = { k: segmentation_labels[k] for k in sorted_labels[:self.max_segments] if seg_id2count.get(segmentation_labels[k], 0) >= self.min_seg_area } instance.update({ 'image': torch.from_numpy(tgt_image.astype(np.float32) / 255.0).permute(2, 0, 1), 'depth': torch.from_numpy(tgt_depth).float(), 'depth_mask': torch.from_numpy(tgt_depth_mask).bool(), 'depth_mask_inf': torch.from_numpy(tgt_depth_mask_inf).bool(), 'intrinsics': torch.from_numpy(tgt_intrinsics).float(), 'points': torch.from_numpy(tgt_points).float(), 'segmentation_mask': torch.from_numpy(tgt_segmentation_mask).long() if tgt_segmentation_mask is not None else None, 'segmentation_labels': segmentation_labels, 'is_metric': is_metric, 'has_sharp_boundary': self.has_sharp_boundary, }) if tgt_normal is not None: instance['normal'] = torch.from_numpy(tgt_normal).float() instance = {k: v for k, v in instance.items() if v is not None} return instance def start(self): """Start asynchronous loader workers. Call this before :meth:`get` when not using the context manager. Returns: ``None``. """ self.pipeline.start() def stop(self): """Stop asynchronous loader workers and release resources. Any prefetched samples are discarded by the pipeline implementation. Returns: ``None``. """ self.pipeline.stop() def __enter__(self): """Start the pipeline and return it as a context-manager value. This is equivalent to an explicit :meth:`start` call. Returns: This :class:`EvalDataLoaderPipeline` instance. """ self.start() return self def __exit__(self, exc_type, exc_value, traceback): """Stop the pipeline when leaving its context. Args: exc_type: Exception class raised inside the context, if any. exc_value: Exception instance raised inside the context, if any. traceback: Associated traceback object, if any. Returns: ``None``; exceptions are not suppressed. """ self.stop() def get(self): """Block until the next processed evaluation sample is available. Worker-side exceptions are surfaced by the underlying pipeline call. Returns: Processed sample dictionary documented by :meth:`_process_instance`. """ return self.pipeline.get()