unienv_data.storages.hdf5¶
HDF5Storage(single_instance_space: HDF5SpaceType, root: Group, capacity: Optional[int] = None, reduce_io: bool = True, check_file: bool = True, is_mutable: Optional[bool] = None, multiprocessing: bool = False, filename: Optional[Union[str, PathLike]] = None, mode: Optional[str] = None, full_name: Optional[str] = None, h5py_kwargs: Optional[Dict[str, Any]] = None, length: Optional[int] = None, owns_root: bool = True)
Bases: SpaceStorage[HDF5BatchType, NumpyArrayType, NumpyDeviceType, NumpyDtypeType, NumpyRNGType]
Stores samples of a (nested) DictSpace / BoxSpace / BinarySpace / TextSpace in one HDF5 file,
one dataset per leaf, numpy backend only.
Dataset layout, chosen by create() from chunks / compression (a mapping sets them per key, "*" for
the remaining keys):
- Fixed capacity, no compression,
chunks=None(the default): contiguous datasets, the fastest layout for per-step appends and random reads. chunks=Noneorchunks=Truewith compression, or an unbounded storage: chunks of whole rows of aboutCHUNK_TARGET_BYTES(64 KiB), e.g. one 128x128x3 uint8 image or ~2300 7-float rows per chunk, so a row access touches (and decompresses) a single chunk. Compressed files come out somewhat larger than with h5py's sub-row chunk guess, which made every image read decompress dozens of chunks.- An explicit tuple is used as given, padded with the row shape:
chunks=(1,)stores every step in its own chunk (the per-step layout: cheapest single-row writes and reads with compression, worst ratio).
Compression ("gzip", "lzf") trades write and random-read speed for size. For camera images prefer
VideoStorage (or ImageStorage with JPEG / PNG) over gzip-compressed HDF5; keep low-dim keys uncompressed.
staticmethod
¶
build_space_from_hdf5_file(root: Union[Group, Dataset]) -> Tuple[int, Optional[int], HDF5SpaceType]
staticmethod
¶
load_replay_buffer_from_raw_hdf5(path: Union[str, PathLike]) -> ReplayBuffer[HDF5BatchType, NumpyArrayType, NumpyDeviceType, NumpyDtypeType, NumpyRNGType]
staticmethod
¶
call_function_on_first_dataset(root: Group, function: Callable[[Dataset], Any]) -> Any
staticmethod
¶
call_function_on_every_dataset(root: Group, function: Callable[[Dataset], None]) -> None
staticmethod
¶
get_from(root: Union[Group, Dataset], single_instance_space: HDF5SpaceType, index: Union[int, slice, Sequence[int], BArrayType], reduce_io: bool = True) -> Any
staticmethod
¶
set_to(root: Union[Group, Dataset], single_instance_space: HDF5SpaceType, index: Union[int, slice, Sequence[int], BArrayType], value: HDF5BatchType) -> None
classmethod
¶
create(single_instance_space, capacity, cache_path=None, multiprocessing: bool = False, initial_capacity: Optional[int] = None, compression: Union[Dict[str, Any], Optional[str]] = None, compression_level: Union[Dict[str, Any], Optional[int]] = None, chunks: Union[Dict[str, Any], Optional[Union[bool, Tuple[int, ...]]]] = None, reduce_io: bool = True, **kwargs) -> HDF5Storage
classmethod
¶
load_from(path, single_instance_space, *, capacity=None, read_only=True, multiprocessing: bool = False, reduce_io: bool = True, **kwargs) -> HDF5Storage
open_segment_writer() -> SegmentWriter
A writer that builds one segment of this storage off to the side; see SegmentWriter.
Unlike mark_segment_start() / append(), any number of writers can be open at a time.
The default keeps the samples in memory until the commit; storages override it to stream instead.
clear() -> None
Clear all data inside the storage and set the length to 0 if the storage has unlimited capacity. For storages with fixed capacity, this should reset the storage to its initial state.
is_fancy_index(index: Any) -> bool
Check if the given index is fancy indexing. Args: index (Any): Index to check Returns: bool: True if fancy indexing, False otherwise
fancy_indexing_to_supported_indexing(index: ndarray) -> Tuple[np.ndarray, np.ndarray]
Convert fancy indexing to limited fancy indexing that h5py supports. Args: index (np.ndarray): Fancy indexing array of shape (N,) Returns: np.ndarray: Limited fancy indexing array of shape (M,) np.ndarray: Indices to reconstruct the resulting data to original order from the fetched data, shape (N,)
fancy_indexing_to_supported_set_indexing(index: ndarray) -> Tuple[np.ndarray, np.ndarray]
Convert fancy indexing to limited fancy indexing that h5py supports for setting values. Args: index (np.ndarray): Fancy indexing array of shape (N,) Returns: np.ndarray: Limited fancy indexing array of shape (M,) np.ndarray: Indices to read data required to construct the limited set data, shape (M,)
io_reduced_indexing(original_index: ndarray, max_gap_size: int = 8) -> List[Union[slice, np.ndarray, int]]
Convert a supported fancy-index array (sorted, unique values) to a list of slices and fancy indices to reduce IO operations. Args: original_index (np.ndarray): Original indexing array of shape (N,) Returns: List[Union[slice, np.ndarray, int]]: List of slices and fancy indices
read_sorted_unique_rows(dataset: Dataset, index: ndarray) -> np.ndarray
dataset[index] for a sorted, duplicate-free integer array, without h5py point / fancy selections
(which build one hyperslab per element and are 10-100x slower than plain slices).
Dense or small spans are read as one bounding slice; otherwise every run of consecutive rows is one slice read.
write_rows(dataset: Dataset, start: int, value: Any, num_rows: Optional[int] = None) -> bool
dataset[start:start + num_rows] = value through the low-level h5py API when value is a plain array of exactly
num_rows whole rows: skips h5py's selection / broadcasting machinery (~30us of Python per call, 3-4x the cost of
the write itself). Returns False (writing nothing) when the fast path does not apply.
write_sorted_unique_rows(dataset: Dataset, index: ndarray, value: ndarray) -> None
dataset[index] = value for a sorted, duplicate-free integer array, as one slice write per consecutive run.