""" Base class for MapAnything datasets. """ from typing import List, Tuple, Union import numpy as np import PIL import torch import torchvision.transforms as tvf from scipy.spatial.transform import Rotation from mapanything.datasets.base.easy_dataset import EasyDataset from mapanything.utils.cropping import ( bbox_from_intrinsics_in_out, camera_matrix_of_crop, crop_image_and_other_optional_info, rescale_image_and_other_optional_info, ) from mapanything.utils.geometry import ( depthmap_to_camera_coordinates, get_absolute_pointmaps_and_rays_info, ) from uniception.models.encoders.image_normalizations import IMAGE_NORMALIZATION_DICT class BaseDataset(EasyDataset): """ Define all basic options. Usage: class MyDataset(BaseDataset): def _get_views(self, idx): views = [] views.append(dict(img=, ...)) return views """ def __init__( self, num_views: int, variable_num_views: bool = False, split: str = None, covisibility_thres: float = None, resolution: Union[int, Tuple[int, int], List[Tuple[int, int]]] = None, principal_point_centered: bool = False, transform: str = None, data_norm_type: str = None, aug_crop: int = 0, seed: int = None, max_num_retries: int = 5, ): """ PyTorch dataset for multi-view images sampled from scenes, where the images form a single connected component. Args: num_views (int): Number of views. variable_num_views (bool): If True, the number of views can vary from batch to batch. The maximum number of views is num_views and minimum is 2. On by default for N-view train dataloader (hydra config). split (str): 'train', 'val', 'test', etc. covisibility_thres (float): Covisibility (%) threshold to determine if another image is a neighbor or not resolution (int or tuple or list of tuples): Resolution of the images principal_point_centered (bool): If True, the principal point is centered in the image. transform (str): Transform to apply to the images. Options: - 'colorjitter+grayscale+gaublur': tvf.Compose([ tvf.RandomApply([tvf.ColorJittter(0.3, 0.4, 0.2, 0.1)], p=0.75), tvf.RandomGrayscale(p=0.05), tvf.RandomApply([tvf.GaussianBlur(5, sigma=(0.1, 1.0))], p=0.05), ]) after ImgNorm - 'colorjitter': tvf.ColorJittter(0.5, 0.5, 0.5, 0.1) after ImgNorm - 'imgnorm': ImgNorm only data_norm_type (str): Image normalization type. For options, see UniCeption image normalization dict. aug_crop (int): Augment crop. If int greater than 0, indicates the number of pixels to increase in target resolution. seed (int): Seed for the random number generator. max_num_retries (int): Maximum number of retries for loading a different sample from the dataset, if provided idx fails. """ self.num_views = num_views self.variable_num_views = variable_num_views self.num_views_min = 2 self.split = split self.covisibility_thres = covisibility_thres self._set_resolutions(resolution) self.principal_point_centered = principal_point_centered # Update the number of views if necessary and make it a list if variable_num_views is True if self.variable_num_views and self.num_views > self.num_views_min: self.num_views = list(range(self.num_views_min, self.num_views + 1)) # Initialize the image normalization type if data_norm_type in IMAGE_NORMALIZATION_DICT.keys(): self.data_norm_type = data_norm_type image_norm = IMAGE_NORMALIZATION_DICT[data_norm_type] ImgNorm = tvf.Compose( [ tvf.ToTensor(), tvf.Normalize(mean=image_norm.mean, std=image_norm.std), ] ) elif data_norm_type == "identity": self.data_norm_type = data_norm_type ImgNorm = tvf.Compose([tvf.ToTensor()]) else: raise ValueError( f"Unknown data_norm_type: {data_norm_type}. Available options: identity or {list(IMAGE_NORMALIZATION_DICT.keys())}" ) # Initialize torchvision transforms if transform == "imgnorm": self.transform = ImgNorm elif transform == "colorjitter": self.transform = tvf.Compose([tvf.ColorJitter(0.5, 0.5, 0.5, 0.1), ImgNorm]) elif transform == "colorjitter+grayscale+gaublur": self.transform = tvf.Compose( [ tvf.RandomApply([tvf.ColorJitter(0.3, 0.4, 0.2, 0.1)], p=0.75), tvf.RandomGrayscale(p=0.05), tvf.RandomApply([tvf.GaussianBlur(5, sigma=(0.1, 1.0))], p=0.05), ImgNorm, ] ) else: raise ValueError( 'Unknown transform. Available options: "imgnorm", "colorjitter", "colorjitter+grayscale+gaublur"' ) # Initialize the augmentation parameters self.aug_crop = aug_crop # Initialize the seed for the random number generator self.seed = seed self._seed_offset = 0 # Initialize the maximum number of retries for loading a different sample from the dataset, if the first idx fails self.max_num_retries = max_num_retries # Initialize the dataset type flags self.is_metric_scale = False # by default a dataset is not metric scale, subclasses can overwrite this self.is_synthetic = False # by default a dataset is not synthetic, subclasses can overwrite this def _load_data(self): self.scenes = [] self.num_of_scenes = len(self.scenes) def __len__(self): "Length of the dataset is determined by the number of scenes in the dataset split" return self.num_of_scenes def get_stats(self): "Get the number of scenes in the dataset split" return f"{self.num_of_scenes} scenes" def __repr__(self): resolutions_str = "[" + ";".join(f"{w}x{h}" for w, h in self._resolutions) + "]" return ( f"""{type(self).__name__}({self.get_stats()}, {self.num_views=} {self.split=}, {self.seed=}, resolutions={resolutions_str}, {self.transform=})""".replace("self.", "") .replace("\n", "") .replace(" ", "") ) def _get_views(self, idx, num_views_to_sample, resolution): raise NotImplementedError() def _set_seed_offset(self, idx): """ Set the seed offset. This is directly added to self.seed when setting the random seed. """ self._seed_offset = idx def _set_resolutions(self, resolutions): assert resolutions is not None, "undefined resolution" if isinstance(resolutions, int): resolutions = [resolutions] elif isinstance(resolutions, tuple): resolutions = [resolutions] elif isinstance(resolutions, list): assert all(isinstance(res, tuple) for res in resolutions), ( f"Bad type for {resolutions=}, should be int or tuple of ints or list of tuples of ints" ) else: raise ValueError( f"Bad type for {resolutions=}, should be int or tuple of ints or list of tuples of ints" ) self._resolutions = [] for resolution in resolutions: if isinstance(resolution, int): width = height = resolution else: width, height = resolution assert isinstance(width, int), ( f"Bad type for {width=} {type(width)=}, should be int" ) assert isinstance(height, int), ( f"Bad type for {height=} {type(height)=}, should be int" ) self._resolutions.append((width, height)) def _crop_resize_if_necessary( self, image, resolution, depthmap, intrinsics, additional_quantities=None, ): """ Process an image by downsampling and cropping as needed to match the target resolution. This method performs the following operations: 1. Converts the image to PIL.Image if necessary 2. Crops the image centered on the principal point if requested 3. Downsamples the image using high-quality Lanczos filtering 4. Performs final cropping to match the target resolution Args: image (numpy.ndarray or PIL.Image.Image): Input image to be processed resolution (tuple): Target resolution as (width, height) depthmap (numpy.ndarray): Depth map corresponding to the image intrinsics (numpy.ndarray): Camera intrinsics matrix (3x3) additional_quantities (dict, optional): Additional image-related data to be processed alongside the main image with nearest interpolation. Defaults to None. Returns: tuple: Processed image, depthmap, and updated intrinsics matrix. If additional_quantities is provided, it returns those as well. """ if not isinstance(image, PIL.Image.Image): image = PIL.Image.fromarray(image) # Cropping centered on the principal point if necessary if self.principal_point_centered: W, H = image.size cx, cy = intrinsics[:2, 2].round().astype(int) if cx < 0 or cx >= W or cy < 0 or cy >= H: # Skip centered cropping if principal point is outside image bounds pass else: min_margin_x = min(cx, W - cx) min_margin_y = min(cy, H - cy) left, top = cx - min_margin_x, cy - min_margin_y right, bottom = cx + min_margin_x, cy + min_margin_y crop_bbox = (left, top, right, bottom) # Only perform the centered crop if the crop_bbox is larger than the target resolution crop_width = right - left crop_height = bottom - top if crop_width > resolution[0] and crop_height > resolution[1]: image, depthmap, intrinsics, additional_quantities = ( crop_image_and_other_optional_info( image=image, crop_bbox=crop_bbox, depthmap=depthmap, camera_intrinsics=intrinsics, additional_quantities=additional_quantities, ) ) # Get the target resolution for re-scaling target_rescale_resolution = np.array(resolution) if self.aug_crop > 1: target_rescale_resolution += self._rng.integers(0, self.aug_crop) # High-quality Lanczos down-scaling if necessary image, depthmap, intrinsics, additional_quantities = ( rescale_image_and_other_optional_info( image=image, output_resolution=target_rescale_resolution, depthmap=depthmap, camera_intrinsics=intrinsics, additional_quantities_to_be_resized_with_nearest=additional_quantities, ) ) # Actual cropping (if necessary) new_intrinsics = camera_matrix_of_crop( input_camera_matrix=intrinsics, input_resolution=image.size, output_resolution=resolution, offset_factor=0.5, ) crop_bbox = bbox_from_intrinsics_in_out( input_camera_matrix=intrinsics, output_camera_matrix=new_intrinsics, output_resolution=resolution, ) image, depthmap, new_intrinsics, additional_quantities = ( crop_image_and_other_optional_info( image=image, crop_bbox=crop_bbox, depthmap=depthmap, camera_intrinsics=intrinsics, additional_quantities=additional_quantities, ) ) # Return the output if additional_quantities is not None: return image, depthmap, new_intrinsics, additional_quantities else: return image, depthmap, new_intrinsics def _random_walk_sampling( self, scene_pairwise_covisibility, num_of_samples, max_retries=4, use_bidirectional_covis=True, ): """ Randomly samples S indices from an N x N covisbility matrix by forming adjacency edges such that the resulting subgraph (given by the indices) is connected. If the current node has no new unvisited neighbors, backtracking occurs. Retries with different starting indices if the desired number of samples is not reached, excluding previously visited components. Args: scene_pairwise_covisibility : np.ndarray (mmap) N x N covisibility matrix for the scene, where N is the number of views in the scene. num_of_samples : int The desired number of nodes to sample (num_of_samples < N). max_retries : int The maximum number of retries with different starting indices. use_bidirectional_covis : bool Whether to compute bidirectional covisibility by averaging row and column values. If False, uses only row access (faster for large memory-mapped arrays). Defaults to True. Returns: np.ndarray An array of sampled indices forming a connected subgraph. """ excluded_nodes = set() best_walk = [] # To keep track of the best walk found for _ in range(max_retries): visited = set() walk = [] # List to store the random walk sampling order stack = [] # Stack for backtracking # Choose a random starting index that is not in the excluded set all_nodes = set(range(len(scene_pairwise_covisibility))) available_nodes = list(all_nodes - excluded_nodes) if not available_nodes: break # No more nodes to try start = self._rng.choice(available_nodes) walk.append(start) visited.add(start) stack.append(start) # Continue until we have sampled S indices or all expandable nodes are exhausted while len(walk) < num_of_samples and stack: current = stack[-1] # Get the pairwise covisibility for the current node if use_bidirectional_covis: # Use bidirectional covisibility (slower for large memory-mapped arrays) pairwise_covisibility = ( scene_pairwise_covisibility[current, :] + scene_pairwise_covisibility[:, current].T ) / 2 else: # Use only row access (faster for large memory-mapped arrays) pairwise_covisibility = scene_pairwise_covisibility[current, :] # Normalize the covisibility using self covisibility pairwise_covisibility = pairwise_covisibility / ( pairwise_covisibility[current] + 1e-8 ) # Assign overlap score of zero to self-pairs pairwise_covisibility[current] = 0 # Threshold the covisibility to get adjacency list for the current node adjacency_list_for_current = ( pairwise_covisibility > self.covisibility_thres ).astype(int) adjacency_list_for_current = np.flatnonzero(adjacency_list_for_current) # Get all unvisited neighbors candidates = [ idx for idx in adjacency_list_for_current if idx not in visited ] # Remove visited nodes if candidates: # Randomly select one of the unvisited overlapping neighbors next_node = self._rng.choice(candidates) walk.append(next_node) visited.add(next_node) stack.append(next_node) else: # If no unvisited neighbor is available, backtrack stack.pop() # Update the best walk if the current walk is larger if len(walk) > len(best_walk): best_walk = walk # If we have enough samples, return the result if len(walk) >= num_of_samples: return np.array(walk) # Add all visited nodes to the excluded set excluded_nodes.update(visited) # If all retries are exhausted and we still don't have enough samples, return the best walk found return np.array(best_walk) def _sample_view_indices( self, num_views_to_sample, num_views_in_scene, scene_pairwise_covisibility, use_bidirectional_covis=True, ): """ Sample view indices from a scene based on the adjacency list and the number of views to sample. Args: num_views_to_sample (int): Number of views to sample. num_views_in_scene (int): Total number of views available in the scene. scene_pairwise_covisibility (np.ndarray): N x N covisibility matrix for the scene, where N is the number of views in the scene. use_bidirectional_covis (bool): Whether to compute bidirectional covisibility by averaging row and column values. If False, uses only row access (faster for large memory-mapped arrays). Returns: numpy.ndarray: Array of sampled view indices. """ if num_views_to_sample == num_views_in_scene: # Select all views in the scene view_indices = self._rng.permutation(num_views_in_scene) elif num_views_to_sample > num_views_in_scene: # Select all views in the scene and repeat them to get the desired number of views view_indices = self._rng.choice( num_views_in_scene, size=num_views_to_sample, replace=True ) else: # Select a subset of single component connected views in the scene using random walk sampling view_indices = self._random_walk_sampling( scene_pairwise_covisibility, num_views_to_sample, use_bidirectional_covis=use_bidirectional_covis, ) # If the required num of views can't be obtained even with 4 retries, repeat existing indices to get the desired number of views if len(view_indices) < num_views_to_sample: view_indices = self._rng.choice( view_indices, size=num_views_to_sample, replace=True ) return view_indices def _getitem_fn(self, idx): if isinstance(idx, tuple): # The idx is a tuple if specifying the aspect-ratio or/and the number of views if isinstance(self.num_views, int): idx, ar_idx = idx else: idx, ar_idx, num_views_to_sample_idx = idx else: assert len(self._resolutions) == 1 assert isinstance(self.num_views, int) ar_idx = 0 # Setup the rng if self.seed: # reseed for each _getitem_fn # Leads to deterministic sampling where repeating self.seed and self._seed_offset yields the same multi-view set again # Scenes will be repeated if size of dataset is artificially increased using "N @" or "N *" # When scenes are repeated, self._seed_offset is increased to ensure new multi-view sets # This is useful for evaluation if the number of dataset scenes is < N, yet we want unique multi-view sets each iter self._rng = np.random.default_rng(seed=self.seed + self._seed_offset + idx) elif not hasattr(self, "_rng"): seed = torch.initial_seed() # this is different for each dataloader process self._rng = np.random.default_rng(seed=seed) # Get the views for the given index and check that the number of views is correct resolution = self._resolutions[ar_idx] if isinstance(self.num_views, int): num_views_to_sample = self.num_views else: num_views_to_sample = self.num_views[num_views_to_sample_idx] views = self._get_views(idx, num_views_to_sample, resolution) if isinstance(self.num_views, int): assert len(views) == self.num_views else: assert len(views) in self.num_views for v, view in enumerate(views): # Store the index and other metadata view["idx"] = (idx, ar_idx, v) view["is_metric_scale"] = self.is_metric_scale view["is_synthetic"] = self.is_synthetic # Check the depth, intrinsics, and pose data (also other data if present) assert "camera_intrinsics" in view assert "camera_pose" in view assert np.isfinite(view["camera_pose"]).all(), ( f"NaN or infinite values in camera pose for view {view_name(view)}" ) assert np.isfinite(view["depthmap"]).all(), ( f"NaN or infinite values in depthmap for view {view_name(view)}" ) assert "valid_mask" not in view assert "pts3d" not in view, ( f"pts3d should not be there, they will be computed afterwards based on intrinsics+depthmap for view {view_name(view)}" ) if "prior_depth_z" in view: assert np.isfinite(view["prior_depth_z"]).all(), ( f"NaN or infinite values in prior_depth_z for view {view_name(view)}" ) if "non_ambiguous_mask" in view: assert np.isfinite(view["non_ambiguous_mask"]).all(), ( f"NaN or infinite values in non_ambiguous_mask for view {view_name(view)}" ) # Encode the image width, height = view["img"].size view["true_shape"] = np.int32((height, width)) view["img"] = self.transform(view["img"]) view["data_norm_type"] = self.data_norm_type # Compute the pointmaps, raymap and depth along ray ( pts3d, valid_mask, ray_origins_world, ray_directions_world, depth_along_ray, ray_directions_cam, pts3d_cam, ) = get_absolute_pointmaps_and_rays_info(**view) view["pts3d"] = pts3d view["valid_mask"] = valid_mask & np.isfinite(pts3d).all(axis=-1) view["depth_along_ray"] = depth_along_ray view["ray_directions_cam"] = ray_directions_cam view["pts3d_cam"] = pts3d_cam # Compute the prior depth along ray if present if "prior_depth_z" in view: prior_pts3d, _ = depthmap_to_camera_coordinates( view["prior_depth_z"], view["camera_intrinsics"] ) view["prior_depth_along_ray"] = np.linalg.norm(prior_pts3d, axis=-1) view["prior_depth_along_ray"] = view["prior_depth_along_ray"][..., None] del view["prior_depth_z"] # Convert ambiguous mask dtype to match valid mask dtype if "non_ambiguous_mask" in view: view["non_ambiguous_mask"] = view["non_ambiguous_mask"].astype( view["valid_mask"].dtype ) else: ambiguous_mask = view["depthmap"] < 0 view["non_ambiguous_mask"] = ~ambiguous_mask view["non_ambiguous_mask"] = view["non_ambiguous_mask"].astype( view["valid_mask"].dtype ) # Check all datatypes for key, val in view.items(): res, err_msg = is_good_type(val) assert res, f"{err_msg} with {key}={val} for view {view_name(view)}" # Check shapes assert view["depthmap"].shape == view["img"].shape[1:] assert view["depthmap"].shape == view["pts3d"].shape[:2] assert view["depthmap"].shape == view["valid_mask"].shape assert view["depthmap"].shape == view["depth_along_ray"].shape[:2] assert view["depthmap"].shape == view["ray_directions_cam"].shape[:2] assert view["depthmap"].shape == view["pts3d_cam"].shape[:2] if "prior_depth_along_ray" in view: assert view["depthmap"].shape == view["prior_depth_along_ray"].shape[:2] if "non_ambiguous_mask" in view: assert view["depthmap"].shape == view["non_ambiguous_mask"].shape # Expand the last dimennsion of the depthmap view["depthmap"] = view["depthmap"][..., None] # Append RNG state to the views, this allows to check whether the RNG is in the same state each time view["rng"] = int.from_bytes(self._rng.bytes(4), "big") # Compute and store the quaternions and translation for the camera poses # Notation is (x, y, z, w) for quaternions # This also ensures that the camera poses have a positive determinant (right-handed coordinate system) view["camera_pose_quats"] = ( Rotation.from_matrix(view["camera_pose"][:3, :3]) .as_quat() .astype(view["camera_pose"].dtype) ) view["camera_pose_trans"] = view["camera_pose"][:3, 3].astype( view["camera_pose"].dtype ) # Check the pointmaps, rays, depth along ray, and camera pose quaternions and translation to ensure they are finite assert np.isfinite(view["pts3d"]).all(), ( f"NaN in pts3d for view {view_name(view)}" ) assert np.isfinite(view["valid_mask"]).all(), ( f"NaN in valid_mask for view {view_name(view)}" ) assert np.isfinite(view["depth_along_ray"]).all(), ( f"NaN in depth_along_ray for view {view_name(view)}" ) assert np.isfinite(view["ray_directions_cam"]).all(), ( f"NaN in ray_directions_cam for view {view_name(view)}" ) assert np.isfinite(view["pts3d_cam"]).all(), ( f"NaN in pts3d_cam for view {view_name(view)}" ) assert np.isfinite(view["camera_pose_quats"]).all(), ( f"NaN in camera_pose_quats for view {view_name(view)}" ) assert np.isfinite(view["camera_pose_trans"]).all(), ( f"NaN in camera_pose_trans for view {view_name(view)}" ) if "prior_depth_along_ray" in view: assert np.isfinite(view["prior_depth_along_ray"]).all(), ( f"NaN in prior_depth_along_ray for view {view_name(view)}" ) return views def __getitem__(self, idx): if self.max_num_retries == 0: return self._getitem_fn(idx) num_retries = 0 while num_retries <= self.max_num_retries: try: return self._getitem_fn(idx) except Exception as e: scene_idx = idx[0] if isinstance(idx, tuple) else idx print( f"Error in {type(self).__name__}.__getitem__ for scene_idx={scene_idx}: {e}" ) if num_retries >= self.max_num_retries: print( f"Max retries ({self.max_num_retries}) reached, raising the exception" ) raise e # Retry with a different scene index num_retries += 1 if isinstance(idx, tuple): # The scene index is the first element of the tuple idx_list = list(idx) idx_list[0] = np.random.randint(0, len(self)) idx = tuple(idx_list) else: # The scene index is idx idx = np.random.randint(0, len(self)) scene_idx = idx[0] if isinstance(idx, tuple) else idx print( f"Retrying with scene_idx={scene_idx} ({num_retries} of {self.max_num_retries})" ) def is_good_type(v): """ Check if a value has an acceptable data type for processing in the dataset. Args: v: The value to check. Returns: tuple: A tuple containing: - bool: True if the type is acceptable, False otherwise. - str or None: Error message if the type is not acceptable, None otherwise. """ if isinstance(v, (str, int, tuple)): return True, None if v.dtype not in (np.float32, torch.float32, bool, np.int32, np.int64, np.uint8): return False, f"bad {v.dtype=}" return True, None def view_name(view, batch_index=None): """ Generate a string identifier for a view based on its dataset, label, and instance. Args: view (dict): Dictionary containing view information with 'dataset', 'label', and 'instance' keys. batch_index (int, optional): Index to select from batched data. Defaults to None. Returns: str: A formatted string in the form "dataset/label/instance". """ def sel(x): return x[batch_index] if batch_index not in (None, slice(None)) else x db = sel(view["dataset"]) label = sel(view["label"]) instance = sel(view["instance"]) return f"{db}/{label}/{instance}"