unienv_data.storages.video_storage¶
PyAvVideoReader
¶
PyAvVideoReader(backend: ComputeBackend, filename: str, buffer_pixel_format: Optional[str] = None, hwaccel: Optional[Union[HWAccel, Literal['auto']]] = None, seek_mode: Literal['exact', 'approximate'] = 'exact', device: Optional[BDeviceType] = None)
TorchCodecVideoReader
¶
TorchCodecVideoReader(backend: ComputeBackend, filename: str, buffer_pixel_format: Optional[str] = None, hwaccel: Optional[Union[HWAccel, Literal['auto']]] = None, seek_mode: Literal['exact', 'approximate'] = 'exact', device: Optional[BDeviceType] = None)
VideoStorage
¶
VideoStorage(single_instance_space: BoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType], cache_filename: Union[str, PathLike], seek_mode: Literal['exact', 'approximate'] = 'exact', hardware_acceleration: Optional[Union[Any, Literal['auto']]] = 'auto', codec: Union[str, Literal['auto']] = 'auto', decode_backend: Literal['torchcodec', 'pyav', 'auto'] = 'auto', file_ext: str = 'mp4', file_pixel_format: Optional[str] = None, buffer_pixel_format: str = 'rgb24', fps: int = 15, mutable: bool = True, capacity: Optional[int] = None, length: int = 0)
Bases: EpisodeStorageBase[BArrayType, BArrayType, BDeviceType, BDtypeType, BRNGType]
A storage for RGB or depth video data using video files
If encoding RGB video
- Set buffer_pixel_format to rgb24
- Set file_pixel_format to None (especially when running with nvenc codec)
- Set file_ext to anything you like (e.g., "mp4", "avi", "mkv", etc.)
If encoding depth video
- Set buffer_pixel_format to gray16le (You can use rescale transform inside a TransformedStorage to convert depth values to this format, where dtype should be np.uint16) - if in meters, set min to 0 and max to 65.535 as the multiplication factor is 1000 (i.e., depth in mm)
- Set file_pixel_format to gray16le
- Set codec to a lossless codec that supports gray16le, e.g., ffv1
- Set file_ext to mkv
create
classmethod
¶
create(single_instance_space: BoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType], *args, seek_mode: Literal['exact', 'approximate'] = 'exact', hardware_acceleration: Optional[Union[Any, Literal['auto']]] = 'auto', codec: Union[str, Literal['auto']] = 'auto', file_ext: str = 'mp4', file_pixel_format: Optional[str] = None, buffer_pixel_format: str = 'rgb24', fps: int = 15, capacity: Optional[int] = None, cache_path: Optional[str] = None, multiprocessing: bool = False, **kwargs) -> VideoStorage[BArrayType, BDeviceType, BDtypeType, BRNGType]
load_from
classmethod
¶
load_from(path: Union[str, PathLike], single_instance_space: BoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType], *, seek_mode: Literal['exact', 'approximate'] = 'exact', hardware_acceleration: Optional[Union[Any, Literal['auto']]] = 'auto', codec: Union[str, Literal['auto']] = 'auto', capacity: Optional[int] = None, read_only: bool = True, multiprocessing: bool = False, **kwargs) -> VideoStorage[BArrayType, BDeviceType, BDtypeType, BRNGType]
get_from_file
¶
get_from_file(filename: str, index: Union[IndexableType, BArrayType], total_length: int) -> BArrayType
get_start_end_filename_iter
¶
get_start_end_filename_iter() -> Iterable[Tuple[int, int, str]]
Iterate over (start_idx, end_idx, filename) tuples, constructing filenames on the fly.
convert_read_index_to_filenames_and_offsets
¶
convert_read_index_to_filenames_and_offsets(index: Union[IndexableType, BArrayType]) -> Tuple[int, List[Tuple[Union[str, os.PathLike], Union[int, BArrayType], Union[int, BArrayType]]]]
Convert an index (which can be an integer, slice, list of integers, or backend array) to a list of filenames and offsets. Each filename corresponds to an episode file, and the offset indicates the position within that episode.
remove_index_range
¶
remove_index_range(start_index: int, end_index: int)
Remove data in the index range [start_index, end_index] (inclusive). This handles both regular files and wrap-around files. If start_index > end_index, this is a wrap-around removal covering [start_index, capacity) and [0, end_index].