import copy import h5py import torch import numpy as np from time import time from typing import List, Tuple, Union, Any import src from src.utils import tensor_idx, is_sorted, indices_to_pointers, \ sizes_to_pointers, fast_repeat, save_tensor, load_tensor __all__ = ['CSRData', 'CSRBatch'] # TODO : all CSR reasoning could maybe be ported to PyG.utils.sparse and # torch_sparse.SparseTensor ? Would be good to refactor everything by # leveraging established dependencies. Still, need to look into the # features these offer and make sure they cover all our needs for the # project. In particular, we need efficient fancy indexing and batching # of sparse tensors. # One thing is for sure, as of Nov 2024, torch.sparse still does not do # the job: # - CSR tensors cannot be fancy-indexed # - batches of CSR tensors must contain the same number of elements # BUT if we move away from CSR in favor of COO: # - COO tensors do support index_select() # - COO tensors support cat(), stack(), vstack(), hstack() # Meanwhile, torch_sparse.SparseTensor seems to support CSR and COO, # fancy indexing, and concatenation. BUT, although feasible, some # basic elementwise operations on the values like addition or # multiplication requires non-trivial syntax class CSRData: """Implements the CSRData format and associated mechanisms in Torch. When defining a subclass A of CSRData, it is recommended to create an associated CSRBatch subclass by doing the following: - ABatch inherits from (A, CSRBatch) - A.get_base_class() returns A - A.get_batch_class() returns ABatch """ __value_serialization_keys__ = None __pointer_serialization_key__ = 'pointers' __is_index_value_serialization_key__ = 'is_index_value' def __init__( self, pointers: torch.Tensor, *args, dense: bool = False, is_index_value: List[bool] = None): """Initialize the pointers and values. Values are passed as args and stored in a list. They are expected to all have the same size and support torch tensor indexing (i.e. they can be torch tensor or CSRData objects themselves). If `dense=True`, pointers are treated as a dense tensor of indices to be converted into pointer indices. Optionally, a list of booleans `is_index_value` can be passed. It must be the same size as *args and indicates, for each value, whether it holds elements that should be treated as indices when stacking CSRData objects into a CSRBatch. If so, the indices will be updated wrt the cumulative size of the batched values. """ if dense: self.pointers, order = indices_to_pointers(pointers) args = [a[order] for a in args] else: self.pointers = pointers self.values = [*args] if len(args) > 0 else None if is_index_value is None or is_index_value == []: self.is_index_value = torch.zeros(self.num_values, dtype=torch.bool) else: self.is_index_value = torch.BoolTensor(is_index_value) if src.is_debug_enabled(): self.debug() def debug(self): if self.pointer_key in self.value_keys: raise ValueError( f"Cannot serialize {self.__class__.__name__} object because" f"'{self.pointer_key}' is both in `self.pointer_key` and " f"`self.value_keys`.") if len(self.value_keys) != self.num_values: raise ValueError( f"Cannot serialize {self.__class__.__name__} object because" f"`self.value_keys` has length {len(self.value_keys)} but " f"`self.num_values` is {self.num_values}.") # assert self.num_groups >= 1, \ # "pointer indices must cover at least one group." assert self.pointers[0] == 0, \ "The first pointer element must always be 0." assert torch.all(self.sizes >= 0), \ "pointer indices must be increasing." if self.values is not None: assert isinstance(self.values, list), \ "Values must be held in a list." assert all([len(v) == self.num_items for v in self.values]), \ "All value objects must have the same size." assert len(self.values[0]) == self.num_items, \ "pointers must cover the entire range of values." for v in self.values: if isinstance(v, CSRData): v.debug() if self.values is not None and self.is_index_value is not None: assert isinstance(self.is_index_value, torch.BoolTensor), \ "is_index_value must be a torch.BoolTensor." assert self.is_index_value.dtype == torch.bool, \ "is_index_value must be an tensor of booleans." assert self.is_index_value.ndim == 1, \ "is_index_value must be a 1D tensor." assert self.is_index_value.shape[0] == self.num_values, \ "is_index_value size must match the number of value tensors." def detach(self) -> 'CSRData': """Detach all tensors in the CSRData.""" self.pointers = self.pointers.detach() for i in range(self.num_values): self.values[i] = self.values[i].detach() return self def to(self, device, **kwargs) -> 'CSRData': """Move the CSRData to the specified device.""" self.pointers = self.pointers.to(device, **kwargs) for i in range(self.num_values): self.values[i] = self.values[i].to(device, **kwargs) return self def cpu(self, **kwargs) -> 'CSRData': """Move the CSRData to the CPU.""" return self.to('cpu', **kwargs) def cuda(self, **kwargs) -> 'CSRData': """Move the CSRData to the first available GPU.""" return self.to('cuda', **kwargs) @property def device(self) -> torch.device: return self.pointers.device @property def num_groups(self): return self.pointers.shape[0] - 1 @property def num_values(self): return len(self.values) if self.values is not None else 0 @property def num_items(self): return self.pointers[-1] @property def sizes(self) -> torch.Tensor: """Returns the size of each group (i.e. the pointer jumps). """ return self.pointers[1:] - self.pointers[:-1] @property def indices(self) -> torch.Tensor: """Returns the dense indices corresponding to the pointers. """ return fast_repeat( torch.arange(self.num_groups, device=self.device), self.sizes) @classmethod def get_base_class(cls) -> type: """Helps `self.from_list()` and `self.to_list()` identify which classes to use for batch collation and un-collation. """ return CSRData @classmethod def get_batch_class(cls) -> type: """Helps `self.from_list()` and `self.to_list()` identify which classes to use for batch collation and un-collation. """ return CSRBatch def clone(self) -> 'CSRData': """Shallow copy of self. This may cause issues for certain types of downstream operations, but it saves time and memory. In practice, it shouldn't be problematic in this project. """ out = copy.copy(self) out.pointers = copy.copy(self.pointers) out.values = copy.copy(self.values) return out def reindex_groups( self, group_indices: torch.Tensor, order: torch.Tensor = None, num_groups: int = None ) -> 'CSRData': """Returns a copy of self with modified pointers to account for new groups. Affects the num_groups and the order of groups. Injects 0-length pointers where need be. By default, pointers are implicitly linked to the group indices in range(0, self.num_groups). Here we provide new group_indices for the existing pointers, with group_indices[i] corresponding to the position of existing group i in the new tensor. The indices missing from group_indices account for empty groups to be injected. The num_groups specifies the number of groups in the new tensor. If not provided, it is inferred from the size of group_indices. """ if order is None: order = torch.argsort(group_indices) csr_new = self[order].insert_empty_groups( group_indices[order], num_groups=num_groups) return csr_new def insert_empty_groups( self, group_indices: torch.Tensor, num_groups: int = None ) -> 'CSRData': """Method called when in-place reindexing groups. The group_indices are assumed to be sorted and group_indices[i] corresponds to the position of existing group i in the new tensor. The indices missing from group_indices correspond to empty groups to be injected. The num_groups specifies the number of groups in the new tensor. If not provided, it is inferred from the size of group_indices. """ assert self.num_groups == group_indices.shape[0], \ "New group indices must correspond to the existing number " \ "of groups" assert is_sorted(group_indices), "New group indices must be sorted." if num_groups is not None: num_groups = max(group_indices.max() + 1, num_groups) else: num_groups = group_indices.max() + 1 starts = torch.cat([ torch.LongTensor([-1]).to(self.device), group_indices.to(self.device)]) ends = torch.cat([ group_indices.to(self.device), torch.LongTensor([num_groups]).to(self.device)]) repeats = ends - starts self.pointers = self.pointers.repeat_interleave(repeats) return self @staticmethod def index_select_pointers( pointers: torch.Tensor, indices: torch.Tensor ) -> Tuple[torch.Tensor, torch.Tensor]: """Index selection of pointers. Returns a new pointer tensor with updated pointers, along with an index tensor to be used to update any values tensor associated with the input pointers. """ assert indices.max() <= pointers.shape[0] - 2 device = pointers.device # Create the new pointers pointers_new = torch.cat([ torch.zeros(1, dtype=pointers.dtype, device=device), torch.cumsum(pointers[indices + 1] - pointers[indices], 0)]) # Create the indexing tensor to select and order values. # Simply, we could have used a list of slices but we want to # avoid for loops and list concatenations to benefit from torch # capabilities. sizes = pointers_new[1:] - pointers_new[:-1] val_idx = torch.arange(pointers_new[-1], device=device) val_idx -= torch.arange(pointers_new[-1] + 1, device=device)[ pointers_new[:-1]].repeat_interleave(sizes) val_idx += pointers[indices].repeat_interleave(sizes).to(device) return pointers_new, val_idx def __getitem__( self, idx: Union[int, List[int], torch.Tensor, np.ndarray] ) -> 'CSRData': """Indexing CSRData format. Supports Numpy and torch indexing mechanisms. Return a copy of self with updated pointers and values. """ idx = tensor_idx(idx).to(self.device) # Shallow copy self and edit pointers and values. This # preserves the class for CSRData subclasses. out = self.clone() # If idx is empty, return an empty CSRData with empty values # of consistent type if idx.shape[0] == 0: out.pointers = torch.LongTensor([0]) out.values = [v[[]] for v in self.values] else: # Select the pointers and prepare the values indexing pointers, val_idx = self.__class__.index_select_pointers( self.pointers, idx) out.pointers = pointers out.values = [v[val_idx] for v in self.values] if src.is_debug_enabled(): out.debug() return out def select( self, idx: Union[int, List[int], torch.Tensor, np.ndarray], *args, **kwargs ) -> 'CSRData': """Returns a new CSRData which indexes `self` using entries in `idx`. Supports torch and numpy fancy indexing. :parameter idx: int or 1D torch.LongTensor or numpy.NDArray Cluster indices to select from 'self'. Must NOT contain duplicates """ # Normal CSRData indexing, creates a new object in memory return self[idx] def __len__(self): return self.num_groups def __repr__(self): info = [ f"{key}={int(getattr(self, key))}" for key in ['num_groups', 'num_items']] info.append(f"device={self.device}") return f"{self.__class__.__name__}({', '.join(info)})" def __eq__(self, other: Any) -> bool: if not isinstance(other, self.__class__): if src.is_debug_enabled(): print(f'{self.__class__.__name__}.__eq__: classes differ') return False if not torch.equal(self.pointers, other.pointers): if src.is_debug_enabled(): print(f'{self.__class__.__name__}.__eq__: pointers differ') return False if not torch.equal(self.is_index_value, other.is_index_value): if src.is_debug_enabled(): print(f'{self.__class__.__name__}.__eq__: is_index_value differ') return False if self.num_values != other.num_values: if src.is_debug_enabled(): print(f'{self.__class__.__name__}.__eq__: num_values differ') return False for v1, v2 in zip(self.values, other.values): # NB: this may be a bit strong a constraint for Cluster # where values are well-attributed to the proper clusters # but their order differ inside the cluster. In reality, we # want a set, the order does not matter. We could normalize # things by using a lexsort on cluster and point indices but # this be a bit costly... if not torch.equal(v1, v2): if src.is_debug_enabled(): print(f'{self.__class__.__name__}.__eq__: values differ') return False return True def __hash__(self) -> int: """Hashing for an CSRData. """ return hash(( self.__class__.__name__, self.pointers, *(v for v in self.values))) @property def pointer_key(self) -> str: """Key name for pointers. This will be used as labels for serialization. """ return self.__pointer_serialization_key__ @property def value_keys(self) -> List[str]: """List of names for each value. These will be used as labels for serialization. """ if self.__value_serialization_keys__ is None: return [str(i) for i in range(self.num_values)] return self.__value_serialization_keys__ @property def is_index_value_key(self) -> str: """Key name for is_index_value. This will be used as labels for serialization. """ return self.__is_index_value_serialization_key__ def save( self, f: Union[str, h5py.File, h5py.Group], fp_dtype: torch.dtype = torch.float): """Save CSRData to HDF5 file. :param f: h5 file path of h5py.File or h5py.Group :param fp_dtype: torch dtype Data type to which floating point tensors will be cast before saving :return: """ if not isinstance(f, (h5py.File, h5py.Group)): with h5py.File(f, 'w') as file: self.save(file, fp_dtype=fp_dtype) return save_tensor(self.pointers, f, self.pointer_key, fp_dtype=fp_dtype) if self.is_index_value_key is not None: save_tensor( self.is_index_value, f, self.is_index_value_key, fp_dtype=fp_dtype) if self.values is None: return for k, v in zip(self.value_keys, self.values): save_tensor(v, f, k, fp_dtype=fp_dtype) @classmethod def load( cls, f: Union[str, h5py.File, h5py.Group], idx: Union[int, List, np.ndarray, torch.Tensor] = None, verbose: bool = False ) -> 'CSRData': """Load CSRData from an HDF5 file. See `CSRData.save` for writing such file. Options allow reading only part of the clusters. :param f: h5 file path of h5py.File or h5py.Group :param idx: int, list, numpy.ndarray, torch.Tensor Used to select clusters when reading. Supports fancy indexing :param verbose: bool """ if not isinstance(f, (h5py.File, h5py.Group)): with h5py.File(f, 'r') as file: out = cls.load(file, idx=idx, verbose=verbose) return out start = time() idx = tensor_idx(idx) if verbose: print(f'{cls.__name__}.load tensor_idx : {time() - start:0.5f}s') # Check if the file actually corresponds to a batch object # rather than its corresponding base object has_sizes = '__sizes__' in f.keys() is_not_batch = cls != cls.get_batch_class() has_no_indexing = idx is None or idx.shape[0] == 0 if has_sizes and is_not_batch and has_no_indexing: return cls.get_batch_class().load(f, idx=idx, verbose=verbose) # Check expected keys are in the file pointer_key = cls.__pointer_serialization_key__ value_keys = cls.__value_serialization_keys__ value_keys = value_keys if value_keys is not None else [] is_index_value_key = cls.__is_index_value_serialization_key__ assert pointer_key in f.keys() assert all(k in f.keys() for k in value_keys) assert is_index_value_key is None or is_index_value_key in f.keys() # If no value keys are provided, CSRData.save() falls back to # using integers to index values. So, need to infer the number # of values from the consecutive integer keys in the file if len(value_keys) == 0: num_values = 0 while str(num_values) in f.keys(): num_values += 1 value_keys = [str(i) for i in range(num_values)] if idx is None or idx.shape[0] == 0: start = time() pointers = load_tensor(f[pointer_key]) values = [load_tensor(f[k]) for k in value_keys] if verbose: print(f'{cls.__name__}.load read all : {time() - start:0.5f}s') start = time() out = cls(pointers, *values) if is_index_value_key is not None: out.is_index_value = load_tensor(f[is_index_value_key]).bool() if verbose: print(f'{cls.__name__}.load init : {time() - start:0.5f}s') return out # Read only pointers start and end indices based on idx start = time() ptr_start = load_tensor(f[pointer_key], idx=idx) ptr_end = load_tensor(f[pointer_key], idx=idx + 1) if verbose: print(f'{cls.__name__}.load read ptr : {time() - start:0.5f}s') # Create the new pointers start = time() pointers = torch.cat([ torch.zeros(1, dtype=ptr_start.dtype), torch.cumsum(ptr_end - ptr_start, 0)]) if verbose: print(f'{cls.__name__}.load new pointers : {time() - start:0.5f}s') # Create the indexing tensor to select and order values. # Simply, we could have used a list of slices, but we want to # avoid for loops and list concatenations to benefit from torch # capabilities. start = time() sizes = pointers[1:] - pointers[:-1] val_idx = torch.arange(pointers[-1]) val_idx -= torch.arange(pointers[-1] + 1)[ pointers[:-1]].repeat_interleave(sizes) val_idx += ptr_start.repeat_interleave(sizes) if verbose: print(f'{cls.__name__}.load val_idx : {time() - start:0.5f}s') # Read the values now we have computed the val_idx start = time() values = [load_tensor(f[k], idx=val_idx) for k in value_keys] if verbose: print(f'{cls.__name__}.load read values : {time() - start:0.5f}s') # Build the CSRData object start = time() out = cls(pointers, *values) if is_index_value_key is not None: out.is_index_value = load_tensor(f[is_index_value_key]).bool() if verbose: print(f'{cls.__name__}.load init : {time() - start:0.5f}s') return out class CSRBatch(CSRData): """ Wrapper class of CSRData to build a batch from a list of CSRData data and reconstruct it afterward. When defining a subclass A of CSRData, it is recommended to create an associated CSRBatch subclass by doing the following: - ABatch inherits from (A, CSRBatch) - A.get_base_class() returns A - A.get_batch_class() returns ABatch """ def __init__( self, pointers: torch.Tensor, *args, dense: bool = False, is_index_value: List[bool] = None): """Basic constructor for a CSRBatch. Batches are rather intended to be built using the from_list() method. """ super(CSRBatch, self).__init__( pointers, *args, dense=dense, is_index_value=is_index_value) self.__sizes__ = None @property def batch_pointers(self) -> torch.Tensor: return sizes_to_pointers(self.__sizes__) if self.__sizes__ is not None \ else None @property def batch_items_sizes(self) -> torch.Tensor: return self.__sizes__ if self.__sizes__ is not None else None @property def num_batch_items(self): return len(self.__sizes__) if self.__sizes__ is not None else 0 def detach(self) -> 'CSRBatch': """Detach all tensors in the CSRBatch.""" self = super().detach() self.__sizes__ = self.__sizes__.detach() if self.__sizes__ is not None \ else None return self def to(self, device, **kwargs) -> 'CSRBatch': """Move the CSRBatch to the specified device.""" out = super().to(device, **kwargs) out.__sizes__ = self.__sizes__.to(device, **kwargs) \ if self.__sizes__ is not None else None return out @classmethod def from_list(cls, csr_list: List['CSRData']) -> 'CSRBatch': assert isinstance(csr_list, list) and len(csr_list) > 0 assert isinstance(csr_list[0], CSRData), \ "All provided items must be CSRData objects." csr_cls = type(csr_list[0]) assert all([isinstance(csr, csr_cls) for csr in csr_list]), \ "All provided items must have the same class." device = csr_list[0].device assert all([csr.device == device for csr in csr_list]), \ "All provided items must be on the same device." num_values = csr_list[0].num_values assert all([csr.num_values == num_values for csr in csr_list]), \ "All provided items must have the same number of values." is_index_value = csr_list[0].is_index_value if is_index_value is not None: assert all([ np.array_equal(csr.is_index_value, is_index_value) for csr in csr_list]), \ "All provided items must have the same is_index_value." else: assert all([csr.is_index_value is None for csr in csr_list]), \ "All provided items must have the same is_index_value." if src.is_debug_enabled(): for csr in csr_list: csr.debug() # Offsets are used to stack pointer indices and values # identified as "index" value by `is_index_value` without # losing the indexing information they carry. offsets = torch.cumsum(torch.LongTensor( [0] + [csr.num_items for csr in csr_list[:-1]]), dim=0).to(device) # Stack pointers pointers = torch.cat(( torch.LongTensor([0]).to(device), *[csr.pointers[1:] + offset for csr, offset in zip(csr_list, offsets)]), dim=0) # Stack values values = [] for i in range(num_values): val_list = [csr.values[i] for csr in csr_list] if len(val_list) > 0 and isinstance(val_list[0], CSRData): val = val_list[0].from_list(val_list) elif is_index_value[i]: # "Index" values are stacked with updated indices. # For Clusters, this implies all point indices are # assumed to be present in the Cluster.points. There can # be no point with no cluster offsets = torch.LongTensor( [0] + [ v.max() + 1 if v.shape[0] > 0 else 0 for v in val_list[:-1]]) cum_offsets = torch.cumsum(offsets, dim=0).to(device) val = torch.cat([ v + o for v, o in zip(val_list, cum_offsets)], dim=0) else: val = torch.cat(val_list, dim=0) values.append(val) # Create the Batch object, depending on the data type # Default of CSRData is CSRBatch, but subclasses of CSRData # may define their own batch class inheriting from CSRBatch. batch = csr_list[0].get_batch_class()( pointers, *values, dense=False, is_index_value=is_index_value) batch.__sizes__ = torch.LongTensor([csr.num_groups for csr in csr_list]) return batch def to_list(self) -> List['CSRData']: if self.__sizes__ is None: raise RuntimeError( 'Cannot reconstruct CSRData data list from batch because the ' 'CSRBatch was not created using `CSRBatch.from_list()`.') group_pointers = self.batch_pointers item_pointers = self.pointers[group_pointers] # Recover pointers and index offsets for each CSRData item pointers = [ self.pointers[group_pointers[i]:group_pointers[i + 1] + 1] - item_pointers[i] for i in range(self.num_batch_items)] # Recover the values for each CSRData item values = [] for i in range(self.num_values): batch_value = self.values[i] if isinstance(batch_value, CSRData): val = batch_value.to_list() elif self.is_index_value[i]: val = [ batch_value[item_pointers[j]:item_pointers[j + 1]] - (batch_value[:item_pointers[j]].max() + 1 if j > 0 else 0) for j in range(self.num_batch_items)] # Hacky fix for a pesky edge case. When a `CSRBatch` is # "manually" created and `__sizes__` is populated (this # happens in multiple places across the project), # `to_list()` may produce negative items for # `is_index_value` has not been constructed using the # expecting offsetting. To fix this, we enforce that the # returned is `is_index_value` indices start at 0. This # is a subjective choice and may very well break the # meaning of the corresponding indices. To reproduce # this issue, create an `InstanceData` with # dense `obj` attribute, manually convert it to a # `InstanceBatch` with `__sizes__`, then call # `to_list()`. The `obj` attributes of the second and # later items will contain negatives for j in range(self.num_batch_items): if val[j].min() < 0: val[j] -= val[j].min() else: val = [batch_value[item_pointers[j]:item_pointers[j + 1]] for j in range(self.num_batch_items)] values.append(val) values = [list(x) for x in zip(*values)] csr_list = [ self.get_base_class()( j, *v, dense=False, is_index_value=self.is_index_value) for j, v in zip(pointers, values)] return csr_list def __repr__(self): info = [f"{key}={getattr(self, key)}" for key in [ 'num_batch_items', 'num_groups', 'num_items', 'device']] return f"{self.__class__.__name__}({', '.join(info)})" def select( self, idx: Union[int, List[int], torch.Tensor, np.ndarray], *args, **kwargs ) -> 'CSRData': """Indexing CSRBatch format. Supports Numpy and torch indexing mechanisms. Since indexing breaks batching, this will return a CSRData object with updated pointers and values. """ # Default indexing will return a CSRBatch object out_batch = super().select(idx, *args, **kwargs) # Convert to a base object, since batching mechanism is lost # after indexing. For this, we populate an object of the proper # class, initialized with fake data out = self.get_base_class()( torch.arange(1), *[torch.empty(0, dtype=v.dtype) for v in self.values]) out.pointers = out_batch.pointers out.values = out_batch.values out.is_index_value = out_batch.is_index_value return out def save( self, f: Union[str, h5py.File, h5py.Group], fp_dtype: torch.dtype = torch.float): """Save CSRBatch to HDF5 file. :param f: h5 file path of h5py.File or h5py.Group :param fp_dtype: torch dtype Data type to which floating point tensors will be cast before saving :return: """ if not isinstance(f, (h5py.File, h5py.Group)): with h5py.File(f, 'w') as file: self.save(file, fp_dtype=fp_dtype) return # CSRData.save stores pointers, values, and is_index_value super().save(f, fp_dtype=fp_dtype) # Need to additionally save __sizes__ to be able to maintain # the batching mechanism throughout serialization save_tensor(self.__sizes__, f, '__sizes__', fp_dtype=fp_dtype) @classmethod def load( cls, f: Union[str, h5py.File, h5py.Group], idx: Union[int, List, np.ndarray, torch.Tensor] = None, verbose: bool = False ) -> Union['CSRBatch', 'CSRData']: """Load CSRBatch from an HDF5 file. See `CSRData.save` for writing such file. Options allow reading only part of the clusters. :param f: h5 file path of h5py.File or h5py.Group :param idx: int, list, numpy.ndarray, torch.Tensor Used to select clusters when reading. Supports fancy indexing :param verbose: bool """ # Indexing breaks batching, so we return a base object if # indexing is required idx = tensor_idx(idx) if idx is not None and idx.shape[0] != 0: return cls.get_base_class().load(f, idx=idx, verbose=verbose) if not isinstance(f, (h5py.File, h5py.Group)): with h5py.File(f, 'r') as file: out = cls.load(file, idx=idx, verbose=verbose) return out # Check if the file actually corresponds to a batch object # rather than its corresponding base object if '__sizes__' not in f.keys(): return cls.get_base_class().load(f, idx=idx, verbose=verbose) # Load all attributes like the parent class, and also load # attributes necessary for batching out = super().load(f, idx=idx, verbose=verbose) out.__sizes__ = load_tensor(f['__sizes__']) return out