video
sleap_io.model.video
¶
Data model for videos.
The Video class is a SLEAP data structure that stores information regarding
a video and its components used in SLEAP.
Classes:
| Name | Description |
|---|---|
HDF5Video |
Video backend for reading videos stored in HDF5 files. |
ImageVideo |
Video backend for reading videos stored as image files. |
MediaVideo |
Video backend for reading videos stored as common media files. |
Video |
|
VideoBackend |
Base class for video backends. |
VideoWriter |
Simple video writer using imageio and FFMPEG. |
Functions:
| Name | Description |
|---|---|
crop_points |
Adjust point coordinates for a crop transformation. |
is_file_accessible |
Check if a file is accessible. |
uncrop_points |
Map crop-local point coordinates back to source coordinates. |
Attributes:
| Name | Type | Description |
|---|---|---|
__cached__ |
str(object='') -> str |
|
__doc__ |
str(object='') -> str |
|
__file__ |
str(object='') -> str |
|
__name__ |
str(object='') -> str |
|
__package__ |
str(object='') -> str |
__cached__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/model/__pycache__/video.cpython-313.pyc'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__doc__ = 'Data model for videos.\n\nThe `Video` class is a SLEAP data structure that stores information regarding\na video and its components used in SLEAP.\n'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__file__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/model/video.py'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__name__ = 'sleap_io.model.video'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__package__ = 'sleap_io.model'
module-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
HDF5Video
¶
Bases: sleap_io.io.video_reading.VideoBackend
Video backend for reading videos stored in HDF5 files.
This backend supports reading videos stored in HDF5 files, both in rank-4 datasets as well as in datasets with lists of binary-encoded images.
Embedded image datasets are used in SLEAP when exporting package files (.pkg.slp)
with videos embedded in them. This is useful for bundling training or inference data
without having to worry about the videos (or frame images) being moved or deleted.
It is expected that these types of datasets will be in a Group with a int8
variable length dataset called "video". This dataset must also contain an
attribute called "format" with a string describing the image format (e.g., "png" or
"jpg") which will be used to decode it appropriately.
If a frame_numbers dataset is present in the group, it will be used to map from
source video frames to the frames in the dataset. This is useful to preserve frame
indexing when exporting a subset of frames in the video. It will also be used to
populate frame_map and source_inds attributes.
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
Path to HDF5 file (.h5, .hdf5 or .slp). |
|
grayscale |
Whether to force grayscale. If None, autodetect on first frame load. |
|
keep_open |
Whether to keep the video reader open between calls to read frames. If False, will close the reader after each call. If True (the default), it will keep the reader open and cache it for subsequent calls which may enhance the performance of reading multiple frames. |
|
dataset |
Name of dataset to read from. If |
|
input_format |
Format of the data in the dataset. One of "channels_last" (the
default) in |
|
frame_map |
Mapping from frame indices to indices in the dataset. This is used to translate between the frame indices of the images within their source video and the indices of the images in the dataset. This is only used when reading embedded image datasets. |
|
source_filename |
Path to the source video file. This is metadata and only used when reading embedded image datasets. |
|
source_inds |
Indices of the frames in the source video file. This is metadata and only used when reading embedded image datasets. |
|
image_format |
Format of the images in the embedded dataset. This is metadata and only used when reading embedded image datasets. |
|
channel_order |
Channel order of embedded images, either "RGB" or "BGR". This is used to ensure consistent color channel ordering when decoding embedded images. If the encoding and decoding plugins have different channel orders, the channels will be automatically flipped during decoding. |
|
plugin |
Plugin to use for decoding embedded images. One of "opencv" or "FFMPEG". If None, uses the global default or auto-detects based on available packages. Note that "pyav" is automatically mapped to "FFMPEG" since PyAV doesn't support image decoding. |
Notes
Concurrent reads of a single remote (URL-backed) HDF5Video from
multiple threads are safe: although all reads share one cached fsspec
file-like (a single byte position), h5py serializes every HDF5 C-library
call under a global recursive lock (h5py._objects.phil), so the
seek+read pair a frame read performs is never interleaved across threads.
For true read parallelism (rather than just safety), construct
independent Video/HDF5Video instances per worker; each gets its own
fsspec file and block cache.
Methods:
| Name | Description |
|---|---|
__attrs_post_init__ |
Auto-detect dataset and frame map heuristically. |
__eq__ |
Method generated by attrs for class HDF5Video. |
__getstate__ |
Return state for pickling/deepcopy, dropping unpicklable handles. |
__init__ |
Method generated by attrs for class HDF5Video. |
__repr__ |
Method generated by attrs for class HDF5Video. |
__setattr__ |
Method generated by attrs for class HDF5Video. |
close |
Release the cached HDF5 reader and the cached fsspec URL file-like. |
decode_embedded |
Decode an embedded image string into a numpy array. |
get_frame_raw_bytes |
Get raw encoded bytes for a frame without decoding. |
has_frame |
Check if a frame index is contained in the video. |
read_crop |
Read a spatial hyperslab of a frame, padded to the crop shape. |
read_crops |
Batched :meth: |
read_test_frame |
Read a single frame from the video to test for grayscale. |
Source code in sleap_io/io/video_reading.py
@attrs.define
class HDF5Video(VideoBackend):
"""Video backend for reading videos stored in HDF5 files.
This backend supports reading videos stored in HDF5 files, both in rank-4 datasets
as well as in datasets with lists of binary-encoded images.
Embedded image datasets are used in SLEAP when exporting package files (`.pkg.slp`)
with videos embedded in them. This is useful for bundling training or inference data
without having to worry about the videos (or frame images) being moved or deleted.
It is expected that these types of datasets will be in a `Group` with a `int8`
variable length dataset called `"video"`. This dataset must also contain an
attribute called "format" with a string describing the image format (e.g., "png" or
"jpg") which will be used to decode it appropriately.
If a `frame_numbers` dataset is present in the group, it will be used to map from
source video frames to the frames in the dataset. This is useful to preserve frame
indexing when exporting a subset of frames in the video. It will also be used to
populate `frame_map` and `source_inds` attributes.
Attributes:
filename: Path to HDF5 file (.h5, .hdf5 or .slp).
grayscale: Whether to force grayscale. If None, autodetect on first frame load.
keep_open: Whether to keep the video reader open between calls to read frames.
If False, will close the reader after each call. If True (the default), it
will keep the reader open and cache it for subsequent calls which may
enhance the performance of reading multiple frames.
dataset: Name of dataset to read from. If `None`, will try to find a rank-4
dataset by iterating through datasets in the file. If specifying an embedded
dataset, this can be the group containing a "video" dataset or the dataset
itself (e.g., "video0" or "video0/video").
input_format: Format of the data in the dataset. One of "channels_last" (the
default) in `(frames, height, width, channels)` order or "channels_first" in
`(frames, channels, width, height)` order. Embedded datasets should use the
"channels_last" format.
frame_map: Mapping from frame indices to indices in the dataset. This is used to
translate between the frame indices of the images within their source video
and the indices of the images in the dataset. This is only used when reading
embedded image datasets.
source_filename: Path to the source video file. This is metadata and only used
when reading embedded image datasets.
source_inds: Indices of the frames in the source video file. This is metadata
and only used when reading embedded image datasets.
image_format: Format of the images in the embedded dataset. This is metadata and
only used when reading embedded image datasets.
channel_order: Channel order of embedded images, either "RGB" or "BGR". This is
used to ensure consistent color channel ordering when decoding embedded
images. If the encoding and decoding plugins have different channel orders,
the channels will be automatically flipped during decoding.
plugin: Plugin to use for decoding embedded images. One of "opencv" or
"FFMPEG". If None, uses the global default or auto-detects based on
available packages. Note that "pyav" is automatically mapped to "FFMPEG"
since PyAV doesn't support image decoding.
Notes:
Concurrent reads of a single remote (URL-backed) `HDF5Video` from
multiple threads are safe: although all reads share one cached fsspec
file-like (a single byte position), h5py serializes every HDF5 C-library
call under a global recursive lock (`h5py._objects.phil`), so the
seek+read pair a frame read performs is never interleaved across threads.
For true read *parallelism* (rather than just safety), construct
independent `Video`/`HDF5Video` instances per worker; each gets its own
fsspec file and block cache.
"""
dataset: str | None = None
input_format: str = attrs.field(
default="channels_last",
validator=attrs.validators.in_(["channels_last", "channels_first"]),
)
frame_map: dict[int, int] = attrs.field(init=False, default=attrs.Factory(dict))
_can_push_crop_cached: bool | None = attrs.field(
init=False, default=None, repr=False, eq=False
)
source_filename: str | None = None
source_inds: np.ndarray | None = None
image_format: str = "hdf5"
channel_order: str = "RGB"
plugin: str | None = None
_url_file: object | None = attrs.field(
init=False, default=None, repr=False, eq=False
)
# ``_url_headers`` / ``_url_stream_mode`` are ``init=True`` (attrs derives the
# constructor aliases ``url_headers`` / ``url_stream_mode`` by stripping the
# leading underscore) so the metadata probe in ``__attrs_post_init__`` runs
# *authenticated*: an embedded ``pkg.slp`` over an auth-gated URL would
# otherwise probe with no headers and silently lose the embedded-image
# metadata. They remain ``repr=False, eq=False`` and, because the attribute
# names keep the leading underscore, the name-based ``__getstate__`` pickle
# contract is unchanged.
_url_headers: dict[str, str] | None = attrs.field(
default=None, repr=False, eq=False
)
_url_stream_mode: str = attrs.field(default="blockcache", repr=False, eq=False)
EXTS = ("h5", "hdf5", "slp")
def _open_h5(self) -> h5py.File:
"""Open the backing HDF5 file as an ``h5py.File`` in read mode.
For local paths this opens ``self.filename`` directly. For URLs it lazily
opens (and caches on ``self._url_file``) an fsspec-backed file-like object
via :func:`sleap_io.io._remote.open_url` and wraps it with ``h5py``.
Returns:
An open ``h5py.File`` handle. The caller owns closing the returned
handle; the cached ``self._url_file`` is reused across reads and
dropped on pickling.
"""
from sleap_io.io import _remote
if _remote._is_url(self.filename):
if self._url_file is None:
self._url_file = _remote.open_url(
self.filename,
headers=self._url_headers,
stream_mode=self._url_stream_mode,
)
return h5py.File(self._url_file, "r")
return h5py.File(self.filename, "r")
def _close_url_file(self) -> None:
"""Close and drop the cached fsspec URL file-like, if any (idempotent).
A no-op for local files (where ``_url_file`` is never set) and when it
has already been closed/dropped.
"""
if self._url_file is None:
return
try:
self._url_file.close()
except Exception: # pragma: no cover - defensive: close() should not raise
pass
self._url_file = None
def _release_probe_url_file(self, preexisting: bool) -> None:
"""Close and drop ``self._url_file`` if a probe opened it.
Used by :meth:`__attrs_post_init__` so that a remote handle opened just
to sniff the dataset/frame-map does not leak and is not reused by later
(possibly authenticated) reads. A no-op for local files and when the
cached file-like already existed before the probe.
Args:
preexisting: Whether ``self._url_file`` was already set before the
probe opened the file (in which case it is left untouched).
"""
if preexisting:
return
self._close_url_file()
def close(self) -> None:
"""Release the cached HDF5 reader and the cached fsspec URL file-like.
Extends :meth:`VideoBackend.close` (which drops the cached ``h5py.File``
reader) by also closing the fsspec-backed ``_url_file`` shared across
reads, which ``h5py.File.close()`` does not close on its own. Both are
lazily reopened on the next read, so this is safe to call between reads.
"""
super().close()
self._close_url_file()
def __getstate__(self) -> dict:
"""Return state for pickling/deepcopy, dropping unpicklable handles.
Extends :meth:`VideoBackend.__getstate__` to also drop the cached
fsspec-backed ``_url_file`` (reopened lazily by :meth:`_open_h5`).
"""
state = super().__getstate__()
state["_url_file"] = None
return state
def __attrs_post_init__(self):
"""Auto-detect dataset and frame map heuristically."""
# Check if the file accessible before applying heuristics.
# For URLs, track whether this probe opened the cached fsspec file-like
# so it can be released afterwards (it would otherwise leak the handle on
# an early return / exception, and a probe-time open may predate the
# final auth headers being applied).
url_file_preexisting = self._url_file is not None
try:
f = self._open_h5()
except OSError:
self._release_probe_url_file(url_file_preexisting)
return
try:
if self.dataset is None:
# Iterate through datasets to find a rank 4 array.
def find_movies(name, obj):
if isinstance(obj, h5py.Dataset) and obj.ndim == 4:
self.dataset = name
return True
f.visititems(find_movies)
if self.dataset is None:
# Iterate through datasets to find an embedded video dataset.
def find_embedded(name, obj):
if isinstance(obj, h5py.Dataset) and name.endswith("/video"):
self.dataset = name
return True
f.visititems(find_embedded)
if self.dataset is None:
# Couldn't find video datasets.
return
if isinstance(f[self.dataset], h5py.Group):
# If this is a group, assume it's an embedded video dataset.
if "video" in f[self.dataset]:
self.dataset = f"{self.dataset}/video"
if self.dataset.split("/")[-1] == "video":
# This may be an embedded video dataset. Check for frame map.
ds = f[self.dataset]
if "format" in ds.attrs:
self.image_format = ds.attrs["format"]
# Read channel_order, with backwards compatibility
if "channel_order" in ds.attrs:
self.channel_order = ds.attrs["channel_order"]
else:
# Backwards compatibility: Check format_id for older files
# Prior to format 1.4, embedded images were primarily encoded
# with OpenCV which uses BGR, so default to BGR for older
# formats
if "metadata" in f and "format_id" in f["metadata"].attrs:
format_id = f["metadata"].attrs["format_id"]
if format_id < 1.4:
self.channel_order = "BGR" # Legacy default
# If no format_id found, assume BGR (safest legacy default)
# since most embedded images before this change used OpenCV
if "frame_numbers" in ds.parent:
frame_numbers = ds.parent["frame_numbers"][:].astype(int)
self.frame_map = {
frame: idx for idx, frame in enumerate(frame_numbers)
}
self.source_inds = frame_numbers
if "source_video" in ds.parent:
source_grp = ds.parent["source_video"]
# Source metadata is normally in the "json" attribute, but
# oversized metadata (e.g. an image-sequence source with many
# thousands of filenames, exceeding HDF5's 64 KB attribute limit)
# is stored in a "json" *dataset* instead (see
# ``slp._write_source_video_json``). Read whichever is present so
# such packages remain openable -- otherwise the backend fails to
# open, ``Video.backend`` is left ``None``, and embedded frames
# cannot be read.
if "json" in source_grp:
source_json = source_grp["json"][()]
else:
source_json = source_grp.attrs["json"]
self.source_filename = json.loads(source_json)["backend"][
"filename"
]
# Read FPS from attributes if present
if "fps" in ds.attrs:
self._fps = float(ds.attrs["fps"])
elif "fps" in ds.parent.attrs:
self._fps = float(ds.parent.attrs["fps"])
finally:
f.close()
self._release_probe_url_file(url_file_preexisting)
# Set default plugin if not specified (use image plugin, not video plugin)
if self.plugin is None:
# Check image plugin default first (for embedded images)
if _default_image_plugin is not None:
self.plugin = _default_image_plugin
# Otherwise auto-detect (for embedded image decoding)
elif "cv2" in sys.modules:
self.plugin = "opencv"
else:
self.plugin = "imageio" # imageio fallback
@property
def num_frames(self) -> int:
"""Number of frames in the video."""
with self._open_h5() as f:
return f[self.dataset].shape[0]
@property
def img_shape(self) -> tuple[int, int, int]:
"""Shape of a single frame in the video as `(height, width, channels)`."""
with self._open_h5() as f:
ds = f[self.dataset]
img_shape = None
if "height" in ds.attrs:
# Try to get shape from the attributes.
img_shape = (
ds.attrs["height"],
ds.attrs["width"],
ds.attrs["channels"],
)
if img_shape[0] == 0 or img_shape[1] == 0:
# Invalidate the shape if the attributes are zero.
img_shape = None
if img_shape is None and self.image_format == "hdf5" and ds.ndim == 4:
# Use the dataset shape if just stored as a rank-4 array.
img_shape = ds.shape[1:]
if self.input_format == "channels_first":
img_shape = img_shape[::-1]
if img_shape is None:
# Fall back to reading a test frame.
return super().img_shape
return int(img_shape[0]), int(img_shape[1]), int(img_shape[2])
def read_test_frame(self) -> np.ndarray:
"""Read a single frame from the video to test for grayscale."""
if self.frame_map:
frame_idx = list(self.frame_map.keys())[0]
else:
frame_idx = 0
return self._read_frame(frame_idx)
@property
def has_embedded_images(self) -> bool:
"""Return True if the dataset contains embedded images."""
return self.image_format is not None and self.image_format != "hdf5"
@property
def embedded_frame_inds(self) -> list[int]:
"""Return the frame indices of the embedded images."""
return list(self.frame_map.keys())
def decode_embedded(self, img_string: np.ndarray) -> np.ndarray:
"""Decode an embedded image string into a numpy array.
Args:
img_string: Binary string of the image as a `int8` numpy vector with the
bytes as values corresponding to the format-encoded image.
Returns:
The decoded image as a numpy array of shape `(height, width, channels)`. If
a rank-2 image is decoded, it will be expanded such that channels will be 1.
This method does not apply grayscale conversion as per the `grayscale`
attribute. Use the `get_frame` or `get_frames` methods of the `VideoBackend`
to apply grayscale conversion rather than calling this function directly.
"""
# Decode based on plugin
if self.plugin == "opencv":
img = cv2.imdecode(img_string, cv2.IMREAD_UNCHANGED)
decoder_order = "BGR" # OpenCV decodes to BGR
else:
# Use imageio for FFMPEG or any other plugin
img = iio.imread(BytesIO(img_string), extension=f".{self.image_format}")
decoder_order = "RGB" # imageio decodes to RGB
if img.ndim == 2:
img = np.expand_dims(img, axis=-1)
# Convert channel order if needed
# If the stored order doesn't match the decoder order, flip channels
if img.shape[-1] == 3 and self.channel_order != decoder_order:
img = img[..., ::-1] # Flip RGB <-> BGR
return img
def has_frame(self, frame_idx: int) -> bool:
"""Check if a frame index is contained in the video.
Args:
frame_idx: Index of frame to check.
Returns:
`True` if the index is contained in the video, otherwise `False`.
"""
if self.frame_map:
return frame_idx in self.frame_map
else:
return frame_idx < len(self)
def get_frame_raw_bytes(self, frame_idx: int) -> np.ndarray | None:
"""Get raw encoded bytes for a frame without decoding.
This method reads the raw compressed image data (PNG/JPEG bytes) directly
from the HDF5 dataset without decoding it. This is useful for fast copying
of embedded images when the target format matches the source format.
Args:
frame_idx: Index of the frame to read.
Returns:
Raw encoded bytes as int8 numpy array, or None if:
- The backend doesn't have embedded images (including "hdf5" format which
stores raw numpy arrays, not encoded images)
- The frame index is not available
Notes:
For variable-length datasets, returns the raw bytes directly.
For fixed-length datasets, returns bytes with trailing zeros stripped.
"""
if not self.has_embedded_images:
return None
if not self.has_frame(frame_idx):
return None
# Get the internal index (handle frame_map)
internal_idx = (
self.frame_map.get(frame_idx, frame_idx) if self.frame_map else frame_idx
)
# Read directly from dataset
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
else:
f = self._open_h5()
ds = f[self.dataset]
raw_bytes = ds[internal_idx]
# Handle fixed-length padding (strip trailing zeros)
is_vlen = h5py.check_vlen_dtype(ds.dtype) is not None
if not is_vlen:
# Find last non-zero byte
non_zero_mask = raw_bytes != 0
if non_zero_mask.any():
last_non_zero = np.where(non_zero_mask)[0][-1]
raw_bytes = raw_bytes[: last_non_zero + 1]
if not self.keep_open:
f.close()
return raw_bytes
def _read_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the video.
Args:
frame_idx: Index of frame to read.
Returns:
The frame as a numpy array of shape `(height, width, channels)`.
Notes:
This does not apply grayscale conversion. It is recommended to use the
`get_frame` method of the `VideoBackend` class instead.
"""
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
else:
f = self._open_h5()
ds = f[self.dataset]
if self.frame_map:
frame_idx = self.frame_map[frame_idx]
img = ds[frame_idx]
if self.has_embedded_images:
img = self.decode_embedded(img)
if self.input_format == "channels_first":
img = np.transpose(img, (2, 1, 0))
if not self.keep_open:
f.close()
return img
def _read_frames(self, frame_inds: list) -> np.ndarray:
"""Read a list of frames from the video.
Args:
frame_inds: List of indices of frames to read.
Returns:
The frame as a numpy array of shape `(frames, height, width, channels)`.
Notes:
This does not apply grayscale conversion. It is recommended to use the
`get_frames` method of the `VideoBackend` class instead.
"""
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
else:
f = self._open_h5()
if self.frame_map:
frame_inds = [self.frame_map[idx] for idx in frame_inds]
ds = f[self.dataset]
imgs = ds[frame_inds]
if "format" in ds.attrs:
imgs = np.stack(
[self.decode_embedded(img) for img in imgs],
axis=0,
)
if self.input_format == "channels_first":
imgs = np.transpose(imgs, (0, 3, 2, 1))
if not self.keep_open:
f.close()
return imgs
@property
def _can_push_crop(self) -> bool:
"""Whether this dataset supports HDF5 crop pushdown (dataset-level gate).
Pushdown reads only a spatial hyperslab of a frame instead of decoding the
whole frame, but it is only valid (and beneficial) for raw rank-4 chunked
datasets with sub-frame spatial chunking and no embedded/frame-mapped
subset. This is the dataset-level gate only (the per-call "crop smaller than
the chunk span" predicate is evaluated in :meth:`read_crop`/:meth:`read_crops`).
The probe reflects the immutable on-disk layout, so the result is cached
after the first call (no file open per read).
Returns:
``True`` if the dataset is a raw (``image_format == "hdf5"``) rank-4
chunked array with sub-frame spatial chunking and an empty
``frame_map``; ``False`` otherwise (including any error while probing,
so a non-applicable dataset never raises).
"""
# Cheap short-circuit (no file open) for embedded/frame-mapped datasets.
if self.image_format != "hdf5" or self.frame_map:
return False
if self._can_push_crop_cached is None:
self._can_push_crop_cached = self._probe_can_push_crop()
return self._can_push_crop_cached
def _probe_can_push_crop(self) -> bool:
"""Probe the on-disk layout for pushdown eligibility (opens the file once)."""
try:
with self._open_h5() as f:
ds = f[self.dataset]
if ds.ndim != 4 or ds.chunks is None:
return False
chunks = ds.chunks
if self.input_format == "channels_first":
# On-disk layout is (F, C, W, H).
disk_w, disk_h = ds.shape[2], ds.shape[3]
return chunks[2] < disk_w or chunks[3] < disk_h
# channels_last on-disk layout is (F, H, W, C).
height, width = ds.shape[1], ds.shape[2]
return chunks[1] < height or chunks[2] < width
except (OSError, KeyError, TypeError): # pragma: no cover - defensive
return False
def read_crop(
self,
frame_idx: int,
crop: tuple[int, int, int, int],
fill: int | tuple[int, ...] = 0,
) -> np.ndarray | None:
"""Read a spatial hyperslab of a frame, padded to the crop shape.
This is the single-frame HDF5 crop pushdown hook consumed by
:class:`CropVideoBackend`. When applicable, it reads only the spatial
region of the frame that overlaps ``crop`` directly from the chunked
dataset (avoiding a full-frame decode) and pads out-of-bounds regions
exactly as :func:`sleap_io.transform.frame.crop_frame` would.
Args:
frame_idx: Index of the frame to read (source-video index; mapped
through ``frame_map`` if present, though pushdown is gated off when
a ``frame_map`` exists).
crop: Crop region ``(x1, y1, x2, y2)`` with ``x2``/``y2`` exclusive.
May be negative or exceed the frame bounds (padded with ``fill``).
fill: Fill value for out-of-bounds regions.
Returns:
A ``(y2 - y1, x2 - x1, C)`` array (pre-grayscale, ``dtype == ds.dtype``)
byte-identical to ``crop_frame(self._read_frame(frame_idx), crop,
fill)`` when pushdown is applicable; otherwise ``None`` to signal the
caller should fall back to a full-frame decode plus ``crop_frame``.
Never raises for out-of-bounds crops.
"""
if not self._can_push_crop:
return None
try:
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
ds = f[self.dataset]
return self._read_crop_from_ds(ds, frame_idx, crop, fill)
else:
with self._open_h5() as f:
ds = f[self.dataset]
return self._read_crop_from_ds(ds, frame_idx, crop, fill)
except (OSError, KeyError, IndexError): # pragma: no cover - defensive
return None
def read_crops(
self,
frame_inds: list,
crop: tuple[int, int, int, int],
fill: int | tuple[int, ...] = 0,
) -> np.ndarray | None:
"""Batched :meth:`read_crop`.
Args:
frame_inds: List of source-video frame indices to read.
crop: Crop region ``(x1, y1, x2, y2)`` with ``x2``/``y2`` exclusive.
fill: Fill value for out-of-bounds regions.
Returns:
A ``(N, y2 - y1, x2 - x1, C)`` array byte-identical to stacking
per-frame ``crop_frame`` results, or ``None`` to fall back to a
full-frame decode plus ``crop_frame``.
"""
if not self._can_push_crop:
return None
try:
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
ds = f[self.dataset]
return self._stack_crops(ds, frame_inds, crop, fill)
else:
with self._open_h5() as f:
ds = f[self.dataset]
return self._stack_crops(ds, frame_inds, crop, fill)
except (OSError, KeyError, IndexError): # pragma: no cover - defensive
return None
def _stack_crops(
self,
ds: h5py.Dataset,
frame_inds: list,
crop: tuple[int, int, int, int],
fill: int | tuple[int, ...],
) -> np.ndarray | None:
"""Stack per-frame crop reads, falling back to ``None`` if the gate declines.
The per-call gate in :meth:`_read_crop_from_ds` is frame-index independent, so
a batch is uniformly all-arrays or all-``None``; returning ``None`` on any
``None`` keeps batched reads byte-for-byte consistent with the scalar path
(the caller then decodes the full frames and crops them).
Args:
ds: The open ``h5py.Dataset`` (raw rank-4).
frame_inds: Frame indices to read.
crop: Crop region ``(x1, y1, x2, y2)``.
fill: Fill value for out-of-bounds regions.
Returns:
A ``(N, y2 - y1, x2 - x1, C)`` array, or ``None`` to signal fallback.
"""
parts = [self._read_crop_from_ds(ds, i, crop, fill) for i in frame_inds]
if any(p is None for p in parts):
return None
return np.stack(parts, axis=0)
def _read_crop_from_ds(
self,
ds: h5py.Dataset,
frame_idx: int,
crop: tuple[int, int, int, int],
fill: int | tuple[int, ...],
) -> np.ndarray | None:
"""Read and pad one frame's crop region from an open dataset.
Performs the per-call gate (crop must be smaller than the chunk span on at
least one spatial axis) and the clamp+pad hyperslab read. Axis ordering is
derived from ``ds.shape`` rather than assumed. Pushdown is structurally gated
off for frame-mapped/embedded datasets (see :attr:`_can_push_crop`), so
``frame_idx`` is always a raw source index here.
Args:
ds: The open ``h5py.Dataset`` (raw rank-4).
frame_idx: Frame index (raw source index; no ``frame_map`` remap needed).
crop: Crop region ``(x1, y1, x2, y2)``.
fill: Fill value for out-of-bounds regions.
Returns:
The ``(y2 - y1, x2 - x1, C)`` cropped/padded frame, or ``None`` if the
per-call gate decides a full read is at least as good.
"""
x1, y1, x2, y2 = crop
chunks = ds.chunks
if self.input_format == "channels_first":
# On-disk layout (F, C, W, H): x maps to axis 2 (W), y to axis 3 (H).
channels = ds.shape[1]
disk_w, disk_h = ds.shape[2], ds.shape[3]
width, height = disk_w, disk_h
chunk_w, chunk_h = chunks[2], chunks[3]
else:
# channels_last (F, H, W, C).
height, width, channels = ds.shape[1], ds.shape[2], ds.shape[3]
chunk_h, chunk_w = chunks[1], chunks[2]
# Per-call gate: only push down when the crop touches fewer spatial chunks
# than the full frame does on at least one axis. If the (in-bounds) crop
# already touches every chunk on both spatial axes, a hyperslab read buys
# nothing over a full read, so fall back.
crop_w, crop_h = x2 - x1, y2 - y1
in_sx1, in_sy1 = max(0, x1), max(0, y1)
in_sx2, in_sy2 = min(width, x2), min(height, y2)
if in_sx2 <= in_sx1 or in_sy2 <= in_sy1:
# Fully outside on at least one axis: no valid source pixels to read,
# so the hyperslab touches no chunks; pushdown is trivially beneficial.
n_chunks_w = n_chunks_h = 0
else:
n_chunks_w = (in_sx2 - 1) // chunk_w - in_sx1 // chunk_w + 1
n_chunks_h = (in_sy2 - 1) // chunk_h - in_sy1 // chunk_h + 1
frame_chunks_w = -(-width // chunk_w)
frame_chunks_h = -(-height // chunk_h)
if n_chunks_w >= frame_chunks_w and n_chunks_h >= frame_chunks_h:
return None
# frame_map is always empty here: _can_push_crop gates pushdown off for
# frame-mapped/embedded datasets, so frame_idx is a raw source index.
out = np.full((crop_h, crop_w, channels), fill, dtype=ds.dtype)
# Clamp the requested rect to the valid frame bounds.
sx1, sy1 = max(0, x1), max(0, y1)
sx2, sy2 = min(width, x2), min(height, y2)
if sx2 > sx1 and sy2 > sy1:
if self.input_format == "channels_first":
region = np.transpose(ds[frame_idx, :, sx1:sx2, sy1:sy2], (2, 1, 0))
else:
region = ds[frame_idx, sy1:sy2, sx1:sx2, :]
out[
sy1 - y1 : sy1 - y1 + (sy2 - sy1),
sx1 - x1 : sx1 - x1 + (sx2 - sx1),
] = region
return out
EXTS = ('h5', 'hdf5', 'slp')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__annotations__ = {'dataset': 'str | None', 'input_format': 'str', 'frame_map': 'dict[int, int]', '_can_push_crop_cached': 'bool | None', 'source_filename': 'str | None', 'source_inds': 'np.ndarray | None', 'image_format': 'str', 'channel_order': 'str', 'plugin': 'str | None', '_url_file': 'object | None', '_url_headers': 'dict[str, str] | None', '_url_stream_mode': 'str'}
class-attribute
¶
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
__attrs_own_setattr__ = True
class-attribute
¶
Returns True when the argument is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.
__attrs_props__ = ClassProps(is_exception=False, is_slotted=True, has_weakref_slot=True, is_frozen=False, kw_only=<KeywordOnly.NO: 'no'>, collected_fields_by_mro=True, added_init=True, added_repr=True, added_eq=True, added_ordering=False, hashability=<Hashability.UNHASHABLE: 'unhashable'>, added_match_args=True, added_str=False, added_pickling=False, on_setattr_hook=<function pipe.<locals>.wrapped_pipe at 0x7f08471ce840>, field_transformer=None)
class-attribute
¶
Effective class properties as derived from parameters to attr.s() or
define() decorators.
This is the same data structure that attrs uses internally to decide how to construct the final class.
Warning:
This feature is currently **experimental** and is not covered by our
strict backwards-compatibility guarantees.
Attributes:
| Name | Type | Description |
|---|---|---|
is_exception |
bool
|
Whether the class is treated as an exception class. |
is_slotted |
bool
|
Whether the class is |
has_weakref_slot |
bool
|
Whether the class has a slot for weak references. |
is_frozen |
bool
|
Whether the class is frozen. |
kw_only |
KeywordOnly
|
Whether / how the class enforces keyword-only arguments on the
|
collected_fields_by_mro |
bool
|
Whether the class fields were collected by method resolution order.
That is, correctly but unlike |
added_init |
bool
|
Whether the class has an attrs-generated |
added_repr |
bool
|
Whether the class has an attrs-generated |
added_eq |
bool
|
Whether the class has attrs-generated equality methods. |
added_ordering |
bool
|
Whether the class has attrs-generated ordering methods. |
hashability |
Hashability
|
How |
added_match_args |
bool
|
Whether the class supports positional |
added_str |
bool
|
Whether the class has an attrs-generated |
added_pickling |
bool
|
Whether the class has attrs-generated |
on_setattr_hook |
Callable[[Any, Attribute[Any], Any], Any] | None
|
The class's |
field_transformer |
Callable[[Attribute[Any]], Attribute[Any]] | None
|
The class's |
.. versionadded:: 25.4.0
__doc__ = 'Video backend for reading videos stored in HDF5 files.\n\nThis backend supports reading videos stored in HDF5 files, both in rank-4 datasets\nas well as in datasets with lists of binary-encoded images.\n\nEmbedded image datasets are used in SLEAP when exporting package files (`.pkg.slp`)\nwith videos embedded in them. This is useful for bundling training or inference data\nwithout having to worry about the videos (or frame images) being moved or deleted.\nIt is expected that these types of datasets will be in a `Group` with a `int8`\nvariable length dataset called `"video"`. This dataset must also contain an\nattribute called "format" with a string describing the image format (e.g., "png" or\n"jpg") which will be used to decode it appropriately.\n\nIf a `frame_numbers` dataset is present in the group, it will be used to map from\nsource video frames to the frames in the dataset. This is useful to preserve frame\nindexing when exporting a subset of frames in the video. It will also be used to\npopulate `frame_map` and `source_inds` attributes.\n\nAttributes:\n filename: Path to HDF5 file (.h5, .hdf5 or .slp).\n grayscale: Whether to force grayscale. If None, autodetect on first frame load.\n keep_open: Whether to keep the video reader open between calls to read frames.\n If False, will close the reader after each call. If True (the default), it\n will keep the reader open and cache it for subsequent calls which may\n enhance the performance of reading multiple frames.\n dataset: Name of dataset to read from. If `None`, will try to find a rank-4\n dataset by iterating through datasets in the file. If specifying an embedded\n dataset, this can be the group containing a "video" dataset or the dataset\n itself (e.g., "video0" or "video0/video").\n input_format: Format of the data in the dataset. One of "channels_last" (the\n default) in `(frames, height, width, channels)` order or "channels_first" in\n `(frames, channels, width, height)` order. Embedded datasets should use the\n "channels_last" format.\n frame_map: Mapping from frame indices to indices in the dataset. This is used to\n translate between the frame indices of the images within their source video\n and the indices of the images in the dataset. This is only used when reading\n embedded image datasets.\n source_filename: Path to the source video file. This is metadata and only used\n when reading embedded image datasets.\n source_inds: Indices of the frames in the source video file. This is metadata\n and only used when reading embedded image datasets.\n image_format: Format of the images in the embedded dataset. This is metadata and\n only used when reading embedded image datasets.\n channel_order: Channel order of embedded images, either "RGB" or "BGR". This is\n used to ensure consistent color channel ordering when decoding embedded\n images. If the encoding and decoding plugins have different channel orders,\n the channels will be automatically flipped during decoding.\n plugin: Plugin to use for decoding embedded images. One of "opencv" or\n "FFMPEG". If None, uses the global default or auto-detects based on\n available packages. Note that "pyav" is automatically mapped to "FFMPEG"\n since PyAV doesn\'t support image decoding.\n\nNotes:\n Concurrent reads of a single remote (URL-backed) `HDF5Video` from\n multiple threads are safe: although all reads share one cached fsspec\n file-like (a single byte position), h5py serializes every HDF5 C-library\n call under a global recursive lock (`h5py._objects.phil`), so the\n seek+read pair a frame read performs is never interleaved across threads.\n For true read *parallelism* (rather than just safety), construct\n independent `Video`/`HDF5Video` instances per worker; each gets its own\n fsspec file and block cache.\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 1126
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__match_args__ = ('filename', 'grayscale', 'keep_open', '_cached_shape', '_open_reader', '_fps', 'dataset', 'input_format', 'source_filename', 'source_inds', 'image_format', 'channel_order', 'plugin', '_url_headers', '_url_stream_mode')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__module__ = 'sleap_io.io.video_reading'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__slots__ = ('dataset', 'input_format', 'frame_map', '_can_push_crop_cached', 'source_filename', 'source_inds', 'image_format', 'channel_order', 'plugin', '_url_file', '_url_headers', '_url_stream_mode')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__static_attributes__ = ('_can_push_crop_cached', '_fps', '_open_reader', '_url_file', 'channel_order', 'dataset', 'frame_map', 'image_format', 'plugin', 'source_filename', 'source_inds')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
embedded_frame_inds
property
¶
Return the frame indices of the embedded images.
has_embedded_images
property
¶
Return True if the dataset contains embedded images.
img_shape
property
¶
Shape of a single frame in the video as (height, width, channels).
num_frames
property
¶
Number of frames in the video.
__attrs_post_init__()
¶
Auto-detect dataset and frame map heuristically.
Source code in sleap_io/io/video_reading.py
def __attrs_post_init__(self):
"""Auto-detect dataset and frame map heuristically."""
# Check if the file accessible before applying heuristics.
# For URLs, track whether this probe opened the cached fsspec file-like
# so it can be released afterwards (it would otherwise leak the handle on
# an early return / exception, and a probe-time open may predate the
# final auth headers being applied).
url_file_preexisting = self._url_file is not None
try:
f = self._open_h5()
except OSError:
self._release_probe_url_file(url_file_preexisting)
return
try:
if self.dataset is None:
# Iterate through datasets to find a rank 4 array.
def find_movies(name, obj):
if isinstance(obj, h5py.Dataset) and obj.ndim == 4:
self.dataset = name
return True
f.visititems(find_movies)
if self.dataset is None:
# Iterate through datasets to find an embedded video dataset.
def find_embedded(name, obj):
if isinstance(obj, h5py.Dataset) and name.endswith("/video"):
self.dataset = name
return True
f.visititems(find_embedded)
if self.dataset is None:
# Couldn't find video datasets.
return
if isinstance(f[self.dataset], h5py.Group):
# If this is a group, assume it's an embedded video dataset.
if "video" in f[self.dataset]:
self.dataset = f"{self.dataset}/video"
if self.dataset.split("/")[-1] == "video":
# This may be an embedded video dataset. Check for frame map.
ds = f[self.dataset]
if "format" in ds.attrs:
self.image_format = ds.attrs["format"]
# Read channel_order, with backwards compatibility
if "channel_order" in ds.attrs:
self.channel_order = ds.attrs["channel_order"]
else:
# Backwards compatibility: Check format_id for older files
# Prior to format 1.4, embedded images were primarily encoded
# with OpenCV which uses BGR, so default to BGR for older
# formats
if "metadata" in f and "format_id" in f["metadata"].attrs:
format_id = f["metadata"].attrs["format_id"]
if format_id < 1.4:
self.channel_order = "BGR" # Legacy default
# If no format_id found, assume BGR (safest legacy default)
# since most embedded images before this change used OpenCV
if "frame_numbers" in ds.parent:
frame_numbers = ds.parent["frame_numbers"][:].astype(int)
self.frame_map = {
frame: idx for idx, frame in enumerate(frame_numbers)
}
self.source_inds = frame_numbers
if "source_video" in ds.parent:
source_grp = ds.parent["source_video"]
# Source metadata is normally in the "json" attribute, but
# oversized metadata (e.g. an image-sequence source with many
# thousands of filenames, exceeding HDF5's 64 KB attribute limit)
# is stored in a "json" *dataset* instead (see
# ``slp._write_source_video_json``). Read whichever is present so
# such packages remain openable -- otherwise the backend fails to
# open, ``Video.backend`` is left ``None``, and embedded frames
# cannot be read.
if "json" in source_grp:
source_json = source_grp["json"][()]
else:
source_json = source_grp.attrs["json"]
self.source_filename = json.loads(source_json)["backend"][
"filename"
]
# Read FPS from attributes if present
if "fps" in ds.attrs:
self._fps = float(ds.attrs["fps"])
elif "fps" in ds.parent.attrs:
self._fps = float(ds.parent.attrs["fps"])
finally:
f.close()
self._release_probe_url_file(url_file_preexisting)
# Set default plugin if not specified (use image plugin, not video plugin)
if self.plugin is None:
# Check image plugin default first (for embedded images)
if _default_image_plugin is not None:
self.plugin = _default_image_plugin
# Otherwise auto-detect (for embedded image decoding)
elif "cv2" in sys.modules:
self.plugin = "opencv"
else:
self.plugin = "imageio" # imageio fallback
__eq__(other)
¶
Method generated by attrs for class HDF5Video.
Source code in sleap_io/io/video_reading.py
from sleap_io.io import _remote
from sleap_io.transform.frame import crop_frame
from sleap_io.transform.points import crop_points, uncrop_points
try:
import cv2
except ImportError:
pass
try:
import imageio_ffmpeg # noqa: F401
except ImportError:
pass
try:
import av # noqa: F401
except ImportError:
pass
__getstate__()
¶
Return state for pickling/deepcopy, dropping unpicklable handles.
Extends :meth:VideoBackend.__getstate__ to also drop the cached
fsspec-backed _url_file (reopened lazily by :meth:_open_h5).
Source code in sleap_io/io/video_reading.py
def __getstate__(self) -> dict:
"""Return state for pickling/deepcopy, dropping unpicklable handles.
Extends :meth:`VideoBackend.__getstate__` to also drop the cached
fsspec-backed ``_url_file`` (reopened lazily by :meth:`_open_h5`).
"""
state = super().__getstate__()
state["_url_file"] = None
return state
__init__(filename, grayscale=None, keep_open=True, cached_shape=None, open_reader=None, fps=None, dataset=None, input_format='channels_last', source_filename=None, source_inds=None, image_format='hdf5', channel_order='RGB', plugin=None, url_headers=None, url_stream_mode='blockcache')
¶
Method generated by attrs for class HDF5Video.
Source code in sleap_io/io/video_reading.py
# Track available backends (populated on module import)
_AVAILABLE_VIDEO_BACKENDS = {
"opencv": "cv2" in sys.modules,
"FFMPEG": "imageio_ffmpeg" in sys.modules,
"pyav": "av" in sys.modules,
}
_AVAILABLE_IMAGE_BACKENDS = {
"opencv": "cv2" in sys.modules,
"imageio": True, # Always available (core dependency)
}
# Global default video plugin
_default_video_plugin: str | None = None
def normalize_plugin_name(plugin: str) -> str:
"""Normalize plugin names to standard format.
Args:
plugin: Plugin name or alias (case-insensitive).
__repr__()
¶
Method generated by attrs for class HDF5Video.
__setattr__(name, val)
¶
Method generated by attrs for class HDF5Video.
Source code in sleap_io/io/video_reading.py
multiple threads are safe: although all reads share one cached fsspec
file-like (a single byte position), h5py serializes every HDF5 C-library
call under a global recursive lock (`h5py._objects.phil`), so the
seek+read pair a frame read performs is never interleaved across threads.
For true read *parallelism* (rather than just safety), construct
independent `Video`/`HDF5Video` instances per worker; each gets its own
fsspec file and block cache.
"""
close()
¶
Release the cached HDF5 reader and the cached fsspec URL file-like.
Extends :meth:VideoBackend.close (which drops the cached h5py.File
reader) by also closing the fsspec-backed _url_file shared across
reads, which h5py.File.close() does not close on its own. Both are
lazily reopened on the next read, so this is safe to call between reads.
Source code in sleap_io/io/video_reading.py
def close(self) -> None:
"""Release the cached HDF5 reader and the cached fsspec URL file-like.
Extends :meth:`VideoBackend.close` (which drops the cached ``h5py.File``
reader) by also closing the fsspec-backed ``_url_file`` shared across
reads, which ``h5py.File.close()`` does not close on its own. Both are
lazily reopened on the next read, so this is safe to call between reads.
"""
super().close()
self._close_url_file()
decode_embedded(img_string)
¶
Decode an embedded image string into a numpy array.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
img_string
|
ndarray
|
Binary string of the image as a |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
The decoded image as a numpy array of shape This method does not apply grayscale conversion as per the |
Source code in sleap_io/io/video_reading.py
def decode_embedded(self, img_string: np.ndarray) -> np.ndarray:
"""Decode an embedded image string into a numpy array.
Args:
img_string: Binary string of the image as a `int8` numpy vector with the
bytes as values corresponding to the format-encoded image.
Returns:
The decoded image as a numpy array of shape `(height, width, channels)`. If
a rank-2 image is decoded, it will be expanded such that channels will be 1.
This method does not apply grayscale conversion as per the `grayscale`
attribute. Use the `get_frame` or `get_frames` methods of the `VideoBackend`
to apply grayscale conversion rather than calling this function directly.
"""
# Decode based on plugin
if self.plugin == "opencv":
img = cv2.imdecode(img_string, cv2.IMREAD_UNCHANGED)
decoder_order = "BGR" # OpenCV decodes to BGR
else:
# Use imageio for FFMPEG or any other plugin
img = iio.imread(BytesIO(img_string), extension=f".{self.image_format}")
decoder_order = "RGB" # imageio decodes to RGB
if img.ndim == 2:
img = np.expand_dims(img, axis=-1)
# Convert channel order if needed
# If the stored order doesn't match the decoder order, flip channels
if img.shape[-1] == 3 and self.channel_order != decoder_order:
img = img[..., ::-1] # Flip RGB <-> BGR
return img
get_frame_raw_bytes(frame_idx)
¶
Get raw encoded bytes for a frame without decoding.
This method reads the raw compressed image data (PNG/JPEG bytes) directly from the HDF5 dataset without decoding it. This is useful for fast copying of embedded images when the target format matches the source format.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Index of the frame to read. |
required |
Returns:
| Type | Description |
|---|---|
ndarray | None
|
Raw encoded bytes as int8 numpy array, or None if: - The backend doesn't have embedded images (including "hdf5" format which stores raw numpy arrays, not encoded images) - The frame index is not available |
Notes
For variable-length datasets, returns the raw bytes directly. For fixed-length datasets, returns bytes with trailing zeros stripped.
Source code in sleap_io/io/video_reading.py
def get_frame_raw_bytes(self, frame_idx: int) -> np.ndarray | None:
"""Get raw encoded bytes for a frame without decoding.
This method reads the raw compressed image data (PNG/JPEG bytes) directly
from the HDF5 dataset without decoding it. This is useful for fast copying
of embedded images when the target format matches the source format.
Args:
frame_idx: Index of the frame to read.
Returns:
Raw encoded bytes as int8 numpy array, or None if:
- The backend doesn't have embedded images (including "hdf5" format which
stores raw numpy arrays, not encoded images)
- The frame index is not available
Notes:
For variable-length datasets, returns the raw bytes directly.
For fixed-length datasets, returns bytes with trailing zeros stripped.
"""
if not self.has_embedded_images:
return None
if not self.has_frame(frame_idx):
return None
# Get the internal index (handle frame_map)
internal_idx = (
self.frame_map.get(frame_idx, frame_idx) if self.frame_map else frame_idx
)
# Read directly from dataset
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
else:
f = self._open_h5()
ds = f[self.dataset]
raw_bytes = ds[internal_idx]
# Handle fixed-length padding (strip trailing zeros)
is_vlen = h5py.check_vlen_dtype(ds.dtype) is not None
if not is_vlen:
# Find last non-zero byte
non_zero_mask = raw_bytes != 0
if non_zero_mask.any():
last_non_zero = np.where(non_zero_mask)[0][-1]
raw_bytes = raw_bytes[: last_non_zero + 1]
if not self.keep_open:
f.close()
return raw_bytes
has_frame(frame_idx)
¶
Check if a frame index is contained in the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Index of frame to check. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Source code in sleap_io/io/video_reading.py
def has_frame(self, frame_idx: int) -> bool:
"""Check if a frame index is contained in the video.
Args:
frame_idx: Index of frame to check.
Returns:
`True` if the index is contained in the video, otherwise `False`.
"""
if self.frame_map:
return frame_idx in self.frame_map
else:
return frame_idx < len(self)
read_crop(frame_idx, crop, fill=0)
¶
Read a spatial hyperslab of a frame, padded to the crop shape.
This is the single-frame HDF5 crop pushdown hook consumed by
:class:CropVideoBackend. When applicable, it reads only the spatial
region of the frame that overlaps crop directly from the chunked
dataset (avoiding a full-frame decode) and pads out-of-bounds regions
exactly as :func:sleap_io.transform.frame.crop_frame would.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Index of the frame to read (source-video index; mapped
through |
required |
crop
|
tuple[int, int, int, int]
|
Crop region |
required |
fill
|
int | tuple[int, ...]
|
Fill value for out-of-bounds regions. |
0
|
Returns:
| Type | Description |
|---|---|
ndarray | None
|
A |
Source code in sleap_io/io/video_reading.py
def read_crop(
self,
frame_idx: int,
crop: tuple[int, int, int, int],
fill: int | tuple[int, ...] = 0,
) -> np.ndarray | None:
"""Read a spatial hyperslab of a frame, padded to the crop shape.
This is the single-frame HDF5 crop pushdown hook consumed by
:class:`CropVideoBackend`. When applicable, it reads only the spatial
region of the frame that overlaps ``crop`` directly from the chunked
dataset (avoiding a full-frame decode) and pads out-of-bounds regions
exactly as :func:`sleap_io.transform.frame.crop_frame` would.
Args:
frame_idx: Index of the frame to read (source-video index; mapped
through ``frame_map`` if present, though pushdown is gated off when
a ``frame_map`` exists).
crop: Crop region ``(x1, y1, x2, y2)`` with ``x2``/``y2`` exclusive.
May be negative or exceed the frame bounds (padded with ``fill``).
fill: Fill value for out-of-bounds regions.
Returns:
A ``(y2 - y1, x2 - x1, C)`` array (pre-grayscale, ``dtype == ds.dtype``)
byte-identical to ``crop_frame(self._read_frame(frame_idx), crop,
fill)`` when pushdown is applicable; otherwise ``None`` to signal the
caller should fall back to a full-frame decode plus ``crop_frame``.
Never raises for out-of-bounds crops.
"""
if not self._can_push_crop:
return None
try:
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
ds = f[self.dataset]
return self._read_crop_from_ds(ds, frame_idx, crop, fill)
else:
with self._open_h5() as f:
ds = f[self.dataset]
return self._read_crop_from_ds(ds, frame_idx, crop, fill)
except (OSError, KeyError, IndexError): # pragma: no cover - defensive
return None
read_crops(frame_inds, crop, fill=0)
¶
Batched :meth:read_crop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_inds
|
list
|
List of source-video frame indices to read. |
required |
crop
|
tuple[int, int, int, int]
|
Crop region |
required |
fill
|
int | tuple[int, ...]
|
Fill value for out-of-bounds regions. |
0
|
Returns:
| Type | Description |
|---|---|
ndarray | None
|
A |
Source code in sleap_io/io/video_reading.py
def read_crops(
self,
frame_inds: list,
crop: tuple[int, int, int, int],
fill: int | tuple[int, ...] = 0,
) -> np.ndarray | None:
"""Batched :meth:`read_crop`.
Args:
frame_inds: List of source-video frame indices to read.
crop: Crop region ``(x1, y1, x2, y2)`` with ``x2``/``y2`` exclusive.
fill: Fill value for out-of-bounds regions.
Returns:
A ``(N, y2 - y1, x2 - x1, C)`` array byte-identical to stacking
per-frame ``crop_frame`` results, or ``None`` to fall back to a
full-frame decode plus ``crop_frame``.
"""
if not self._can_push_crop:
return None
try:
if self.keep_open:
if self._open_reader is None:
self._open_reader = self._open_h5()
f = self._open_reader
ds = f[self.dataset]
return self._stack_crops(ds, frame_inds, crop, fill)
else:
with self._open_h5() as f:
ds = f[self.dataset]
return self._stack_crops(ds, frame_inds, crop, fill)
except (OSError, KeyError, IndexError): # pragma: no cover - defensive
return None
read_test_frame()
¶
Read a single frame from the video to test for grayscale.
ImageVideo
¶
Bases: sleap_io.io.video_reading.VideoBackend
Video backend for reading videos stored as image files.
This backend supports reading videos stored as a list of images.
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
Path to image files. |
|
grayscale |
Whether to force grayscale. If None, autodetect on first frame load. |
|
plugin |
Image plugin to use for reading. One of "opencv" or "imageio". If None, uses global default from get_default_image_plugin(), or auto-detects. |
Methods:
| Name | Description |
|---|---|
__eq__ |
Method generated by attrs for class ImageVideo. |
__init__ |
Method generated by attrs for class ImageVideo. |
__repr__ |
Method generated by attrs for class ImageVideo. |
__setattr__ |
Method generated by attrs for class ImageVideo. |
find_images |
Find images in a folder and return a list of filenames. |
get_frame_raw_bytes |
Return the raw encoded bytes of the source image file for a frame. |
Source code in sleap_io/io/video_reading.py
@attrs.define
class ImageVideo(VideoBackend):
"""Video backend for reading videos stored as image files.
This backend supports reading videos stored as a list of images.
Attributes:
filename: Path to image files.
grayscale: Whether to force grayscale. If None, autodetect on first frame load.
plugin: Image plugin to use for reading. One of "opencv" or "imageio".
If None, uses global default from get_default_image_plugin(), or
auto-detects.
"""
EXTS = ("png", "jpg", "jpeg", "tif", "tiff", "bmp")
plugin: str = attrs.field()
@plugin.validator
def _validate_plugin(self, attribute, value):
"""Validate and normalize plugin name."""
normalized = normalize_image_plugin_name(value)
object.__setattr__(self, attribute.name, normalized)
@plugin.default
def _default_plugin(self) -> str:
"""Get default plugin, checking global default first."""
# Check global default first
if _default_image_plugin is not None:
# Warn if preferred plugin not available
if not _AVAILABLE_IMAGE_BACKENDS.get(_default_image_plugin, False):
import warnings
available = get_available_image_backends()
install_cmd = get_installation_instructions(
_default_image_plugin, "image"
)
warnings.warn(
f"Preferred image plugin '{_default_image_plugin}' is not "
f"available. Available plugins: {available}\n"
f"Install with: {install_cmd}"
)
# Fall through to auto-detection
else:
return _default_image_plugin
# Otherwise auto-detect
if "cv2" in sys.modules:
return "opencv"
else:
return "imageio"
@staticmethod
def find_images(folder: str) -> list[str]:
"""Find images in a folder and return a list of filenames."""
folder = Path(folder)
return sorted(
[f.as_posix() for f in folder.glob("*") if f.suffix[1:] in ImageVideo.EXTS]
)
@property
def num_frames(self) -> int:
"""Number of frames in the video."""
return len(self.filename)
def get_frame_raw_bytes(self, frame_idx: int) -> np.ndarray | None:
"""Return the raw encoded bytes of the source image file for a frame.
Reads the on-disk image file verbatim (no decode/re-encode), enabling a
direct byte-for-byte embed of already-compressed sources. This avoids both
the cost of a decode/re-encode cycle and any additional compression
artifacts (important for lossy JPEG sources).
Only PNG/JPEG sources are supported here -- these are already entropy-coded
and are decodable by the embedded-image reader (`HDF5Video.decode_embedded`).
Other extensions (e.g. TIFF/BMP) return `None` so the caller falls back to
decoding and re-encoding to the requested format.
Args:
frame_idx: Index of the frame to read.
Returns:
The raw file bytes as an `int8` numpy vector, or `None` if the source
file is not a directly-storable compressed image (PNG/JPEG) or cannot
be read.
Notes:
Bytes copied this way decode back to RGB (matching `_read_frame`), so
the embedded dataset should record `channel_order="RGB"`.
"""
filename = self.filename[frame_idx]
ext = Path(filename).suffix.lower().lstrip(".")
if ext not in ("png", "jpg", "jpeg"):
return None
try:
with open(filename, "rb") as f:
data = f.read()
except OSError:
return None
return np.frombuffer(data, dtype="int8")
def _read_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the video.
Args:
frame_idx: Index of frame to read.
Returns:
The frame as a numpy array of shape `(height, width, channels)` in RGB
order.
Notes:
This does not apply grayscale conversion. It is recommended to use the
`get_frame` method of the `VideoBackend` class instead.
Images are always returned in RGB order regardless of plugin:
- imageio: Returns RGB natively
- opencv: Returns BGR, automatically flipped to RGB
"""
if self.plugin == "opencv":
# OpenCV reads as BGR, flip to RGB
img = cv2.imread(self.filename[frame_idx], cv2.IMREAD_UNCHANGED)
if img is None:
raise ValueError(f"Failed to read image: {self.filename[frame_idx]}")
if img.ndim == 3 and img.shape[-1] == 3:
img = img[..., ::-1] # BGR -> RGB
else: # imageio
# imageio reads as RGB natively
img = iio.imread(self.filename[frame_idx])
if img.ndim == 2:
img = np.expand_dims(img, axis=-1)
return img
EXTS = ('png', 'jpg', 'jpeg', 'tif', 'tiff', 'bmp')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__annotations__ = {'plugin': 'str'}
class-attribute
¶
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
__attrs_own_setattr__ = True
class-attribute
¶
Returns True when the argument is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.
__attrs_props__ = ClassProps(is_exception=False, is_slotted=True, has_weakref_slot=True, is_frozen=False, kw_only=<KeywordOnly.NO: 'no'>, collected_fields_by_mro=True, added_init=True, added_repr=True, added_eq=True, added_ordering=False, hashability=<Hashability.UNHASHABLE: 'unhashable'>, added_match_args=True, added_str=False, added_pickling=True, on_setattr_hook=<function pipe.<locals>.wrapped_pipe at 0x7f08471ce840>, field_transformer=None)
class-attribute
¶
Effective class properties as derived from parameters to attr.s() or
define() decorators.
This is the same data structure that attrs uses internally to decide how to construct the final class.
Warning:
This feature is currently **experimental** and is not covered by our
strict backwards-compatibility guarantees.
Attributes:
| Name | Type | Description |
|---|---|---|
is_exception |
bool
|
Whether the class is treated as an exception class. |
is_slotted |
bool
|
Whether the class is |
has_weakref_slot |
bool
|
Whether the class has a slot for weak references. |
is_frozen |
bool
|
Whether the class is frozen. |
kw_only |
KeywordOnly
|
Whether / how the class enforces keyword-only arguments on the
|
collected_fields_by_mro |
bool
|
Whether the class fields were collected by method resolution order.
That is, correctly but unlike |
added_init |
bool
|
Whether the class has an attrs-generated |
added_repr |
bool
|
Whether the class has an attrs-generated |
added_eq |
bool
|
Whether the class has attrs-generated equality methods. |
added_ordering |
bool
|
Whether the class has attrs-generated ordering methods. |
hashability |
Hashability
|
How |
added_match_args |
bool
|
Whether the class supports positional |
added_str |
bool
|
Whether the class has an attrs-generated |
added_pickling |
bool
|
Whether the class has attrs-generated |
on_setattr_hook |
Callable[[Any, Attribute[Any], Any], Any] | None
|
The class's |
field_transformer |
Callable[[Attribute[Any]], Attribute[Any]] | None
|
The class's |
.. versionadded:: 25.4.0
__doc__ = 'Video backend for reading videos stored as image files.\n\nThis backend supports reading videos stored as a list of images.\n\nAttributes:\n filename: Path to image files.\n grayscale: Whether to force grayscale. If None, autodetect on first frame load.\n plugin: Image plugin to use for reading. One of "opencv" or "imageio".\n If None, uses global default from get_default_image_plugin(), or\n auto-detects.\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 1871
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__match_args__ = ('filename', 'grayscale', 'keep_open', '_cached_shape', '_open_reader', '_fps', 'plugin')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__module__ = 'sleap_io.io.video_reading'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__slots__ = ('plugin',)
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__static_attributes__ = ()
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
num_frames
property
¶
Number of frames in the video.
__eq__(other)
¶
Method generated by attrs for class ImageVideo.
__init__(filename, grayscale=None, keep_open=True, cached_shape=None, open_reader=None, fps=None, plugin=NOTHING)
¶
Method generated by attrs for class ImageVideo.
__repr__()
¶
Method generated by attrs for class ImageVideo.
__setattr__(name, val)
¶
Method generated by attrs for class ImageVideo.
Source code in sleap_io/io/video_reading.py
multiple threads are safe: although all reads share one cached fsspec
file-like (a single byte position), h5py serializes every HDF5 C-library
call under a global recursive lock (`h5py._objects.phil`), so the
seek+read pair a frame read performs is never interleaved across threads.
For true read *parallelism* (rather than just safety), construct
independent `Video`/`HDF5Video` instances per worker; each gets its own
fsspec file and block cache.
"""
find_images(folder)
staticmethod
¶
Find images in a folder and return a list of filenames.
get_frame_raw_bytes(frame_idx)
¶
Return the raw encoded bytes of the source image file for a frame.
Reads the on-disk image file verbatim (no decode/re-encode), enabling a direct byte-for-byte embed of already-compressed sources. This avoids both the cost of a decode/re-encode cycle and any additional compression artifacts (important for lossy JPEG sources).
Only PNG/JPEG sources are supported here -- these are already entropy-coded
and are decodable by the embedded-image reader (HDF5Video.decode_embedded).
Other extensions (e.g. TIFF/BMP) return None so the caller falls back to
decoding and re-encoding to the requested format.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Index of the frame to read. |
required |
Returns:
| Type | Description |
|---|---|
ndarray | None
|
The raw file bytes as an |
Notes
Bytes copied this way decode back to RGB (matching _read_frame), so
the embedded dataset should record channel_order="RGB".
Source code in sleap_io/io/video_reading.py
def get_frame_raw_bytes(self, frame_idx: int) -> np.ndarray | None:
"""Return the raw encoded bytes of the source image file for a frame.
Reads the on-disk image file verbatim (no decode/re-encode), enabling a
direct byte-for-byte embed of already-compressed sources. This avoids both
the cost of a decode/re-encode cycle and any additional compression
artifacts (important for lossy JPEG sources).
Only PNG/JPEG sources are supported here -- these are already entropy-coded
and are decodable by the embedded-image reader (`HDF5Video.decode_embedded`).
Other extensions (e.g. TIFF/BMP) return `None` so the caller falls back to
decoding and re-encoding to the requested format.
Args:
frame_idx: Index of the frame to read.
Returns:
The raw file bytes as an `int8` numpy vector, or `None` if the source
file is not a directly-storable compressed image (PNG/JPEG) or cannot
be read.
Notes:
Bytes copied this way decode back to RGB (matching `_read_frame`), so
the embedded dataset should record `channel_order="RGB"`.
"""
filename = self.filename[frame_idx]
ext = Path(filename).suffix.lower().lstrip(".")
if ext not in ("png", "jpg", "jpeg"):
return None
try:
with open(filename, "rb") as f:
data = f.read()
except OSError:
return None
return np.frombuffer(data, dtype="int8")
MediaVideo
¶
Bases: sleap_io.io.video_reading.VideoBackend
Video backend for reading videos stored as common media files.
This backend supports reading through FFMPEG (the default), pyav, or OpenCV. Here are their trade-offs:
- "opencv": Fastest video reader, but only supports a limited number of codecs
and may not be able to read some videos. It requires `opencv-python` to be
installed. It is the fastest because it uses the OpenCV C++ library to read
videos, but is limited by the version of FFMPEG that was linked into it at
build time as well as the OpenCV version used.
- "FFMPEG": Slowest, but most reliable. This is the default backend. It requires
`imageio-ffmpeg` and a `ffmpeg` executable on the system path (which can be
installed via conda). The `imageio` plugin for FFMPEG reads frames into raw
bytes which are communicated to Python through STDOUT on a subprocess pipe,
which can be slow. However, it is the most reliable and feature-complete. If
you install the conda-forge version of ffmpeg, it will be compiled with
support for many codecs, including GPU-accelerated codecs like NVDEC for
H264 and others.
- "pyav": Supports most codecs that FFMPEG does, but not as complete or reliable
of an implementation in `imageio` as FFMPEG for some video types. It is
faster than FFMPEG because it uses the `av` package to read frames directly
into numpy arrays in memory without the need for a subprocess pipe. These
are Python bindings for the C library libav, which is the same library that
FFMPEG uses under the hood.
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
Path to video file. |
|
grayscale |
Whether to force grayscale. If None, autodetect on first frame load. |
|
keep_open |
Whether to keep the video reader open between calls to read frames. If False, will close the reader after each call. If True (the default), it will keep the reader open and cache it for subsequent calls which may enhance the performance of reading multiple frames. |
|
plugin |
Video plugin to use. One of "opencv", "FFMPEG", or "pyav". If |
Methods:
| Name | Description |
|---|---|
__eq__ |
Method generated by attrs for class MediaVideo. |
__init__ |
Method generated by attrs for class MediaVideo. |
__repr__ |
Method generated by attrs for class MediaVideo. |
__setattr__ |
Method generated by attrs for class MediaVideo. |
Source code in sleap_io/io/video_reading.py
@attrs.define
class MediaVideo(VideoBackend):
"""Video backend for reading videos stored as common media files.
This backend supports reading through FFMPEG (the default), pyav, or OpenCV. Here
are their trade-offs:
- "opencv": Fastest video reader, but only supports a limited number of codecs
and may not be able to read some videos. It requires `opencv-python` to be
installed. It is the fastest because it uses the OpenCV C++ library to read
videos, but is limited by the version of FFMPEG that was linked into it at
build time as well as the OpenCV version used.
- "FFMPEG": Slowest, but most reliable. This is the default backend. It requires
`imageio-ffmpeg` and a `ffmpeg` executable on the system path (which can be
installed via conda). The `imageio` plugin for FFMPEG reads frames into raw
bytes which are communicated to Python through STDOUT on a subprocess pipe,
which can be slow. However, it is the most reliable and feature-complete. If
you install the conda-forge version of ffmpeg, it will be compiled with
support for many codecs, including GPU-accelerated codecs like NVDEC for
H264 and others.
- "pyav": Supports most codecs that FFMPEG does, but not as complete or reliable
of an implementation in `imageio` as FFMPEG for some video types. It is
faster than FFMPEG because it uses the `av` package to read frames directly
into numpy arrays in memory without the need for a subprocess pipe. These
are Python bindings for the C library libav, which is the same library that
FFMPEG uses under the hood.
Attributes:
filename: Path to video file.
grayscale: Whether to force grayscale. If None, autodetect on first frame load.
keep_open: Whether to keep the video reader open between calls to read frames.
If False, will close the reader after each call. If True (the default), it
will keep the reader open and cache it for subsequent calls which may
enhance the performance of reading multiple frames.
plugin: Video plugin to use. One of "opencv", "FFMPEG", or "pyav". If `None`,
will use the first available plugin in the order listed above.
"""
plugin: str = attrs.field()
@plugin.validator
def _validate_plugin(self, attribute, value):
# Normalize the plugin name
normalized = normalize_plugin_name(value)
# Update the actual value to the normalized version
object.__setattr__(self, attribute.name, normalized)
EXTS = ("mp4", "avi", "mov", "mj2", "mkv")
@plugin.default
def _default_plugin(self) -> str:
# Check global default first
if _default_video_plugin is not None:
# Warn if preferred plugin not available
if not _AVAILABLE_VIDEO_BACKENDS.get(_default_video_plugin, False):
import warnings
available = get_available_video_backends()
install_cmd = get_installation_instructions(_default_video_plugin)
warnings.warn(
f"Preferred video plugin '{_default_video_plugin}' is not "
f"available. Available plugins: {available}\n"
f"Install with: {install_cmd}"
)
# Fall through to auto-detection
else:
return _default_video_plugin
# Auto-detect based on what's available
if "cv2" in sys.modules:
return "opencv"
elif "imageio_ffmpeg" in sys.modules:
return "FFMPEG"
elif "av" in sys.modules:
return "pyav"
else:
# Enhanced error message with installation instructions
raise ImportError(
"No video backend plugins are available.\n\n"
"The bundled imageio-ffmpeg should be available by default.\n"
"If you see this error, try reinstalling sleap-io:\n"
" pip install --force-reinstall sleap-io\n\n"
"Alternative backends:\n"
" opencv (fastest): pip install sleap-io[opencv]\n"
" pyav (balanced): pip install sleap-io[pyav]\n\n"
"For more information, see: https://io.sleap.ai"
)
@property
def reader(self) -> object:
"""Return the reader object for the video, caching if necessary."""
if self.keep_open:
if self._open_reader is None:
if self.plugin == "opencv":
self._open_reader = cv2.VideoCapture(self.filename)
elif self.plugin == "pyav" or self.plugin == "FFMPEG":
self._open_reader = iio.imopen(
self.filename, "r", plugin=self.plugin
)
return self._open_reader
else:
if self.plugin == "opencv":
return cv2.VideoCapture(self.filename)
elif self.plugin == "pyav" or self.plugin == "FFMPEG":
return iio.imopen(self.filename, "r", plugin=self.plugin)
@property
def num_frames(self) -> int:
"""Number of frames in the video."""
if self.plugin == "opencv":
return int(self.reader.get(cv2.CAP_PROP_FRAME_COUNT))
else:
props = iio.improps(self.filename, plugin=self.plugin)
n_frames = props.n_images
if np.isinf(n_frames):
legacy_reader = self.reader.legacy_get_reader()
# Note: This might be super slow for some videos, so maybe we should
# defer evaluation of this or give the user control over it.
n_frames = legacy_reader.count_frames()
return n_frames
@property
def fps(self) -> float | None:
"""Frames per second from video container metadata.
Returns:
The FPS from the video container, or None if it cannot be determined.
Notes:
This reads the FPS from the video file metadata using the appropriate
method for the current plugin:
- OpenCV: cv2.CAP_PROP_FPS
- FFMPEG/pyav: imageio metadata
For remote (URL) filenames the FPS is read directly from the pyav
container via ``av.open(url)``. imageio's v2 FFMPEG reader (used for
local files) requires the ``imageio-ffmpeg`` package and an ffmpeg
executable, which are not guaranteed in a pyav-only install, whereas
``av`` is already required for remote loading.
"""
# Return cached/explicit value if set
if self._fps is not None:
return self._fps
# Read from container metadata and cache the result so repeated access
# is O(1). This matters most for the remote (URL) path: ``av.open`` over
# http does not use Range requests, so each uncached read re-streams the
# entire video. Public helpers such as ``Video.frame_to_seconds`` read
# ``fps`` multiple times per call, which would otherwise re-download the
# whole video each time. Container fps is immutable for a given file, and
# the cache slot is cleared when the filename changes (see
# ``Video.replace_filename``), so caching is safe.
try:
if self.plugin == "opencv":
fps = self.reader.get(cv2.CAP_PROP_FPS)
rate = fps if fps > 0 else None
elif _remote._is_url(self.filename):
rate = _fps_from_av_container(self.filename)
else:
# Use imageio v2 API to get metadata (v3 improps doesn't include fps)
import imageio.v2 as iio_v2
reader = iio_v2.get_reader(self.filename, format="FFMPEG")
meta = reader.get_meta_data()
reader.close()
fps = meta.get("fps")
rate = float(fps) if fps is not None else None
except Exception:
return None
self._fps = rate
return rate
@fps.setter
def fps(self, value: float | None) -> None:
"""Set an explicit FPS override.
Args:
value: Frames per second. Must be positive if not None.
Raises:
ValueError: If value is not positive.
Notes:
Setting FPS on MediaVideo overrides the value from container metadata.
This can be useful when the container metadata is incorrect or missing.
"""
if value is not None and value <= 0:
raise ValueError(f"FPS must be positive, got {value}")
self._fps = value
def _read_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the video.
Args:
frame_idx: Index of frame to read.
Returns:
The frame as a numpy array of shape `(height, width, channels)`.
Notes:
This does not apply grayscale conversion. It is recommended to use the
`get_frame` method of the `VideoBackend` class instead.
"""
if self.plugin == "opencv":
if self.keep_open:
if self._open_reader is None:
self._open_reader = cv2.VideoCapture(self.filename)
reader = self._open_reader
else:
reader = cv2.VideoCapture(self.filename)
if reader.get(cv2.CAP_PROP_POS_FRAMES) != frame_idx:
reader.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
success, img = reader.read()
if success:
img = img[..., ::-1] # BGR -> RGB
elif self.plugin == "pyav" or self.plugin == "FFMPEG":
if self.keep_open:
img = self.reader.read(index=frame_idx)
else:
with iio.imopen(self.filename, "r", plugin=self.plugin) as reader:
img = reader.read(index=frame_idx)
success = img is not None
if not success:
raise IndexError(f"Failed to read frame index {frame_idx}.")
return img
def _read_frames(self, frame_inds: list) -> np.ndarray:
"""Read a list of frames from the video.
Args:
frame_inds: List of indices of frames to read.
Returns:
The frame as a numpy array of shape `(frames, height, width, channels)`.
Notes:
This does not apply grayscale conversion. It is recommended to use the
`get_frames` method of the `VideoBackend` class instead.
"""
if self.plugin == "opencv":
if self.keep_open:
if self._open_reader is None:
self._open_reader = cv2.VideoCapture(self.filename)
reader = self._open_reader
else:
reader = cv2.VideoCapture(self.filename)
reader.set(cv2.CAP_PROP_POS_FRAMES, frame_inds[0])
imgs = []
for idx in frame_inds:
if reader.get(cv2.CAP_PROP_POS_FRAMES) != idx:
reader.set(cv2.CAP_PROP_POS_FRAMES, idx)
_, img = reader.read()
imgs.append(img)
imgs = np.stack(imgs, axis=0)
imgs = imgs[..., ::-1] # BGR -> RGB
elif self.plugin == "pyav" or self.plugin == "FFMPEG":
if self.keep_open:
if self._open_reader is None:
self._open_reader = iio.imopen(
self.filename, "r", plugin=self.plugin
)
reader = self._open_reader
imgs = np.stack([reader.read(index=idx) for idx in frame_inds], axis=0)
else:
with iio.imopen(self.filename, "r", plugin=self.plugin) as reader:
imgs = np.stack(
[reader.read(index=idx) for idx in frame_inds], axis=0
)
return imgs
EXTS = ('mp4', 'avi', 'mov', 'mj2', 'mkv')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__annotations__ = {'plugin': 'str'}
class-attribute
¶
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
__attrs_own_setattr__ = True
class-attribute
¶
Returns True when the argument is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.
__attrs_props__ = ClassProps(is_exception=False, is_slotted=True, has_weakref_slot=True, is_frozen=False, kw_only=<KeywordOnly.NO: 'no'>, collected_fields_by_mro=True, added_init=True, added_repr=True, added_eq=True, added_ordering=False, hashability=<Hashability.UNHASHABLE: 'unhashable'>, added_match_args=True, added_str=False, added_pickling=True, on_setattr_hook=<function pipe.<locals>.wrapped_pipe at 0x7f08471ce840>, field_transformer=None)
class-attribute
¶
Effective class properties as derived from parameters to attr.s() or
define() decorators.
This is the same data structure that attrs uses internally to decide how to construct the final class.
Warning:
This feature is currently **experimental** and is not covered by our
strict backwards-compatibility guarantees.
Attributes:
| Name | Type | Description |
|---|---|---|
is_exception |
bool
|
Whether the class is treated as an exception class. |
is_slotted |
bool
|
Whether the class is |
has_weakref_slot |
bool
|
Whether the class has a slot for weak references. |
is_frozen |
bool
|
Whether the class is frozen. |
kw_only |
KeywordOnly
|
Whether / how the class enforces keyword-only arguments on the
|
collected_fields_by_mro |
bool
|
Whether the class fields were collected by method resolution order.
That is, correctly but unlike |
added_init |
bool
|
Whether the class has an attrs-generated |
added_repr |
bool
|
Whether the class has an attrs-generated |
added_eq |
bool
|
Whether the class has attrs-generated equality methods. |
added_ordering |
bool
|
Whether the class has attrs-generated ordering methods. |
hashability |
Hashability
|
How |
added_match_args |
bool
|
Whether the class supports positional |
added_str |
bool
|
Whether the class has an attrs-generated |
added_pickling |
bool
|
Whether the class has attrs-generated |
on_setattr_hook |
Callable[[Any, Attribute[Any], Any], Any] | None
|
The class's |
field_transformer |
Callable[[Attribute[Any]], Attribute[Any]] | None
|
The class's |
.. versionadded:: 25.4.0
__doc__ = 'Video backend for reading videos stored as common media files.\n\nThis backend supports reading through FFMPEG (the default), pyav, or OpenCV. Here\nare their trade-offs:\n\n - "opencv": Fastest video reader, but only supports a limited number of codecs\n and may not be able to read some videos. It requires `opencv-python` to be\n installed. It is the fastest because it uses the OpenCV C++ library to read\n videos, but is limited by the version of FFMPEG that was linked into it at\n build time as well as the OpenCV version used.\n - "FFMPEG": Slowest, but most reliable. This is the default backend. It requires\n `imageio-ffmpeg` and a `ffmpeg` executable on the system path (which can be\n installed via conda). The `imageio` plugin for FFMPEG reads frames into raw\n bytes which are communicated to Python through STDOUT on a subprocess pipe,\n which can be slow. However, it is the most reliable and feature-complete. If\n you install the conda-forge version of ffmpeg, it will be compiled with\n support for many codecs, including GPU-accelerated codecs like NVDEC for\n H264 and others.\n - "pyav": Supports most codecs that FFMPEG does, but not as complete or reliable\n of an implementation in `imageio` as FFMPEG for some video types. It is\n faster than FFMPEG because it uses the `av` package to read frames directly\n into numpy arrays in memory without the need for a subprocess pipe. These\n are Python bindings for the C library libav, which is the same library that\n FFMPEG uses under the hood.\n\nAttributes:\n filename: Path to video file.\n grayscale: Whether to force grayscale. If None, autodetect on first frame load.\n keep_open: Whether to keep the video reader open between calls to read frames.\n If False, will close the reader after each call. If True (the default), it\n will keep the reader open and cache it for subsequent calls which may\n enhance the performance of reading multiple frames.\n plugin: Video plugin to use. One of "opencv", "FFMPEG", or "pyav". If `None`,\n will use the first available plugin in the order listed above.\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 847
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__match_args__ = ('filename', 'grayscale', 'keep_open', '_cached_shape', '_open_reader', '_fps', 'plugin')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__module__ = 'sleap_io.io.video_reading'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__slots__ = ('plugin',)
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__static_attributes__ = ('_fps', '_open_reader')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
fps
property
¶
Frames per second from video container metadata.
Returns:
| Type | Description |
|---|---|
|
The FPS from the video container, or None if it cannot be determined. |
Notes
This reads the FPS from the video file metadata using the appropriate method for the current plugin: - OpenCV: cv2.CAP_PROP_FPS - FFMPEG/pyav: imageio metadata
For remote (URL) filenames the FPS is read directly from the pyav
container via av.open(url). imageio's v2 FFMPEG reader (used for
local files) requires the imageio-ffmpeg package and an ffmpeg
executable, which are not guaranteed in a pyav-only install, whereas
av is already required for remote loading.
num_frames
property
¶
Number of frames in the video.
reader
property
¶
Return the reader object for the video, caching if necessary.
__eq__(other)
¶
Method generated by attrs for class MediaVideo.
__init__(filename, grayscale=None, keep_open=True, cached_shape=None, open_reader=None, fps=None, plugin=NOTHING)
¶
Method generated by attrs for class MediaVideo.
__repr__()
¶
Method generated by attrs for class MediaVideo.
__setattr__(name, val)
¶
Method generated by attrs for class MediaVideo.
Source code in sleap_io/io/video_reading.py
multiple threads are safe: although all reads share one cached fsspec
file-like (a single byte position), h5py serializes every HDF5 C-library
call under a global recursive lock (`h5py._objects.phil`), so the
seek+read pair a frame read performs is never interleaved across threads.
For true read *parallelism* (rather than just safety), construct
independent `Video`/`HDF5Video` instances per worker; each gets its own
fsspec file and block cache.
"""
Video
¶
Video class used by sleap to represent videos and data associated with them.
This class is used to store information regarding a video and its components.
It is used to store the video's filename, shape, and the video's backend.
To create a Video object, use the from_filename method which will select the
backend appropriately.
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
The filename(s) of the video. Supported extensions: "mp4", "avi", "mov", "mj2", "mkv", "h5", "hdf5", "slp", "png", "jpg", "jpeg", "tif", "tiff", "bmp", "seq". If the filename is a list, a list of image filenames are expected. If filename is a folder, it will be searched for images. |
|
backend |
An object that implements the basic methods for reading and manipulating frames of a specific video type. |
|
backend_metadata |
A dictionary of metadata specific to the backend. This is useful for storing metadata that requires an open backend (e.g., shape information) without having access to the video file itself. |
|
source_video |
The source video object if this is a proxy video. This is present when the video contains an embedded subset of frames from another video. |
|
open_backend |
Whether to open the backend when the video is available. If |
|
_exists_cache |
Per-instance TTL cache for the result of |
Notes
Instances of this class are hashed by identity, not by value. This means that
two Video instances with the same attributes will NOT be considered equal in a
set or dict.
Media Video Plugin Support
For media files (mp4, avi, etc.), the following plugins are supported: - "opencv": Uses OpenCV (cv2) for video reading - "FFMPEG": Uses imageio-ffmpeg for video reading - "pyav": Uses PyAV for video reading
Plugin aliases (case-insensitive): - opencv: "opencv", "cv", "cv2", "ocv" - FFMPEG: "FFMPEG", "ffmpeg", "imageio-ffmpeg", "imageio_ffmpeg" - pyav: "pyav", "av"
Plugin selection priority: 1. Explicitly specified plugin parameter 2. Backend metadata plugin value 3. Global default (set via sio.set_default_video_plugin) 4. Auto-detection based on available packages
See Also
VideoBackend: The backend interface for reading video data. sleap_io.set_default_video_plugin: Set global default plugin. sleap_io.get_default_video_plugin: Get current default plugin.
Methods:
| Name | Description |
|---|---|
__attrs_post_init__ |
Post init syntactic sugar. |
__deepcopy__ |
Deep copy the video object. |
__getitem__ |
Return the frames of the video at the given indices. |
__init__ |
Method generated by attrs for class Video. |
__len__ |
Return the length of the video as the number of frames. |
__repr__ |
Informal string representation (for print or format). |
__str__ |
Informal string representation (for print or format). |
apply_crop |
Bake this video's virtual crop into a new physical video file. |
close |
Close the video backend. |
crop |
Return a virtual, on-read cropped view of this video. |
deduplicate_with |
Create a new video with duplicate images removed. |
exists |
Check if the video file exists and is accessible. |
frame_to_seconds |
Convert a frame index to timestamp in seconds. |
from_crop |
Open |
from_filename |
Create a Video from a filename. |
has_overlapping_images |
Check if this video has overlapping images with another video. |
matches_content |
Check if this video has the same content as another video. |
matches_path |
Check if this video has the same path as another video. |
matches_shape |
Check if this video has the same shape as another video. |
merge_with |
Merge another video's images into this one. |
open |
Open the video backend for reading. |
replace_filename |
Update the filename of the video, optionally opening the backend. |
save |
Save video frames to a new video file. |
seconds_to_frame |
Convert a timestamp in seconds to frame index. |
set_video_plugin |
Set the video plugin and reopen the video. |
to_crop_coords |
Map source-frame |
to_source_coords |
Map cropped-frame |
Source code in sleap_io/model/video.py
@attrs.define(eq=False)
class Video:
"""`Video` class used by sleap to represent videos and data associated with them.
This class is used to store information regarding a video and its components.
It is used to store the video's `filename`, `shape`, and the video's `backend`.
To create a `Video` object, use the `from_filename` method which will select the
backend appropriately.
Attributes:
filename: The filename(s) of the video. Supported extensions: "mp4", "avi",
"mov", "mj2", "mkv", "h5", "hdf5", "slp", "png", "jpg", "jpeg", "tif",
"tiff", "bmp", "seq". If the filename is a list, a list of image filenames
are expected. If filename is a folder, it will be searched for images.
backend: An object that implements the basic methods for reading and
manipulating frames of a specific video type.
backend_metadata: A dictionary of metadata specific to the backend. This is
useful for storing metadata that requires an open backend (e.g., shape
information) without having access to the video file itself.
source_video: The source video object if this is a proxy video. This is present
when the video contains an embedded subset of frames from another video.
open_backend: Whether to open the backend when the video is available. If `True`
(the default), the backend will be automatically opened if the video exists.
Set this to `False` when you want to manually open the backend, or when the
you know the video file does not exist and you want to avoid trying to open
the file.
_exists_cache: Per-instance TTL cache for the result of `exists()` when the
`filename` is a remote URL. Keyed by `(filename, dataset)` and storing
`(exists_bool, monotonic_timestamp)`. This avoids issuing a network probe
on every call (e.g. from the `is_open` property, which GUIs poll on each
render). The TTL defaults to 60 seconds and can be overridden via the
`SLEAP_IO_EXISTS_TTL` environment variable. The cache is cleared on
`replace_filename`.
Notes:
Instances of this class are hashed by identity, not by value. This means that
two `Video` instances with the same attributes will NOT be considered equal in a
set or dict.
Media Video Plugin Support:
For media files (mp4, avi, etc.), the following plugins are supported:
- "opencv": Uses OpenCV (cv2) for video reading
- "FFMPEG": Uses imageio-ffmpeg for video reading
- "pyav": Uses PyAV for video reading
Plugin aliases (case-insensitive):
- opencv: "opencv", "cv", "cv2", "ocv"
- FFMPEG: "FFMPEG", "ffmpeg", "imageio-ffmpeg", "imageio_ffmpeg"
- pyav: "pyav", "av"
Plugin selection priority:
1. Explicitly specified plugin parameter
2. Backend metadata plugin value
3. Global default (set via sio.set_default_video_plugin)
4. Auto-detection based on available packages
See Also:
VideoBackend: The backend interface for reading video data.
sleap_io.set_default_video_plugin: Set global default plugin.
sleap_io.get_default_video_plugin: Get current default plugin.
"""
filename: str | list[str]
backend: VideoBackend | None = None
backend_metadata: dict[str, any] = attrs.field(factory=dict)
source_video: "Video | None" = None
open_backend: bool = True
_exists_cache: dict[tuple[str, str | None], tuple[bool, float]] = attrs.field(
init=False, factory=dict, repr=False, eq=False
)
# URL auth context, threaded in by `make_video` for remote loads. Persisted
# on the Video (not just the backend) so existence probes and a later
# `open()` reconstruction stay authenticated after the backend is closed.
_url_headers: dict[str, str] | None = attrs.field(
init=False, default=None, repr=False, eq=False
)
_url_stream_mode: str = attrs.field(
init=False, default="blockcache", repr=False, eq=False
)
EXTS = MediaVideo.EXTS + HDF5Video.EXTS + ImageVideo.EXTS + ("seq",)
def _backend_url_headers(self) -> dict[str, str] | None:
"""Return the HTTP headers to authenticate remote existence probes.
Prefers the URL auth context stored on this `Video` (set by `make_video`
at load time); falls back to the live backend's headers when present.
Returns `None` for local files and unauthenticated URLs.
"""
if self._url_headers is not None:
return self._url_headers
if isinstance(self.backend, HDF5Video):
return getattr(self.backend, "_url_headers", None)
return None
@property
def original_video(self) -> "Video | None":
"""The root video in the provenance chain.
For embedded videos, this returns the ultimate source video by
traversing the source_video chain. Returns None if this video
has no source_video (i.e., it IS an original).
This property is computed by following the source_video chain to find
the root. For a single-level embedding (A embeds from B), original_video
returns B. For multi-level embedding (A <- B <- C), it returns C.
"""
if self.source_video is None:
return None # This IS the original
# Traverse to root
v = self.source_video
while v.source_video is not None:
v = v.source_video
return v
def __attrs_post_init__(self):
"""Post init syntactic sugar."""
if self.open_backend and self.backend is None and self.exists():
try:
self.open()
except Exception:
# If we can't open the backend, just ignore it for now so we don't
# prevent the user from building the Video object entirely.
pass
def __deepcopy__(self, memo):
"""Deep copy the video object."""
if id(self) in memo:
return memo[id(self)]
reopen = False
if self.is_open:
reopen = True
self.close()
new_video = Video(
filename=self.filename,
backend=None,
backend_metadata=self.backend_metadata.copy(),
source_video=self.source_video,
open_backend=self.open_backend,
)
memo[id(self)] = new_video
if reopen:
self.open()
return new_video
@classmethod
def from_filename(
cls,
filename: str | list[str],
dataset: str | None = None,
grayscale: bool | None = None,
keep_open: bool = True,
source_video: "Video | None" = None,
**kwargs,
) -> VideoBackend:
"""Create a Video from a filename.
Args:
filename: The filename(s) of the video. Supported extensions: "mp4", "avi",
"mov", "mj2", "mkv", "h5", "hdf5", "slp", "png", "jpg", "jpeg", "tif",
"tiff", "bmp". If the filename is a list, a list of image filenames are
expected. If filename is a folder, it will be searched for images.
dataset: Name of dataset in HDF5 file.
grayscale: Whether to force grayscale. If None, autodetect on first frame
load.
keep_open: Whether to keep the video reader open between calls to read
frames. If False, will close the reader after each call. If True (the
default), it will keep the reader open and cache it for subsequent calls
which may enhance the performance of reading multiple frames.
source_video: The source video object if this is a proxy video. This is
present when the video contains an embedded subset of frames from
another video.
**kwargs: Additional backend-specific arguments passed to
VideoBackend.from_filename. See VideoBackend.from_filename for supported
arguments.
Returns:
Video instance with the appropriate backend instantiated.
"""
backend = VideoBackend.from_filename(
filename,
dataset=dataset,
grayscale=grayscale,
keep_open=keep_open,
**kwargs,
)
# If filename is a directory, VideoBackend.from_filename will expand it
# to a list of paths to images contained within the directory. In this
# case we want to use the expanded list as filename
return cls(
filename=backend.filename,
backend=backend,
source_video=source_video,
)
def crop(
self,
crop: tuple[int, int, int, int] | None = None,
*,
bbox: tuple[float, float, float, float] | None = None,
roi: object | None = None,
center: tuple[float, float] | None = None,
size: tuple[int, int] | None = None,
margin: int = 0,
fill: int | tuple[int, ...] = 0,
share_decode: bool = True,
) -> "Video":
"""Return a virtual, on-read cropped view of this video.
Exactly one region spec must be given: ``crop`` (explicit
``(x1, y1, x2, y2)`` rect), ``bbox``, ``roi`` (its axis-aligned bounds +
``margin``), or (``center``, ``size``) for a fixed-size centered/
centroid-following window. The returned ``Video`` shares no pixels with
this one; frames are decoded on read and cropped (byte-identical to
:func:`sleap_io.transform.frame.crop_frame`). Out-of-bounds regions are
pad-filled with ``fill`` (never clamped), so the output shape is always
exactly ``(y2 - y1, x2 - x1)``.
The crop composes (FLATTENS when fills agree and the region is in-bounds)
with any existing crop on this video via
:meth:`CropVideoBackend.wrap`. ``source_video`` is set to this video for
provenance. When ``share_decode`` (the default), the new crop reuses this
video's backend instance as the shared inner so a mosaic of tiles over
one file decodes each source frame once; in that case the new tile does
NOT own the shared decoder (this video does).
Args:
crop: Explicit crop region ``(x1, y1, x2, y2)``, ``x2``/``y2``
exclusive.
bbox: A bounding box ``(x1, y1, x2, y2)``; bounds may be float.
roi: Any object exposing axis-aligned ``.bounds`` as
``(minx, miny, maxx, maxy)`` (e.g. a shapely geometry).
center: Window center ``(cx, cy)`` (used with ``size``).
size: Fixed output ``(width, height)`` (used with ``center``).
margin: Pixels added around the ``roi`` bounds on every side.
fill: Fill value for out-of-bounds regions.
share_decode: If ``True`` (the default), reuse this video's backend
as the shared inner so tiles decode each frame once; the new tile
does not own the shared decoder.
Returns:
A new ``Video`` exposing the cropped view.
"""
from sleap_io.io.video_reading import CropVideoBackend
rect = _resolve_crop_rect(crop, bbox, roi, center, size, margin)
if self.backend is None and self.open_backend:
self.open()
if self.backend is None:
raise ValueError(
"Cannot crop a video with no open backend. Open it first (set "
"open_backend=True or call .open()) before cropping."
)
inner = self.backend
cropped_backend = CropVideoBackend.wrap(
inner=inner, crop=rect, fill=fill, owns_inner=not share_decode
)
cropped = Video(
filename=self.filename,
backend=cropped_backend,
source_video=self,
open_backend=self.open_backend,
)
x1, y1, x2, y2 = cropped_backend.crop
src_shape = self.shape
cropped.backend_metadata = {
**self.backend_metadata,
"shape": (src_shape[0], y2 - y1, x2 - x1, src_shape[3])
if src_shape is not None
else None,
# The uncropped source shape, so a closed re-serialize keeps videos_json
# describing the full frame even without a live source_video (D-120/DI-2).
"source_shape": list(src_shape) if src_shape is not None else None,
# COMPOSED source rect from wrap (D-120): keeps open/closed crop keys
# identical and root-canonical, and survives close()->open().
"crop": list(cropped_backend.crop),
"crop_fill": cropped_backend.fill,
}
return cropped
@classmethod
def from_crop(
cls,
video: "str | Path | Video",
crop: tuple[int, int, int, int] | None = None,
*,
bbox: tuple[float, float, float, float] | None = None,
roi: object | None = None,
center: tuple[float, float] | None = None,
size: tuple[int, int] | None = None,
margin: int = 0,
fill: int | tuple[int, ...] = 0,
share_decode: bool = True,
**kwargs,
) -> "Video":
"""Open ``video`` (path or ``Video``) and return a virtual crop.
Accepts the same region specs as :meth:`crop` (``crop``/``bbox``/``roi``/
``center``+``size``); extra keyword arguments are forwarded to
:meth:`from_filename` when ``video`` is a path (ignored when it is already
a ``Video``).
Args:
video: A path/filename to open, or an existing ``Video`` to crop.
crop: Explicit crop region ``(x1, y1, x2, y2)``, ``x2``/``y2`` exclusive.
bbox: A bounding box ``(x1, y1, x2, y2)``; bounds may be float.
roi: An object exposing axis-aligned ``.bounds`` (e.g. a shapely
geometry); ``margin`` is applied around it.
center: Window center ``(cx, cy)`` (with ``size``).
size: Fixed output ``(width, height)`` (with ``center``).
margin: Pixels added around the ``roi`` bounds on every side.
fill: Fill value for out-of-bounds regions.
share_decode: If ``True`` (default), reuse the source decoder.
**kwargs: Forwarded to :meth:`from_filename` for a path input.
Returns:
A new ``Video`` exposing the cropped view.
"""
if isinstance(video, (str, Path)):
video = cls.from_filename(video, **kwargs)
return video.crop(
crop,
bbox=bbox,
roi=roi,
center=center,
size=size,
margin=margin,
fill=fill,
share_decode=share_decode,
)
def _crop_tuple(self) -> tuple[int, int, int, int] | None:
"""Return this video's crop rect ``(x1, y1, x2, y2)`` or ``None``.
Reads ``backend.crop`` when the backend is a ``CropVideoBackend`` (open
path), else ``backend_metadata["crop"]`` (closed path), else ``None``
(uncropped).
"""
from sleap_io.io.video_reading import CropVideoBackend
if isinstance(self.backend, CropVideoBackend):
return tuple(self.backend.crop)
crop = self.backend_metadata.get("crop")
return tuple(crop) if crop is not None else None
def _crop_fill(self) -> int | tuple[int, ...]:
"""Return this video's crop fill value (open: backend; closed: metadata).
Returns ``0`` for an uncropped video. Mirrors :meth:`_crop_tuple`.
"""
from sleap_io.io.video_reading import CropVideoBackend
if isinstance(self.backend, CropVideoBackend):
return self.backend.fill
return self.backend_metadata.get("crop_fill", 0)
@property
def is_cropped(self) -> bool:
"""Whether this video is a virtual crop of another video."""
return self._crop_tuple() is not None
@property
def crop_rect(self) -> tuple[int, int, int, int] | None:
"""Crop rect ``(x1, y1, x2, y2)`` in source coords, or ``None`` if uncropped."""
return self._crop_tuple()
@property
def crop_fill(self) -> int | tuple[int, ...]:
"""The out-of-bounds fill value for this video's crop (``0`` if uncropped)."""
return self._crop_fill()
def to_crop_coords(self, points: np.ndarray) -> np.ndarray:
"""Map source-frame ``(x, y)`` into this video's cropped frame.
Args:
points: Coordinate array of shape ``(..., 2)``. NaN values are
preserved.
Returns:
Coordinates translated into the cropped frame. If this video is not
cropped, a copy of ``points`` is returned unchanged.
"""
crop = self._crop_tuple()
return points.copy() if crop is None else crop_points(points, crop)
def to_source_coords(self, points: np.ndarray) -> np.ndarray:
"""Map cropped-frame ``(x, y)`` back to source-frame coordinates.
Inverse of :meth:`to_crop_coords`.
Args:
points: Coordinate array of shape ``(..., 2)``. NaN values are
preserved.
Returns:
Coordinates translated back to source coordinates. If this video is
not cropped, a copy of ``points`` is returned unchanged.
"""
crop = self._crop_tuple()
return points.copy() if crop is None else uncrop_points(points, crop)
@property
def shape(self) -> tuple[int, int, int, int] | None:
"""Return the shape of the video as (num_frames, height, width, channels).
If the video backend is not set or it cannot determine the shape of the video,
this will return None.
"""
return self._get_shape()
def _get_shape(self) -> tuple[int, int, int, int] | None:
"""Return the shape of the video as (num_frames, height, width, channels).
This suppresses errors related to querying the backend for the video shape, such
as when it has not been set or when the video file is not found.
"""
try:
return self.backend.shape
except Exception:
if "shape" in self.backend_metadata:
return self.backend_metadata["shape"]
return None
@property
def grayscale(self) -> bool | None:
"""Return whether the video is grayscale.
If the video backend is not set or it cannot determine whether the video is
grayscale, this will return None.
"""
shape = self.shape
if shape is not None:
return shape[-1] == 1
else:
grayscale = None
if "grayscale" in self.backend_metadata:
grayscale = self.backend_metadata["grayscale"]
return grayscale
@grayscale.setter
def grayscale(self, value: bool):
"""Set the grayscale value and adjust the backend."""
if self.backend is not None:
self.backend.grayscale = value
self.backend._cached_shape = None
self.backend_metadata["grayscale"] = value
@property
def fps(self) -> float | None:
"""Return the frames per second of the video.
For MediaVideo backends, this reads FPS from the video container metadata.
For other backends (ImageVideo, HDF5Video, TiffVideo), this returns the
explicitly set value or None if not set.
Returns:
The FPS if known, or None if unavailable/unknown.
"""
if self.backend is not None:
return self.backend.fps
return self.backend_metadata.get("fps")
@fps.setter
def fps(self, value: float | None):
"""Set the frames per second.
Args:
value: Frames per second. Must be positive if not None.
Raises:
ValueError: If value is not positive.
Notes:
For MediaVideo backends, setting FPS overrides the value from container
metadata. For other backends, this sets the FPS directly.
"""
if value is not None and value <= 0:
raise ValueError(f"FPS must be positive, got {value}")
if self.backend is not None:
self.backend.fps = value
self.backend_metadata["fps"] = value
def frame_to_seconds(self, frame_idx: int) -> float | None:
"""Convert a frame index to timestamp in seconds.
Args:
frame_idx: Zero-indexed frame number.
Returns:
Time in seconds, or None if FPS is unknown.
Notes:
This assumes constant frame rate. For variable frame rate videos,
the returned timestamp may be approximate.
"""
if self.fps is None or self.fps <= 0:
return None
return frame_idx / self.fps
def seconds_to_frame(self, seconds: float) -> int | None:
"""Convert a timestamp in seconds to frame index.
Args:
seconds: Time in seconds from video start.
Returns:
Zero-indexed frame number (rounded down), or None if FPS unknown.
"""
if self.fps is None or self.fps <= 0:
return None
return int(seconds * self.fps)
def __len__(self) -> int:
"""Return the length of the video as the number of frames."""
shape = self.shape
return 0 if shape is None else shape[0]
def __repr__(self) -> str:
"""Informal string representation (for print or format)."""
dataset = (
f"dataset={self.backend.dataset}, "
if getattr(self.backend, "dataset", "")
else ""
)
return (
"Video("
f'filename="{self.filename}", '
f"shape={self.shape}, "
f"{dataset}"
f"backend={type(self.backend).__name__}"
")"
)
def __str__(self) -> str:
"""Informal string representation (for print or format)."""
return self.__repr__()
def __getitem__(self, inds: int | list[int] | slice) -> np.ndarray:
"""Return the frames of the video at the given indices.
Args:
inds: Index or list of indices of frames to read.
Returns:
Frame or frames as a numpy array of shape `(height, width, channels)` if a
scalar index is provided, or `(frames, height, width, channels)` if a list
of indices is provided.
See also: VideoBackend.get_frame, VideoBackend.get_frames
"""
if not self.is_open:
if self.open_backend:
self.open()
else:
raise ValueError(
"Video backend is not open. Call video.open() or set "
"video.open_backend to True to do automatically on frame read."
)
return self.backend[inds]
def exists(self, check_all: bool = False, dataset: str | None = None) -> bool:
"""Check if the video file exists and is accessible.
Args:
check_all: If `True`, check that all filenames in a list exist. If `False`
(the default), check that the first filename exists.
dataset: Name of dataset in HDF5 file. If specified, this will function will
return `False` if the dataset does not exist.
Returns:
`True` if the file exists and is accessible, `False` otherwise.
"""
if isinstance(self.filename, list):
if check_all:
for f in self.filename:
if not is_file_accessible(f):
return False
return True
else:
return is_file_accessible(self.filename[0])
# URL fast path: must run BEFORE `is_file_accessible`, which treats the
# filename as a local path and would spuriously return False for a URL.
from sleap_io.io._remote import _is_url
if _is_url(self.filename):
return self._url_exists(dataset)
file_is_accessible = is_file_accessible(self.filename)
if not file_is_accessible:
# Check if it's a directory (ImageVideo source)
if Path(self.filename).is_dir():
return True
return False
if dataset is None or dataset == "":
dataset = self.backend_metadata.get("dataset", None)
if dataset is not None and dataset != "":
has_dataset = False
if (
self.backend is not None
and type(self.backend) is HDF5Video
and self.backend._open_reader is not None
):
has_dataset = dataset in self.backend._open_reader
else:
with h5py.File(self.filename, "r") as f:
has_dataset = dataset in f
return has_dataset
return True
def _url_exists(self, dataset: str | None) -> bool:
"""Check whether a remote URL `filename` exists, with a TTL cache.
Args:
dataset: Name of dataset in the (remote) HDF5 file. If specified (or
derivable from `backend_metadata`), existence additionally requires
that the dataset be present in the file.
Returns:
`True` if the URL is reachable (and, if a dataset was requested, the
dataset exists), `False` otherwise.
Notes:
Results are cached per instance keyed by `(filename, dataset)` for a
TTL (default 60s, overridable via the `SLEAP_IO_EXISTS_TTL` env var) so
repeated calls (e.g. from the `is_open` property in a GUI render loop)
do not issue a network probe each time.
"""
from sleap_io.io._remote import _head_or_range_probe
key = (self.filename, dataset)
try:
ttl = float(os.environ.get("SLEAP_IO_EXISTS_TTL", "60"))
except ValueError:
# A malformed env value must not break the never-raise bool
# contract of exists()/is_open; fall back to the 60s default.
ttl = 60.0
cached = self._exists_cache.get(key)
if cached is not None and (time.monotonic() - cached[1]) < ttl:
return cached[0]
try:
if not _head_or_range_probe(
self.filename, headers=self._backend_url_headers()
):
result = False
else:
if dataset is None or dataset == "":
dataset = self.backend_metadata.get("dataset", None)
if dataset is None or dataset == "":
result = True
else:
result = self._url_dataset_exists(dataset)
except Exception:
result = False
self._exists_cache[key] = (result, time.monotonic())
return result
def _url_dataset_exists(self, dataset: str) -> bool:
"""Check whether `dataset` is present in the remote HDF5 file.
Reuses the backend's already-open HDF5 reader when available; otherwise
opens the remote file via fsspec for a single membership check.
Args:
dataset: Name of dataset in the remote HDF5 file.
Returns:
`True` if the dataset is present, `False` otherwise.
"""
if (
self.backend is not None
and type(self.backend) is HDF5Video
and self.backend._open_reader is not None
):
return dataset in self.backend._open_reader
from sleap_io.io._remote import open_remote_h5
url_file = open_remote_h5(self.filename, headers=self._backend_url_headers())
try:
with h5py.File(url_file, "r") as f:
return dataset in f
finally:
url_file.close()
@property
def is_open(self) -> bool:
"""Check if the video backend is open."""
return self.exists() and self.backend is not None
def open(
self,
filename: str | None = None,
dataset: str | None = None,
grayscale: str | None = None,
keep_open: bool = True,
plugin: str | None = None,
):
"""Open the video backend for reading.
Args:
filename: Filename to open. If not specified, will use the filename set on
the video object.
dataset: Name of dataset in HDF5 file.
grayscale: Whether to force grayscale. If None, autodetect on first frame
load.
keep_open: Whether to keep the video reader open between calls to read
frames. If False, will close the reader after each call. If True (the
default), it will keep the reader open and cache it for subsequent calls
which may enhance the performance of reading multiple frames.
plugin: Video plugin to use for MediaVideo files. One of "opencv",
"FFMPEG", or "pyav". Also accepts aliases (case-insensitive).
If not specified, uses the backend metadata, global default,
or auto-detection in that order.
Notes:
This is useful for opening the video backend to read frames and then closing
it after reading all the necessary frames.
If the backend was already open, it will be closed before opening a new one.
Values for the HDF5 dataset and grayscale will be remembered if not
specified.
"""
if filename is not None:
self.replace_filename(filename, open=False)
# Try to remember values from previous backend if available and not specified.
if self.backend is not None:
if dataset is None:
dataset = getattr(self.backend, "dataset", None)
if grayscale is None:
grayscale = getattr(self.backend, "grayscale", None)
else:
if dataset is None and "dataset" in self.backend_metadata:
dataset = self.backend_metadata["dataset"]
if grayscale is None:
if "grayscale" in self.backend_metadata:
grayscale = self.backend_metadata["grayscale"]
elif "shape" in self.backend_metadata:
grayscale = self.backend_metadata["shape"][-1] == 1
if not self.exists(dataset=dataset):
from sleap_io.io._remote import _is_url, _redact_url
# Redact credential-bearing URLs (e.g. presigned ``?token=`` links)
# so they never surface in tracebacks/logs. Local paths are shown
# verbatim.
name = (
_redact_url(self.filename)
if isinstance(self.filename, str) and _is_url(self.filename)
else self.filename
)
msg = f"Video does not exist or cannot be opened for reading: {name}"
if dataset is not None:
msg += f" (dataset: {dataset})"
raise FileNotFoundError(msg)
# Close previous backend if open.
self.close()
# Handle plugin parameter
backend_kwargs = {}
if plugin is not None:
from sleap_io.io.video_reading import normalize_plugin_name
plugin = normalize_plugin_name(plugin)
self.backend_metadata["plugin"] = plugin
if "plugin" in self.backend_metadata:
backend_kwargs["plugin"] = self.backend_metadata["plugin"]
# Create new backend. Forward the URL auth context so a reopened remote
# HDF5Video stays authenticated (the previous backend, and its headers,
# were dropped by self.close() above).
self.backend = VideoBackend.from_filename(
self.filename,
dataset=dataset,
grayscale=grayscale,
keep_open=keep_open,
url_headers=self._url_headers,
url_stream_mode=self._url_stream_mode,
**backend_kwargs,
)
# Re-wrap as a crop view if this video records a crop in its metadata.
# The rebuilt backend above is always a plain backend, so this wraps
# exactly once (idempotent across close()->open() and deepcopy).
if "crop" in self.backend_metadata:
from sleap_io.io.video_reading import CropVideoBackend
self.backend = CropVideoBackend.wrap(
inner=self.backend,
crop=tuple(self.backend_metadata["crop"]),
fill=self.backend_metadata.get("crop_fill", 0),
)
def close(self):
"""Close the video backend."""
if self.backend is not None:
# Try to remember values from previous backend if available and not
# specified.
try:
self.backend_metadata["dataset"] = getattr(
self.backend, "dataset", None
)
self.backend_metadata["grayscale"] = getattr(
self.backend, "grayscale", None
)
self.backend_metadata["shape"] = getattr(self.backend, "shape", None)
self.backend_metadata["fps"] = getattr(self.backend, "fps", None)
# Persist the crop so a Video cropped in-memory (never loaded
# from disk) survives a close()->open() and deepcopy: open()
# re-wraps from these keys (the closed-path shape above is
# already the cropped shape).
from sleap_io.io.video_reading import CropVideoBackend
if isinstance(self.backend, CropVideoBackend):
self.backend_metadata["crop"] = list(self.backend.crop)
self.backend_metadata["crop_fill"] = self.backend.fill
except Exception:
pass
# Deterministically release the backend's open handles (the cached
# reader and, for a remote HDF5Video, the fsspec URL file-like)
# rather than relying on garbage collection.
try:
self.backend.close()
except Exception:
pass
del self.backend
self.backend = None
def replace_filename(
self, new_filename: str | Path | list[str] | list[Path], open: bool = True
):
"""Update the filename of the video, optionally opening the backend.
Args:
new_filename: New filename to set for the video.
open: If `True` (the default), open the backend with the new filename. If
the new filename does not exist, no error is raised.
"""
if isinstance(new_filename, Path):
new_filename = new_filename.as_posix()
if isinstance(new_filename, list):
new_filename = [
p.as_posix() if isinstance(p, Path) else p for p in new_filename
]
# A relink to a different file makes the recorded shape/grayscale/fps in
# ``backend_metadata`` stale: they describe the OLD file but the new file
# may have a different resolution/channels/frame rate. They must not be
# serialized under the new filename (regression from #483, where
# ``save_slp(prefer_metadata=True)`` prefers these recorded values), so
# invalidate them on a real relink and let them be recomputed from the new
# backend. The no-relink path leaves metadata untouched so golden
# byte-identical saves stay byte-identical.
filename_changed = new_filename != self.filename
self.filename = new_filename
self.backend_metadata["filename"] = new_filename
# Invalidate any cached URL existence results for the previous filename.
self._exists_cache.clear()
if open:
if self.exists():
self.open()
else:
self.close()
# Drop stale metadata AFTER (re)opening: ``open()`` internally calls
# ``close()``, which would otherwise re-stamp the OLD backend's
# shape/grayscale/fps back into ``backend_metadata``.
if filename_changed:
for key in ("shape", "grayscale", "fps"):
self.backend_metadata.pop(key, None)
def matches_path(self, other: "Video", strict: bool = False) -> bool:
"""Check if this video has the same path as another video.
Args:
other: Another video to compare with.
strict: If True, require exact path match. If False, consider videos
with the same filename (basename) as matching.
Returns:
True if the videos have matching paths, False otherwise.
Notes:
For HDF5 video backends (e.g., embedded videos in .pkg.slp files),
matching prioritizes the source_filename attribute since multiple
videos can share the same HDF5 file path but reference different
source videos. Falls back to dataset name matching if source_filename
is not available.
"""
# Handle HDF5 backends specially - prioritize source_filename matching
self_is_hdf5 = isinstance(self.backend, HDF5Video)
other_is_hdf5 = isinstance(other.backend, HDF5Video)
if self_is_hdf5 and other_is_hdf5:
# Both are HDF5 videos - must match by BOTH source_filename AND dataset
# to distinguish different videos embedded in the same pkg.slp file
self_source = self.backend.source_filename
other_source = other.backend.source_filename
self_dataset = self.backend.dataset
other_dataset = other.backend.dataset
# If both have datasets, they must match
if self_dataset is not None and other_dataset is not None:
if self_dataset != other_dataset:
return False # Different datasets = different videos
# If both have source_filenames, compare them
if self_source is not None and other_source is not None:
if strict:
# For HDF5 videos, just compare normalized path strings
# (avoid slow resolve() on network paths)
return Path(self_source).as_posix() == Path(other_source).as_posix()
else:
return Path(self_source).name == Path(other_source).name
# If only datasets available (no source_filename), they must match
if self_dataset is not None and other_dataset is not None:
return self_dataset == other_dataset
# If neither source_filename nor dataset available, cannot match
return False
if isinstance(self.filename, list) and isinstance(other.filename, list):
# Both are image sequences
if strict:
return self.filename == other.filename
else:
# Compare basenames
self_basenames = [Path(f).name for f in self.filename]
other_basenames = [Path(f).name for f in other.filename]
return self_basenames == other_basenames
elif isinstance(self.filename, list) or isinstance(other.filename, list):
# One is image sequence, other is single file
return False
else:
# Both are single files - use resolve() for symlink handling
if strict:
p1, p2 = Path(self.filename), Path(other.filename)
# Fast string comparison first
if p1.as_posix() == p2.as_posix():
return True
# Only resolve if both exist locally (avoid slow network timeouts)
try:
if p1.exists() and p2.exists():
return p1.resolve() == p2.resolve()
except OSError:
pass
return False
else:
return Path(self.filename).name == Path(other.filename).name
def matches_content(self, other: "Video") -> bool:
"""Check if this video has the same content as another video.
Args:
other: Another video to compare with.
Returns:
True if the videos have the same shape and backend type.
Notes:
This compares metadata like shape and backend type, not actual frame data.
"""
# Compare shapes
self_shape = self.shape
other_shape = other.shape
if self_shape != other_shape:
return False
# Compare backend types
if self.backend is None and other.backend is None:
return True
elif self.backend is None or other.backend is None:
return False
return type(self.backend).__name__ == type(other.backend).__name__
def matches_shape(self, other: "Video") -> bool:
"""Check if this video has the same shape as another video.
Args:
other: Another video to compare with.
Returns:
True if the videos have the same height, width, and channels.
Notes:
This only compares spatial dimensions, not the number of frames.
"""
# Try to get shape from backend metadata first if shape is not available
if self.backend is None and "shape" in self.backend_metadata:
self_shape = self.backend_metadata["shape"]
else:
self_shape = self.shape
if other.backend is None and "shape" in other.backend_metadata:
other_shape = other.backend_metadata["shape"]
else:
other_shape = other.shape
# Handle None shapes
if self_shape is None or other_shape is None:
return False
# Compare only height, width, channels (not frames)
return self_shape[1:] == other_shape[1:]
def has_overlapping_images(self, other: "Video") -> bool:
"""Check if this video has overlapping images with another video.
This method is specifically for ImageVideo backends (image sequences).
Args:
other: Another video to compare with.
Returns:
True if both are ImageVideo instances with overlapping image files.
False if either video is not an ImageVideo or no overlap exists.
Notes:
Only works with ImageVideo backends where filename is a list.
Compares individual image filenames (basenames only).
"""
# Both must be image sequences
if not (isinstance(self.filename, list) and isinstance(other.filename, list)):
return False
# Get basenames for comparison
self_basenames = set(Path(f).name for f in self.filename)
other_basenames = set(Path(f).name for f in other.filename)
# Check if there's any overlap
return len(self_basenames & other_basenames) > 0
def deduplicate_with(self, other: "Video") -> "Video":
"""Create a new video with duplicate images removed.
This method is specifically for ImageVideo backends (image sequences).
Args:
other: Another video to deduplicate against. Must also be ImageVideo.
Returns:
A new Video object with duplicate images removed from this video,
or None if all images were duplicates.
Raises:
ValueError: If either video is not an ImageVideo backend.
Notes:
Only works with ImageVideo backends where filename is a list.
Images are considered duplicates if they have the same basename.
The returned video contains only images from this video that are
not present in the other video.
"""
if not isinstance(self.filename, list):
raise ValueError("deduplicate_with only works with ImageVideo backends")
if not isinstance(other.filename, list):
raise ValueError("Other video must also be ImageVideo backend")
# Get basenames from other video
other_basenames = set(Path(f).name for f in other.filename)
# Keep only non-duplicate images
deduplicated_paths = [
f for f in self.filename if Path(f).name not in other_basenames
]
if not deduplicated_paths:
# All images were duplicates
return None
# Create new video with deduplicated images
return Video.from_filename(deduplicated_paths, grayscale=self.grayscale)
def merge_with(self, other: "Video") -> "Video":
"""Merge another video's images into this one.
This method is specifically for ImageVideo backends (image sequences).
Args:
other: Another video to merge with. Must also be ImageVideo.
Returns:
A new Video object with unique images from both videos.
Raises:
ValueError: If either video is not an ImageVideo backend.
Notes:
Only works with ImageVideo backends where filename is a list.
The merged video contains all unique images from both videos,
with automatic deduplication based on image basename.
"""
if not isinstance(self.filename, list):
raise ValueError("merge_with only works with ImageVideo backends")
if not isinstance(other.filename, list):
raise ValueError("Other video must also be ImageVideo backend")
# Get all unique images (by basename) preserving order
seen_basenames = set()
merged_paths = []
for path in self.filename:
basename = Path(path).name
if basename not in seen_basenames:
merged_paths.append(path)
seen_basenames.add(basename)
for path in other.filename:
basename = Path(path).name
if basename not in seen_basenames:
merged_paths.append(path)
seen_basenames.add(basename)
# Create new video with merged images
return Video.from_filename(merged_paths, grayscale=self.grayscale)
def save(
self,
save_path: str | Path,
frame_inds: list[int] | np.ndarray | None = None,
fps: float | None = None,
video_kwargs: dict[str, Any] | None = None,
) -> "Video":
"""Save video frames to a new video file.
Args:
save_path: Path to the new video file. Should end in MP4.
frame_inds: Frame indices to save. Can be specified as a list or array of
frame integers. If not specified, saves all video frames.
fps: Frames per second for the output video. If not specified, uses the
source video's FPS if available, otherwise defaults to 30.
video_kwargs: A dictionary of keyword arguments to provide to
`sio.save_video` for video compression.
Returns:
A new `Video` object pointing to the new video file.
"""
video_kwargs = {} if video_kwargs is None else video_kwargs.copy()
frame_inds = np.arange(len(self)) if frame_inds is None else frame_inds
# Use source video FPS if not explicitly specified
if fps is None:
fps = self.fps
if fps is not None and "fps" not in video_kwargs:
video_kwargs["fps"] = fps
with VideoWriter(save_path, **video_kwargs) as vw:
for frame_ind in frame_inds:
vw(self[frame_ind])
new_video = Video.from_filename(save_path, grayscale=self.grayscale)
return new_video
def apply_crop(
self,
path: str | Path,
*,
frame_inds: list[int] | np.ndarray | None = None,
fps: float | None = None,
video_kwargs: dict[str, Any] | None = None,
) -> "Video":
"""Bake this video's virtual crop into a new physical video file.
Materializes the cropped frames (``self[i]``, already cropped by the
virtual :class:`~sleap_io.io.video_reading.CropVideoBackend`) to ``path``
via :class:`~sleap_io.io.video_writing.VideoWriter`. The crop becomes
physical: the returned video has no ``CropVideoBackend`` / ``/video_crops``
entry. ``baked.shape`` equals this video's cropped shape when the cropped
width and height are multiples of 16; otherwise the H.264 encoder pads the
bottom/right edges up to the next multiple of 16 (the macro-block size),
so ``baked.shape`` may exceed the cropped shape on those edges. The
top-left content is preserved, so coordinates stay aligned regardless.
This operation is coordinate-neutral. A virtual crop already presents
cropped-frame coordinates, so baking the cropped pixels does not change
any point coordinates (unlike ``sio transform --crop``, which applies a
new crop and adjusts coordinates).
Provenance is preserved: the returned video's ``source_video`` is the
uncropped original — ``self.source_video`` (the parent a virtual crop is
created against), or, for a manually-built crop with no parent, an
uncropped view reconstructed from the crop backend's inner. So
``baked.source_video.shape`` is the uncropped shape while ``baked.shape``
is the cropped shape, and ``baked.grayscale`` is carried from this video.
Args:
path: Path to the new video file. Should end in MP4.
frame_inds: Frame indices to bake. Can be specified as a list or array
of frame integers. If not specified, bakes all video frames.
fps: Frames per second for the output video. If not specified, uses
this video's FPS if available, otherwise defaults to 30.
video_kwargs: A dictionary of keyword arguments to provide to
``sio.save_video`` for video compression.
Returns:
A new ``Video`` pointing to the baked file, with ``source_video`` set
to the uncropped original (or this video) and ``grayscale`` carried
from this video.
Raises:
ValueError: If this video has no virtual crop to apply (i.e.,
:meth:`_crop_tuple` returns ``None``). Use :meth:`save` to
re-encode an uncropped video.
"""
if self._crop_tuple() is None:
raise ValueError(
"apply_crop requires a cropped video (a virtual crop created via "
"Video.crop / Video.from_crop), but this video has no crop to "
"apply. Use Video.save to re-encode an uncropped video."
)
video_kwargs = {} if video_kwargs is None else video_kwargs.copy()
if frame_inds is None:
# A crop over a SPARSELY embedded video (frame_map keys are not the dense
# range 0..N-1, e.g. {5, 9}) cannot be baked by default: writing the frames
# compacts them to 0..k-1, so any labeled frame referencing a source index
# (5, 9) would dangle. Refuse with a clear error rather than crash or
# silently misalign. An explicit frame_inds bypasses this for advanced use.
inner = getattr(self.backend, "inner", None)
frame_map = getattr(inner, "frame_map", None)
if frame_map:
keys = sorted(frame_map.keys())
if keys != list(range(len(keys))):
raise ValueError(
"Cannot bake a virtual crop over a video with sparsely "
f"embedded frames (frame_map keys {keys}): baking would "
"compact frames to a contiguous range and break frame_idx "
"references. Pass explicit frame_inds to override, or "
"materialize from the original source video."
)
frame_inds = np.arange(len(self))
# Use this video's FPS if not explicitly specified.
if fps is None:
fps = self.fps
if fps is not None and "fps" not in video_kwargs:
video_kwargs["fps"] = fps
with VideoWriter(path, **video_kwargs) as vw:
for frame_ind in frame_inds:
vw(self[frame_ind])
baked = Video.from_filename(path, grayscale=self.grayscale)
# Provenance: the uncropped original. Walk past any still-virtual crop
# ancestors (a flattened crop-of-crop's source_video may itself be a crop)
# to the first uncropped ancestor. For a manually-built crop with no parent,
# reconstruct an uncropped view from the crop backend's inner, so
# source_video is never a cropped video.
source = self.source_video
while source is not None and source._crop_tuple() is not None:
source = source.source_video
if source is None:
inner = getattr(self.backend, "inner", None)
source = (
Video(filename=inner.filename, backend=inner)
if inner is not None
else self
)
baked.source_video = source
return baked
def set_video_plugin(self, plugin: str) -> None:
"""Set the video plugin and reopen the video.
Args:
plugin: Video plugin to use. One of "opencv", "FFMPEG", or "pyav".
Also accepts aliases (case-insensitive).
Raises:
ValueError: If the video is not a MediaVideo type.
Examples:
>>> video.set_video_plugin("opencv")
>>> video.set_video_plugin("CV2") # Same as "opencv"
"""
from sleap_io.io.video_reading import MediaVideo, normalize_plugin_name
if not self.filename.endswith(MediaVideo.EXTS):
raise ValueError(f"Cannot set plugin for non-media video: {self.filename}")
plugin = normalize_plugin_name(plugin)
# Close current backend if open
was_open = self.is_open
if was_open:
self.close()
# Update backend metadata
self.backend_metadata["plugin"] = plugin
# Reopen with new plugin if it was open
if was_open:
self.open()
EXTS = ('mp4', 'avi', 'mov', 'mj2', 'mkv', 'h5', 'hdf5', 'slp', 'png', 'jpg', 'jpeg', 'tif', 'tiff', 'bmp', 'seq')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__annotations__ = {'filename': 'str | list[str]', 'backend': 'VideoBackend | None', 'backend_metadata': 'dict[str, any]', 'source_video': "'Video | None'", 'open_backend': 'bool', '_exists_cache': 'dict[tuple[str, str | None], tuple[bool, float]]', '_url_headers': 'dict[str, str] | None', '_url_stream_mode': 'str'}
class-attribute
¶
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
__attrs_own_setattr__ = False
class-attribute
¶
Returns True when the argument is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.
__attrs_props__ = ClassProps(is_exception=False, is_slotted=True, has_weakref_slot=True, is_frozen=False, kw_only=<KeywordOnly.NO: 'no'>, collected_fields_by_mro=True, added_init=True, added_repr=False, added_eq=False, added_ordering=False, hashability=<Hashability.LEAVE_ALONE: 'leave_alone'>, added_match_args=True, added_str=False, added_pickling=True, on_setattr_hook=<function pipe.<locals>.wrapped_pipe at 0x7f08471ce840>, field_transformer=None)
class-attribute
¶
Effective class properties as derived from parameters to attr.s() or
define() decorators.
This is the same data structure that attrs uses internally to decide how to construct the final class.
Warning:
This feature is currently **experimental** and is not covered by our
strict backwards-compatibility guarantees.
Attributes:
| Name | Type | Description |
|---|---|---|
is_exception |
bool
|
Whether the class is treated as an exception class. |
is_slotted |
bool
|
Whether the class is |
has_weakref_slot |
bool
|
Whether the class has a slot for weak references. |
is_frozen |
bool
|
Whether the class is frozen. |
kw_only |
KeywordOnly
|
Whether / how the class enforces keyword-only arguments on the
|
collected_fields_by_mro |
bool
|
Whether the class fields were collected by method resolution order.
That is, correctly but unlike |
added_init |
bool
|
Whether the class has an attrs-generated |
added_repr |
bool
|
Whether the class has an attrs-generated |
added_eq |
bool
|
Whether the class has attrs-generated equality methods. |
added_ordering |
bool
|
Whether the class has attrs-generated ordering methods. |
hashability |
Hashability
|
How |
added_match_args |
bool
|
Whether the class supports positional |
added_str |
bool
|
Whether the class has an attrs-generated |
added_pickling |
bool
|
Whether the class has attrs-generated |
on_setattr_hook |
Callable[[Any, Attribute[Any], Any], Any] | None
|
The class's |
field_transformer |
Callable[[Attribute[Any]], Attribute[Any]] | None
|
The class's |
.. versionadded:: 25.4.0
__doc__ = '`Video` class used by sleap to represent videos and data associated with them.\n\nThis class is used to store information regarding a video and its components.\nIt is used to store the video\'s `filename`, `shape`, and the video\'s `backend`.\n\nTo create a `Video` object, use the `from_filename` method which will select the\nbackend appropriately.\n\nAttributes:\n filename: The filename(s) of the video. Supported extensions: "mp4", "avi",\n "mov", "mj2", "mkv", "h5", "hdf5", "slp", "png", "jpg", "jpeg", "tif",\n "tiff", "bmp", "seq". If the filename is a list, a list of image filenames\n are expected. If filename is a folder, it will be searched for images.\n backend: An object that implements the basic methods for reading and\n manipulating frames of a specific video type.\n backend_metadata: A dictionary of metadata specific to the backend. This is\n useful for storing metadata that requires an open backend (e.g., shape\n information) without having access to the video file itself.\n source_video: The source video object if this is a proxy video. This is present\n when the video contains an embedded subset of frames from another video.\n open_backend: Whether to open the backend when the video is available. If `True`\n (the default), the backend will be automatically opened if the video exists.\n Set this to `False` when you want to manually open the backend, or when the\n you know the video file does not exist and you want to avoid trying to open\n the file.\n _exists_cache: Per-instance TTL cache for the result of `exists()` when the\n `filename` is a remote URL. Keyed by `(filename, dataset)` and storing\n `(exists_bool, monotonic_timestamp)`. This avoids issuing a network probe\n on every call (e.g. from the `is_open` property, which GUIs poll on each\n render). The TTL defaults to 60 seconds and can be overridden via the\n `SLEAP_IO_EXISTS_TTL` environment variable. The cache is cleared on\n `replace_filename`.\n\nNotes:\n Instances of this class are hashed by identity, not by value. This means that\n two `Video` instances with the same attributes will NOT be considered equal in a\n set or dict.\n\nMedia Video Plugin Support:\n For media files (mp4, avi, etc.), the following plugins are supported:\n - "opencv": Uses OpenCV (cv2) for video reading\n - "FFMPEG": Uses imageio-ffmpeg for video reading\n - "pyav": Uses PyAV for video reading\n\n Plugin aliases (case-insensitive):\n - opencv: "opencv", "cv", "cv2", "ocv"\n - FFMPEG: "FFMPEG", "ffmpeg", "imageio-ffmpeg", "imageio_ffmpeg"\n - pyav: "pyav", "av"\n\n Plugin selection priority:\n 1. Explicitly specified plugin parameter\n 2. Backend metadata plugin value\n 3. Global default (set via sio.set_default_video_plugin)\n 4. Auto-detection based on available packages\n\nSee Also:\n VideoBackend: The backend interface for reading video data.\n sleap_io.set_default_video_plugin: Set global default plugin.\n sleap_io.get_default_video_plugin: Get current default plugin.\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 102
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__match_args__ = ('filename', 'backend', 'backend_metadata', 'source_video', 'open_backend')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__module__ = 'sleap_io.model.video'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__slots__ = ('filename', 'backend', 'backend_metadata', 'source_video', 'open_backend', '_exists_cache', '_url_headers', '_url_stream_mode', '__weakref__')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__static_attributes__ = ('backend', 'filename')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__weakref__
property
¶
list of weak references to the object
crop_fill
property
¶
The out-of-bounds fill value for this video's crop (0 if uncropped).
crop_rect
property
¶
Crop rect (x1, y1, x2, y2) in source coords, or None if uncropped.
fps
property
¶
Return the frames per second of the video.
For MediaVideo backends, this reads FPS from the video container metadata. For other backends (ImageVideo, HDF5Video, TiffVideo), this returns the explicitly set value or None if not set.
Returns:
| Type | Description |
|---|---|
|
The FPS if known, or None if unavailable/unknown. |
grayscale
property
¶
Return whether the video is grayscale.
If the video backend is not set or it cannot determine whether the video is grayscale, this will return None.
is_cropped
property
¶
Whether this video is a virtual crop of another video.
is_open
property
¶
Check if the video backend is open.
original_video
property
¶
The root video in the provenance chain.
For embedded videos, this returns the ultimate source video by traversing the source_video chain. Returns None if this video has no source_video (i.e., it IS an original).
This property is computed by following the source_video chain to find the root. For a single-level embedding (A embeds from B), original_video returns B. For multi-level embedding (A <- B <- C), it returns C.
shape
property
¶
Return the shape of the video as (num_frames, height, width, channels).
If the video backend is not set or it cannot determine the shape of the video, this will return None.
__attrs_post_init__()
¶
Post init syntactic sugar.
Source code in sleap_io/model/video.py
__deepcopy__(memo)
¶
Deep copy the video object.
Source code in sleap_io/model/video.py
def __deepcopy__(self, memo):
"""Deep copy the video object."""
if id(self) in memo:
return memo[id(self)]
reopen = False
if self.is_open:
reopen = True
self.close()
new_video = Video(
filename=self.filename,
backend=None,
backend_metadata=self.backend_metadata.copy(),
source_video=self.source_video,
open_backend=self.open_backend,
)
memo[id(self)] = new_video
if reopen:
self.open()
return new_video
__getitem__(inds)
¶
Return the frames of the video at the given indices.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inds
|
int | list[int] | slice
|
Index or list of indices of frames to read. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Frame or frames as a numpy array of shape |
See also: VideoBackend.get_frame, VideoBackend.get_frames
Source code in sleap_io/model/video.py
def __getitem__(self, inds: int | list[int] | slice) -> np.ndarray:
"""Return the frames of the video at the given indices.
Args:
inds: Index or list of indices of frames to read.
Returns:
Frame or frames as a numpy array of shape `(height, width, channels)` if a
scalar index is provided, or `(frames, height, width, channels)` if a list
of indices is provided.
See also: VideoBackend.get_frame, VideoBackend.get_frames
"""
if not self.is_open:
if self.open_backend:
self.open()
else:
raise ValueError(
"Video backend is not open. Call video.open() or set "
"video.open_backend to True to do automatically on frame read."
)
return self.backend[inds]
__init__(filename, backend=None, backend_metadata=NOTHING, source_video=None, open_backend=True)
¶
Method generated by attrs for class Video.
__len__()
¶
__repr__()
¶
Informal string representation (for print or format).
Source code in sleap_io/model/video.py
def __repr__(self) -> str:
"""Informal string representation (for print or format)."""
dataset = (
f"dataset={self.backend.dataset}, "
if getattr(self.backend, "dataset", "")
else ""
)
return (
"Video("
f'filename="{self.filename}", '
f"shape={self.shape}, "
f"{dataset}"
f"backend={type(self.backend).__name__}"
")"
)
__str__()
¶
apply_crop(path, *, frame_inds=None, fps=None, video_kwargs=None)
¶
Bake this video's virtual crop into a new physical video file.
Materializes the cropped frames (self[i], already cropped by the
virtual :class:~sleap_io.io.video_reading.CropVideoBackend) to path
via :class:~sleap_io.io.video_writing.VideoWriter. The crop becomes
physical: the returned video has no CropVideoBackend / /video_crops
entry. baked.shape equals this video's cropped shape when the cropped
width and height are multiples of 16; otherwise the H.264 encoder pads the
bottom/right edges up to the next multiple of 16 (the macro-block size),
so baked.shape may exceed the cropped shape on those edges. The
top-left content is preserved, so coordinates stay aligned regardless.
This operation is coordinate-neutral. A virtual crop already presents
cropped-frame coordinates, so baking the cropped pixels does not change
any point coordinates (unlike sio transform --crop, which applies a
new crop and adjusts coordinates).
Provenance is preserved: the returned video's source_video is the
uncropped original — self.source_video (the parent a virtual crop is
created against), or, for a manually-built crop with no parent, an
uncropped view reconstructed from the crop backend's inner. So
baked.source_video.shape is the uncropped shape while baked.shape
is the cropped shape, and baked.grayscale is carried from this video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to the new video file. Should end in MP4. |
required |
frame_inds
|
list[int] | ndarray | None
|
Frame indices to bake. Can be specified as a list or array of frame integers. If not specified, bakes all video frames. |
None
|
fps
|
float | None
|
Frames per second for the output video. If not specified, uses this video's FPS if available, otherwise defaults to 30. |
None
|
video_kwargs
|
dict[str, Any] | None
|
A dictionary of keyword arguments to provide to
|
None
|
Returns:
| Type | Description |
|---|---|
Video
|
A new |
Raises:
| Type | Description |
|---|---|
ValueError
|
If this video has no virtual crop to apply (i.e.,
:meth: |
Source code in sleap_io/model/video.py
def apply_crop(
self,
path: str | Path,
*,
frame_inds: list[int] | np.ndarray | None = None,
fps: float | None = None,
video_kwargs: dict[str, Any] | None = None,
) -> "Video":
"""Bake this video's virtual crop into a new physical video file.
Materializes the cropped frames (``self[i]``, already cropped by the
virtual :class:`~sleap_io.io.video_reading.CropVideoBackend`) to ``path``
via :class:`~sleap_io.io.video_writing.VideoWriter`. The crop becomes
physical: the returned video has no ``CropVideoBackend`` / ``/video_crops``
entry. ``baked.shape`` equals this video's cropped shape when the cropped
width and height are multiples of 16; otherwise the H.264 encoder pads the
bottom/right edges up to the next multiple of 16 (the macro-block size),
so ``baked.shape`` may exceed the cropped shape on those edges. The
top-left content is preserved, so coordinates stay aligned regardless.
This operation is coordinate-neutral. A virtual crop already presents
cropped-frame coordinates, so baking the cropped pixels does not change
any point coordinates (unlike ``sio transform --crop``, which applies a
new crop and adjusts coordinates).
Provenance is preserved: the returned video's ``source_video`` is the
uncropped original — ``self.source_video`` (the parent a virtual crop is
created against), or, for a manually-built crop with no parent, an
uncropped view reconstructed from the crop backend's inner. So
``baked.source_video.shape`` is the uncropped shape while ``baked.shape``
is the cropped shape, and ``baked.grayscale`` is carried from this video.
Args:
path: Path to the new video file. Should end in MP4.
frame_inds: Frame indices to bake. Can be specified as a list or array
of frame integers. If not specified, bakes all video frames.
fps: Frames per second for the output video. If not specified, uses
this video's FPS if available, otherwise defaults to 30.
video_kwargs: A dictionary of keyword arguments to provide to
``sio.save_video`` for video compression.
Returns:
A new ``Video`` pointing to the baked file, with ``source_video`` set
to the uncropped original (or this video) and ``grayscale`` carried
from this video.
Raises:
ValueError: If this video has no virtual crop to apply (i.e.,
:meth:`_crop_tuple` returns ``None``). Use :meth:`save` to
re-encode an uncropped video.
"""
if self._crop_tuple() is None:
raise ValueError(
"apply_crop requires a cropped video (a virtual crop created via "
"Video.crop / Video.from_crop), but this video has no crop to "
"apply. Use Video.save to re-encode an uncropped video."
)
video_kwargs = {} if video_kwargs is None else video_kwargs.copy()
if frame_inds is None:
# A crop over a SPARSELY embedded video (frame_map keys are not the dense
# range 0..N-1, e.g. {5, 9}) cannot be baked by default: writing the frames
# compacts them to 0..k-1, so any labeled frame referencing a source index
# (5, 9) would dangle. Refuse with a clear error rather than crash or
# silently misalign. An explicit frame_inds bypasses this for advanced use.
inner = getattr(self.backend, "inner", None)
frame_map = getattr(inner, "frame_map", None)
if frame_map:
keys = sorted(frame_map.keys())
if keys != list(range(len(keys))):
raise ValueError(
"Cannot bake a virtual crop over a video with sparsely "
f"embedded frames (frame_map keys {keys}): baking would "
"compact frames to a contiguous range and break frame_idx "
"references. Pass explicit frame_inds to override, or "
"materialize from the original source video."
)
frame_inds = np.arange(len(self))
# Use this video's FPS if not explicitly specified.
if fps is None:
fps = self.fps
if fps is not None and "fps" not in video_kwargs:
video_kwargs["fps"] = fps
with VideoWriter(path, **video_kwargs) as vw:
for frame_ind in frame_inds:
vw(self[frame_ind])
baked = Video.from_filename(path, grayscale=self.grayscale)
# Provenance: the uncropped original. Walk past any still-virtual crop
# ancestors (a flattened crop-of-crop's source_video may itself be a crop)
# to the first uncropped ancestor. For a manually-built crop with no parent,
# reconstruct an uncropped view from the crop backend's inner, so
# source_video is never a cropped video.
source = self.source_video
while source is not None and source._crop_tuple() is not None:
source = source.source_video
if source is None:
inner = getattr(self.backend, "inner", None)
source = (
Video(filename=inner.filename, backend=inner)
if inner is not None
else self
)
baked.source_video = source
return baked
close()
¶
Close the video backend.
Source code in sleap_io/model/video.py
def close(self):
"""Close the video backend."""
if self.backend is not None:
# Try to remember values from previous backend if available and not
# specified.
try:
self.backend_metadata["dataset"] = getattr(
self.backend, "dataset", None
)
self.backend_metadata["grayscale"] = getattr(
self.backend, "grayscale", None
)
self.backend_metadata["shape"] = getattr(self.backend, "shape", None)
self.backend_metadata["fps"] = getattr(self.backend, "fps", None)
# Persist the crop so a Video cropped in-memory (never loaded
# from disk) survives a close()->open() and deepcopy: open()
# re-wraps from these keys (the closed-path shape above is
# already the cropped shape).
from sleap_io.io.video_reading import CropVideoBackend
if isinstance(self.backend, CropVideoBackend):
self.backend_metadata["crop"] = list(self.backend.crop)
self.backend_metadata["crop_fill"] = self.backend.fill
except Exception:
pass
# Deterministically release the backend's open handles (the cached
# reader and, for a remote HDF5Video, the fsspec URL file-like)
# rather than relying on garbage collection.
try:
self.backend.close()
except Exception:
pass
del self.backend
self.backend = None
crop(crop=None, *, bbox=None, roi=None, center=None, size=None, margin=0, fill=0, share_decode=True)
¶
Return a virtual, on-read cropped view of this video.
Exactly one region spec must be given: crop (explicit
(x1, y1, x2, y2) rect), bbox, roi (its axis-aligned bounds +
margin), or (center, size) for a fixed-size centered/
centroid-following window. The returned Video shares no pixels with
this one; frames are decoded on read and cropped (byte-identical to
:func:sleap_io.transform.frame.crop_frame). Out-of-bounds regions are
pad-filled with fill (never clamped), so the output shape is always
exactly (y2 - y1, x2 - x1).
The crop composes (FLATTENS when fills agree and the region is in-bounds)
with any existing crop on this video via
:meth:CropVideoBackend.wrap. source_video is set to this video for
provenance. When share_decode (the default), the new crop reuses this
video's backend instance as the shared inner so a mosaic of tiles over
one file decodes each source frame once; in that case the new tile does
NOT own the shared decoder (this video does).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
crop
|
tuple[int, int, int, int] | None
|
Explicit crop region |
None
|
bbox
|
tuple[float, float, float, float] | None
|
A bounding box |
None
|
roi
|
object | None
|
Any object exposing axis-aligned |
None
|
center
|
tuple[float, float] | None
|
Window center |
None
|
size
|
tuple[int, int] | None
|
Fixed output |
None
|
margin
|
int
|
Pixels added around the |
0
|
fill
|
int | tuple[int, ...]
|
Fill value for out-of-bounds regions. |
0
|
share_decode
|
bool
|
If |
True
|
Returns:
| Type | Description |
|---|---|
Video
|
A new |
Source code in sleap_io/model/video.py
def crop(
self,
crop: tuple[int, int, int, int] | None = None,
*,
bbox: tuple[float, float, float, float] | None = None,
roi: object | None = None,
center: tuple[float, float] | None = None,
size: tuple[int, int] | None = None,
margin: int = 0,
fill: int | tuple[int, ...] = 0,
share_decode: bool = True,
) -> "Video":
"""Return a virtual, on-read cropped view of this video.
Exactly one region spec must be given: ``crop`` (explicit
``(x1, y1, x2, y2)`` rect), ``bbox``, ``roi`` (its axis-aligned bounds +
``margin``), or (``center``, ``size``) for a fixed-size centered/
centroid-following window. The returned ``Video`` shares no pixels with
this one; frames are decoded on read and cropped (byte-identical to
:func:`sleap_io.transform.frame.crop_frame`). Out-of-bounds regions are
pad-filled with ``fill`` (never clamped), so the output shape is always
exactly ``(y2 - y1, x2 - x1)``.
The crop composes (FLATTENS when fills agree and the region is in-bounds)
with any existing crop on this video via
:meth:`CropVideoBackend.wrap`. ``source_video`` is set to this video for
provenance. When ``share_decode`` (the default), the new crop reuses this
video's backend instance as the shared inner so a mosaic of tiles over
one file decodes each source frame once; in that case the new tile does
NOT own the shared decoder (this video does).
Args:
crop: Explicit crop region ``(x1, y1, x2, y2)``, ``x2``/``y2``
exclusive.
bbox: A bounding box ``(x1, y1, x2, y2)``; bounds may be float.
roi: Any object exposing axis-aligned ``.bounds`` as
``(minx, miny, maxx, maxy)`` (e.g. a shapely geometry).
center: Window center ``(cx, cy)`` (used with ``size``).
size: Fixed output ``(width, height)`` (used with ``center``).
margin: Pixels added around the ``roi`` bounds on every side.
fill: Fill value for out-of-bounds regions.
share_decode: If ``True`` (the default), reuse this video's backend
as the shared inner so tiles decode each frame once; the new tile
does not own the shared decoder.
Returns:
A new ``Video`` exposing the cropped view.
"""
from sleap_io.io.video_reading import CropVideoBackend
rect = _resolve_crop_rect(crop, bbox, roi, center, size, margin)
if self.backend is None and self.open_backend:
self.open()
if self.backend is None:
raise ValueError(
"Cannot crop a video with no open backend. Open it first (set "
"open_backend=True or call .open()) before cropping."
)
inner = self.backend
cropped_backend = CropVideoBackend.wrap(
inner=inner, crop=rect, fill=fill, owns_inner=not share_decode
)
cropped = Video(
filename=self.filename,
backend=cropped_backend,
source_video=self,
open_backend=self.open_backend,
)
x1, y1, x2, y2 = cropped_backend.crop
src_shape = self.shape
cropped.backend_metadata = {
**self.backend_metadata,
"shape": (src_shape[0], y2 - y1, x2 - x1, src_shape[3])
if src_shape is not None
else None,
# The uncropped source shape, so a closed re-serialize keeps videos_json
# describing the full frame even without a live source_video (D-120/DI-2).
"source_shape": list(src_shape) if src_shape is not None else None,
# COMPOSED source rect from wrap (D-120): keeps open/closed crop keys
# identical and root-canonical, and survives close()->open().
"crop": list(cropped_backend.crop),
"crop_fill": cropped_backend.fill,
}
return cropped
deduplicate_with(other)
¶
Create a new video with duplicate images removed.
This method is specifically for ImageVideo backends (image sequences).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
Video
|
Another video to deduplicate against. Must also be ImageVideo. |
required |
Returns:
| Type | Description |
|---|---|
Video
|
A new Video object with duplicate images removed from this video, or None if all images were duplicates. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If either video is not an ImageVideo backend. |
Notes
Only works with ImageVideo backends where filename is a list. Images are considered duplicates if they have the same basename. The returned video contains only images from this video that are not present in the other video.
Source code in sleap_io/model/video.py
def deduplicate_with(self, other: "Video") -> "Video":
"""Create a new video with duplicate images removed.
This method is specifically for ImageVideo backends (image sequences).
Args:
other: Another video to deduplicate against. Must also be ImageVideo.
Returns:
A new Video object with duplicate images removed from this video,
or None if all images were duplicates.
Raises:
ValueError: If either video is not an ImageVideo backend.
Notes:
Only works with ImageVideo backends where filename is a list.
Images are considered duplicates if they have the same basename.
The returned video contains only images from this video that are
not present in the other video.
"""
if not isinstance(self.filename, list):
raise ValueError("deduplicate_with only works with ImageVideo backends")
if not isinstance(other.filename, list):
raise ValueError("Other video must also be ImageVideo backend")
# Get basenames from other video
other_basenames = set(Path(f).name for f in other.filename)
# Keep only non-duplicate images
deduplicated_paths = [
f for f in self.filename if Path(f).name not in other_basenames
]
if not deduplicated_paths:
# All images were duplicates
return None
# Create new video with deduplicated images
return Video.from_filename(deduplicated_paths, grayscale=self.grayscale)
exists(check_all=False, dataset=None)
¶
Check if the video file exists and is accessible.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
check_all
|
bool
|
If |
False
|
dataset
|
str | None
|
Name of dataset in HDF5 file. If specified, this will function will
return |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
|
Source code in sleap_io/model/video.py
def exists(self, check_all: bool = False, dataset: str | None = None) -> bool:
"""Check if the video file exists and is accessible.
Args:
check_all: If `True`, check that all filenames in a list exist. If `False`
(the default), check that the first filename exists.
dataset: Name of dataset in HDF5 file. If specified, this will function will
return `False` if the dataset does not exist.
Returns:
`True` if the file exists and is accessible, `False` otherwise.
"""
if isinstance(self.filename, list):
if check_all:
for f in self.filename:
if not is_file_accessible(f):
return False
return True
else:
return is_file_accessible(self.filename[0])
# URL fast path: must run BEFORE `is_file_accessible`, which treats the
# filename as a local path and would spuriously return False for a URL.
from sleap_io.io._remote import _is_url
if _is_url(self.filename):
return self._url_exists(dataset)
file_is_accessible = is_file_accessible(self.filename)
if not file_is_accessible:
# Check if it's a directory (ImageVideo source)
if Path(self.filename).is_dir():
return True
return False
if dataset is None or dataset == "":
dataset = self.backend_metadata.get("dataset", None)
if dataset is not None and dataset != "":
has_dataset = False
if (
self.backend is not None
and type(self.backend) is HDF5Video
and self.backend._open_reader is not None
):
has_dataset = dataset in self.backend._open_reader
else:
with h5py.File(self.filename, "r") as f:
has_dataset = dataset in f
return has_dataset
return True
frame_to_seconds(frame_idx)
¶
Convert a frame index to timestamp in seconds.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Zero-indexed frame number. |
required |
Returns:
| Type | Description |
|---|---|
float | None
|
Time in seconds, or None if FPS is unknown. |
Notes
This assumes constant frame rate. For variable frame rate videos, the returned timestamp may be approximate.
Source code in sleap_io/model/video.py
def frame_to_seconds(self, frame_idx: int) -> float | None:
"""Convert a frame index to timestamp in seconds.
Args:
frame_idx: Zero-indexed frame number.
Returns:
Time in seconds, or None if FPS is unknown.
Notes:
This assumes constant frame rate. For variable frame rate videos,
the returned timestamp may be approximate.
"""
if self.fps is None or self.fps <= 0:
return None
return frame_idx / self.fps
from_crop(video, crop=None, *, bbox=None, roi=None, center=None, size=None, margin=0, fill=0, share_decode=True, **kwargs)
classmethod
¶
Open video (path or Video) and return a virtual crop.
Accepts the same region specs as :meth:crop (crop/bbox/roi/
center+size); extra keyword arguments are forwarded to
:meth:from_filename when video is a path (ignored when it is already
a Video).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
video
|
str | Path | Video
|
A path/filename to open, or an existing |
required |
crop
|
tuple[int, int, int, int] | None
|
Explicit crop region |
None
|
bbox
|
tuple[float, float, float, float] | None
|
A bounding box |
None
|
roi
|
object | None
|
An object exposing axis-aligned |
None
|
center
|
tuple[float, float] | None
|
Window center |
None
|
size
|
tuple[int, int] | None
|
Fixed output |
None
|
margin
|
int
|
Pixels added around the |
0
|
fill
|
int | tuple[int, ...]
|
Fill value for out-of-bounds regions. |
0
|
share_decode
|
bool
|
If |
True
|
**kwargs
|
Forwarded to :meth: |
required |
Returns:
| Type | Description |
|---|---|
Video
|
A new |
Source code in sleap_io/model/video.py
@classmethod
def from_crop(
cls,
video: "str | Path | Video",
crop: tuple[int, int, int, int] | None = None,
*,
bbox: tuple[float, float, float, float] | None = None,
roi: object | None = None,
center: tuple[float, float] | None = None,
size: tuple[int, int] | None = None,
margin: int = 0,
fill: int | tuple[int, ...] = 0,
share_decode: bool = True,
**kwargs,
) -> "Video":
"""Open ``video`` (path or ``Video``) and return a virtual crop.
Accepts the same region specs as :meth:`crop` (``crop``/``bbox``/``roi``/
``center``+``size``); extra keyword arguments are forwarded to
:meth:`from_filename` when ``video`` is a path (ignored when it is already
a ``Video``).
Args:
video: A path/filename to open, or an existing ``Video`` to crop.
crop: Explicit crop region ``(x1, y1, x2, y2)``, ``x2``/``y2`` exclusive.
bbox: A bounding box ``(x1, y1, x2, y2)``; bounds may be float.
roi: An object exposing axis-aligned ``.bounds`` (e.g. a shapely
geometry); ``margin`` is applied around it.
center: Window center ``(cx, cy)`` (with ``size``).
size: Fixed output ``(width, height)`` (with ``center``).
margin: Pixels added around the ``roi`` bounds on every side.
fill: Fill value for out-of-bounds regions.
share_decode: If ``True`` (default), reuse the source decoder.
**kwargs: Forwarded to :meth:`from_filename` for a path input.
Returns:
A new ``Video`` exposing the cropped view.
"""
if isinstance(video, (str, Path)):
video = cls.from_filename(video, **kwargs)
return video.crop(
crop,
bbox=bbox,
roi=roi,
center=center,
size=size,
margin=margin,
fill=fill,
share_decode=share_decode,
)
from_filename(filename, dataset=None, grayscale=None, keep_open=True, source_video=None, **kwargs)
classmethod
¶
Create a Video from a filename.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str | list[str]
|
The filename(s) of the video. Supported extensions: "mp4", "avi", "mov", "mj2", "mkv", "h5", "hdf5", "slp", "png", "jpg", "jpeg", "tif", "tiff", "bmp". If the filename is a list, a list of image filenames are expected. If filename is a folder, it will be searched for images. |
required |
dataset
|
str | None
|
Name of dataset in HDF5 file. |
None
|
grayscale
|
bool | None
|
Whether to force grayscale. If None, autodetect on first frame load. |
None
|
keep_open
|
bool
|
Whether to keep the video reader open between calls to read frames. If False, will close the reader after each call. If True (the default), it will keep the reader open and cache it for subsequent calls which may enhance the performance of reading multiple frames. |
True
|
source_video
|
Video | None
|
The source video object if this is a proxy video. This is present when the video contains an embedded subset of frames from another video. |
None
|
**kwargs
|
Additional backend-specific arguments passed to VideoBackend.from_filename. See VideoBackend.from_filename for supported arguments. |
required |
Returns:
| Type | Description |
|---|---|
VideoBackend
|
Video instance with the appropriate backend instantiated. |
Source code in sleap_io/model/video.py
@classmethod
def from_filename(
cls,
filename: str | list[str],
dataset: str | None = None,
grayscale: bool | None = None,
keep_open: bool = True,
source_video: "Video | None" = None,
**kwargs,
) -> VideoBackend:
"""Create a Video from a filename.
Args:
filename: The filename(s) of the video. Supported extensions: "mp4", "avi",
"mov", "mj2", "mkv", "h5", "hdf5", "slp", "png", "jpg", "jpeg", "tif",
"tiff", "bmp". If the filename is a list, a list of image filenames are
expected. If filename is a folder, it will be searched for images.
dataset: Name of dataset in HDF5 file.
grayscale: Whether to force grayscale. If None, autodetect on first frame
load.
keep_open: Whether to keep the video reader open between calls to read
frames. If False, will close the reader after each call. If True (the
default), it will keep the reader open and cache it for subsequent calls
which may enhance the performance of reading multiple frames.
source_video: The source video object if this is a proxy video. This is
present when the video contains an embedded subset of frames from
another video.
**kwargs: Additional backend-specific arguments passed to
VideoBackend.from_filename. See VideoBackend.from_filename for supported
arguments.
Returns:
Video instance with the appropriate backend instantiated.
"""
backend = VideoBackend.from_filename(
filename,
dataset=dataset,
grayscale=grayscale,
keep_open=keep_open,
**kwargs,
)
# If filename is a directory, VideoBackend.from_filename will expand it
# to a list of paths to images contained within the directory. In this
# case we want to use the expanded list as filename
return cls(
filename=backend.filename,
backend=backend,
source_video=source_video,
)
has_overlapping_images(other)
¶
Check if this video has overlapping images with another video.
This method is specifically for ImageVideo backends (image sequences).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
Video
|
Another video to compare with. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if both are ImageVideo instances with overlapping image files. False if either video is not an ImageVideo or no overlap exists. |
Notes
Only works with ImageVideo backends where filename is a list. Compares individual image filenames (basenames only).
Source code in sleap_io/model/video.py
def has_overlapping_images(self, other: "Video") -> bool:
"""Check if this video has overlapping images with another video.
This method is specifically for ImageVideo backends (image sequences).
Args:
other: Another video to compare with.
Returns:
True if both are ImageVideo instances with overlapping image files.
False if either video is not an ImageVideo or no overlap exists.
Notes:
Only works with ImageVideo backends where filename is a list.
Compares individual image filenames (basenames only).
"""
# Both must be image sequences
if not (isinstance(self.filename, list) and isinstance(other.filename, list)):
return False
# Get basenames for comparison
self_basenames = set(Path(f).name for f in self.filename)
other_basenames = set(Path(f).name for f in other.filename)
# Check if there's any overlap
return len(self_basenames & other_basenames) > 0
matches_content(other)
¶
Check if this video has the same content as another video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
Video
|
Another video to compare with. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the videos have the same shape and backend type. |
Notes
This compares metadata like shape and backend type, not actual frame data.
Source code in sleap_io/model/video.py
def matches_content(self, other: "Video") -> bool:
"""Check if this video has the same content as another video.
Args:
other: Another video to compare with.
Returns:
True if the videos have the same shape and backend type.
Notes:
This compares metadata like shape and backend type, not actual frame data.
"""
# Compare shapes
self_shape = self.shape
other_shape = other.shape
if self_shape != other_shape:
return False
# Compare backend types
if self.backend is None and other.backend is None:
return True
elif self.backend is None or other.backend is None:
return False
return type(self.backend).__name__ == type(other.backend).__name__
matches_path(other, strict=False)
¶
Check if this video has the same path as another video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
Video
|
Another video to compare with. |
required |
strict
|
bool
|
If True, require exact path match. If False, consider videos with the same filename (basename) as matching. |
False
|
Returns:
| Type | Description |
|---|---|
bool
|
True if the videos have matching paths, False otherwise. |
Notes
For HDF5 video backends (e.g., embedded videos in .pkg.slp files), matching prioritizes the source_filename attribute since multiple videos can share the same HDF5 file path but reference different source videos. Falls back to dataset name matching if source_filename is not available.
Source code in sleap_io/model/video.py
def matches_path(self, other: "Video", strict: bool = False) -> bool:
"""Check if this video has the same path as another video.
Args:
other: Another video to compare with.
strict: If True, require exact path match. If False, consider videos
with the same filename (basename) as matching.
Returns:
True if the videos have matching paths, False otherwise.
Notes:
For HDF5 video backends (e.g., embedded videos in .pkg.slp files),
matching prioritizes the source_filename attribute since multiple
videos can share the same HDF5 file path but reference different
source videos. Falls back to dataset name matching if source_filename
is not available.
"""
# Handle HDF5 backends specially - prioritize source_filename matching
self_is_hdf5 = isinstance(self.backend, HDF5Video)
other_is_hdf5 = isinstance(other.backend, HDF5Video)
if self_is_hdf5 and other_is_hdf5:
# Both are HDF5 videos - must match by BOTH source_filename AND dataset
# to distinguish different videos embedded in the same pkg.slp file
self_source = self.backend.source_filename
other_source = other.backend.source_filename
self_dataset = self.backend.dataset
other_dataset = other.backend.dataset
# If both have datasets, they must match
if self_dataset is not None and other_dataset is not None:
if self_dataset != other_dataset:
return False # Different datasets = different videos
# If both have source_filenames, compare them
if self_source is not None and other_source is not None:
if strict:
# For HDF5 videos, just compare normalized path strings
# (avoid slow resolve() on network paths)
return Path(self_source).as_posix() == Path(other_source).as_posix()
else:
return Path(self_source).name == Path(other_source).name
# If only datasets available (no source_filename), they must match
if self_dataset is not None and other_dataset is not None:
return self_dataset == other_dataset
# If neither source_filename nor dataset available, cannot match
return False
if isinstance(self.filename, list) and isinstance(other.filename, list):
# Both are image sequences
if strict:
return self.filename == other.filename
else:
# Compare basenames
self_basenames = [Path(f).name for f in self.filename]
other_basenames = [Path(f).name for f in other.filename]
return self_basenames == other_basenames
elif isinstance(self.filename, list) or isinstance(other.filename, list):
# One is image sequence, other is single file
return False
else:
# Both are single files - use resolve() for symlink handling
if strict:
p1, p2 = Path(self.filename), Path(other.filename)
# Fast string comparison first
if p1.as_posix() == p2.as_posix():
return True
# Only resolve if both exist locally (avoid slow network timeouts)
try:
if p1.exists() and p2.exists():
return p1.resolve() == p2.resolve()
except OSError:
pass
return False
else:
return Path(self.filename).name == Path(other.filename).name
matches_shape(other)
¶
Check if this video has the same shape as another video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
Video
|
Another video to compare with. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the videos have the same height, width, and channels. |
Notes
This only compares spatial dimensions, not the number of frames.
Source code in sleap_io/model/video.py
def matches_shape(self, other: "Video") -> bool:
"""Check if this video has the same shape as another video.
Args:
other: Another video to compare with.
Returns:
True if the videos have the same height, width, and channels.
Notes:
This only compares spatial dimensions, not the number of frames.
"""
# Try to get shape from backend metadata first if shape is not available
if self.backend is None and "shape" in self.backend_metadata:
self_shape = self.backend_metadata["shape"]
else:
self_shape = self.shape
if other.backend is None and "shape" in other.backend_metadata:
other_shape = other.backend_metadata["shape"]
else:
other_shape = other.shape
# Handle None shapes
if self_shape is None or other_shape is None:
return False
# Compare only height, width, channels (not frames)
return self_shape[1:] == other_shape[1:]
merge_with(other)
¶
Merge another video's images into this one.
This method is specifically for ImageVideo backends (image sequences).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
Video
|
Another video to merge with. Must also be ImageVideo. |
required |
Returns:
| Type | Description |
|---|---|
Video
|
A new Video object with unique images from both videos. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If either video is not an ImageVideo backend. |
Notes
Only works with ImageVideo backends where filename is a list. The merged video contains all unique images from both videos, with automatic deduplication based on image basename.
Source code in sleap_io/model/video.py
def merge_with(self, other: "Video") -> "Video":
"""Merge another video's images into this one.
This method is specifically for ImageVideo backends (image sequences).
Args:
other: Another video to merge with. Must also be ImageVideo.
Returns:
A new Video object with unique images from both videos.
Raises:
ValueError: If either video is not an ImageVideo backend.
Notes:
Only works with ImageVideo backends where filename is a list.
The merged video contains all unique images from both videos,
with automatic deduplication based on image basename.
"""
if not isinstance(self.filename, list):
raise ValueError("merge_with only works with ImageVideo backends")
if not isinstance(other.filename, list):
raise ValueError("Other video must also be ImageVideo backend")
# Get all unique images (by basename) preserving order
seen_basenames = set()
merged_paths = []
for path in self.filename:
basename = Path(path).name
if basename not in seen_basenames:
merged_paths.append(path)
seen_basenames.add(basename)
for path in other.filename:
basename = Path(path).name
if basename not in seen_basenames:
merged_paths.append(path)
seen_basenames.add(basename)
# Create new video with merged images
return Video.from_filename(merged_paths, grayscale=self.grayscale)
open(filename=None, dataset=None, grayscale=None, keep_open=True, plugin=None)
¶
Open the video backend for reading.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str | None
|
Filename to open. If not specified, will use the filename set on the video object. |
None
|
dataset
|
str | None
|
Name of dataset in HDF5 file. |
None
|
grayscale
|
str | None
|
Whether to force grayscale. If None, autodetect on first frame load. |
None
|
keep_open
|
bool
|
Whether to keep the video reader open between calls to read frames. If False, will close the reader after each call. If True (the default), it will keep the reader open and cache it for subsequent calls which may enhance the performance of reading multiple frames. |
True
|
plugin
|
str | None
|
Video plugin to use for MediaVideo files. One of "opencv", "FFMPEG", or "pyav". Also accepts aliases (case-insensitive). If not specified, uses the backend metadata, global default, or auto-detection in that order. |
None
|
Notes
This is useful for opening the video backend to read frames and then closing it after reading all the necessary frames.
If the backend was already open, it will be closed before opening a new one. Values for the HDF5 dataset and grayscale will be remembered if not specified.
Source code in sleap_io/model/video.py
def open(
self,
filename: str | None = None,
dataset: str | None = None,
grayscale: str | None = None,
keep_open: bool = True,
plugin: str | None = None,
):
"""Open the video backend for reading.
Args:
filename: Filename to open. If not specified, will use the filename set on
the video object.
dataset: Name of dataset in HDF5 file.
grayscale: Whether to force grayscale. If None, autodetect on first frame
load.
keep_open: Whether to keep the video reader open between calls to read
frames. If False, will close the reader after each call. If True (the
default), it will keep the reader open and cache it for subsequent calls
which may enhance the performance of reading multiple frames.
plugin: Video plugin to use for MediaVideo files. One of "opencv",
"FFMPEG", or "pyav". Also accepts aliases (case-insensitive).
If not specified, uses the backend metadata, global default,
or auto-detection in that order.
Notes:
This is useful for opening the video backend to read frames and then closing
it after reading all the necessary frames.
If the backend was already open, it will be closed before opening a new one.
Values for the HDF5 dataset and grayscale will be remembered if not
specified.
"""
if filename is not None:
self.replace_filename(filename, open=False)
# Try to remember values from previous backend if available and not specified.
if self.backend is not None:
if dataset is None:
dataset = getattr(self.backend, "dataset", None)
if grayscale is None:
grayscale = getattr(self.backend, "grayscale", None)
else:
if dataset is None and "dataset" in self.backend_metadata:
dataset = self.backend_metadata["dataset"]
if grayscale is None:
if "grayscale" in self.backend_metadata:
grayscale = self.backend_metadata["grayscale"]
elif "shape" in self.backend_metadata:
grayscale = self.backend_metadata["shape"][-1] == 1
if not self.exists(dataset=dataset):
from sleap_io.io._remote import _is_url, _redact_url
# Redact credential-bearing URLs (e.g. presigned ``?token=`` links)
# so they never surface in tracebacks/logs. Local paths are shown
# verbatim.
name = (
_redact_url(self.filename)
if isinstance(self.filename, str) and _is_url(self.filename)
else self.filename
)
msg = f"Video does not exist or cannot be opened for reading: {name}"
if dataset is not None:
msg += f" (dataset: {dataset})"
raise FileNotFoundError(msg)
# Close previous backend if open.
self.close()
# Handle plugin parameter
backend_kwargs = {}
if plugin is not None:
from sleap_io.io.video_reading import normalize_plugin_name
plugin = normalize_plugin_name(plugin)
self.backend_metadata["plugin"] = plugin
if "plugin" in self.backend_metadata:
backend_kwargs["plugin"] = self.backend_metadata["plugin"]
# Create new backend. Forward the URL auth context so a reopened remote
# HDF5Video stays authenticated (the previous backend, and its headers,
# were dropped by self.close() above).
self.backend = VideoBackend.from_filename(
self.filename,
dataset=dataset,
grayscale=grayscale,
keep_open=keep_open,
url_headers=self._url_headers,
url_stream_mode=self._url_stream_mode,
**backend_kwargs,
)
# Re-wrap as a crop view if this video records a crop in its metadata.
# The rebuilt backend above is always a plain backend, so this wraps
# exactly once (idempotent across close()->open() and deepcopy).
if "crop" in self.backend_metadata:
from sleap_io.io.video_reading import CropVideoBackend
self.backend = CropVideoBackend.wrap(
inner=self.backend,
crop=tuple(self.backend_metadata["crop"]),
fill=self.backend_metadata.get("crop_fill", 0),
)
replace_filename(new_filename, open=True)
¶
Update the filename of the video, optionally opening the backend.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
new_filename
|
str | Path | list[str] | list[Path]
|
New filename to set for the video. |
required |
open
|
bool
|
If |
True
|
Source code in sleap_io/model/video.py
def replace_filename(
self, new_filename: str | Path | list[str] | list[Path], open: bool = True
):
"""Update the filename of the video, optionally opening the backend.
Args:
new_filename: New filename to set for the video.
open: If `True` (the default), open the backend with the new filename. If
the new filename does not exist, no error is raised.
"""
if isinstance(new_filename, Path):
new_filename = new_filename.as_posix()
if isinstance(new_filename, list):
new_filename = [
p.as_posix() if isinstance(p, Path) else p for p in new_filename
]
# A relink to a different file makes the recorded shape/grayscale/fps in
# ``backend_metadata`` stale: they describe the OLD file but the new file
# may have a different resolution/channels/frame rate. They must not be
# serialized under the new filename (regression from #483, where
# ``save_slp(prefer_metadata=True)`` prefers these recorded values), so
# invalidate them on a real relink and let them be recomputed from the new
# backend. The no-relink path leaves metadata untouched so golden
# byte-identical saves stay byte-identical.
filename_changed = new_filename != self.filename
self.filename = new_filename
self.backend_metadata["filename"] = new_filename
# Invalidate any cached URL existence results for the previous filename.
self._exists_cache.clear()
if open:
if self.exists():
self.open()
else:
self.close()
# Drop stale metadata AFTER (re)opening: ``open()`` internally calls
# ``close()``, which would otherwise re-stamp the OLD backend's
# shape/grayscale/fps back into ``backend_metadata``.
if filename_changed:
for key in ("shape", "grayscale", "fps"):
self.backend_metadata.pop(key, None)
save(save_path, frame_inds=None, fps=None, video_kwargs=None)
¶
Save video frames to a new video file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
save_path
|
str | Path
|
Path to the new video file. Should end in MP4. |
required |
frame_inds
|
list[int] | ndarray | None
|
Frame indices to save. Can be specified as a list or array of frame integers. If not specified, saves all video frames. |
None
|
fps
|
float | None
|
Frames per second for the output video. If not specified, uses the source video's FPS if available, otherwise defaults to 30. |
None
|
video_kwargs
|
dict[str, Any] | None
|
A dictionary of keyword arguments to provide to
|
None
|
Returns:
| Type | Description |
|---|---|
Video
|
A new |
Source code in sleap_io/model/video.py
def save(
self,
save_path: str | Path,
frame_inds: list[int] | np.ndarray | None = None,
fps: float | None = None,
video_kwargs: dict[str, Any] | None = None,
) -> "Video":
"""Save video frames to a new video file.
Args:
save_path: Path to the new video file. Should end in MP4.
frame_inds: Frame indices to save. Can be specified as a list or array of
frame integers. If not specified, saves all video frames.
fps: Frames per second for the output video. If not specified, uses the
source video's FPS if available, otherwise defaults to 30.
video_kwargs: A dictionary of keyword arguments to provide to
`sio.save_video` for video compression.
Returns:
A new `Video` object pointing to the new video file.
"""
video_kwargs = {} if video_kwargs is None else video_kwargs.copy()
frame_inds = np.arange(len(self)) if frame_inds is None else frame_inds
# Use source video FPS if not explicitly specified
if fps is None:
fps = self.fps
if fps is not None and "fps" not in video_kwargs:
video_kwargs["fps"] = fps
with VideoWriter(save_path, **video_kwargs) as vw:
for frame_ind in frame_inds:
vw(self[frame_ind])
new_video = Video.from_filename(save_path, grayscale=self.grayscale)
return new_video
seconds_to_frame(seconds)
¶
Convert a timestamp in seconds to frame index.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
seconds
|
float
|
Time in seconds from video start. |
required |
Returns:
| Type | Description |
|---|---|
int | None
|
Zero-indexed frame number (rounded down), or None if FPS unknown. |
Source code in sleap_io/model/video.py
def seconds_to_frame(self, seconds: float) -> int | None:
"""Convert a timestamp in seconds to frame index.
Args:
seconds: Time in seconds from video start.
Returns:
Zero-indexed frame number (rounded down), or None if FPS unknown.
"""
if self.fps is None or self.fps <= 0:
return None
return int(seconds * self.fps)
set_video_plugin(plugin)
¶
Set the video plugin and reopen the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
plugin
|
str
|
Video plugin to use. One of "opencv", "FFMPEG", or "pyav". Also accepts aliases (case-insensitive). |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the video is not a MediaVideo type. |
Examples:
Source code in sleap_io/model/video.py
def set_video_plugin(self, plugin: str) -> None:
"""Set the video plugin and reopen the video.
Args:
plugin: Video plugin to use. One of "opencv", "FFMPEG", or "pyav".
Also accepts aliases (case-insensitive).
Raises:
ValueError: If the video is not a MediaVideo type.
Examples:
>>> video.set_video_plugin("opencv")
>>> video.set_video_plugin("CV2") # Same as "opencv"
"""
from sleap_io.io.video_reading import MediaVideo, normalize_plugin_name
if not self.filename.endswith(MediaVideo.EXTS):
raise ValueError(f"Cannot set plugin for non-media video: {self.filename}")
plugin = normalize_plugin_name(plugin)
# Close current backend if open
was_open = self.is_open
if was_open:
self.close()
# Update backend metadata
self.backend_metadata["plugin"] = plugin
# Reopen with new plugin if it was open
if was_open:
self.open()
to_crop_coords(points)
¶
Map source-frame (x, y) into this video's cropped frame.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
points
|
ndarray
|
Coordinate array of shape |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Coordinates translated into the cropped frame. If this video is not
cropped, a copy of |
Source code in sleap_io/model/video.py
def to_crop_coords(self, points: np.ndarray) -> np.ndarray:
"""Map source-frame ``(x, y)`` into this video's cropped frame.
Args:
points: Coordinate array of shape ``(..., 2)``. NaN values are
preserved.
Returns:
Coordinates translated into the cropped frame. If this video is not
cropped, a copy of ``points`` is returned unchanged.
"""
crop = self._crop_tuple()
return points.copy() if crop is None else crop_points(points, crop)
to_source_coords(points)
¶
Map cropped-frame (x, y) back to source-frame coordinates.
Inverse of :meth:to_crop_coords.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
points
|
ndarray
|
Coordinate array of shape |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Coordinates translated back to source coordinates. If this video is
not cropped, a copy of |
Source code in sleap_io/model/video.py
def to_source_coords(self, points: np.ndarray) -> np.ndarray:
"""Map cropped-frame ``(x, y)`` back to source-frame coordinates.
Inverse of :meth:`to_crop_coords`.
Args:
points: Coordinate array of shape ``(..., 2)``. NaN values are
preserved.
Returns:
Coordinates translated back to source coordinates. If this video is
not cropped, a copy of ``points`` is returned unchanged.
"""
crop = self._crop_tuple()
return points.copy() if crop is None else uncrop_points(points, crop)
VideoBackend
¶
Base class for video backends.
This class is not meant to be used directly. Instead, use the from_filename
constructor to create a backend instance.
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
Path to video file(s). |
|
grayscale |
Whether to force grayscale. If None, autodetect on first frame load. |
|
keep_open |
Whether to keep the video reader open between calls to read frames. If False, will close the reader after each call. If True (the default), it will keep the reader open and cache it for subsequent calls which may enhance the performance of reading multiple frames. |
|
fps |
Frames per second of the video. For MediaVideo, this is read from container metadata. For other backends (ImageVideo, HDF5Video, TiffVideo), this must be set explicitly or will be None. |
Methods:
| Name | Description |
|---|---|
__eq__ |
Method generated by attrs for class VideoBackend. |
__getitem__ |
Return a single frame or a list of frames from the video. |
__getstate__ |
Return state for pickling/deepcopy, dropping the open reader handle. |
__init__ |
Method generated by attrs for class VideoBackend. |
__len__ |
Return number of frames in the video. |
__repr__ |
Method generated by attrs for class VideoBackend. |
__setstate__ |
Restore state from pickling/deepcopy. |
close |
Release the cached open reader handle, if any. |
detect_grayscale |
Detect whether the video is grayscale. |
from_filename |
Create a VideoBackend from a filename. |
get_frame |
Read a single frame from the video. |
get_frames |
Read a list of frames from the video. |
has_frame |
Check if a frame index is contained in the video. |
read_test_frame |
Read a single frame from the video to test for grayscale. |
Source code in sleap_io/io/video_reading.py
@attrs.define
class VideoBackend:
"""Base class for video backends.
This class is not meant to be used directly. Instead, use the `from_filename`
constructor to create a backend instance.
Attributes:
filename: Path to video file(s).
grayscale: Whether to force grayscale. If None, autodetect on first frame load.
keep_open: Whether to keep the video reader open between calls to read frames.
If False, will close the reader after each call. If True (the default), it
will keep the reader open and cache it for subsequent calls which may
enhance the performance of reading multiple frames.
fps: Frames per second of the video. For MediaVideo, this is read from container
metadata. For other backends (ImageVideo, HDF5Video, TiffVideo), this must
be set explicitly or will be None.
"""
filename: str | Path | list[str] | list[Path]
grayscale: bool | None = None
keep_open: bool = True
_cached_shape: tuple[int, int, int, int] | None = None
_open_reader: object | None = None
_fps: float | None = None
@property
def fps(self) -> float | None:
"""Frames per second of the video.
Returns:
The FPS if known, or None if unavailable/unknown.
Notes:
For MediaVideo, this is read from container metadata.
For ImageVideo, HDF5Video, and TiffVideo, this must be set explicitly
or inherited from source_video.
"""
return self._fps
@fps.setter
def fps(self, value: float | None) -> None:
"""Set the FPS.
Args:
value: Frames per second. Must be positive if not None.
Raises:
ValueError: If value is not positive.
"""
if value is not None and value <= 0:
raise ValueError(f"FPS must be positive, got {value}")
self._fps = value
def __getstate__(self) -> dict:
"""Return state for pickling/deepcopy, dropping the open reader handle.
The cached ``_open_reader`` (e.g. an ``h5py.File`` or video container) is
not picklable and is reopened lazily on next access, so it is excluded.
"""
import attr
state = {a.name: getattr(self, a.name) for a in attr.fields(type(self))}
state["_open_reader"] = None
return state
def __setstate__(self, state: dict) -> None:
"""Restore state from pickling/deepcopy.
attrs slotted classes need ``object.__setattr__`` to set slots directly.
Validators are skipped, which is safe since state came from a valid object.
"""
for key, value in state.items():
object.__setattr__(self, key, value)
def close(self) -> None:
"""Release the cached open reader handle, if any.
Closes (``.close()``) or releases (``.release()`` for an OpenCV
``VideoCapture``) the cached ``_open_reader`` and drops the reference so
a long-lived backend does not leak the underlying file/container handle.
The reader is lazily reopened on the next read, so this is safe to call
between reads. A no-op when nothing is cached. Subclasses that hold
additional handles (e.g. :class:`HDF5Video`'s URL file-like) override
this and call ``super().close()``.
"""
reader = self._open_reader
self._open_reader = None
if reader is None:
return
# Every real reader is an h5py.File / imageio reader (``.close()``) or an
# OpenCV VideoCapture (``.release()``); the None case is purely defensive.
closer = getattr(reader, "close", None) or getattr(reader, "release", None)
if closer is None: # pragma: no cover - defensive: reader always closeable
return
try:
closer()
except Exception: # pragma: no cover - defensive: close should not raise
pass
@classmethod
def from_filename(
cls,
filename: str | list[str],
dataset: str | None = None,
grayscale: bool | None = None,
keep_open: bool = True,
url_headers: dict[str, str] | None = None,
url_stream_mode: str = "blockcache",
**kwargs,
) -> "VideoBackend":
"""Create a VideoBackend from a filename.
Args:
filename: Path to video file(s).
dataset: Name of dataset in HDF5 file.
grayscale: Whether to force grayscale. If None, autodetect on first frame
load.
keep_open: Whether to keep the video reader open between calls to read
frames. If False, will close the reader after each call. If True (the
default), it will keep the reader open and cache it for subsequent calls
which may enhance the performance of reading multiple frames.
url_headers: HTTP headers forwarded to the remote backend when
``filename`` is a URL (HDF5Video only). Set at construction so the
metadata probe is authenticated; ignored for local files and other
backends.
url_stream_mode: Remote streaming strategy for a URL-backed HDF5Video
(one of ``"blockcache"``/``"cache"``/``"filecache"``/``"download"``).
Ignored for local files and other backends.
**kwargs: Additional backend-specific arguments. These are filtered to only
include parameters that are valid for the specific backend being
created:
- For ImageVideo: plugin (str): Image plugin to use. One of "opencv"
or "imageio". Also accepts aliases (case-insensitive).
If None, uses global default if set, otherwise auto-detects.
- For MediaVideo: plugin (str): Video plugin to use. One of "opencv",
"FFMPEG", or "pyav". Also accepts aliases (case-insensitive).
If None, uses global default if set, otherwise auto-detects.
- For HDF5Video: input_format (str), frame_map (dict),
source_filename (str),
source_inds (np.ndarray), image_format (str). See HDF5Video for
details.
Returns:
VideoBackend subclass instance.
"""
if isinstance(filename, Path):
filename = filename.as_posix()
is_url = type(filename) is str and _remote._is_url(filename)
if is_url:
from sleap_io.io._gdrive import _is_gdrive_url
if _is_gdrive_url(filename):
# Drive download URLs carry no extension and Drive rejects the
# range/HEAD requests video decoding relies on, so streaming a
# Drive video is not supported. Drive *labels* (.slp) loading is
# supported via load_slp/load_file.
raise NotImplementedError(
"Loading videos directly from Google Drive URLs is not "
"supported (Drive download links carry no file extension and "
"reject the range requests video decoding needs). Download "
"the video file first, or load Drive .slp label files with "
f"load_slp/load_file. (URL: {_remote._redact_url(filename)})"
)
# Skip local-filesystem dir detection for URLs (``Path.is_dir`` on a URL
# is meaningless and would just return False, but avoid the syscall).
if type(filename) is str and not is_url and Path(filename).is_dir():
filename = ImageVideo.find_images(filename)
# Match extensions against the URL *path* (query/fragment stripped) for
# URLs, and the lowercased filename otherwise.
ext_token = _extension_token(filename) if type(filename) is str else ""
if type(filename) is list:
filename = [Path(f).as_posix() for f in filename]
return ImageVideo(
filename, grayscale=grayscale, **_get_valid_kwargs(ImageVideo, kwargs)
)
elif ext_token.endswith(("tif", "tiff")):
# Detect TIFF format
format_type, metadata = TiffVideo.detect_format(filename)
if format_type in ("multi_page", "rank3_video", "rank4_video"):
# Use TiffVideo for multi-page or multi-dimensional TIFFs
tiff_kwargs = _get_valid_kwargs(TiffVideo, kwargs)
# Add format if detected
if format_type in ("rank3_video", "rank4_video"):
tiff_kwargs["format"] = metadata.get("format")
return TiffVideo(
filename,
grayscale=grayscale,
keep_open=keep_open,
**tiff_kwargs,
)
else:
# Single-page TIFF, treat as regular image
return ImageVideo(
[filename],
grayscale=grayscale,
**_get_valid_kwargs(ImageVideo, kwargs),
)
elif ext_token.endswith(tuple(ext.lower() for ext in ImageVideo.EXTS)):
return ImageVideo(
[filename], grayscale=grayscale, **_get_valid_kwargs(ImageVideo, kwargs)
)
elif ext_token.endswith(".seq"):
from sleap_io.io.seq import SeqVideo
return SeqVideo(
filename,
grayscale=grayscale,
keep_open=keep_open,
**_get_valid_kwargs(SeqVideo, kwargs),
)
elif ext_token.endswith(tuple(ext.lower() for ext in MediaVideo.EXTS)):
media_kwargs = _get_valid_kwargs(MediaVideo, kwargs)
if is_url:
# Remote media videos are read via pyav (imageio's pyav plugin
# forwards http(s) URIs to ``av.open`` natively). Enforce the
# documented contract that only http/https URLs are supported:
# cloud schemes (s3/gs/gcs/az/abfs) are recognized as remote by
# ``_is_url`` but are not safe to hand to ``av.open``, so reject
# them cleanly here rather than letting the raw URL reach the
# decoder.
scheme = urllib.parse.urlparse(filename).scheme.lower()
if scheme not in ("http", "https"):
raise NotImplementedError(
"Remote video loading only supports http/https URLs; "
f"got scheme '{scheme}' for "
f"{_remote._redact_url(filename)}. Download the file "
"locally first."
)
# Remote media video is decoded by handing the raw URL to
# ``av.open`` (via imageio's pyav plugin), which has no hook for
# forwarding HTTP request headers or selecting an fsspec stream
# mode. Auth/streaming kwargs that work for remote .slp/.pkg.slp
# (HDF5Video) therefore cannot be honored here. Rather than
# silently drop them and return an unauthenticated backend,
# reject them with an actionable error. ``url_headers`` /
# ``url_stream_mode`` are the explicit ``from_filename``
# parameters; ``headers`` / ``stream_mode`` arrive via
# ``**kwargs`` (e.g. from ``load_video(url, headers=...)``).
if (
url_headers is not None
or url_stream_mode != "blockcache"
or kwargs.get("headers") is not None
or kwargs.get("stream_mode") not in (None, "auto")
):
raise ValueError(
"Remote media video cannot be authenticated with "
"'headers'/'url_headers' or configured with a stream "
"mode: it is decoded by handing the URL directly to "
"FFmpeg (via pyav), which does not support custom HTTP "
"headers or fsspec streaming. Use a pre-signed URL that "
"embeds credentials in the query string, or download "
"the file locally first. (These options do work for "
"remote .slp/.pkg.slp labels.) (URL: "
f"{_remote._redact_url(filename)})"
)
# Default to pyav when the caller did not request a specific
# plugin, and require the ``av`` package up front for a clear
# error.
if media_kwargs.get("plugin") is None:
media_kwargs["plugin"] = "pyav"
if (
normalize_plugin_name(media_kwargs["plugin"]) == "pyav"
and not _is_pyav_available()
):
# Defensive: ``av`` is required for remote loading and is
# always present in the test/CI environment, so this guard
# only fires for an install lacking the ``[pyav]`` extra.
raise ImportError( # pragma: no cover
"Loading videos from URLs requires the 'av' package "
"(pyav). Install with: pip install 'sleap-io[pyav]'. "
f"(URL: {_remote._redact_url(filename)})"
)
return MediaVideo(
filename,
grayscale=grayscale,
keep_open=keep_open,
**media_kwargs,
)
elif ext_token.endswith(tuple(ext.lower() for ext in HDF5Video.EXTS)):
# Pass ``url_headers`` / ``url_stream_mode`` explicitly (not via
# ``_get_valid_kwargs``, which keys on the underscored field *name*
# and would drop the alias) so the construction-time probe in
# ``HDF5Video.__attrs_post_init__`` is authenticated for remote URLs.
return HDF5Video(
filename,
dataset=dataset,
grayscale=grayscale,
keep_open=keep_open,
url_headers=url_headers,
url_stream_mode=url_stream_mode,
**_get_valid_kwargs(HDF5Video, kwargs),
)
else:
raise ValueError(f"Unknown video file type: {filename}")
def _read_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the video. Must be implemented in subclasses."""
raise NotImplementedError
def _read_frames(self, frame_inds: list) -> np.ndarray:
"""Read a list of frames from the video."""
return np.stack([self.get_frame(i) for i in frame_inds], axis=0)
def read_test_frame(self) -> np.ndarray:
"""Read a single frame from the video to test for grayscale.
Note:
This reads the frame at index 0. This may not be appropriate if the first
frame is not available in a given backend.
"""
return self._read_frame(0)
def detect_grayscale(self, test_img: np.ndarray | None = None) -> bool:
"""Detect whether the video is grayscale.
This works by reading in a test frame and comparing the first and last channel
for equality. It may fail in cases where, due to compression, the first and
last channels are not exactly the same.
Args:
test_img: Optional test image to use. If not provided, a test image will be
loaded via the `read_test_frame` method.
Returns:
Whether the video is grayscale. This value is also cached in the `grayscale`
attribute of the class.
"""
if test_img is None:
test_img = self.read_test_frame()
is_grayscale = np.array_equal(test_img[..., 0], test_img[..., -1])
self.grayscale = is_grayscale
return is_grayscale
@property
def num_frames(self) -> int:
"""Number of frames in the video. Must be implemented in subclasses."""
raise NotImplementedError
@property
def img_shape(self) -> tuple[int, int, int]:
"""Shape of a single frame in the video."""
height, width, channels = self.read_test_frame().shape
if self.grayscale is None:
self.detect_grayscale()
if self.grayscale is False:
channels = 3
elif self.grayscale is True:
channels = 1
return int(height), int(width), int(channels)
@property
def shape(self) -> tuple[int, int, int, int]:
"""Shape of the video as a tuple of `(frames, height, width, channels)`.
On first call, this will defer to `num_frames` and `img_shape` to determine the
full shape. This call may be expensive for some subclasses, so the result is
cached and returned on subsequent calls.
"""
if self._cached_shape is not None:
return self._cached_shape
else:
shape = (self.num_frames,) + self.img_shape
self._cached_shape = shape
return shape
@property
def frames(self) -> int:
"""Number of frames in the video."""
return self.shape[0]
def __len__(self) -> int:
"""Return number of frames in the video."""
return self.shape[0]
def has_frame(self, frame_idx: int) -> bool:
"""Check if a frame index is contained in the video.
Args:
frame_idx: Index of frame to check.
Returns:
`True` if the index is contained in the video, otherwise `False`.
"""
return frame_idx < len(self)
def get_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the video.
Args:
frame_idx: Index of frame to read.
Returns:
Frame as a numpy array of shape `(height, width, channels)` where the
`channels` dimension is 1 for grayscale videos and 3 for color videos.
Notes:
If the `grayscale` attribute is set to `True`, the `channels` dimension will
be reduced to 1 if an RGB frame is loaded from the backend.
If the `grayscale` attribute is set to `None`, the `grayscale` attribute
will be automatically set based on the first frame read.
See also: `get_frames`
"""
if not self.has_frame(frame_idx):
raise IndexError(f"Frame index {frame_idx} out of range.")
img = self._read_frame(frame_idx)
if self.grayscale is None:
self.detect_grayscale(img)
if self.grayscale:
img = img[..., [0]]
return img
def get_frames(self, frame_inds: list[int]) -> np.ndarray:
"""Read a list of frames from the video.
Depending on the backend implementation, this may be faster than reading frames
individually using `get_frame`.
Args:
frame_inds: List of frame indices to read.
Returns:
Frames as a numpy array of shape `(frames, height, width, channels)` where
`channels` dimension is 1 for grayscale videos and 3 for color videos.
Notes:
If the `grayscale` attribute is set to `True`, the `channels` dimension will
be reduced to 1 if an RGB frame is loaded from the backend.
If the `grayscale` attribute is set to `None`, the `grayscale` attribute
will be automatically set based on the first frame read.
See also: `get_frame`
"""
imgs = self._read_frames(frame_inds)
if self.grayscale is None:
self.detect_grayscale(imgs[0])
if self.grayscale:
imgs = imgs[..., [0]]
return imgs
def __getitem__(self, ind: int | list[int] | slice) -> np.ndarray:
"""Return a single frame or a list of frames from the video.
Args:
ind: Index or list of indices of frames to read.
Returns:
Frame or frames as a numpy array of shape `(height, width, channels)` if a
scalar index is provided, or `(frames, height, width, channels)` if a list
of indices is provided.
See also: get_frame, get_frames
"""
if np.isscalar(ind):
return self.get_frame(ind)
else:
if type(ind) is slice:
start = (ind.start or 0) % len(self)
stop = ind.stop or len(self)
if stop < 0:
stop = len(self) + stop
step = ind.step or 1
ind = range(start, stop, step)
return self.get_frames(ind)
__annotations__ = {'filename': 'str | Path | list[str] | list[Path]', 'grayscale': 'bool | None', 'keep_open': 'bool', '_cached_shape': 'tuple[int, int, int, int] | None', '_open_reader': 'object | None', '_fps': 'float | None'}
class-attribute
¶
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
__attrs_own_setattr__ = False
class-attribute
¶
Returns True when the argument is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.
__attrs_props__ = ClassProps(is_exception=False, is_slotted=True, has_weakref_slot=True, is_frozen=False, kw_only=<KeywordOnly.NO: 'no'>, collected_fields_by_mro=True, added_init=True, added_repr=True, added_eq=True, added_ordering=False, hashability=<Hashability.UNHASHABLE: 'unhashable'>, added_match_args=True, added_str=False, added_pickling=False, on_setattr_hook=<function pipe.<locals>.wrapped_pipe at 0x7f08471ce840>, field_transformer=None)
class-attribute
¶
Effective class properties as derived from parameters to attr.s() or
define() decorators.
This is the same data structure that attrs uses internally to decide how to construct the final class.
Warning:
This feature is currently **experimental** and is not covered by our
strict backwards-compatibility guarantees.
Attributes:
| Name | Type | Description |
|---|---|---|
is_exception |
bool
|
Whether the class is treated as an exception class. |
is_slotted |
bool
|
Whether the class is |
has_weakref_slot |
bool
|
Whether the class has a slot for weak references. |
is_frozen |
bool
|
Whether the class is frozen. |
kw_only |
KeywordOnly
|
Whether / how the class enforces keyword-only arguments on the
|
collected_fields_by_mro |
bool
|
Whether the class fields were collected by method resolution order.
That is, correctly but unlike |
added_init |
bool
|
Whether the class has an attrs-generated |
added_repr |
bool
|
Whether the class has an attrs-generated |
added_eq |
bool
|
Whether the class has attrs-generated equality methods. |
added_ordering |
bool
|
Whether the class has attrs-generated ordering methods. |
hashability |
Hashability
|
How |
added_match_args |
bool
|
Whether the class supports positional |
added_str |
bool
|
Whether the class has an attrs-generated |
added_pickling |
bool
|
Whether the class has attrs-generated |
on_setattr_hook |
Callable[[Any, Attribute[Any], Any], Any] | None
|
The class's |
field_transformer |
Callable[[Attribute[Any]], Attribute[Any]] | None
|
The class's |
.. versionadded:: 25.4.0
__doc__ = 'Base class for video backends.\n\nThis class is not meant to be used directly. Instead, use the `from_filename`\nconstructor to create a backend instance.\n\nAttributes:\n filename: Path to video file(s).\n grayscale: Whether to force grayscale. If None, autodetect on first frame load.\n keep_open: Whether to keep the video reader open between calls to read frames.\n If False, will close the reader after each call. If True (the default), it\n will keep the reader open and cache it for subsequent calls which may\n enhance the performance of reading multiple frames.\n fps: Frames per second of the video. For MediaVideo, this is read from container\n metadata. For other backends (ImageVideo, HDF5Video, TiffVideo), this must\n be set explicitly or will be None.\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 365
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__match_args__ = ('filename', 'grayscale', 'keep_open', '_cached_shape', '_open_reader', '_fps')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__module__ = 'sleap_io.io.video_reading'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__slots__ = ('filename', 'grayscale', 'keep_open', '_cached_shape', '_open_reader', '_fps', '__weakref__')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__static_attributes__ = ('_cached_shape', '_fps', '_open_reader', 'grayscale')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__weakref__
property
¶
list of weak references to the object
fps
property
¶
Frames per second of the video.
Returns:
| Type | Description |
|---|---|
|
The FPS if known, or None if unavailable/unknown. |
Notes
For MediaVideo, this is read from container metadata. For ImageVideo, HDF5Video, and TiffVideo, this must be set explicitly or inherited from source_video.
frames
property
¶
Number of frames in the video.
img_shape
property
¶
Shape of a single frame in the video.
num_frames
property
¶
Number of frames in the video. Must be implemented in subclasses.
shape
property
¶
Shape of the video as a tuple of (frames, height, width, channels).
On first call, this will defer to num_frames and img_shape to determine the
full shape. This call may be expensive for some subclasses, so the result is
cached and returned on subsequent calls.
__eq__(other)
¶
Method generated by attrs for class VideoBackend.
__getitem__(ind)
¶
Return a single frame or a list of frames from the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ind
|
int | list[int] | slice
|
Index or list of indices of frames to read. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Frame or frames as a numpy array of shape |
See also: get_frame, get_frames
Source code in sleap_io/io/video_reading.py
def __getitem__(self, ind: int | list[int] | slice) -> np.ndarray:
"""Return a single frame or a list of frames from the video.
Args:
ind: Index or list of indices of frames to read.
Returns:
Frame or frames as a numpy array of shape `(height, width, channels)` if a
scalar index is provided, or `(frames, height, width, channels)` if a list
of indices is provided.
See also: get_frame, get_frames
"""
if np.isscalar(ind):
return self.get_frame(ind)
else:
if type(ind) is slice:
start = (ind.start or 0) % len(self)
stop = ind.stop or len(self)
if stop < 0:
stop = len(self) + stop
step = ind.step or 1
ind = range(start, stop, step)
return self.get_frames(ind)
__getstate__()
¶
Return state for pickling/deepcopy, dropping the open reader handle.
The cached _open_reader (e.g. an h5py.File or video container) is
not picklable and is reopened lazily on next access, so it is excluded.
Source code in sleap_io/io/video_reading.py
def __getstate__(self) -> dict:
"""Return state for pickling/deepcopy, dropping the open reader handle.
The cached ``_open_reader`` (e.g. an ``h5py.File`` or video container) is
not picklable and is reopened lazily on next access, so it is excluded.
"""
import attr
state = {a.name: getattr(self, a.name) for a in attr.fields(type(self))}
state["_open_reader"] = None
return state
__init__(filename, grayscale=None, keep_open=True, cached_shape=None, open_reader=None, fps=None)
¶
__len__()
¶
__repr__()
¶
Method generated by attrs for class VideoBackend.
__setstate__(state)
¶
Restore state from pickling/deepcopy.
attrs slotted classes need object.__setattr__ to set slots directly.
Validators are skipped, which is safe since state came from a valid object.
Source code in sleap_io/io/video_reading.py
def __setstate__(self, state: dict) -> None:
"""Restore state from pickling/deepcopy.
attrs slotted classes need ``object.__setattr__`` to set slots directly.
Validators are skipped, which is safe since state came from a valid object.
"""
for key, value in state.items():
object.__setattr__(self, key, value)
close()
¶
Release the cached open reader handle, if any.
Closes (.close()) or releases (.release() for an OpenCV
VideoCapture) the cached _open_reader and drops the reference so
a long-lived backend does not leak the underlying file/container handle.
The reader is lazily reopened on the next read, so this is safe to call
between reads. A no-op when nothing is cached. Subclasses that hold
additional handles (e.g. :class:HDF5Video's URL file-like) override
this and call super().close().
Source code in sleap_io/io/video_reading.py
def close(self) -> None:
"""Release the cached open reader handle, if any.
Closes (``.close()``) or releases (``.release()`` for an OpenCV
``VideoCapture``) the cached ``_open_reader`` and drops the reference so
a long-lived backend does not leak the underlying file/container handle.
The reader is lazily reopened on the next read, so this is safe to call
between reads. A no-op when nothing is cached. Subclasses that hold
additional handles (e.g. :class:`HDF5Video`'s URL file-like) override
this and call ``super().close()``.
"""
reader = self._open_reader
self._open_reader = None
if reader is None:
return
# Every real reader is an h5py.File / imageio reader (``.close()``) or an
# OpenCV VideoCapture (``.release()``); the None case is purely defensive.
closer = getattr(reader, "close", None) or getattr(reader, "release", None)
if closer is None: # pragma: no cover - defensive: reader always closeable
return
try:
closer()
except Exception: # pragma: no cover - defensive: close should not raise
pass
detect_grayscale(test_img=None)
¶
Detect whether the video is grayscale.
This works by reading in a test frame and comparing the first and last channel for equality. It may fail in cases where, due to compression, the first and last channels are not exactly the same.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
test_img
|
ndarray | None
|
Optional test image to use. If not provided, a test image will be
loaded via the |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
Whether the video is grayscale. This value is also cached in the |
Source code in sleap_io/io/video_reading.py
def detect_grayscale(self, test_img: np.ndarray | None = None) -> bool:
"""Detect whether the video is grayscale.
This works by reading in a test frame and comparing the first and last channel
for equality. It may fail in cases where, due to compression, the first and
last channels are not exactly the same.
Args:
test_img: Optional test image to use. If not provided, a test image will be
loaded via the `read_test_frame` method.
Returns:
Whether the video is grayscale. This value is also cached in the `grayscale`
attribute of the class.
"""
if test_img is None:
test_img = self.read_test_frame()
is_grayscale = np.array_equal(test_img[..., 0], test_img[..., -1])
self.grayscale = is_grayscale
return is_grayscale
from_filename(filename, dataset=None, grayscale=None, keep_open=True, url_headers=None, url_stream_mode='blockcache', **kwargs)
classmethod
¶
Create a VideoBackend from a filename.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str | list[str]
|
Path to video file(s). |
required |
dataset
|
str | None
|
Name of dataset in HDF5 file. |
None
|
grayscale
|
bool | None
|
Whether to force grayscale. If None, autodetect on first frame load. |
None
|
keep_open
|
bool
|
Whether to keep the video reader open between calls to read frames. If False, will close the reader after each call. If True (the default), it will keep the reader open and cache it for subsequent calls which may enhance the performance of reading multiple frames. |
True
|
url_headers
|
dict[str, str] | None
|
HTTP headers forwarded to the remote backend when
|
None
|
url_stream_mode
|
str
|
Remote streaming strategy for a URL-backed HDF5Video
(one of |
'blockcache'
|
**kwargs
|
Additional backend-specific arguments. These are filtered to only include parameters that are valid for the specific backend being created: - For ImageVideo: plugin (str): Image plugin to use. One of "opencv" or "imageio". Also accepts aliases (case-insensitive). If None, uses global default if set, otherwise auto-detects. - For MediaVideo: plugin (str): Video plugin to use. One of "opencv", "FFMPEG", or "pyav". Also accepts aliases (case-insensitive). If None, uses global default if set, otherwise auto-detects. - For HDF5Video: input_format (str), frame_map (dict), source_filename (str), source_inds (np.ndarray), image_format (str). See HDF5Video for details. |
required |
Returns:
| Type | Description |
|---|---|
VideoBackend
|
VideoBackend subclass instance. |
Source code in sleap_io/io/video_reading.py
@classmethod
def from_filename(
cls,
filename: str | list[str],
dataset: str | None = None,
grayscale: bool | None = None,
keep_open: bool = True,
url_headers: dict[str, str] | None = None,
url_stream_mode: str = "blockcache",
**kwargs,
) -> "VideoBackend":
"""Create a VideoBackend from a filename.
Args:
filename: Path to video file(s).
dataset: Name of dataset in HDF5 file.
grayscale: Whether to force grayscale. If None, autodetect on first frame
load.
keep_open: Whether to keep the video reader open between calls to read
frames. If False, will close the reader after each call. If True (the
default), it will keep the reader open and cache it for subsequent calls
which may enhance the performance of reading multiple frames.
url_headers: HTTP headers forwarded to the remote backend when
``filename`` is a URL (HDF5Video only). Set at construction so the
metadata probe is authenticated; ignored for local files and other
backends.
url_stream_mode: Remote streaming strategy for a URL-backed HDF5Video
(one of ``"blockcache"``/``"cache"``/``"filecache"``/``"download"``).
Ignored for local files and other backends.
**kwargs: Additional backend-specific arguments. These are filtered to only
include parameters that are valid for the specific backend being
created:
- For ImageVideo: plugin (str): Image plugin to use. One of "opencv"
or "imageio". Also accepts aliases (case-insensitive).
If None, uses global default if set, otherwise auto-detects.
- For MediaVideo: plugin (str): Video plugin to use. One of "opencv",
"FFMPEG", or "pyav". Also accepts aliases (case-insensitive).
If None, uses global default if set, otherwise auto-detects.
- For HDF5Video: input_format (str), frame_map (dict),
source_filename (str),
source_inds (np.ndarray), image_format (str). See HDF5Video for
details.
Returns:
VideoBackend subclass instance.
"""
if isinstance(filename, Path):
filename = filename.as_posix()
is_url = type(filename) is str and _remote._is_url(filename)
if is_url:
from sleap_io.io._gdrive import _is_gdrive_url
if _is_gdrive_url(filename):
# Drive download URLs carry no extension and Drive rejects the
# range/HEAD requests video decoding relies on, so streaming a
# Drive video is not supported. Drive *labels* (.slp) loading is
# supported via load_slp/load_file.
raise NotImplementedError(
"Loading videos directly from Google Drive URLs is not "
"supported (Drive download links carry no file extension and "
"reject the range requests video decoding needs). Download "
"the video file first, or load Drive .slp label files with "
f"load_slp/load_file. (URL: {_remote._redact_url(filename)})"
)
# Skip local-filesystem dir detection for URLs (``Path.is_dir`` on a URL
# is meaningless and would just return False, but avoid the syscall).
if type(filename) is str and not is_url and Path(filename).is_dir():
filename = ImageVideo.find_images(filename)
# Match extensions against the URL *path* (query/fragment stripped) for
# URLs, and the lowercased filename otherwise.
ext_token = _extension_token(filename) if type(filename) is str else ""
if type(filename) is list:
filename = [Path(f).as_posix() for f in filename]
return ImageVideo(
filename, grayscale=grayscale, **_get_valid_kwargs(ImageVideo, kwargs)
)
elif ext_token.endswith(("tif", "tiff")):
# Detect TIFF format
format_type, metadata = TiffVideo.detect_format(filename)
if format_type in ("multi_page", "rank3_video", "rank4_video"):
# Use TiffVideo for multi-page or multi-dimensional TIFFs
tiff_kwargs = _get_valid_kwargs(TiffVideo, kwargs)
# Add format if detected
if format_type in ("rank3_video", "rank4_video"):
tiff_kwargs["format"] = metadata.get("format")
return TiffVideo(
filename,
grayscale=grayscale,
keep_open=keep_open,
**tiff_kwargs,
)
else:
# Single-page TIFF, treat as regular image
return ImageVideo(
[filename],
grayscale=grayscale,
**_get_valid_kwargs(ImageVideo, kwargs),
)
elif ext_token.endswith(tuple(ext.lower() for ext in ImageVideo.EXTS)):
return ImageVideo(
[filename], grayscale=grayscale, **_get_valid_kwargs(ImageVideo, kwargs)
)
elif ext_token.endswith(".seq"):
from sleap_io.io.seq import SeqVideo
return SeqVideo(
filename,
grayscale=grayscale,
keep_open=keep_open,
**_get_valid_kwargs(SeqVideo, kwargs),
)
elif ext_token.endswith(tuple(ext.lower() for ext in MediaVideo.EXTS)):
media_kwargs = _get_valid_kwargs(MediaVideo, kwargs)
if is_url:
# Remote media videos are read via pyav (imageio's pyav plugin
# forwards http(s) URIs to ``av.open`` natively). Enforce the
# documented contract that only http/https URLs are supported:
# cloud schemes (s3/gs/gcs/az/abfs) are recognized as remote by
# ``_is_url`` but are not safe to hand to ``av.open``, so reject
# them cleanly here rather than letting the raw URL reach the
# decoder.
scheme = urllib.parse.urlparse(filename).scheme.lower()
if scheme not in ("http", "https"):
raise NotImplementedError(
"Remote video loading only supports http/https URLs; "
f"got scheme '{scheme}' for "
f"{_remote._redact_url(filename)}. Download the file "
"locally first."
)
# Remote media video is decoded by handing the raw URL to
# ``av.open`` (via imageio's pyav plugin), which has no hook for
# forwarding HTTP request headers or selecting an fsspec stream
# mode. Auth/streaming kwargs that work for remote .slp/.pkg.slp
# (HDF5Video) therefore cannot be honored here. Rather than
# silently drop them and return an unauthenticated backend,
# reject them with an actionable error. ``url_headers`` /
# ``url_stream_mode`` are the explicit ``from_filename``
# parameters; ``headers`` / ``stream_mode`` arrive via
# ``**kwargs`` (e.g. from ``load_video(url, headers=...)``).
if (
url_headers is not None
or url_stream_mode != "blockcache"
or kwargs.get("headers") is not None
or kwargs.get("stream_mode") not in (None, "auto")
):
raise ValueError(
"Remote media video cannot be authenticated with "
"'headers'/'url_headers' or configured with a stream "
"mode: it is decoded by handing the URL directly to "
"FFmpeg (via pyav), which does not support custom HTTP "
"headers or fsspec streaming. Use a pre-signed URL that "
"embeds credentials in the query string, or download "
"the file locally first. (These options do work for "
"remote .slp/.pkg.slp labels.) (URL: "
f"{_remote._redact_url(filename)})"
)
# Default to pyav when the caller did not request a specific
# plugin, and require the ``av`` package up front for a clear
# error.
if media_kwargs.get("plugin") is None:
media_kwargs["plugin"] = "pyav"
if (
normalize_plugin_name(media_kwargs["plugin"]) == "pyav"
and not _is_pyav_available()
):
# Defensive: ``av`` is required for remote loading and is
# always present in the test/CI environment, so this guard
# only fires for an install lacking the ``[pyav]`` extra.
raise ImportError( # pragma: no cover
"Loading videos from URLs requires the 'av' package "
"(pyav). Install with: pip install 'sleap-io[pyav]'. "
f"(URL: {_remote._redact_url(filename)})"
)
return MediaVideo(
filename,
grayscale=grayscale,
keep_open=keep_open,
**media_kwargs,
)
elif ext_token.endswith(tuple(ext.lower() for ext in HDF5Video.EXTS)):
# Pass ``url_headers`` / ``url_stream_mode`` explicitly (not via
# ``_get_valid_kwargs``, which keys on the underscored field *name*
# and would drop the alias) so the construction-time probe in
# ``HDF5Video.__attrs_post_init__`` is authenticated for remote URLs.
return HDF5Video(
filename,
dataset=dataset,
grayscale=grayscale,
keep_open=keep_open,
url_headers=url_headers,
url_stream_mode=url_stream_mode,
**_get_valid_kwargs(HDF5Video, kwargs),
)
else:
raise ValueError(f"Unknown video file type: {filename}")
get_frame(frame_idx)
¶
Read a single frame from the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Index of frame to read. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Frame as a numpy array of shape |
Notes
If the grayscale attribute is set to True, the channels dimension will
be reduced to 1 if an RGB frame is loaded from the backend.
If the grayscale attribute is set to None, the grayscale attribute
will be automatically set based on the first frame read.
See also: get_frames
Source code in sleap_io/io/video_reading.py
def get_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the video.
Args:
frame_idx: Index of frame to read.
Returns:
Frame as a numpy array of shape `(height, width, channels)` where the
`channels` dimension is 1 for grayscale videos and 3 for color videos.
Notes:
If the `grayscale` attribute is set to `True`, the `channels` dimension will
be reduced to 1 if an RGB frame is loaded from the backend.
If the `grayscale` attribute is set to `None`, the `grayscale` attribute
will be automatically set based on the first frame read.
See also: `get_frames`
"""
if not self.has_frame(frame_idx):
raise IndexError(f"Frame index {frame_idx} out of range.")
img = self._read_frame(frame_idx)
if self.grayscale is None:
self.detect_grayscale(img)
if self.grayscale:
img = img[..., [0]]
return img
get_frames(frame_inds)
¶
Read a list of frames from the video.
Depending on the backend implementation, this may be faster than reading frames
individually using get_frame.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_inds
|
list[int]
|
List of frame indices to read. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Frames as a numpy array of shape |
Notes
If the grayscale attribute is set to True, the channels dimension will
be reduced to 1 if an RGB frame is loaded from the backend.
If the grayscale attribute is set to None, the grayscale attribute
will be automatically set based on the first frame read.
See also: get_frame
Source code in sleap_io/io/video_reading.py
def get_frames(self, frame_inds: list[int]) -> np.ndarray:
"""Read a list of frames from the video.
Depending on the backend implementation, this may be faster than reading frames
individually using `get_frame`.
Args:
frame_inds: List of frame indices to read.
Returns:
Frames as a numpy array of shape `(frames, height, width, channels)` where
`channels` dimension is 1 for grayscale videos and 3 for color videos.
Notes:
If the `grayscale` attribute is set to `True`, the `channels` dimension will
be reduced to 1 if an RGB frame is loaded from the backend.
If the `grayscale` attribute is set to `None`, the `grayscale` attribute
will be automatically set based on the first frame read.
See also: `get_frame`
"""
imgs = self._read_frames(frame_inds)
if self.grayscale is None:
self.detect_grayscale(imgs[0])
if self.grayscale:
imgs = imgs[..., [0]]
return imgs
has_frame(frame_idx)
¶
Check if a frame index is contained in the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Index of frame to check. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
read_test_frame()
¶
Read a single frame from the video to test for grayscale.
Note
This reads the frame at index 0. This may not be appropriate if the first frame is not available in a given backend.
VideoWriter
¶
Simple video writer using imageio and FFMPEG.
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
Path to output video file. |
|
fps |
Frames per second. Defaults to 30. |
|
pixelformat |
Pixel format for video. Defaults to "yuv420p". |
|
codec |
Codec to use for encoding. Defaults to "libx264". |
|
crf |
Constant rate factor to control lossiness of video. Values go from 2 to 32, with numbers in the 18 to 30 range being most common. Lower values mean less compressed/higher quality. Defaults to 25. No effect if codec is not "libx264". |
|
preset |
H264 encoding preset. Defaults to "superfast". No effect if codec is not "libx264". |
|
keyframe_interval |
Interval between keyframes in seconds. If None, uses encoder default. Lower values improve seeking but increase file size. Defaults to None. |
|
no_audio |
If True, strips audio from the output. Defaults to False. |
|
output_params |
Additional output parameters for FFMPEG. This should be a list of
strings corresponding to command line arguments for FFMPEG and libx264. Use
|
Notes
This class can be used as a context manager to ensure the video is properly closed after writing. For example:
Methods:
| Name | Description |
|---|---|
__call__ |
Write a frame to the video. |
__enter__ |
Context manager entry. |
__eq__ |
Method generated by attrs for class VideoWriter. |
__exit__ |
Context manager exit. |
__init__ |
Method generated by attrs for class VideoWriter. |
__repr__ |
Method generated by attrs for class VideoWriter. |
__setattr__ |
Method generated by attrs for class VideoWriter. |
build_output_params |
Build the output parameters for FFMPEG. |
close |
Close the video writer. |
open |
Open the video writer. |
write_frame |
Write a frame to the video. |
Source code in sleap_io/io/video_writing.py
@attrs.define
class VideoWriter:
"""Simple video writer using imageio and FFMPEG.
Attributes:
filename: Path to output video file.
fps: Frames per second. Defaults to 30.
pixelformat: Pixel format for video. Defaults to "yuv420p".
codec: Codec to use for encoding. Defaults to "libx264".
crf: Constant rate factor to control lossiness of video. Values go from 2 to 32,
with numbers in the 18 to 30 range being most common. Lower values mean less
compressed/higher quality. Defaults to 25. No effect if codec is not
"libx264".
preset: H264 encoding preset. Defaults to "superfast". No effect if codec is not
"libx264".
keyframe_interval: Interval between keyframes in seconds. If None, uses encoder
default. Lower values improve seeking but increase file size. Defaults to
None.
no_audio: If True, strips audio from the output. Defaults to False.
output_params: Additional output parameters for FFMPEG. This should be a list of
strings corresponding to command line arguments for FFMPEG and libx264. Use
`ffmpeg -h encoder=libx264` to see all options for libx264 output_params.
Notes:
This class can be used as a context manager to ensure the video is properly
closed after writing. For example:
```python
with VideoWriter("output.mp4") as writer:
for frame in frames:
writer(frame)
```
"""
filename: Path = attrs.field(converter=Path)
fps: float = 30
pixelformat: str = "yuv420p"
codec: str = "libx264"
crf: int = 25
preset: str = "superfast"
keyframe_interval: float | None = None
no_audio: bool = False
output_params: list[str] = attrs.field(factory=list)
_writer: "imageio.plugins.ffmpeg.FfmpegFormat.Writer | None" = None
def build_output_params(self) -> list[str]:
"""Build the output parameters for FFMPEG."""
output_params = []
if self.codec == "libx264":
output_params.extend(
[
"-crf",
str(self.crf),
"-preset",
self.preset,
]
)
# Add keyframe interval (GOP size)
if self.keyframe_interval is not None:
gop_size = max(1, int(self.fps * self.keyframe_interval))
output_params.extend(["-g", str(gop_size)])
# Strip audio if requested
if self.no_audio:
output_params.extend(["-an"])
return output_params + self.output_params
def open(self):
"""Open the video writer."""
self.close()
self.filename.parent.mkdir(parents=True, exist_ok=True)
self._writer = iio_v2.get_writer(
self.filename.as_posix(),
format="FFMPEG",
fps=self.fps,
codec=self.codec,
pixelformat=self.pixelformat,
output_params=self.build_output_params(),
# Disable imageio's auto-scaling for non-divisible frame sizes.
# We handle padding manually in write_frame() to preserve coordinates.
macro_block_size=1,
)
def close(self):
"""Close the video writer."""
if self._writer is not None:
self._writer.close()
self._writer = None
def write_frame(self, frame: np.ndarray):
"""Write a frame to the video.
Args:
frame: Frame to write to video. Should be a 2D or 3D numpy array with
dimensions (height, width) or (height, width, channels).
Notes:
For libx264 codec, frames are automatically padded to dimensions divisible
by 16 (the macro block size). Padding is only added to the bottom and right
edges to preserve coordinate alignment.
"""
if self._writer is None:
self.open()
if self.codec == "libx264":
frame = _pad_to_macro_block(frame, macro_block_size=16)
self._writer.append_data(frame)
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> bool | None:
"""Context manager exit."""
self.close()
return False
def __call__(self, frame: np.ndarray):
"""Write a frame to the video.
Args:
frame: Frame to write to video. Should be a 2D or 3D numpy array with
dimensions (height, width) or (height, width, channels).
"""
self.write_frame(frame)
__annotations__ = {'filename': 'Path', 'fps': 'float', 'pixelformat': 'str', 'codec': 'str', 'crf': 'int', 'preset': 'str', 'keyframe_interval': 'float | None', 'no_audio': 'bool', 'output_params': 'list[str]', '_writer': "'imageio.plugins.ffmpeg.FfmpegFormat.Writer | None'"}
class-attribute
¶
dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
__attrs_own_setattr__ = True
class-attribute
¶
Returns True when the argument is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.
__attrs_props__ = ClassProps(is_exception=False, is_slotted=True, has_weakref_slot=True, is_frozen=False, kw_only=<KeywordOnly.NO: 'no'>, collected_fields_by_mro=True, added_init=True, added_repr=True, added_eq=True, added_ordering=False, hashability=<Hashability.UNHASHABLE: 'unhashable'>, added_match_args=True, added_str=False, added_pickling=True, on_setattr_hook=<function pipe.<locals>.wrapped_pipe at 0x7f08471ce840>, field_transformer=None)
class-attribute
¶
Effective class properties as derived from parameters to attr.s() or
define() decorators.
This is the same data structure that attrs uses internally to decide how to construct the final class.
Warning:
This feature is currently **experimental** and is not covered by our
strict backwards-compatibility guarantees.
Attributes:
| Name | Type | Description |
|---|---|---|
is_exception |
bool
|
Whether the class is treated as an exception class. |
is_slotted |
bool
|
Whether the class is |
has_weakref_slot |
bool
|
Whether the class has a slot for weak references. |
is_frozen |
bool
|
Whether the class is frozen. |
kw_only |
KeywordOnly
|
Whether / how the class enforces keyword-only arguments on the
|
collected_fields_by_mro |
bool
|
Whether the class fields were collected by method resolution order.
That is, correctly but unlike |
added_init |
bool
|
Whether the class has an attrs-generated |
added_repr |
bool
|
Whether the class has an attrs-generated |
added_eq |
bool
|
Whether the class has attrs-generated equality methods. |
added_ordering |
bool
|
Whether the class has attrs-generated ordering methods. |
hashability |
Hashability
|
How |
added_match_args |
bool
|
Whether the class supports positional |
added_str |
bool
|
Whether the class has an attrs-generated |
added_pickling |
bool
|
Whether the class has attrs-generated |
on_setattr_hook |
Callable[[Any, Attribute[Any], Any], Any] | None
|
The class's |
field_transformer |
Callable[[Attribute[Any]], Attribute[Any]] | None
|
The class's |
.. versionadded:: 25.4.0
__doc__ = 'Simple video writer using imageio and FFMPEG.\n\nAttributes:\n filename: Path to output video file.\n fps: Frames per second. Defaults to 30.\n pixelformat: Pixel format for video. Defaults to "yuv420p".\n codec: Codec to use for encoding. Defaults to "libx264".\n crf: Constant rate factor to control lossiness of video. Values go from 2 to 32,\n with numbers in the 18 to 30 range being most common. Lower values mean less\n compressed/higher quality. Defaults to 25. No effect if codec is not\n "libx264".\n preset: H264 encoding preset. Defaults to "superfast". No effect if codec is not\n "libx264".\n keyframe_interval: Interval between keyframes in seconds. If None, uses encoder\n default. Lower values improve seeking but increase file size. Defaults to\n None.\n no_audio: If True, strips audio from the output. Defaults to False.\n output_params: Additional output parameters for FFMPEG. This should be a list of\n strings corresponding to command line arguments for FFMPEG and libx264. Use\n `ffmpeg -h encoder=libx264` to see all options for libx264 output_params.\n\nNotes:\n This class can be used as a context manager to ensure the video is properly\n closed after writing. For example:\n\n ```python\n with VideoWriter("output.mp4") as writer:\n for frame in frames:\n writer(frame)\n ```\n'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__firstlineno__ = 46
class-attribute
¶
int([x]) -> integer int(x, base=10) -> integer
Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.
If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.
int('0b100', base=0) 4
__match_args__ = ('filename', 'fps', 'pixelformat', 'codec', 'crf', 'preset', 'keyframe_interval', 'no_audio', 'output_params', '_writer')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__module__ = 'sleap_io.io.video_writing'
class-attribute
¶
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.
__slots__ = ('filename', 'fps', 'pixelformat', 'codec', 'crf', 'preset', 'keyframe_interval', 'no_audio', 'output_params', '_writer', '__weakref__')
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__static_attributes__ = ('_writer',)
class-attribute
¶
Built-in immutable sequence.
If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.
If the argument is a tuple, the return value is the same object.
__weakref__
property
¶
list of weak references to the object
__call__(frame)
¶
Write a frame to the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame
|
ndarray
|
Frame to write to video. Should be a 2D or 3D numpy array with dimensions (height, width) or (height, width, channels). |
required |
__enter__()
¶
__eq__(other)
¶
Method generated by attrs for class VideoWriter.
Source code in sleap_io/io/video_writing.py
This preserves coordinate alignment by only adding padding to the bottom and right
edges of the frame. Without this, encoding with x264 may scale or pad symmetrically,
causing coordinate shifts.
Args:
frame: Frame to pad. Should be a 2D or 3D numpy array with dimensions
(height, width) or (height, width, channels).
macro_block_size: Block size to align to. Defaults to 16 for x264.
Returns:
Padded frame with dimensions divisible by macro_block_size, or the original
frame if no padding is needed.
"""
h, w = frame.shape[:2]
__exit__(exc_type, exc_value, traceback)
¶
__init__(filename, fps=30, pixelformat='yuv420p', codec='libx264', crf=25, preset='superfast', keyframe_interval=None, no_audio=False, output_params=NOTHING, writer=None)
¶
Method generated by attrs for class VideoWriter.
Source code in sleap_io/io/video_writing.py
# Calculate padding needed (only bottom/right)
pad_h = (macro_block_size - (h % macro_block_size)) % macro_block_size
pad_w = (macro_block_size - (w % macro_block_size)) % macro_block_size
if pad_h == 0 and pad_w == 0:
return frame
# Pad only bottom and right
if frame.ndim == 2:
return np.pad(frame, ((0, pad_h), (0, pad_w)), mode="constant")
else:
return np.pad(frame, ((0, pad_h), (0, pad_w), (0, 0)), mode="constant")
__repr__()
¶
Method generated by attrs for class VideoWriter.
Source code in sleap_io/io/video_writing.py
"""Utilities for writing videos."""
from __future__ import annotations
from pathlib import Path
from types import TracebackType
import attrs
import imageio
import imageio.v2 as iio_v2
import numpy as np
def _pad_to_macro_block(frame: np.ndarray, macro_block_size: int = 16) -> np.ndarray:
"""Pad frame to be divisible by macro_block_size, padding only bottom/right.
__setattr__(name, val)
¶
Method generated by attrs for class VideoWriter.
build_output_params()
¶
Build the output parameters for FFMPEG.
Source code in sleap_io/io/video_writing.py
def build_output_params(self) -> list[str]:
"""Build the output parameters for FFMPEG."""
output_params = []
if self.codec == "libx264":
output_params.extend(
[
"-crf",
str(self.crf),
"-preset",
self.preset,
]
)
# Add keyframe interval (GOP size)
if self.keyframe_interval is not None:
gop_size = max(1, int(self.fps * self.keyframe_interval))
output_params.extend(["-g", str(gop_size)])
# Strip audio if requested
if self.no_audio:
output_params.extend(["-an"])
return output_params + self.output_params
close()
¶
open()
¶
Open the video writer.
Source code in sleap_io/io/video_writing.py
def open(self):
"""Open the video writer."""
self.close()
self.filename.parent.mkdir(parents=True, exist_ok=True)
self._writer = iio_v2.get_writer(
self.filename.as_posix(),
format="FFMPEG",
fps=self.fps,
codec=self.codec,
pixelformat=self.pixelformat,
output_params=self.build_output_params(),
# Disable imageio's auto-scaling for non-divisible frame sizes.
# We handle padding manually in write_frame() to preserve coordinates.
macro_block_size=1,
)
write_frame(frame)
¶
Write a frame to the video.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame
|
ndarray
|
Frame to write to video. Should be a 2D or 3D numpy array with dimensions (height, width) or (height, width, channels). |
required |
Notes
For libx264 codec, frames are automatically padded to dimensions divisible by 16 (the macro block size). Padding is only added to the bottom and right edges to preserve coordinate alignment.
Source code in sleap_io/io/video_writing.py
def write_frame(self, frame: np.ndarray):
"""Write a frame to the video.
Args:
frame: Frame to write to video. Should be a 2D or 3D numpy array with
dimensions (height, width) or (height, width, channels).
Notes:
For libx264 codec, frames are automatically padded to dimensions divisible
by 16 (the macro block size). Padding is only added to the bottom and right
edges to preserve coordinate alignment.
"""
if self._writer is None:
self.open()
if self.codec == "libx264":
frame = _pad_to_macro_block(frame, macro_block_size=16)
self._writer.append_data(frame)
crop_points(points, crop)
¶
Adjust point coordinates for a crop transformation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
points
|
ndarray
|
Coordinate array of shape (..., 2) where the last dimension contains (x, y) coordinates. NaN values are preserved. |
required |
crop
|
tuple[int, int, int, int]
|
Crop region as (x1, y1, x2, y2) pixel coordinates. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Adjusted coordinates with same shape as input. |
Source code in sleap_io/transform/points.py
def crop_points(
points: np.ndarray,
crop: tuple[int, int, int, int],
) -> np.ndarray:
"""Adjust point coordinates for a crop transformation.
Args:
points: Coordinate array of shape (..., 2) where the last dimension
contains (x, y) coordinates. NaN values are preserved.
crop: Crop region as (x1, y1, x2, y2) pixel coordinates.
Returns:
Adjusted coordinates with same shape as input.
"""
x1, y1, x2, y2 = crop
result = points.copy()
result[..., 0] = points[..., 0] - x1
result[..., 1] = points[..., 1] - y1
return result
is_file_accessible(filename)
¶
Check if a file is accessible.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
filename
|
str | Path
|
Path to a file. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Notes
This checks if the file readable by the current user by reading one byte from the file.
Source code in sleap_io/io/utils.py
def is_file_accessible(filename: str | Path) -> bool:
"""Check if a file is accessible.
Args:
filename: Path to a file.
Returns:
`True` if the file is accessible, `False` otherwise.
Notes:
This checks if the file readable by the current user by reading one byte from
the file.
"""
filename = Path(filename)
try:
with open(filename, "rb") as f:
f.read(1)
return True
except (FileNotFoundError, PermissionError, OSError, ValueError):
return False
uncrop_points(points, crop)
¶
Map crop-local point coordinates back to source coordinates.
Inverse of :func:crop_points: maps crop-local (x, y) coordinates back to
source coordinates by adding the crop origin (x1, y1).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
points
|
ndarray
|
Coordinate array of shape (..., 2) where the last dimension contains (x, y) coordinates. NaN values are preserved. |
required |
crop
|
tuple[int, int, int, int]
|
Crop region as (x1, y1, x2, y2) pixel coordinates. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
Adjusted coordinates with same shape as input. |
Source code in sleap_io/transform/points.py
def uncrop_points(
points: np.ndarray,
crop: tuple[int, int, int, int],
) -> np.ndarray:
"""Map crop-local point coordinates back to source coordinates.
Inverse of :func:`crop_points`: maps crop-local (x, y) coordinates back to
source coordinates by adding the crop origin (x1, y1).
Args:
points: Coordinate array of shape (..., 2) where the last dimension
contains (x, y) coordinates. NaN values are preserved.
crop: Crop region as (x1, y1, x2, y2) pixel coordinates.
Returns:
Adjusted coordinates with same shape as input.
"""
x1, y1, x2, y2 = crop
result = points.copy()
result[..., 0] = points[..., 0] + x1
result[..., 1] = points[..., 1] + y1
return result