seq
sleap_io.io.seq
¶
Backend for reading Norpix .seq video files.
The .seq format is used by StreamPix / Norpix for high-speed video recording in
behavioral neuroscience. This module provides a SeqVideo backend that integrates
with sleap-io's VideoBackend interface for seamless access to .seq files.
Format overview
- 1024-byte binary header (little-endian, magic 0xFEED)
- Supports uncompressed (raw grayscale/BGR) and compressed (JPEG, PNG) codecs
- Per-frame timestamps (seconds + milliseconds + optional microseconds)
- Compressed formats use variable-length frames requiring a seek index
Reference implementation by Ann Kennedy
https://gist.github.com/talmo/6d577dccb01a6eb739a6d61c973f41cd
Classes:
| Name | Description |
|---|---|
SeqHeader |
Parsed header of a Norpix .seq file. |
SeqIndex |
Frame seek index for a .seq file. |
SeqVideo |
Video backend for reading Norpix .seq files. |
VideoBackend |
Base class for video backends. |
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/io/__pycache__/seq.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__ = "Backend for reading Norpix .seq video files.\n\nThe .seq format is used by StreamPix / Norpix for high-speed video recording in\nbehavioral neuroscience. This module provides a `SeqVideo` backend that integrates\nwith sleap-io's `VideoBackend` interface for seamless access to .seq files.\n\nFormat overview:\n - 1024-byte binary header (little-endian, magic 0xFEED)\n - Supports uncompressed (raw grayscale/BGR) and compressed (JPEG, PNG) codecs\n - Per-frame timestamps (seconds + milliseconds + optional microseconds)\n - Compressed formats use variable-length frames requiring a seek index\n\nReference implementation by Ann Kennedy:\n https://gist.github.com/talmo/6d577dccb01a6eb739a6d61c973f41cd\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/io/seq.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.io.seq'
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.io'
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'.
SeqHeader
¶
Parsed header of a Norpix .seq file.
Attributes:
| Name | Type | Description |
|---|---|---|
magic |
Magic number (must be 0xFEED). |
|
name |
Sequence name string. |
|
version |
Format version number. |
|
header_size |
Size of the header in bytes (always 1024). |
|
description |
User-provided description string. |
|
width |
Frame width in pixels. |
|
height |
Frame height in pixels. |
|
bit_depth |
Total bit depth (e.g., 8 for mono, 24 for color). |
|
bit_depth_real |
Bits per channel (e.g., 8). |
|
image_size_bytes |
Size of a single uncompressed frame in bytes. |
|
image_format |
Numeric codec identifier. |
|
num_frames |
Number of frames declared in the header. |
|
true_image_size |
Stride between frames for uncompressed formats. |
|
fps |
Frame rate from the header. |
|
codec |
Codec string identifier (e.g., "imageFormat100"). |
Methods:
| Name | Description |
|---|---|
from_file |
Read and parse the 1024-byte header from an open file handle. |
Source code in sleap_io/io/seq.py
@dataclass
class SeqHeader:
"""Parsed header of a Norpix .seq file.
Attributes:
magic: Magic number (must be 0xFEED).
name: Sequence name string.
version: Format version number.
header_size: Size of the header in bytes (always 1024).
description: User-provided description string.
width: Frame width in pixels.
height: Frame height in pixels.
bit_depth: Total bit depth (e.g., 8 for mono, 24 for color).
bit_depth_real: Bits per channel (e.g., 8).
image_size_bytes: Size of a single uncompressed frame in bytes.
image_format: Numeric codec identifier.
num_frames: Number of frames declared in the header.
true_image_size: Stride between frames for uncompressed formats.
fps: Frame rate from the header.
codec: Codec string identifier (e.g., "imageFormat100").
"""
magic: int = _MAGIC
name: str = "Norpix seq"
version: int = 0
header_size: int = _HEADER_SIZE
description: str = ""
width: int = 0
height: int = 0
bit_depth: int = 8
bit_depth_real: int = 8
image_size_bytes: int = 0
image_format: int = 100
num_frames: int = 0
true_image_size: int = 0
fps: float = 30.0
codec: str = ""
@property
def codec_name(self) -> str:
"""Human-readable codec name."""
return _IMAGE_FORMAT_CODES.get(
self.image_format, f"unknown({self.image_format})"
)
@property
def is_compressed(self) -> bool:
"""Whether frames use variable-length compression."""
return self.codec_name in _COMPRESSED_CODECS
@property
def num_channels(self) -> int:
"""Number of color channels."""
return self.bit_depth // (self.bit_depth_real or 8)
@classmethod
def from_file(cls, f) -> SeqHeader:
"""Read and parse the 1024-byte header from an open file handle.
Args:
f: Open binary file handle positioned at the start of the file.
Returns:
Parsed SeqHeader instance.
Raises:
ValueError: If the file is too small or has an invalid magic number.
"""
f.seek(0)
raw = f.read(_HEADER_SIZE)
if len(raw) < _HEADER_SIZE:
raise ValueError("File too small to contain a valid .seq header")
# Magic number (bytes 0-3)
magic = struct.unpack_from("<I", raw, 0)[0]
if magic != _MAGIC:
raise ValueError(
f"Invalid .seq magic: 0x{magic:08X} (expected 0x{_MAGIC:08X})"
)
# Name string (bytes 4-23, 10 uint16 chars)
name_chars = struct.unpack_from("<10H", raw, 4)
name = "".join(chr(c) for c in name_chars if 0 < c < 128).strip()
# Version and header size (bytes 28-35)
version, header_size = struct.unpack_from("<iI", raw, 28)
# Description (bytes 36-547, 256 uint16 chars)
desc_chars = struct.unpack_from("<256H", raw, 36)
description = "".join(chr(c) for c in desc_chars if 0 < c < 128).strip()
# 9 uint32 fields (bytes 548-583)
fields = struct.unpack_from("<9I", raw, 548)
width = fields[0]
height = fields[1]
bit_depth = fields[2]
bit_depth_real = fields[3]
image_size_bytes = fields[4]
image_format = fields[5]
num_frames = fields[6]
true_image_size = fields[8]
# Frame rate (bytes 584-591)
fps = struct.unpack_from("<d", raw, 584)[0]
codec = f"imageFormat{image_format:03d}"
return cls(
magic=magic,
name=name,
version=version,
header_size=header_size,
description=description,
width=width,
height=height,
bit_depth=bit_depth,
bit_depth_real=bit_depth_real,
image_size_bytes=image_size_bytes,
image_format=image_format,
num_frames=num_frames,
true_image_size=true_image_size,
fps=fps,
codec=codec,
)
__annotations__ = {'magic': 'int', 'name': 'str', 'version': 'int', 'header_size': 'int', 'description': 'str', 'width': 'int', 'height': 'int', 'bit_depth': 'int', 'bit_depth_real': 'int', 'image_size_bytes': 'int', 'image_format': 'int', 'num_frames': 'int', 'true_image_size': 'int', 'fps': 'float', 'codec': '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)
__dataclass_fields__ = {'magic': Field(name='magic',type='int',default=65261,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'name': Field(name='name',type='str',default='Norpix seq',default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'version': Field(name='version',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'header_size': Field(name='header_size',type='int',default=1024,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'description': Field(name='description',type='str',default='',default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'width': Field(name='width',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'height': Field(name='height',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'bit_depth': Field(name='bit_depth',type='int',default=8,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'bit_depth_real': Field(name='bit_depth_real',type='int',default=8,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'image_size_bytes': Field(name='image_size_bytes',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'image_format': Field(name='image_format',type='int',default=100,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'num_frames': Field(name='num_frames',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'true_image_size': Field(name='true_image_size',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'fps': Field(name='fps',type='float',default=30.0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'codec': Field(name='codec',type='str',default='',default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}
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)
__dict__ = mappingproxy({'__module__': 'sleap_io.io.seq', '__firstlineno__': 49, '__annotations__': {'magic': 'int', 'name': 'str', 'version': 'int', 'header_size': 'int', 'description': 'str', 'width': 'int', 'height': 'int', 'bit_depth': 'int', 'bit_depth_real': 'int', 'image_size_bytes': 'int', 'image_format': 'int', 'num_frames': 'int', 'true_image_size': 'int', 'fps': 'float', 'codec': 'str'}, '__doc__': 'Parsed header of a Norpix .seq file.\n\nAttributes:\n magic: Magic number (must be 0xFEED).\n name: Sequence name string.\n version: Format version number.\n header_size: Size of the header in bytes (always 1024).\n description: User-provided description string.\n width: Frame width in pixels.\n height: Frame height in pixels.\n bit_depth: Total bit depth (e.g., 8 for mono, 24 for color).\n bit_depth_real: Bits per channel (e.g., 8).\n image_size_bytes: Size of a single uncompressed frame in bytes.\n image_format: Numeric codec identifier.\n num_frames: Number of frames declared in the header.\n true_image_size: Stride between frames for uncompressed formats.\n fps: Frame rate from the header.\n codec: Codec string identifier (e.g., "imageFormat100").\n', 'magic': 65261, 'name': 'Norpix seq', 'version': 0, 'header_size': 1024, 'description': '', 'width': 0, 'height': 0, 'bit_depth': 8, 'bit_depth_real': 8, 'image_size_bytes': 0, 'image_format': 100, 'num_frames': 0, 'true_image_size': 0, 'fps': 30.0, 'codec': '', 'codec_name': <property object at 0x7f08281e9170>, 'is_compressed': <property object at 0x7f08281366b0>, 'num_channels': <property object at 0x7f0828136160>, 'from_file': <classmethod(<function SeqHeader.from_file at 0x7f0827df4680>)>, '__static_attributes__': (), '__dict__': <attribute '__dict__' of 'SeqHeader' objects>, '__weakref__': <attribute '__weakref__' of 'SeqHeader' objects>, '__dataclass_params__': _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False,match_args=True,kw_only=False,slots=False,weakref_slot=False), '__dataclass_fields__': {'magic': Field(name='magic',type='int',default=65261,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'name': Field(name='name',type='str',default='Norpix seq',default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'version': Field(name='version',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'header_size': Field(name='header_size',type='int',default=1024,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'description': Field(name='description',type='str',default='',default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'width': Field(name='width',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'height': Field(name='height',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'bit_depth': Field(name='bit_depth',type='int',default=8,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'bit_depth_real': Field(name='bit_depth_real',type='int',default=8,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'image_size_bytes': Field(name='image_size_bytes',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'image_format': Field(name='image_format',type='int',default=100,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'num_frames': Field(name='num_frames',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'true_image_size': Field(name='true_image_size',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'fps': Field(name='fps',type='float',default=30.0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'codec': Field(name='codec',type='str',default='',default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}, '__replace__': <function _replace at 0x7f08482d3560>, '__hash__': None, '__init__': <function SeqHeader.__init__ at 0x7f082811ede0>, '__repr__': <function SeqHeader.__repr__ at 0x7f082811eb60>, '__eq__': <function SeqHeader.__eq__ at 0x7f082811eca0>, '__match_args__': ('magic', 'name', 'version', 'header_size', 'description', 'width', 'height', 'bit_depth', 'bit_depth_real', 'image_size_bytes', 'image_format', 'num_frames', 'true_image_size', 'fps', 'codec')})
class-attribute
¶
Read-only proxy of a mapping.
__doc__ = 'Parsed header of a Norpix .seq file.\n\nAttributes:\n magic: Magic number (must be 0xFEED).\n name: Sequence name string.\n version: Format version number.\n header_size: Size of the header in bytes (always 1024).\n description: User-provided description string.\n width: Frame width in pixels.\n height: Frame height in pixels.\n bit_depth: Total bit depth (e.g., 8 for mono, 24 for color).\n bit_depth_real: Bits per channel (e.g., 8).\n image_size_bytes: Size of a single uncompressed frame in bytes.\n image_format: Numeric codec identifier.\n num_frames: Number of frames declared in the header.\n true_image_size: Stride between frames for uncompressed formats.\n fps: Frame rate from the header.\n codec: Codec string identifier (e.g., "imageFormat100").\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__ = 49
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__ = ('magic', 'name', 'version', 'header_size', 'description', 'width', 'height', 'bit_depth', 'bit_depth_real', 'image_size_bytes', 'image_format', 'num_frames', 'true_image_size', 'fps', 'codec')
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.seq'
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'.
__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.
__weakref__
property
¶
list of weak references to the object
bit_depth = 8
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
bit_depth_real = 8
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
codec = ''
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'.
codec_name
property
¶
Human-readable codec name.
description = ''
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'.
fps = 30.0
class-attribute
¶
Convert a string or number to a floating-point number, if possible.
header_size = 1024
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
height = 0
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
image_format = 100
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
image_size_bytes = 0
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
is_compressed
property
¶
Whether frames use variable-length compression.
magic = 65261
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
name = 'Norpix seq'
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'.
num_channels
property
¶
Number of color channels.
num_frames = 0
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
true_image_size = 0
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
version = 0
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
width = 0
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
from_file(f)
classmethod
¶
Read and parse the 1024-byte header from an open file handle.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
f
|
Open binary file handle positioned at the start of the file. |
required |
Returns:
| Type | Description |
|---|---|
SeqHeader
|
Parsed SeqHeader instance. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the file is too small or has an invalid magic number. |
Source code in sleap_io/io/seq.py
@classmethod
def from_file(cls, f) -> SeqHeader:
"""Read and parse the 1024-byte header from an open file handle.
Args:
f: Open binary file handle positioned at the start of the file.
Returns:
Parsed SeqHeader instance.
Raises:
ValueError: If the file is too small or has an invalid magic number.
"""
f.seek(0)
raw = f.read(_HEADER_SIZE)
if len(raw) < _HEADER_SIZE:
raise ValueError("File too small to contain a valid .seq header")
# Magic number (bytes 0-3)
magic = struct.unpack_from("<I", raw, 0)[0]
if magic != _MAGIC:
raise ValueError(
f"Invalid .seq magic: 0x{magic:08X} (expected 0x{_MAGIC:08X})"
)
# Name string (bytes 4-23, 10 uint16 chars)
name_chars = struct.unpack_from("<10H", raw, 4)
name = "".join(chr(c) for c in name_chars if 0 < c < 128).strip()
# Version and header size (bytes 28-35)
version, header_size = struct.unpack_from("<iI", raw, 28)
# Description (bytes 36-547, 256 uint16 chars)
desc_chars = struct.unpack_from("<256H", raw, 36)
description = "".join(chr(c) for c in desc_chars if 0 < c < 128).strip()
# 9 uint32 fields (bytes 548-583)
fields = struct.unpack_from("<9I", raw, 548)
width = fields[0]
height = fields[1]
bit_depth = fields[2]
bit_depth_real = fields[3]
image_size_bytes = fields[4]
image_format = fields[5]
num_frames = fields[6]
true_image_size = fields[8]
# Frame rate (bytes 584-591)
fps = struct.unpack_from("<d", raw, 584)[0]
codec = f"imageFormat{image_format:03d}"
return cls(
magic=magic,
name=name,
version=version,
header_size=header_size,
description=description,
width=width,
height=height,
bit_depth=bit_depth,
bit_depth_real=bit_depth_real,
image_size_bytes=image_size_bytes,
image_format=image_format,
num_frames=num_frames,
true_image_size=true_image_size,
fps=fps,
codec=codec,
)
SeqIndex
¶
Frame seek index for a .seq file.
For uncompressed formats, frame offsets are computed analytically from the header. For compressed formats, the file must be scanned to build the index, which is then cached as a JSON file alongside the .seq file.
Attributes:
| Name | Type | Description |
|---|---|---|
offsets |
Byte offset for each frame in the file. |
|
num_frames |
Number of indexed frames. |
|
timestamp_size |
Size of per-frame timestamp in bytes (6 or 8). |
Methods:
| Name | Description |
|---|---|
build_compressed |
Build index for compressed formats by scanning the file. |
build_uncompressed |
Build index for uncompressed formats (constant frame stride). |
frame_offset |
Get the byte offset for a given frame number. |
load |
Load a seek index from a JSON file. |
save |
Save the seek index to a JSON file. |
Source code in sleap_io/io/seq.py
@dataclass
class SeqIndex:
"""Frame seek index for a .seq file.
For uncompressed formats, frame offsets are computed analytically from the
header. For compressed formats, the file must be scanned to build the index,
which is then cached as a JSON file alongside the .seq file.
Attributes:
offsets: Byte offset for each frame in the file.
num_frames: Number of indexed frames.
timestamp_size: Size of per-frame timestamp in bytes (6 or 8).
"""
offsets: list[int] = field(default_factory=list)
num_frames: int = 0
timestamp_size: int = 8
def frame_offset(self, frame: int) -> int:
"""Get the byte offset for a given frame number.
Args:
frame: Zero-based frame index.
Returns:
Byte offset of the frame in the file.
Raises:
IndexError: If frame index is out of range.
"""
if frame < 0 or frame >= self.num_frames:
raise IndexError(f"Frame {frame} out of range [0, {self.num_frames})")
return self.offsets[frame]
def save(self, path: str | Path) -> None:
"""Save the seek index to a JSON file.
Args:
path: Path to write the JSON index file.
"""
data = {
"num_frames": self.num_frames,
"timestamp_size": self.timestamp_size,
"offsets": self.offsets,
}
with open(path, "w") as f:
json.dump(data, f)
@classmethod
def load(cls, path: str | Path) -> SeqIndex:
"""Load a seek index from a JSON file.
Args:
path: Path to the JSON index file.
Returns:
Loaded SeqIndex instance.
"""
with open(path) as f:
data = json.load(f)
return cls(
offsets=data["offsets"],
num_frames=data["num_frames"],
timestamp_size=data.get("timestamp_size", 8),
)
@classmethod
def build_uncompressed(cls, header: SeqHeader) -> SeqIndex:
"""Build index for uncompressed formats (constant frame stride).
Args:
header: Parsed SeqHeader.
Returns:
SeqIndex with analytically computed offsets.
"""
offsets = [
_HEADER_SIZE + i * header.true_image_size for i in range(header.num_frames)
]
return cls(
offsets=offsets,
num_frames=header.num_frames,
timestamp_size=8 if header.version >= 5 else 6,
)
@classmethod
def build_compressed(cls, f, header: SeqHeader) -> SeqIndex:
"""Build index for compressed formats by scanning the file.
Compressed frames have variable sizes, so the file must be scanned
sequentially to locate each frame boundary.
Args:
f: Open binary file handle.
header: Parsed SeqHeader.
Returns:
SeqIndex with scanned offsets.
"""
file_size = f.seek(0, 2)
n_max = header.num_frames if header.num_frames > 0 else 10_000_000
ts_size = 8 if header.version >= 5 else 6
extra = None
JPEG_SOI = b"\xff\xd8"
PNG_SIG = b"\x89\x50"
offsets = [_HEADER_SIZE]
for i in range(1, n_max):
prev = offsets[i - 1]
f.seek(prev)
size_bytes = f.read(4)
if len(size_bytes) < 4:
break
frame_size = struct.unpack("<I", size_bytes)[0]
if frame_size == 0 or frame_size > file_size:
break
if extra is not None:
next_offset = prev + frame_size + extra
else:
search_start = prev + frame_size + ts_size
found = False
for pad in range(0, 32, 2):
candidate = search_start + pad
if candidate + 6 > file_size:
break
f.seek(candidate)
probe = f.read(6)
if len(probe) < 6:
break
cand_size = struct.unpack("<I", probe[:4])[0]
cand_magic = probe[4:6]
if 0 < cand_size < file_size and cand_magic in (
JPEG_SOI,
PNG_SIG,
):
extra = ts_size + pad
next_offset = candidate
found = True
break
if not found:
break
if next_offset >= file_size:
break
f.seek(next_offset)
check = f.read(6)
if len(check) < 6:
break
check_size = struct.unpack("<I", check[:4])[0]
if check_size == 0 or check_size > file_size:
break
offsets.append(next_offset)
return cls(
offsets=offsets,
num_frames=len(offsets),
timestamp_size=ts_size,
)
__annotations__ = {'offsets': 'list[int]', 'num_frames': 'int', 'timestamp_size': 'int'}
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)
__dataclass_fields__ = {'offsets': Field(name='offsets',type='list[int]',default=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,default_factory=<class 'list'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'num_frames': Field(name='num_frames',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'timestamp_size': Field(name='timestamp_size',type='int',default=8,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}
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)
__dict__ = mappingproxy({'__module__': 'sleap_io.io.seq', '__firstlineno__': 176, '__annotations__': {'offsets': 'list[int]', 'num_frames': 'int', 'timestamp_size': 'int'}, '__doc__': 'Frame seek index for a .seq file.\n\nFor uncompressed formats, frame offsets are computed analytically from the\nheader. For compressed formats, the file must be scanned to build the index,\nwhich is then cached as a JSON file alongside the .seq file.\n\nAttributes:\n offsets: Byte offset for each frame in the file.\n num_frames: Number of indexed frames.\n timestamp_size: Size of per-frame timestamp in bytes (6 or 8).\n', 'num_frames': 0, 'timestamp_size': 8, 'frame_offset': <function SeqIndex.frame_offset at 0x7f082811eac0>, 'save': <function SeqIndex.save at 0x7f082811ea20>, 'load': <classmethod(<function SeqIndex.load at 0x7f082811e980>)>, 'build_uncompressed': <classmethod(<function SeqIndex.build_uncompressed at 0x7f082811e8e0>)>, 'build_compressed': <classmethod(<function SeqIndex.build_compressed at 0x7f082811e840>)>, '__static_attributes__': (), '__dict__': <attribute '__dict__' of 'SeqIndex' objects>, '__weakref__': <attribute '__weakref__' of 'SeqIndex' objects>, '__dataclass_params__': _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False,match_args=True,kw_only=False,slots=False,weakref_slot=False), '__dataclass_fields__': {'offsets': Field(name='offsets',type='list[int]',default=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,default_factory=<class 'list'>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'num_frames': Field(name='num_frames',type='int',default=0,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD), 'timestamp_size': Field(name='timestamp_size',type='int',default=8,default_factory=<dataclasses._MISSING_TYPE object at 0x7f08483f6cf0>,init=True,repr=True,hash=None,compare=True,metadata=mappingproxy({}),kw_only=False,_field_type=_FIELD)}, '__replace__': <function _replace at 0x7f08482d3560>, '__hash__': None, '__init__': <function SeqIndex.__init__ at 0x7f082811e700>, '__repr__': <function SeqIndex.__repr__ at 0x7f082811e020>, '__eq__': <function SeqIndex.__eq__ at 0x7f082811e0c0>, '__match_args__': ('offsets', 'num_frames', 'timestamp_size')})
class-attribute
¶
Read-only proxy of a mapping.
__doc__ = 'Frame seek index for a .seq file.\n\nFor uncompressed formats, frame offsets are computed analytically from the\nheader. For compressed formats, the file must be scanned to build the index,\nwhich is then cached as a JSON file alongside the .seq file.\n\nAttributes:\n offsets: Byte offset for each frame in the file.\n num_frames: Number of indexed frames.\n timestamp_size: Size of per-frame timestamp in bytes (6 or 8).\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__ = 176
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__ = ('offsets', 'num_frames', 'timestamp_size')
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.seq'
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'.
__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.
__weakref__
property
¶
list of weak references to the object
num_frames = 0
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
timestamp_size = 8
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
build_compressed(f, header)
classmethod
¶
Build index for compressed formats by scanning the file.
Compressed frames have variable sizes, so the file must be scanned sequentially to locate each frame boundary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
f
|
Open binary file handle. |
required | |
header
|
SeqHeader
|
Parsed SeqHeader. |
required |
Returns:
| Type | Description |
|---|---|
SeqIndex
|
SeqIndex with scanned offsets. |
Source code in sleap_io/io/seq.py
@classmethod
def build_compressed(cls, f, header: SeqHeader) -> SeqIndex:
"""Build index for compressed formats by scanning the file.
Compressed frames have variable sizes, so the file must be scanned
sequentially to locate each frame boundary.
Args:
f: Open binary file handle.
header: Parsed SeqHeader.
Returns:
SeqIndex with scanned offsets.
"""
file_size = f.seek(0, 2)
n_max = header.num_frames if header.num_frames > 0 else 10_000_000
ts_size = 8 if header.version >= 5 else 6
extra = None
JPEG_SOI = b"\xff\xd8"
PNG_SIG = b"\x89\x50"
offsets = [_HEADER_SIZE]
for i in range(1, n_max):
prev = offsets[i - 1]
f.seek(prev)
size_bytes = f.read(4)
if len(size_bytes) < 4:
break
frame_size = struct.unpack("<I", size_bytes)[0]
if frame_size == 0 or frame_size > file_size:
break
if extra is not None:
next_offset = prev + frame_size + extra
else:
search_start = prev + frame_size + ts_size
found = False
for pad in range(0, 32, 2):
candidate = search_start + pad
if candidate + 6 > file_size:
break
f.seek(candidate)
probe = f.read(6)
if len(probe) < 6:
break
cand_size = struct.unpack("<I", probe[:4])[0]
cand_magic = probe[4:6]
if 0 < cand_size < file_size and cand_magic in (
JPEG_SOI,
PNG_SIG,
):
extra = ts_size + pad
next_offset = candidate
found = True
break
if not found:
break
if next_offset >= file_size:
break
f.seek(next_offset)
check = f.read(6)
if len(check) < 6:
break
check_size = struct.unpack("<I", check[:4])[0]
if check_size == 0 or check_size > file_size:
break
offsets.append(next_offset)
return cls(
offsets=offsets,
num_frames=len(offsets),
timestamp_size=ts_size,
)
build_uncompressed(header)
classmethod
¶
Build index for uncompressed formats (constant frame stride).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
header
|
SeqHeader
|
Parsed SeqHeader. |
required |
Returns:
| Type | Description |
|---|---|
SeqIndex
|
SeqIndex with analytically computed offsets. |
Source code in sleap_io/io/seq.py
@classmethod
def build_uncompressed(cls, header: SeqHeader) -> SeqIndex:
"""Build index for uncompressed formats (constant frame stride).
Args:
header: Parsed SeqHeader.
Returns:
SeqIndex with analytically computed offsets.
"""
offsets = [
_HEADER_SIZE + i * header.true_image_size for i in range(header.num_frames)
]
return cls(
offsets=offsets,
num_frames=header.num_frames,
timestamp_size=8 if header.version >= 5 else 6,
)
frame_offset(frame)
¶
Get the byte offset for a given frame number.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame
|
int
|
Zero-based frame index. |
required |
Returns:
| Type | Description |
|---|---|
int
|
Byte offset of the frame in the file. |
Raises:
| Type | Description |
|---|---|
IndexError
|
If frame index is out of range. |
Source code in sleap_io/io/seq.py
def frame_offset(self, frame: int) -> int:
"""Get the byte offset for a given frame number.
Args:
frame: Zero-based frame index.
Returns:
Byte offset of the frame in the file.
Raises:
IndexError: If frame index is out of range.
"""
if frame < 0 or frame >= self.num_frames:
raise IndexError(f"Frame {frame} out of range [0, {self.num_frames})")
return self.offsets[frame]
load(path)
classmethod
¶
Load a seek index from a JSON file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to the JSON index file. |
required |
Returns:
| Type | Description |
|---|---|
SeqIndex
|
Loaded SeqIndex instance. |
Source code in sleap_io/io/seq.py
@classmethod
def load(cls, path: str | Path) -> SeqIndex:
"""Load a seek index from a JSON file.
Args:
path: Path to the JSON index file.
Returns:
Loaded SeqIndex instance.
"""
with open(path) as f:
data = json.load(f)
return cls(
offsets=data["offsets"],
num_frames=data["num_frames"],
timestamp_size=data.get("timestamp_size", 8),
)
save(path)
¶
Save the seek index to a JSON file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to write the JSON index file. |
required |
Source code in sleap_io/io/seq.py
SeqVideo
¶
Bases: sleap_io.io.video_reading.VideoBackend
Video backend for reading Norpix .seq files.
This backend supports reading .seq files produced by StreamPix / Norpix, commonly used for high-speed video recording in behavioral neuroscience.
Supported codecs
- monoraw (100): Grayscale uncompressed
- raw (200): Color BGR uncompressed (converted to RGB)
- monojpg (102): Grayscale JPEG compressed
- jpg (201): Color JPEG compressed
- monopng (1): Grayscale PNG compressed
- png (2): Color PNG compressed
Attributes:
| Name | Type | Description |
|---|---|---|
filename |
Path to the .seq file. |
|
grayscale |
Whether to force grayscale. If None, autodetect on first frame. |
|
keep_open |
Whether to keep the file handle open between reads. |
Methods:
| Name | Description |
|---|---|
__attrs_post_init__ |
Parse header, build seek index, and compute FPS. |
__del__ |
Clean up file handle on garbage collection. |
__eq__ |
Method generated by attrs for class SeqVideo. |
__init__ |
Method generated by attrs for class SeqVideo. |
__repr__ |
Method generated by attrs for class SeqVideo. |
close |
Close the underlying file handle. |
get_timestamp |
Get the timestamp for a single frame. |
get_timestamps |
Get all frame timestamps as an array. |
Source code in sleap_io/io/seq.py
@attrs.define
class SeqVideo(VideoBackend):
"""Video backend for reading Norpix .seq files.
This backend supports reading .seq files produced by StreamPix / Norpix,
commonly used for high-speed video recording in behavioral neuroscience.
Supported codecs:
- monoraw (100): Grayscale uncompressed
- raw (200): Color BGR uncompressed (converted to RGB)
- monojpg (102): Grayscale JPEG compressed
- jpg (201): Color JPEG compressed
- monopng (1): Grayscale PNG compressed
- png (2): Color PNG compressed
Attributes:
filename: Path to the .seq file.
grayscale: Whether to force grayscale. If None, autodetect on first frame.
keep_open: Whether to keep the file handle open between reads.
"""
EXTS = ("seq",)
_header: SeqHeader | None = attrs.field(
default=None, alias="_header", repr=False, eq=False
)
_index: SeqIndex | None = attrs.field(
default=None, alias="_index", repr=False, eq=False
)
_file_handle: object | None = attrs.field(
default=None, alias="_file_handle", repr=False, eq=False
)
def __attrs_post_init__(self):
"""Parse header, build seek index, and compute FPS."""
path = Path(self.filename)
if not path.exists():
raise FileNotFoundError(f"File not found: {path}")
f = open(path, "rb")
try:
self._header = SeqHeader.from_file(f)
if self._header.codec_name in _BAYER_CODECS:
raise NotImplementedError(
f"Bayer codec '{self._header.codec_name}' is not supported. "
f"Convert the .seq file to a standard format first."
)
self._index = self._load_or_build_index(f)
self._recompute_fps(f)
except Exception:
f.close()
raise
if self.keep_open:
self._file_handle = f
else:
f.close()
def _get_file_handle(self):
"""Get an open file handle, opening one if necessary.
Returns:
Open binary file handle.
"""
if self._file_handle is not None and not self._file_handle.closed:
return self._file_handle
f = open(self.filename, "rb")
if self.keep_open:
self._file_handle = f
return f
def _maybe_close(self, f):
"""Close the file handle if keep_open is False.
Args:
f: File handle to potentially close.
"""
if not self.keep_open and f is not self._file_handle:
f.close()
def _load_or_build_index(self, f) -> SeqIndex:
"""Load cached index or build one by scanning the file.
Args:
f: Open binary file handle.
Returns:
SeqIndex for frame access.
"""
if not self._header.is_compressed:
return SeqIndex.build_uncompressed(self._header)
cache_path = Path(self.filename).with_suffix(".seq-index.json")
if cache_path.exists():
try:
idx = SeqIndex.load(cache_path)
if idx.num_frames > 0:
return idx
except (json.JSONDecodeError, KeyError):
pass
idx = SeqIndex.build_compressed(f, self._header)
try:
idx.save(cache_path)
except OSError:
pass
return idx
def _recompute_fps(self, f) -> None:
"""Recompute FPS from actual frame timestamps.
Uses the first 100 frames to compute a robust median-filtered FPS
estimate from inter-frame intervals.
Args:
f: Open binary file handle.
"""
try:
n = min(100, self._index.num_frames)
if n < 2:
self._fps = self._header.fps if self._header.fps >= 1.0 else None
return
ts = np.array([self._read_timestamp(f, i) for i in range(n)])
ds = np.diff(ts)
median_ds = np.median(ds)
# 5ms tolerance — tuned for high-speed video (>100 fps); for slow
# framerates where jitter exceeds this, falls back to header FPS.
ds = ds[np.abs(ds - median_ds) < 0.005]
if len(ds) > 0:
computed = 1.0 / np.mean(ds)
if np.isfinite(computed) and computed >= 1.0:
self._fps = float(computed)
return
except Exception:
pass
self._fps = self._header.fps if self._header.fps >= 1.0 else None
def _read_raw_frame(self, f, frame_idx: int) -> tuple[bytes, int]:
"""Read raw frame bytes and return data with position after data.
Args:
f: Open binary file handle.
frame_idx: Zero-based frame index.
Returns:
Tuple of (raw_data_bytes, file_position_after_data).
"""
offset = self._index.frame_offset(frame_idx)
f.seek(offset)
if self._header.is_compressed:
nbytes = struct.unpack("<I", f.read(4))[0]
data = f.read(nbytes - 4)
return data, f.tell()
else:
data = f.read(self._header.image_size_bytes)
return data, f.tell()
def _read_timestamp(self, f, frame_idx: int) -> float:
"""Read the timestamp for a single frame.
Args:
f: Open binary file handle.
frame_idx: Zero-based frame index.
Returns:
Timestamp as seconds since epoch (float64).
"""
_, pos_after_data = self._read_raw_frame(f, frame_idx)
f.seek(pos_after_data)
ts_sec = struct.unpack("<I", f.read(4))[0]
ts_ms = struct.unpack("<H", f.read(2))[0]
result = ts_sec + ts_ms / 1000.0
if self._index.timestamp_size == 8:
ts_us = struct.unpack("<H", f.read(2))[0]
result += ts_us / 1_000_000.0
return result
def _decode_frame(self, data: bytes) -> np.ndarray:
"""Decode raw frame bytes into a numpy image array.
Args:
data: Raw frame bytes.
Returns:
Decoded frame as numpy array of shape (height, width, channels).
Raises:
ValueError: If the codec is unsupported.
"""
codec = self._header.codec_name
h, w = self._header.height, self._header.width
nch = self._header.num_channels
if codec in ("monoraw", "raw"):
arr = np.frombuffer(data, dtype=np.uint8)
if nch == 1:
return arr[: h * w].reshape(h, w, 1)
else:
arr = arr[: h * w * nch].reshape(h, w, nch)
# BGR -> RGB
return arr[:, :, ::-1].copy()
elif codec in ("monojpg", "jpg", "monopng", "png"):
try:
from PIL import Image
except ImportError:
raise ImportError(
f"Pillow is required to decode {codec} frames in .seq files. "
f"Install with: pip install Pillow"
)
img = Image.open(io.BytesIO(data))
arr = np.array(img)
if arr.ndim == 2:
return arr[:, :, np.newaxis]
return arr
else:
raise ValueError(f"Unsupported .seq codec: {codec}")
# --- VideoBackend interface ---
@property
def num_frames(self) -> int:
"""Number of frames in the video."""
return self._index.num_frames
@property
def img_shape(self) -> tuple[int, int, int]:
"""Shape of a single frame as (height, width, channels)."""
h = self._header.height
w = self._header.width
nch = self._header.num_channels
if self.grayscale is True or nch == 1:
return (h, w, 1)
return (h, w, 3)
def _read_frame(self, frame_idx: int) -> np.ndarray:
"""Read a single frame from the .seq file.
Args:
frame_idx: Zero-based frame index. Negative indices are supported.
Returns:
Frame as numpy array of shape (height, width, channels).
"""
if frame_idx < 0:
frame_idx = self._index.num_frames + frame_idx
f = self._get_file_handle()
try:
data, _ = self._read_raw_frame(f, frame_idx)
return self._decode_frame(data)
finally:
self._maybe_close(f)
# --- Seq-specific public methods ---
def get_timestamp(self, frame_idx: int) -> float:
"""Get the timestamp for a single frame.
Args:
frame_idx: Zero-based frame index. Negative indices are supported.
Returns:
Timestamp as seconds since epoch (float64).
Raises:
IndexError: If frame index is out of range.
"""
if frame_idx < 0:
frame_idx = self.num_frames + frame_idx
if frame_idx < 0 or frame_idx >= self.num_frames:
raise IndexError(f"Frame {frame_idx} out of range [0, {self.num_frames})")
f = self._get_file_handle()
try:
return self._read_timestamp(f, frame_idx)
finally:
self._maybe_close(f)
def get_timestamps(self) -> np.ndarray:
"""Get all frame timestamps as an array.
Returns:
Array of timestamps as float64 (seconds since epoch).
"""
f = self._get_file_handle()
try:
return np.array(
[self._read_timestamp(f, i) for i in range(self.num_frames)]
)
finally:
self._maybe_close(f)
@property
def header(self) -> SeqHeader:
"""The parsed .seq file header."""
return self._header
def close(self) -> None:
"""Close the underlying file handle."""
if self._file_handle is not None and not self._file_handle.closed:
self._file_handle.close()
self._file_handle = None
def __del__(self):
"""Clean up file handle on garbage collection."""
try:
self.close()
except Exception:
pass
EXTS = ('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__ = {'_header': 'SeqHeader | None', '_index': 'SeqIndex | None', '_file_handle': 'object | 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=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 Norpix .seq files.\n\nThis backend supports reading .seq files produced by StreamPix / Norpix,\ncommonly used for high-speed video recording in behavioral neuroscience.\n\nSupported codecs:\n - monoraw (100): Grayscale uncompressed\n - raw (200): Color BGR uncompressed (converted to RGB)\n - monojpg (102): Grayscale JPEG compressed\n - jpg (201): Color JPEG compressed\n - monopng (1): Grayscale PNG compressed\n - png (2): Color PNG compressed\n\nAttributes:\n filename: Path to the .seq file.\n grayscale: Whether to force grayscale. If None, autodetect on first frame.\n keep_open: Whether to keep the file handle open between reads.\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__ = 349
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', '_header', '_index', '_file_handle')
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.seq'
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__ = ('_header', '_index', '_file_handle')
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__ = ('_file_handle', '_fps', '_header', '_index')
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.
header
property
¶
The parsed .seq file header.
img_shape
property
¶
Shape of a single frame as (height, width, channels).
num_frames
property
¶
Number of frames in the video.
__attrs_post_init__()
¶
Parse header, build seek index, and compute FPS.
Source code in sleap_io/io/seq.py
def __attrs_post_init__(self):
"""Parse header, build seek index, and compute FPS."""
path = Path(self.filename)
if not path.exists():
raise FileNotFoundError(f"File not found: {path}")
f = open(path, "rb")
try:
self._header = SeqHeader.from_file(f)
if self._header.codec_name in _BAYER_CODECS:
raise NotImplementedError(
f"Bayer codec '{self._header.codec_name}' is not supported. "
f"Convert the .seq file to a standard format first."
)
self._index = self._load_or_build_index(f)
self._recompute_fps(f)
except Exception:
f.close()
raise
if self.keep_open:
self._file_handle = f
else:
f.close()
__del__()
¶
__eq__(other)
¶
__init__(filename, grayscale=None, keep_open=True, cached_shape=None, open_reader=None, fps=None, _header=None, _index=None, _file_handle=None)
¶
Method generated by attrs for class SeqVideo.
Source code in sleap_io/io/seq.py
from sleap_io.io.video_reading import VideoBackend
# Image format codec mapping
_IMAGE_FORMAT_CODES = {
100: "monoraw", # Grayscale uncompressed
200: "raw", # Color BGR uncompressed
101: "brgb8", # Bayer pattern raw
102: "monojpg", # Grayscale JPEG compressed
201: "jpg", # Color JPEG compressed
103: "jbrgb", # Bayer JPEG compressed
__repr__()
¶
Method generated by attrs for class SeqVideo.
Source code in sleap_io/io/seq.py
"""Backend for reading Norpix .seq video files.
The .seq format is used by StreamPix / Norpix for high-speed video recording in
behavioral neuroscience. This module provides a `SeqVideo` backend that integrates
with sleap-io's `VideoBackend` interface for seamless access to .seq files.
Format overview:
- 1024-byte binary header (little-endian, magic 0xFEED)
- Supports uncompressed (raw grayscale/BGR) and compressed (JPEG, PNG) codecs
- Per-frame timestamps (seconds + milliseconds + optional microseconds)
- Compressed formats use variable-length frames requiring a seek index
Reference implementation by Ann Kennedy:
https://gist.github.com/talmo/6d577dccb01a6eb739a6d61c973f41cd
"""
close()
¶
get_timestamp(frame_idx)
¶
Get the timestamp for a single frame.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_idx
|
int
|
Zero-based frame index. Negative indices are supported. |
required |
Returns:
| Type | Description |
|---|---|
float
|
Timestamp as seconds since epoch (float64). |
Raises:
| Type | Description |
|---|---|
IndexError
|
If frame index is out of range. |
Source code in sleap_io/io/seq.py
def get_timestamp(self, frame_idx: int) -> float:
"""Get the timestamp for a single frame.
Args:
frame_idx: Zero-based frame index. Negative indices are supported.
Returns:
Timestamp as seconds since epoch (float64).
Raises:
IndexError: If frame index is out of range.
"""
if frame_idx < 0:
frame_idx = self.num_frames + frame_idx
if frame_idx < 0 or frame_idx >= self.num_frames:
raise IndexError(f"Frame {frame_idx} out of range [0, {self.num_frames})")
f = self._get_file_handle()
try:
return self._read_timestamp(f, frame_idx)
finally:
self._maybe_close(f)
get_timestamps()
¶
Get all frame timestamps as an array.
Returns:
| Type | Description |
|---|---|
ndarray
|
Array of timestamps as float64 (seconds since epoch). |
Source code in sleap_io/io/seq.py
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.