Skip to content

unienv_data.storages.video_storage

LOGGER module-attribute

LOGGER = getLogger(__name__)

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)

HWAccel class-attribute instance-attribute

HWAccel = HWAccel

container instance-attribute

container = open(filename, mode='r', hwaccel=hwaccel)

video_stream instance-attribute

video_stream = video[0]

video_reformatter instance-attribute

video_reformatter = VideoReformatter()

buffer_pixel_format instance-attribute

buffer_pixel_format = buffer_pixel_format

total_frames instance-attribute

total_frames = frames

frame_iterator instance-attribute

frame_iterator = decode(video_stream)

backend instance-attribute

backend = backend

device instance-attribute

device = device

available_hwdevices staticmethod

available_hwdevices() -> List[str]

get_auto_hwaccel staticmethod

get_auto_hwaccel() -> Optional[HWAccel]

seek

seek(frame_index: int)

read

read(index: Union[IndexableType, BArrayType], total_length: int) -> BArrayType

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)

HWAccel class-attribute instance-attribute

HWAccel = Literal['beta', 'ffmpeg']

decoder instance-attribute

decoder = VideoDecoder(filename, device='cuda', seek_mode=seek_mode)

backend instance-attribute

backend = backend

device instance-attribute

device = device

buffer_pixel_format instance-attribute

buffer_pixel_format = buffer_pixel_format

get_auto_hwaccel staticmethod

get_auto_hwaccel() -> Optional[HWAccel]

read

read(index: Union[IndexableType, ndarray], total_length: int) -> np.ndarray

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

PyAV_LOG_LEVEL class-attribute instance-attribute

PyAV_LOG_LEVEL = WARNING

single_file_ext class-attribute instance-attribute

single_file_ext = None

seek_mode instance-attribute

seek_mode = seek_mode

hwaccel instance-attribute

hwaccel = hardware_acceleration

codec instance-attribute

codec = get_auto_codec() if codec == 'auto' else codec

decode_backend instance-attribute

decode_backend = decode_backend

fps instance-attribute

fps = fps

file_pixel_format instance-attribute

file_pixel_format = file_pixel_format

buffer_pixel_format instance-attribute

buffer_pixel_format = buffer_pixel_format

capacity instance-attribute

capacity = capacity

cache_filename property

cache_filename: Union[str, PathLike]

is_multiprocessing_safe property

is_multiprocessing_safe: bool

is_mutable instance-attribute

is_mutable = mutable

backend property

backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType]

device property

device: Optional[BDeviceType]

single_instance_space instance-attribute

single_instance_space = single_instance_space

file_ext instance-attribute

file_ext = file_ext

length instance-attribute

length = length if capacity is None else capacity

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_auto_codec staticmethod

get_auto_codec(base: Optional[str] = None) -> str

get_from_file

get_from_file(filename: str, index: Union[IndexableType, BArrayType], total_length: int) -> BArrayType

set_to_file

set_to_file(filename: str, value: BArrayType)

dumps

dumps(path)

close

close()

extend_length

extend_length(length)

shrink_length

shrink_length(length)

get

get(index)

set

set(index, value)

clear

clear()

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].