slp_lazy
sleap_io.io.slp_lazy
¶
Lazy loading support for SLP files.
This module provides LazyDataStore and LazyFrameList classes that enable deferred materialization of LabeledFrame and Instance objects when loading SLP files with lazy=True.
These classes are implementation details - users interact with Labels objects.
Classes:
| Name | Description |
|---|---|
InstanceType |
Enumeration of instance types to integers. |
LazyDataStore |
Holds raw HDF5 data and provides lazy access methods. |
LazyFrameList |
List-like proxy that materializes LabeledFrame objects on access. |
Attributes:
| Name | Type | Description |
|---|---|---|
TYPE_CHECKING |
Returns True when the argument is true, False otherwise. |
|
__cached__ |
str(object='') -> str |
|
__doc__ |
str(object='') -> str |
|
__file__ |
str(object='') -> str |
|
__name__ |
str(object='') -> str |
|
__package__ |
str(object='') -> str |
TYPE_CHECKING = False
module-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.
__cached__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/__pycache__/slp_lazy.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__ = 'Lazy loading support for SLP files.\n\nThis module provides LazyDataStore and LazyFrameList classes that enable\ndeferred materialization of LabeledFrame and Instance objects when loading\nSLP files with lazy=True.\n\nThese classes are implementation details - users interact with Labels objects.\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/slp_lazy.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.slp_lazy'
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'.
InstanceType
¶
Bases: enum.IntEnum
Enumeration of instance types to integers.
Methods:
| Name | Description |
|---|---|
__format__ |
Convert to a string according to format_spec. |
Attributes:
| Name | Type | Description |
|---|---|---|
PREDICTED |
Enumeration of instance types to integers. |
|
USER |
Enumeration of instance types to integers. |
|
__doc__ |
str(object='') -> str |
|
__module__ |
str(object='') -> str |
Source code in sleap_io/io/slp.py
PREDICTED = <InstanceType.PREDICTED: 1>
class-attribute
¶
Enumeration of instance types to integers.
USER = <InstanceType.USER: 0>
class-attribute
¶
Enumeration of instance types to integers.
__doc__ = 'Enumeration of instance types to integers.'
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'.
__module__ = 'sleap_io.io.slp'
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'.
__format__(format_spec)
method descriptor
¶
Convert to a string according to format_spec.
LazyDataStore
¶
Holds raw HDF5 data and provides lazy access methods.
Attributes:
| Name | Type | Description |
|---|---|---|
frames_data |
Structured array from /frames HDF5 dataset. Fields: frame_id, video_id, frame_idx, instance_id_start, instance_id_end. |
|
instances_data |
Structured array from /instances HDF5 dataset. Fields vary by format_id but include: instance_id, instance_type, frame_id, skeleton_id, track_id, from_predicted, instance_score, point_id_start, point_id_end, and optionally tracking_score. |
|
pred_points_data |
Structured array from /pred_points HDF5 dataset. Fields: x, y, visible, complete, score. |
|
points_data |
Structured array from /points HDF5 dataset. Fields: x, y, visible, complete. |
|
videos |
List of eagerly loaded Video objects. |
|
skeletons |
List of eagerly loaded Skeleton objects. |
|
tracks |
List of eagerly loaded Track objects. |
|
format_id |
SLP format version. |
|
_source_path |
Path to source SLP file (for debugging). |
Methods:
| Name | Description |
|---|---|
__attrs_post_init__ |
Validate index bounds on construction. |
__eq__ |
Method generated by attrs for class LazyDataStore. |
__init__ |
Method generated by attrs for class LazyDataStore. |
__len__ |
Return number of frames. |
__replace__ |
Method generated by attrs for class LazyDataStore. |
__repr__ |
Method generated by attrs for class LazyDataStore. |
copy |
Create an independent copy with copied arrays. |
get_user_frame_indices |
Find indices of frames containing user (non-predicted) instances. |
materialize_all |
Materialize all frames. |
materialize_frame |
Create a fully materialized LabeledFrame. |
to_numpy |
Build numpy array directly from raw data (fast path). |
validate |
Check that all indices are within bounds. |
Source code in sleap_io/io/slp_lazy.py
@attrs.define
class LazyDataStore:
"""Holds raw HDF5 data and provides lazy access methods.
Attributes:
frames_data: Structured array from /frames HDF5 dataset.
Fields: frame_id, video_id, frame_idx, instance_id_start, instance_id_end.
instances_data: Structured array from /instances HDF5 dataset.
Fields vary by format_id but include: instance_id, instance_type, frame_id,
skeleton_id, track_id, from_predicted, instance_score, point_id_start,
point_id_end, and optionally tracking_score.
pred_points_data: Structured array from /pred_points HDF5 dataset.
Fields: x, y, visible, complete, score.
points_data: Structured array from /points HDF5 dataset.
Fields: x, y, visible, complete.
videos: List of eagerly loaded Video objects.
skeletons: List of eagerly loaded Skeleton objects.
tracks: List of eagerly loaded Track objects.
format_id: SLP format version.
_source_path: Path to source SLP file (for debugging).
"""
# Raw arrays
frames_data: np.ndarray
instances_data: np.ndarray
pred_points_data: np.ndarray
points_data: np.ndarray
# References
videos: list["Video"]
skeletons: list["Skeleton"]
tracks: list["Track"]
# Metadata
format_id: float
_source_path: str | None = attrs.field(default=None, alias="source_path")
_negative_frames: set[tuple[int, int]] = attrs.field(
factory=set, alias="negative_frames"
)
# Per-frame annotation lookups: (video_idx, frame_idx) -> list[annotation]
# These are eagerly loaded from HDF5 but attached to frames lazily.
_centroid_by_frame: dict = attrs.field(
factory=dict, repr=False, alias="centroid_by_frame"
)
_bbox_by_frame: dict = attrs.field(factory=dict, repr=False, alias="bbox_by_frame")
_mask_by_frame: dict = attrs.field(factory=dict, repr=False, alias="mask_by_frame")
_label_image_by_frame: dict = attrs.field(
factory=dict, repr=False, alias="label_image_by_frame"
)
_roi_by_frame: dict = attrs.field(factory=dict, repr=False, alias="roi_by_frame")
# Undistributed annotations (video=None or frame_idx=None, e.g. static ROIs)
_undistributed_rois: list = attrs.field(factory=list, repr=False)
_undistributed_masks: list = attrs.field(factory=list, repr=False)
_undistributed_bboxes: list = attrs.field(factory=list, repr=False)
_undistributed_centroids: list = attrs.field(factory=list, repr=False)
_undistributed_label_images: list = attrs.field(factory=list, repr=False)
# Global identity catalog and per-instance identity links (format 2.5+).
# ``identities`` are canonical objects shared with Labels.identities;
# ``_instance_identities`` maps global instance_id -> (identity_idx, score).
identities: list["Identity"] = attrs.field(factory=list, repr=False)
_instance_identities: dict = attrs.field(
factory=dict, repr=False, alias="instance_identities"
)
# Per-instance re-ID embeddings (format 2.5+): instance_id -> Embedding.
_instance_embeddings: dict = attrs.field(
factory=dict, repr=False, alias="instance_embeddings"
)
# Global category catalog and per-instance category links (format 2.7+).
# ``categories`` are canonical objects shared with Labels.categories;
# ``_instance_categories`` maps global instance_id -> (category_idx, score).
categories: list["Category"] = attrs.field(factory=list, repr=False)
_instance_categories: dict = attrs.field(
factory=dict, repr=False, alias="instance_categories"
)
# Per-instance category (classification) embeddings (format 2.7+):
# instance_id -> Embedding.
_instance_category_embeddings: dict = attrs.field(
factory=dict, repr=False, alias="instance_category_embeddings"
)
# Raw RecordingSession payload for lossless lazy passthrough (format 2.8+).
# ``sessions_json_raw`` is the verbatim variable-length ``sessions_json`` bytes
# array; ``session_data`` is the dict of columnar ``/session_data`` arrays (see
# slp._read_session_data). Held so a lazy re-save can copy the frame-group / 3D
# tables verbatim without materializing frames -- the eager path drops them.
sessions_json_raw: "np.ndarray | None" = attrs.field(
default=None, repr=False, alias="sessions_json_raw"
)
session_data: "dict | None" = attrs.field(
default=None, repr=False, alias="session_data"
)
# Immutable snapshot of the video identity order at load time. The passthrough's
# sessions_json encodes video indices, so it is only safe to copy verbatim when
# the current video list matches this order (see slp._videos_unchanged).
session_video_ids: tuple = attrs.field(
default=(), repr=False, alias="session_video_ids"
)
def __attrs_post_init__(self) -> None:
"""Validate index bounds on construction."""
self.validate()
def validate(self) -> None:
"""Check that all indices are within bounds.
Raises:
ValueError: If any index is out of bounds.
"""
n_frames = len(self.frames_data)
n_instances = len(self.instances_data)
n_points = len(self.points_data)
n_pred_points = len(self.pred_points_data)
if n_frames == 0:
return # Empty data is valid
# Validate frame -> instance references
max_inst_end = self.frames_data["instance_id_end"].max() if n_frames > 0 else 0
if max_inst_end > n_instances:
raise ValueError(
f"Frame references instance index {max_inst_end} but only "
f"{n_instances} instances exist."
)
if n_instances == 0:
return # No instances means no points to validate
# Validate instance -> point references
# Separate user instances and predicted instances
user_mask = self.instances_data["instance_type"] == InstanceType.USER
pred_mask = self.instances_data["instance_type"] == InstanceType.PREDICTED
if np.any(user_mask):
user_max_end = self.instances_data[user_mask]["point_id_end"].max()
if user_max_end > n_points:
raise ValueError(
f"User instance references point index {user_max_end} but only "
f"{n_points} points exist."
)
if np.any(pred_mask):
pred_max_end = self.instances_data[pred_mask]["point_id_end"].max()
if pred_max_end > n_pred_points:
raise ValueError(
f"Predicted instance references pred_point index {pred_max_end} "
f"but only {n_pred_points} pred_points exist."
)
def __len__(self) -> int:
"""Return number of frames."""
return len(self.frames_data)
def copy(self) -> "LazyDataStore":
"""Create an independent copy with copied arrays.
The returned copy has independent numpy arrays but shares references
to Video, Skeleton, and Track objects. The metadata objects are shared
because they are the canonical objects referenced by the Labels; copying
them here would create inconsistency with Labels.videos/skeletons/tracks.
Returns:
A new LazyDataStore with copied arrays.
"""
from copy import deepcopy
new_store = LazyDataStore(
frames_data=self.frames_data.copy(),
instances_data=self.instances_data.copy(),
pred_points_data=self.pred_points_data.copy(),
points_data=self.points_data.copy(),
videos=self.videos, # Share references (canonical objects)
skeletons=self.skeletons, # Share references (canonical objects)
tracks=self.tracks, # Share references (canonical objects)
identities=self.identities, # Share references (canonical objects)
instance_identities=dict(self._instance_identities),
instance_embeddings=dict(self._instance_embeddings),
categories=self.categories, # Share references (canonical objects)
instance_categories=dict(self._instance_categories),
instance_category_embeddings=dict(self._instance_category_embeddings),
format_id=self.format_id,
source_path=self._source_path,
negative_frames=self._negative_frames.copy(),
centroid_by_frame={
k: [deepcopy(c) for c in v] for k, v in self._centroid_by_frame.items()
},
bbox_by_frame={
k: [deepcopy(b) for b in v] for k, v in self._bbox_by_frame.items()
},
mask_by_frame={
k: [deepcopy(m) for m in v] for k, v in self._mask_by_frame.items()
},
label_image_by_frame={
k: [deepcopy(li) for li in v]
for k, v in self._label_image_by_frame.items()
},
roi_by_frame={
k: [deepcopy(r) for r in v] for k, v in self._roi_by_frame.items()
},
)
# Copy undistributed annotations
new_store._undistributed_rois = [deepcopy(r) for r in self._undistributed_rois]
new_store._undistributed_masks = [
deepcopy(m) for m in self._undistributed_masks
]
new_store._undistributed_bboxes = [
deepcopy(b) for b in self._undistributed_bboxes
]
new_store._undistributed_centroids = [
deepcopy(c) for c in self._undistributed_centroids
]
new_store._undistributed_label_images = [
deepcopy(li) for li in self._undistributed_label_images
]
return new_store
def materialize_frame(self, idx: int) -> "LabeledFrame":
"""Create a fully materialized LabeledFrame.
Args:
idx: Index into frames_data array.
Returns:
A real LabeledFrame with real Instance objects.
"""
from sleap_io.model.labeled_frame import LabeledFrame
frame_row = self.frames_data[idx]
video_id = int(frame_row[1]) # video_id
frame_idx = int(frame_row[2]) # frame_idx
inst_start = int(frame_row[3]) # instance_id_start
inst_end = int(frame_row[4]) # instance_id_end
instances = []
for inst_idx in range(inst_start, inst_end):
inst = self._materialize_instance(inst_idx)
instances.append(inst)
is_negative = (video_id, frame_idx) in self._negative_frames
# Attach per-frame annotations from eagerly-loaded dicts
key = (video_id, frame_idx)
centroids = self._centroid_by_frame.get(key, [])
bboxes = self._bbox_by_frame.get(key, [])
masks = self._mask_by_frame.get(key, [])
label_images = self._label_image_by_frame.get(key, [])
rois = self._roi_by_frame.get(key, [])
return LabeledFrame(
video=self.videos[video_id],
frame_idx=frame_idx,
instances=instances,
is_negative=is_negative,
centroids=centroids,
bboxes=bboxes,
masks=masks,
label_images=label_images,
rois=rois,
)
def _materialize_instance(self, idx: int) -> "Instance | PredictedInstance":
"""Create a single Instance from raw data.
Args:
idx: Index into instances_data array.
Returns:
Instance or PredictedInstance with populated points.
"""
from sleap_io.model.instance import Instance, PredictedInstance
inst_row = self.instances_data[idx]
# Parse instance data - handle format differences
if self.format_id < 1.2:
(
instance_id,
instance_type,
frame_id,
skeleton_id,
track_id,
from_predicted,
instance_score,
point_id_start,
point_id_end,
) = inst_row
tracking_score = 0.0
else:
(
instance_id,
instance_type,
frame_id,
skeleton_id,
track_id,
from_predicted,
instance_score,
point_id_start,
point_id_end,
tracking_score,
) = inst_row
# Cast index values to int for h5wasm compatibility (float64 columns).
instance_id = int(instance_id)
skeleton_id = int(skeleton_id)
track_id = int(track_id)
from_predicted = int(from_predicted)
point_id_start = int(point_id_start)
point_id_end = int(point_id_end)
skeleton = self.skeletons[skeleton_id]
track = self.tracks[track_id] if track_id >= 0 else None
# Resolve optional per-instance global identity (format 2.5+). Joined on
# the global instance_id; absent for older files (empty mapping -> None).
identity = None
identity_score = None
if self._instance_identities:
entry = self._instance_identities.get(instance_id)
if entry is not None:
identity_idx, id_score = entry
if 0 <= identity_idx < len(self.identities):
identity = self.identities[identity_idx]
identity_score = id_score
# Resolve the optional per-instance re-ID embedding (format 2.5+). The
# stored `Embedding` is shared by reference; its vector array is treated as
# immutable.
identity_embedding = self._instance_embeddings.get(instance_id)
# Resolve the optional per-instance global category (format 2.7+). Joined on
# the global instance_id, mirroring identity; absent for older files.
category = None
category_score = None
if self._instance_categories:
entry = self._instance_categories.get(instance_id)
if entry is not None:
category_idx, cat_score = entry
if 0 <= category_idx < len(self.categories):
category = self.categories[category_idx]
category_score = cat_score
# Resolve the optional per-instance classification embedding (format 2.7+).
category_embedding = self._instance_category_embeddings.get(instance_id)
if instance_type == InstanceType.USER:
pts_data = self.points_data[point_id_start:point_id_end]
points_array = self._make_points_array(pts_data, skeleton)
if self.format_id < 1.1:
# Legacy coordinate system adjustment
points_array["xy"] -= 0.5
return Instance(
points=points_array,
skeleton=skeleton,
track=track,
tracking_score=float(tracking_score),
identity=identity,
identity_score=identity_score,
identity_embedding=identity_embedding,
category=category,
category_score=category_score,
category_embedding=category_embedding,
)
else: # PREDICTED
pts_data = self.pred_points_data[point_id_start:point_id_end]
points_array = self._make_predicted_points_array(pts_data, skeleton)
if self.format_id < 1.1:
# Legacy coordinate system adjustment
points_array["xy"] -= 0.5
return PredictedInstance(
points=points_array,
skeleton=skeleton,
track=track,
score=float(instance_score),
tracking_score=float(tracking_score),
identity=identity,
identity_score=identity_score,
identity_embedding=identity_embedding,
category=category,
category_score=category_score,
category_embedding=category_embedding,
)
def _make_points_array(
self, pts_data: np.ndarray, skeleton: "Skeleton"
) -> np.ndarray:
"""Create PointsArray from raw point data.
Args:
pts_data: Structured array with x, y, visible, complete fields.
skeleton: Skeleton defining node structure.
Returns:
Populated PointsArray.
"""
from sleap_io.model.instance import PointsArray
n = len(pts_data)
points = PointsArray.empty(n)
points["xy"][:, 0] = pts_data["x"]
points["xy"][:, 1] = pts_data["y"]
points["visible"] = pts_data["visible"]
points["complete"] = pts_data["complete"]
points["name"] = skeleton.node_names
return points
def _make_predicted_points_array(
self, pts_data: np.ndarray, skeleton: "Skeleton"
) -> np.ndarray:
"""Create PredictedPointsArray from raw point data.
Args:
pts_data: Structured array with x, y, visible, complete, score fields.
skeleton: Skeleton defining node structure.
Returns:
Populated PredictedPointsArray.
"""
from sleap_io.model.instance import PredictedPointsArray
n = len(pts_data)
points = PredictedPointsArray.empty(n)
points["xy"][:, 0] = pts_data["x"]
points["xy"][:, 1] = pts_data["y"]
points["visible"] = pts_data["visible"]
points["complete"] = pts_data["complete"]
points["score"] = pts_data["score"]
points["name"] = skeleton.node_names
return points
def materialize_all(self) -> list["LabeledFrame"]:
"""Materialize all frames.
Returns:
List of all LabeledFrame objects.
"""
return [self.materialize_frame(i) for i in range(len(self))]
def get_user_frame_indices(self) -> list[int]:
"""Find indices of frames containing user (non-predicted) instances.
This also includes frames marked as negative (is_negative=True), since
those are considered user-labeled even though they have no instances.
Returns:
List of frame indices (into frames_data) that have at least one user
instance or are marked as negative.
"""
from sleap_io.io.slp import InstanceType
result_indices: set[int] = set()
# Find all user instances
user_mask = self.instances_data["instance_type"] == InstanceType.USER
if np.any(user_mask):
# Get frame boundaries for binary search
frame_ends = self.frames_data["instance_id_end"]
# Use binary search to find frame for each user instance - O(n log m)
user_instance_indices = np.where(user_mask)[0]
# searchsorted finds insertion point; instance i is in frame fi where
# frame_ends[fi-1] <= i < frame_ends[fi] (with frame_ends[-1] = 0)
frame_indices = np.searchsorted(
frame_ends, user_instance_indices, side="right"
)
# Get unique frame indices (already sorted by searchsorted)
unique_frames = np.unique(frame_indices)
# Filter out any out-of-bounds indices
valid_mask = unique_frames < len(self.frames_data)
result_indices.update(unique_frames[valid_mask].tolist())
# Also include negative frames
if self._negative_frames:
for idx in range(len(self.frames_data)):
frame_row = self.frames_data[idx]
video_id = int(frame_row[1])
frame_idx = int(frame_row[2])
if (video_id, frame_idx) in self._negative_frames:
result_indices.add(idx)
return sorted(result_indices)
def to_numpy(
self,
video: "Video | None" = None,
untracked: bool = False,
return_confidence: bool = False,
user_instances: bool = True,
) -> np.ndarray:
"""Build numpy array directly from raw data (fast path).
This method builds the output array directly from raw HDF5 data without
creating any Instance or LabeledFrame objects, providing significant
performance improvement for workflows that only need numpy output.
Args:
video: Video to filter by. If None, uses the first video.
untracked: If True, index by instance order instead of tracks.
If False (default), organize instances by their track assignment.
return_confidence: If True, include confidence as third coordinate.
For user instances, confidence is set to 1.0.
user_instances: If True (default), prefer user instances over predicted
instances. If False, only include predicted instances.
Returns:
Array of shape (n_frames, n_tracks, n_nodes, 2) or
(n_frames, n_tracks, n_nodes, 3) if return_confidence is True.
Missing data is filled with np.nan.
"""
# Step 1: Determine video_id to filter
if video is None:
video_id = 0
else:
video_id = self.videos.index(video)
# Step 2: Filter frames by video and get frame range
frames_data = self.frames_data
video_mask = frames_data["video"] == video_id
video_frame_data = frames_data[video_mask]
n_frames_data = len(video_frame_data)
if n_frames_data == 0:
# No frames for this video, return empty array
skeleton = self.skeletons[0] if self.skeletons else None
n_nodes = len(skeleton.nodes) if skeleton else 0
n_coords = 3 if return_confidence else 2
return np.full((0, 0, n_nodes, n_coords), np.nan, dtype="float32")
# Get frame index range for this video
frame_indices = video_frame_data["frame_idx"]
first_frame = int(frame_indices.min())
last_frame = int(frame_indices.max())
# Use video length when available so output spans the full video.
video_obj = (
video
if video is not None
else (self.videos[video_id] if video_id < len(self.videos) else None)
)
if video_obj is not None:
video_length = len(video_obj)
if video_length > 0:
last_frame = max(last_frame, video_length - 1)
n_frames = last_frame - first_frame + 1
# Step 3: Determine output dimensions
skeleton = self.skeletons[-1] # Use last skeleton (consistent with eager)
n_nodes = len(skeleton.nodes)
# Count max instances across frames (matches eager behavior)
n_instances = self._count_max_instances_per_frame(
video_frame_data, user_instances
)
# Single instance case forces untracked mode (matches eager behavior)
is_single_instance = n_instances == 1
untracked = untracked or is_single_instance
if untracked:
n_tracks = n_instances
else:
# Use track count (can be 0 if no tracks)
n_tracks = len(self.tracks)
n_coords = 3 if return_confidence else 2
# Step 4: Allocate output array
output = np.full(
(n_frames, n_tracks, n_nodes, n_coords), np.nan, dtype="float32"
)
# Step 5: Build frame_idx to data index mapping
frame_idx_to_data = {}
for data_idx, row in enumerate(video_frame_data):
frame_idx_to_data[int(row["frame_idx"])] = (row, data_idx)
# Step 6: Fill from raw data
for frame_idx in range(first_frame, last_frame + 1):
if frame_idx not in frame_idx_to_data:
continue
frame_row, _ = frame_idx_to_data[frame_idx]
out_idx = frame_idx - first_frame
self._fill_frame_numpy(
output[out_idx],
frame_row,
untracked=untracked,
return_confidence=return_confidence,
user_instances=user_instances,
)
return output
def _count_max_instances_per_frame(
self,
video_frame_data: np.ndarray,
user_instances: bool,
) -> int:
"""Count maximum instances across frames for untracked mode.
This matches the eager implementation: when user_instances=True,
counts max(n_user, n_predicted); when user_instances=False, counts
only predicted instances.
Args:
video_frame_data: Filtered frame data for a single video.
user_instances: Whether to include user instances.
Returns:
Maximum number of instances in any frame.
"""
max_count = 0
for frame_row in video_frame_data:
inst_start = int(frame_row["instance_id_start"])
inst_end = int(frame_row["instance_id_end"])
n_user = 0
n_pred = 0
for i in range(inst_start, inst_end):
if self.instances_data[i]["instance_type"] == InstanceType.USER:
n_user += 1
else:
n_pred += 1
if user_instances:
# When user_instances=True (and predicted_instances=True implicitly),
# count max of either user or predicted (matches eager behavior)
frame_count = max(n_user, n_pred)
else:
# When user_instances=False, only count predicted instances
frame_count = n_pred
max_count = max(max_count, frame_count)
return max_count
def _fill_frame_numpy(
self,
output: np.ndarray,
frame_row: np.ndarray,
untracked: bool,
return_confidence: bool,
user_instances: bool,
) -> None:
"""Fill a single frame's slice of the output array.
Args:
output: Output array slice of shape (n_tracks, n_nodes, n_coords).
frame_row: Row from frames_data for this frame.
untracked: Whether to use untracked (arbitrary order) indexing.
return_confidence: Whether to include confidence scores.
user_instances: Whether to prefer user instances.
"""
inst_start = int(frame_row["instance_id_start"])
inst_end = int(frame_row["instance_id_end"])
if untracked:
# Fill instances in arbitrary order
# When user_instances=True: prefer user instances, then add predicted
# instances that don't overlap
# When user_instances=False: only include predicted instances
j = 0
instances_to_process = []
# Separate instances by type
user_insts = []
pred_insts = []
for i in range(inst_start, inst_end):
if self.instances_data[i]["instance_type"] == InstanceType.USER:
user_insts.append(i)
else:
pred_insts.append(i)
if user_instances:
# First collect user instances
if user_insts:
instances_to_process.extend(user_insts)
# Check if this is single-instance case (n_instances == 1)
# In that case, we don't add predicted instances
is_single_instance = output.shape[0] == 1
if not is_single_instance:
# Add predicted instances that don't overlap with user instances
for pred_idx in pred_insts:
pred_row = self.instances_data[pred_idx]
skip = False
for user_idx in user_insts:
user_row = self.instances_data[user_idx]
# Skip if user and predicted share same track
user_track = int(user_row["track"])
pred_track = int(pred_row["track"])
if (
user_track >= 0
and pred_track >= 0
and user_track == pred_track
):
skip = True
break
# Skip if linked via from_predicted
from_pred = int(user_row["from_predicted"])
inst_id = int(pred_row["instance_id"])
if from_pred == inst_id:
skip = True
break
if not skip:
instances_to_process.append(pred_idx)
else:
# No user instances, use predicted
instances_to_process.extend(pred_insts)
else:
# Only include predicted instances
instances_to_process.extend(pred_insts)
# Fill output
for inst_idx in instances_to_process:
if j >= output.shape[0]:
break
self._fill_instance_numpy(output[j], inst_idx, return_confidence)
j += 1
else:
# Organize by track
# Build track -> instance mapping, preferring user instances
track_to_inst = {}
# First pass: add predicted instances
for i in range(inst_start, inst_end):
inst_row = self.instances_data[i]
if inst_row["instance_type"] == InstanceType.PREDICTED:
track_id = int(inst_row["track"])
if track_id >= 0:
track_to_inst[track_id] = i
# Second pass: add user instances (overwriting predicted if same track)
if user_instances:
for i in range(inst_start, inst_end):
inst_row = self.instances_data[i]
if inst_row["instance_type"] == InstanceType.USER:
track_id = int(inst_row["track"])
if track_id >= 0:
track_to_inst[track_id] = i
# Fill output by track
for track_id, inst_idx in track_to_inst.items():
if track_id < output.shape[0]:
self._fill_instance_numpy(
output[track_id], inst_idx, return_confidence
)
def _fill_instance_numpy(
self,
output: np.ndarray,
inst_idx: int,
return_confidence: bool,
) -> None:
"""Fill instance points into output array.
Args:
output: Output array slice of shape (n_nodes, n_coords).
inst_idx: Index into instances_data.
return_confidence: Whether to include confidence scores.
"""
inst_row = self.instances_data[inst_idx]
# Parse instance data - handle format differences
if self.format_id < 1.2:
(
_instance_id,
instance_type,
_frame_id,
_skeleton_id,
_track_id,
_from_predicted,
_instance_score,
point_id_start,
point_id_end,
) = inst_row
else:
(
_instance_id,
instance_type,
_frame_id,
_skeleton_id,
_track_id,
_from_predicted,
_instance_score,
point_id_start,
point_id_end,
_tracking_score,
) = inst_row
point_id_start = int(point_id_start)
point_id_end = int(point_id_end)
if instance_type == InstanceType.USER:
pts_data = self.points_data[point_id_start:point_id_end]
x = pts_data["x"]
y = pts_data["y"]
# Apply legacy coordinate adjustment if needed
if self.format_id < 1.1:
x = x - 0.5
y = y - 0.5
output[:, 0] = x
output[:, 1] = y
if return_confidence:
# User instances have confidence of 1.0
output[:, 2] = 1.0
else: # PREDICTED
pts_data = self.pred_points_data[point_id_start:point_id_end]
x = pts_data["x"]
y = pts_data["y"]
# Apply legacy coordinate adjustment if needed
if self.format_id < 1.1:
x = x - 0.5
y = y - 0.5
output[:, 0] = x
output[:, 1] = y
if return_confidence:
output[:, 2] = pts_data["score"]
__annotations__ = {'frames_data': 'np.ndarray', 'instances_data': 'np.ndarray', 'pred_points_data': 'np.ndarray', 'points_data': 'np.ndarray', 'videos': "list['Video']", 'skeletons': "list['Skeleton']", 'tracks': "list['Track']", 'format_id': 'float', '_source_path': 'str | None', '_negative_frames': 'set[tuple[int, int]]', '_centroid_by_frame': 'dict', '_bbox_by_frame': 'dict', '_mask_by_frame': 'dict', '_label_image_by_frame': 'dict', '_roi_by_frame': 'dict', '_undistributed_rois': 'list', '_undistributed_masks': 'list', '_undistributed_bboxes': 'list', '_undistributed_centroids': 'list', '_undistributed_label_images': 'list', 'identities': "list['Identity']", '_instance_identities': 'dict', '_instance_embeddings': 'dict', 'categories': "list['Category']", '_instance_categories': 'dict', '_instance_category_embeddings': 'dict', 'sessions_json_raw': "'np.ndarray | None'", 'session_data': "'dict | None'", 'session_video_ids': 'tuple'}
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__ = 'Holds raw HDF5 data and provides lazy access methods.\n\nAttributes:\n frames_data: Structured array from /frames HDF5 dataset.\n Fields: frame_id, video_id, frame_idx, instance_id_start, instance_id_end.\n instances_data: Structured array from /instances HDF5 dataset.\n Fields vary by format_id but include: instance_id, instance_type, frame_id,\n skeleton_id, track_id, from_predicted, instance_score, point_id_start,\n point_id_end, and optionally tracking_score.\n pred_points_data: Structured array from /pred_points HDF5 dataset.\n Fields: x, y, visible, complete, score.\n points_data: Structured array from /points HDF5 dataset.\n Fields: x, y, visible, complete.\n videos: List of eagerly loaded Video objects.\n skeletons: List of eagerly loaded Skeleton objects.\n tracks: List of eagerly loaded Track objects.\n format_id: SLP format version.\n _source_path: Path to source SLP file (for debugging).\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__ = 28
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__ = ('frames_data', 'instances_data', 'pred_points_data', 'points_data', 'videos', 'skeletons', 'tracks', 'format_id', '_source_path', '_negative_frames', '_centroid_by_frame', '_bbox_by_frame', '_mask_by_frame', '_label_image_by_frame', '_roi_by_frame', '_undistributed_rois', '_undistributed_masks', '_undistributed_bboxes', '_undistributed_centroids', '_undistributed_label_images', 'identities', '_instance_identities', '_instance_embeddings', 'categories', '_instance_categories', '_instance_category_embeddings', 'sessions_json_raw', 'session_data', 'session_video_ids')
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.slp_lazy'
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__ = ('frames_data', 'instances_data', 'pred_points_data', 'points_data', 'videos', 'skeletons', 'tracks', 'format_id', '_source_path', '_negative_frames', '_centroid_by_frame', '_bbox_by_frame', '_mask_by_frame', '_label_image_by_frame', '_roi_by_frame', '_undistributed_rois', '_undistributed_masks', '_undistributed_bboxes', '_undistributed_centroids', '_undistributed_label_images', 'identities', '_instance_identities', '_instance_embeddings', 'categories', '_instance_categories', '_instance_category_embeddings', 'sessions_json_raw', 'session_data', 'session_video_ids', '__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__ = ()
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
__attrs_post_init__()
¶
__eq__(other)
¶
Method generated by attrs for class LazyDataStore.
Source code in sleap_io/io/slp_lazy.py
if TYPE_CHECKING:
from sleap_io.model.category import Category
from sleap_io.model.identity import Identity
from sleap_io.model.instance import Instance, PredictedInstance, Track
from sleap_io.model.labeled_frame import LabeledFrame
from sleap_io.model.skeleton import Skeleton
from sleap_io.model.video import Video
from sleap_io.io.slp import InstanceType
@attrs.define
class LazyDataStore:
"""Holds raw HDF5 data and provides lazy access methods.
Attributes:
frames_data: Structured array from /frames HDF5 dataset.
Fields: frame_id, video_id, frame_idx, instance_id_start, instance_id_end.
instances_data: Structured array from /instances HDF5 dataset.
Fields vary by format_id but include: instance_id, instance_type, frame_id,
skeleton_id, track_id, from_predicted, instance_score, point_id_start,
point_id_end, and optionally tracking_score.
pred_points_data: Structured array from /pred_points HDF5 dataset.
Fields: x, y, visible, complete, score.
points_data: Structured array from /points HDF5 dataset.
Fields: x, y, visible, complete.
videos: List of eagerly loaded Video objects.
skeletons: List of eagerly loaded Skeleton objects.
tracks: List of eagerly loaded Track objects.
format_id: SLP format version.
_source_path: Path to source SLP file (for debugging).
"""
__init__(frames_data, instances_data, pred_points_data, points_data, videos, skeletons, tracks, format_id, source_path=None, negative_frames=NOTHING, centroid_by_frame=NOTHING, bbox_by_frame=NOTHING, mask_by_frame=NOTHING, label_image_by_frame=NOTHING, roi_by_frame=NOTHING, undistributed_rois=NOTHING, undistributed_masks=NOTHING, undistributed_bboxes=NOTHING, undistributed_centroids=NOTHING, undistributed_label_images=NOTHING, identities=NOTHING, instance_identities=NOTHING, instance_embeddings=NOTHING, categories=NOTHING, instance_categories=NOTHING, instance_category_embeddings=NOTHING, sessions_json_raw=None, session_data=None, session_video_ids=())
¶
Method generated by attrs for class LazyDataStore.
Source code in sleap_io/io/slp_lazy.py
# Raw arrays
frames_data: np.ndarray
instances_data: np.ndarray
pred_points_data: np.ndarray
points_data: np.ndarray
# References
videos: list["Video"]
skeletons: list["Skeleton"]
tracks: list["Track"]
# Metadata
format_id: float
_source_path: str | None = attrs.field(default=None, alias="source_path")
_negative_frames: set[tuple[int, int]] = attrs.field(
factory=set, alias="negative_frames"
)
# Per-frame annotation lookups: (video_idx, frame_idx) -> list[annotation]
# These are eagerly loaded from HDF5 but attached to frames lazily.
_centroid_by_frame: dict = attrs.field(
factory=dict, repr=False, alias="centroid_by_frame"
)
_bbox_by_frame: dict = attrs.field(factory=dict, repr=False, alias="bbox_by_frame")
_mask_by_frame: dict = attrs.field(factory=dict, repr=False, alias="mask_by_frame")
_label_image_by_frame: dict = attrs.field(
factory=dict, repr=False, alias="label_image_by_frame"
)
_roi_by_frame: dict = attrs.field(factory=dict, repr=False, alias="roi_by_frame")
# Undistributed annotations (video=None or frame_idx=None, e.g. static ROIs)
_undistributed_rois: list = attrs.field(factory=list, repr=False)
_undistributed_masks: list = attrs.field(factory=list, repr=False)
_undistributed_bboxes: list = attrs.field(factory=list, repr=False)
_undistributed_centroids: list = attrs.field(factory=list, repr=False)
_undistributed_label_images: list = attrs.field(factory=list, repr=False)
# Global identity catalog and per-instance identity links (format 2.5+).
# ``identities`` are canonical objects shared with Labels.identities;
# ``_instance_identities`` maps global instance_id -> (identity_idx, score).
identities: list["Identity"] = attrs.field(factory=list, repr=False)
_instance_identities: dict = attrs.field(
factory=dict, repr=False, alias="instance_identities"
)
# Per-instance re-ID embeddings (format 2.5+): instance_id -> Embedding.
_instance_embeddings: dict = attrs.field(
factory=dict, repr=False, alias="instance_embeddings"
)
# Global category catalog and per-instance category links (format 2.7+).
# ``categories`` are canonical objects shared with Labels.categories;
# ``_instance_categories`` maps global instance_id -> (category_idx, score).
categories: list["Category"] = attrs.field(factory=list, repr=False)
_instance_categories: dict = attrs.field(
factory=dict, repr=False, alias="instance_categories"
)
# Per-instance category (classification) embeddings (format 2.7+):
# instance_id -> Embedding.
_instance_category_embeddings: dict = attrs.field(
factory=dict, repr=False, alias="instance_category_embeddings"
)
# Raw RecordingSession payload for lossless lazy passthrough (format 2.8+).
# ``sessions_json_raw`` is the verbatim variable-length ``sessions_json`` bytes
# array; ``session_data`` is the dict of columnar ``/session_data`` arrays (see
# slp._read_session_data). Held so a lazy re-save can copy the frame-group / 3D
# tables verbatim without materializing frames -- the eager path drops them.
sessions_json_raw: "np.ndarray | None" = attrs.field(
default=None, repr=False, alias="sessions_json_raw"
)
session_data: "dict | None" = attrs.field(
default=None, repr=False, alias="session_data"
)
# Immutable snapshot of the video identity order at load time. The passthrough's
# sessions_json encodes video indices, so it is only safe to copy verbatim when
# the current video list matches this order (see slp._videos_unchanged).
session_video_ids: tuple = attrs.field(
default=(), repr=False, alias="session_video_ids"
)
def __attrs_post_init__(self) -> None:
"""Validate index bounds on construction."""
__len__()
¶
__replace__(*args, **changes)
¶
Method generated by attrs for class LazyDataStore.
Source code in sleap_io/io/slp_lazy.py
# Single instance case forces untracked mode (matches eager behavior)
is_single_instance = n_instances == 1
untracked = untracked or is_single_instance
if untracked:
n_tracks = n_instances
else:
# Use track count (can be 0 if no tracks)
n_tracks = len(self.tracks)
n_coords = 3 if return_confidence else 2
# Step 4: Allocate output array
output = np.full(
(n_frames, n_tracks, n_nodes, n_coords), np.nan, dtype="float32"
)
# Step 5: Build frame_idx to data index mapping
frame_idx_to_data = {}
for data_idx, row in enumerate(video_frame_data):
frame_idx_to_data[int(row["frame_idx"])] = (row, data_idx)
# Step 6: Fill from raw data
for frame_idx in range(first_frame, last_frame + 1):
if frame_idx not in frame_idx_to_data:
continue
frame_row, _ = frame_idx_to_data[frame_idx]
out_idx = frame_idx - first_frame
self._fill_frame_numpy(
output[out_idx],
frame_row,
untracked=untracked,
return_confidence=return_confidence,
user_instances=user_instances,
)
return output
def _count_max_instances_per_frame(
self,
video_frame_data: np.ndarray,
user_instances: bool,
) -> int:
"""Count maximum instances across frames for untracked mode.
This matches the eager implementation: when user_instances=True,
counts max(n_user, n_predicted); when user_instances=False, counts
only predicted instances.
Args:
video_frame_data: Filtered frame data for a single video.
user_instances: Whether to include user instances.
Returns:
__repr__()
¶
Method generated by attrs for class LazyDataStore.
Source code in sleap_io/io/slp_lazy.py
"""Lazy loading support for SLP files.
This module provides LazyDataStore and LazyFrameList classes that enable
deferred materialization of LabeledFrame and Instance objects when loading
SLP files with lazy=True.
These classes are implementation details - users interact with Labels objects.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Iterator
import attrs
import numpy as np
copy()
¶
Create an independent copy with copied arrays.
The returned copy has independent numpy arrays but shares references to Video, Skeleton, and Track objects. The metadata objects are shared because they are the canonical objects referenced by the Labels; copying them here would create inconsistency with Labels.videos/skeletons/tracks.
Returns:
| Type | Description |
|---|---|
LazyDataStore
|
A new LazyDataStore with copied arrays. |
Source code in sleap_io/io/slp_lazy.py
def copy(self) -> "LazyDataStore":
"""Create an independent copy with copied arrays.
The returned copy has independent numpy arrays but shares references
to Video, Skeleton, and Track objects. The metadata objects are shared
because they are the canonical objects referenced by the Labels; copying
them here would create inconsistency with Labels.videos/skeletons/tracks.
Returns:
A new LazyDataStore with copied arrays.
"""
from copy import deepcopy
new_store = LazyDataStore(
frames_data=self.frames_data.copy(),
instances_data=self.instances_data.copy(),
pred_points_data=self.pred_points_data.copy(),
points_data=self.points_data.copy(),
videos=self.videos, # Share references (canonical objects)
skeletons=self.skeletons, # Share references (canonical objects)
tracks=self.tracks, # Share references (canonical objects)
identities=self.identities, # Share references (canonical objects)
instance_identities=dict(self._instance_identities),
instance_embeddings=dict(self._instance_embeddings),
categories=self.categories, # Share references (canonical objects)
instance_categories=dict(self._instance_categories),
instance_category_embeddings=dict(self._instance_category_embeddings),
format_id=self.format_id,
source_path=self._source_path,
negative_frames=self._negative_frames.copy(),
centroid_by_frame={
k: [deepcopy(c) for c in v] for k, v in self._centroid_by_frame.items()
},
bbox_by_frame={
k: [deepcopy(b) for b in v] for k, v in self._bbox_by_frame.items()
},
mask_by_frame={
k: [deepcopy(m) for m in v] for k, v in self._mask_by_frame.items()
},
label_image_by_frame={
k: [deepcopy(li) for li in v]
for k, v in self._label_image_by_frame.items()
},
roi_by_frame={
k: [deepcopy(r) for r in v] for k, v in self._roi_by_frame.items()
},
)
# Copy undistributed annotations
new_store._undistributed_rois = [deepcopy(r) for r in self._undistributed_rois]
new_store._undistributed_masks = [
deepcopy(m) for m in self._undistributed_masks
]
new_store._undistributed_bboxes = [
deepcopy(b) for b in self._undistributed_bboxes
]
new_store._undistributed_centroids = [
deepcopy(c) for c in self._undistributed_centroids
]
new_store._undistributed_label_images = [
deepcopy(li) for li in self._undistributed_label_images
]
return new_store
get_user_frame_indices()
¶
Find indices of frames containing user (non-predicted) instances.
This also includes frames marked as negative (is_negative=True), since those are considered user-labeled even though they have no instances.
Returns:
| Type | Description |
|---|---|
list[int]
|
List of frame indices (into frames_data) that have at least one user instance or are marked as negative. |
Source code in sleap_io/io/slp_lazy.py
def get_user_frame_indices(self) -> list[int]:
"""Find indices of frames containing user (non-predicted) instances.
This also includes frames marked as negative (is_negative=True), since
those are considered user-labeled even though they have no instances.
Returns:
List of frame indices (into frames_data) that have at least one user
instance or are marked as negative.
"""
from sleap_io.io.slp import InstanceType
result_indices: set[int] = set()
# Find all user instances
user_mask = self.instances_data["instance_type"] == InstanceType.USER
if np.any(user_mask):
# Get frame boundaries for binary search
frame_ends = self.frames_data["instance_id_end"]
# Use binary search to find frame for each user instance - O(n log m)
user_instance_indices = np.where(user_mask)[0]
# searchsorted finds insertion point; instance i is in frame fi where
# frame_ends[fi-1] <= i < frame_ends[fi] (with frame_ends[-1] = 0)
frame_indices = np.searchsorted(
frame_ends, user_instance_indices, side="right"
)
# Get unique frame indices (already sorted by searchsorted)
unique_frames = np.unique(frame_indices)
# Filter out any out-of-bounds indices
valid_mask = unique_frames < len(self.frames_data)
result_indices.update(unique_frames[valid_mask].tolist())
# Also include negative frames
if self._negative_frames:
for idx in range(len(self.frames_data)):
frame_row = self.frames_data[idx]
video_id = int(frame_row[1])
frame_idx = int(frame_row[2])
if (video_id, frame_idx) in self._negative_frames:
result_indices.add(idx)
return sorted(result_indices)
materialize_all()
¶
Materialize all frames.
Returns:
| Type | Description |
|---|---|
list[LabeledFrame]
|
List of all LabeledFrame objects. |
materialize_frame(idx)
¶
Create a fully materialized LabeledFrame.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
idx
|
int
|
Index into frames_data array. |
required |
Returns:
| Type | Description |
|---|---|
LabeledFrame
|
A real LabeledFrame with real Instance objects. |
Source code in sleap_io/io/slp_lazy.py
def materialize_frame(self, idx: int) -> "LabeledFrame":
"""Create a fully materialized LabeledFrame.
Args:
idx: Index into frames_data array.
Returns:
A real LabeledFrame with real Instance objects.
"""
from sleap_io.model.labeled_frame import LabeledFrame
frame_row = self.frames_data[idx]
video_id = int(frame_row[1]) # video_id
frame_idx = int(frame_row[2]) # frame_idx
inst_start = int(frame_row[3]) # instance_id_start
inst_end = int(frame_row[4]) # instance_id_end
instances = []
for inst_idx in range(inst_start, inst_end):
inst = self._materialize_instance(inst_idx)
instances.append(inst)
is_negative = (video_id, frame_idx) in self._negative_frames
# Attach per-frame annotations from eagerly-loaded dicts
key = (video_id, frame_idx)
centroids = self._centroid_by_frame.get(key, [])
bboxes = self._bbox_by_frame.get(key, [])
masks = self._mask_by_frame.get(key, [])
label_images = self._label_image_by_frame.get(key, [])
rois = self._roi_by_frame.get(key, [])
return LabeledFrame(
video=self.videos[video_id],
frame_idx=frame_idx,
instances=instances,
is_negative=is_negative,
centroids=centroids,
bboxes=bboxes,
masks=masks,
label_images=label_images,
rois=rois,
)
to_numpy(video=None, untracked=False, return_confidence=False, user_instances=True)
¶
Build numpy array directly from raw data (fast path).
This method builds the output array directly from raw HDF5 data without creating any Instance or LabeledFrame objects, providing significant performance improvement for workflows that only need numpy output.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
video
|
Video | None
|
Video to filter by. If None, uses the first video. |
None
|
untracked
|
bool
|
If True, index by instance order instead of tracks. If False (default), organize instances by their track assignment. |
False
|
return_confidence
|
bool
|
If True, include confidence as third coordinate. For user instances, confidence is set to 1.0. |
False
|
user_instances
|
bool
|
If True (default), prefer user instances over predicted instances. If False, only include predicted instances. |
True
|
Returns:
| Type | Description |
|---|---|
ndarray
|
Array of shape (n_frames, n_tracks, n_nodes, 2) or (n_frames, n_tracks, n_nodes, 3) if return_confidence is True. Missing data is filled with np.nan. |
Source code in sleap_io/io/slp_lazy.py
def to_numpy(
self,
video: "Video | None" = None,
untracked: bool = False,
return_confidence: bool = False,
user_instances: bool = True,
) -> np.ndarray:
"""Build numpy array directly from raw data (fast path).
This method builds the output array directly from raw HDF5 data without
creating any Instance or LabeledFrame objects, providing significant
performance improvement for workflows that only need numpy output.
Args:
video: Video to filter by. If None, uses the first video.
untracked: If True, index by instance order instead of tracks.
If False (default), organize instances by their track assignment.
return_confidence: If True, include confidence as third coordinate.
For user instances, confidence is set to 1.0.
user_instances: If True (default), prefer user instances over predicted
instances. If False, only include predicted instances.
Returns:
Array of shape (n_frames, n_tracks, n_nodes, 2) or
(n_frames, n_tracks, n_nodes, 3) if return_confidence is True.
Missing data is filled with np.nan.
"""
# Step 1: Determine video_id to filter
if video is None:
video_id = 0
else:
video_id = self.videos.index(video)
# Step 2: Filter frames by video and get frame range
frames_data = self.frames_data
video_mask = frames_data["video"] == video_id
video_frame_data = frames_data[video_mask]
n_frames_data = len(video_frame_data)
if n_frames_data == 0:
# No frames for this video, return empty array
skeleton = self.skeletons[0] if self.skeletons else None
n_nodes = len(skeleton.nodes) if skeleton else 0
n_coords = 3 if return_confidence else 2
return np.full((0, 0, n_nodes, n_coords), np.nan, dtype="float32")
# Get frame index range for this video
frame_indices = video_frame_data["frame_idx"]
first_frame = int(frame_indices.min())
last_frame = int(frame_indices.max())
# Use video length when available so output spans the full video.
video_obj = (
video
if video is not None
else (self.videos[video_id] if video_id < len(self.videos) else None)
)
if video_obj is not None:
video_length = len(video_obj)
if video_length > 0:
last_frame = max(last_frame, video_length - 1)
n_frames = last_frame - first_frame + 1
# Step 3: Determine output dimensions
skeleton = self.skeletons[-1] # Use last skeleton (consistent with eager)
n_nodes = len(skeleton.nodes)
# Count max instances across frames (matches eager behavior)
n_instances = self._count_max_instances_per_frame(
video_frame_data, user_instances
)
# Single instance case forces untracked mode (matches eager behavior)
is_single_instance = n_instances == 1
untracked = untracked or is_single_instance
if untracked:
n_tracks = n_instances
else:
# Use track count (can be 0 if no tracks)
n_tracks = len(self.tracks)
n_coords = 3 if return_confidence else 2
# Step 4: Allocate output array
output = np.full(
(n_frames, n_tracks, n_nodes, n_coords), np.nan, dtype="float32"
)
# Step 5: Build frame_idx to data index mapping
frame_idx_to_data = {}
for data_idx, row in enumerate(video_frame_data):
frame_idx_to_data[int(row["frame_idx"])] = (row, data_idx)
# Step 6: Fill from raw data
for frame_idx in range(first_frame, last_frame + 1):
if frame_idx not in frame_idx_to_data:
continue
frame_row, _ = frame_idx_to_data[frame_idx]
out_idx = frame_idx - first_frame
self._fill_frame_numpy(
output[out_idx],
frame_row,
untracked=untracked,
return_confidence=return_confidence,
user_instances=user_instances,
)
return output
validate()
¶
Check that all indices are within bounds.
Raises:
| Type | Description |
|---|---|
ValueError
|
If any index is out of bounds. |
Source code in sleap_io/io/slp_lazy.py
def validate(self) -> None:
"""Check that all indices are within bounds.
Raises:
ValueError: If any index is out of bounds.
"""
n_frames = len(self.frames_data)
n_instances = len(self.instances_data)
n_points = len(self.points_data)
n_pred_points = len(self.pred_points_data)
if n_frames == 0:
return # Empty data is valid
# Validate frame -> instance references
max_inst_end = self.frames_data["instance_id_end"].max() if n_frames > 0 else 0
if max_inst_end > n_instances:
raise ValueError(
f"Frame references instance index {max_inst_end} but only "
f"{n_instances} instances exist."
)
if n_instances == 0:
return # No instances means no points to validate
# Validate instance -> point references
# Separate user instances and predicted instances
user_mask = self.instances_data["instance_type"] == InstanceType.USER
pred_mask = self.instances_data["instance_type"] == InstanceType.PREDICTED
if np.any(user_mask):
user_max_end = self.instances_data[user_mask]["point_id_end"].max()
if user_max_end > n_points:
raise ValueError(
f"User instance references point index {user_max_end} but only "
f"{n_points} points exist."
)
if np.any(pred_mask):
pred_max_end = self.instances_data[pred_mask]["point_id_end"].max()
if pred_max_end > n_pred_points:
raise ValueError(
f"Predicted instance references pred_point index {pred_max_end} "
f"but only {n_pred_points} pred_points exist."
)
LazyFrameList
¶
List-like proxy that materializes LabeledFrame objects on access.
This provides backward compatibility for code that accesses labels.labeled_frames directly. Frames are created on-demand when accessed via indexing or iteration.
Mutations are blocked with helpful error messages suggesting to call labels.materialize() first.
Methods:
| Name | Description |
|---|---|
__delitem__ |
Block item deletion with helpful error. |
__getitem__ |
Get frame(s) by index or slice. |
__init__ |
Initialize with a LazyDataStore. |
__iter__ |
Iterate over frames, materializing each. |
__len__ |
Return number of frames. |
__repr__ |
Return informative representation. |
__setitem__ |
Block item assignment with helpful error. |
append |
Block append with helpful error. |
extend |
Block extend with helpful error. |
insert |
Block insert with helpful error. |
Attributes:
| Name | Type | Description |
|---|---|---|
__dict__ |
Read-only proxy of a mapping. |
|
__doc__ |
str(object='') -> str |
|
__firstlineno__ |
int([x]) -> integer |
|
__module__ |
str(object='') -> str |
|
__static_attributes__ |
Built-in immutable sequence. |
|
__weakref__ |
list of weak references to the object |
Source code in sleap_io/io/slp_lazy.py
class LazyFrameList:
"""List-like proxy that materializes LabeledFrame objects on access.
This provides backward compatibility for code that accesses
labels.labeled_frames directly. Frames are created on-demand when
accessed via indexing or iteration.
Mutations are blocked with helpful error messages suggesting to call
labels.materialize() first.
"""
def __init__(self, store: LazyDataStore) -> None:
"""Initialize with a LazyDataStore.
Args:
store: The LazyDataStore containing raw frame data.
"""
self._store = store
self._supplementary: list["LabeledFrame"] = []
def __len__(self) -> int:
"""Return number of frames."""
return len(self._store) + len(self._supplementary)
def __getitem__(self, idx: int | slice) -> "LabeledFrame | list[LabeledFrame]":
"""Get frame(s) by index or slice.
Args:
idx: Integer index or slice object.
Returns:
A single LabeledFrame for integer indexing, or a list of
LabeledFrames for slicing.
Raises:
IndexError: If index is out of range.
"""
n = len(self)
n_store = len(self._store)
if isinstance(idx, slice):
# Handle slice
indices = range(*idx.indices(n))
results = []
for i in indices:
if i < n_store:
results.append(self._store.materialize_frame(i))
else:
results.append(self._supplementary[i - n_store])
return results
# Handle negative indexing
if idx < 0:
idx = n + idx
# Bounds check
if idx < 0 or idx >= n:
raise IndexError(f"Index {idx} out of range for {n} frames")
if idx < n_store:
return self._store.materialize_frame(idx)
return self._supplementary[idx - n_store]
def __iter__(self) -> Iterator["LabeledFrame"]:
"""Iterate over frames, materializing each."""
for i in range(len(self._store)):
yield self._store.materialize_frame(i)
yield from self._supplementary
def __repr__(self) -> str:
"""Return informative representation."""
return f"LazyFrameList(n_frames={len(self)})"
def _mutation_error(self, operation: str) -> RuntimeError:
"""Create a RuntimeError with helpful guidance for blocked mutations.
Args:
operation: Name of the blocked operation.
Returns:
RuntimeError with guidance message.
"""
return RuntimeError(
f"Cannot {operation} on LazyFrameList (lazy-loaded Labels).\n\n"
f"To modify, first materialize the Labels:\n"
f" labels = labels.materialize()\n"
f" labels.labeled_frames.{operation}(...)"
)
def append(self, item: "LabeledFrame") -> None:
"""Block append with helpful error.
Raises:
RuntimeError: Always, with guidance to materialize first.
"""
raise self._mutation_error("append")
def extend(self, items: list["LabeledFrame"]) -> None:
"""Block extend with helpful error.
Raises:
RuntimeError: Always, with guidance to materialize first.
"""
raise self._mutation_error("extend")
def insert(self, idx: int, item: "LabeledFrame") -> None:
"""Block insert with helpful error.
Raises:
RuntimeError: Always, with guidance to materialize first.
"""
raise self._mutation_error("insert")
def __setitem__(self, idx: int, value: "LabeledFrame") -> None:
"""Block item assignment with helpful error.
Raises:
RuntimeError: Always, with guidance to materialize first.
"""
raise self._mutation_error("__setitem__")
def __delitem__(self, idx: int) -> None:
"""Block item deletion with helpful error.
Raises:
RuntimeError: Always, with guidance to materialize first.
"""
raise self._mutation_error("__delitem__")
__dict__ = mappingproxy({'__module__': 'sleap_io.io.slp_lazy', '__firstlineno__': 856, '__doc__': 'List-like proxy that materializes LabeledFrame objects on access.\n\nThis provides backward compatibility for code that accesses\nlabels.labeled_frames directly. Frames are created on-demand when\naccessed via indexing or iteration.\n\nMutations are blocked with helpful error messages suggesting to call\nlabels.materialize() first.\n', '__init__': <function LazyFrameList.__init__ at 0x7f07ecdf9800>, '__len__': <function LazyFrameList.__len__ at 0x7f07ecdf98a0>, '__getitem__': <function LazyFrameList.__getitem__ at 0x7f07ecdf9d00>, '__iter__': <function LazyFrameList.__iter__ at 0x7f07ecdfb1a0>, '__repr__': <function LazyFrameList.__repr__ at 0x7f07ecdfb240>, '_mutation_error': <function LazyFrameList._mutation_error at 0x7f07ecdfb2e0>, 'append': <function LazyFrameList.append at 0x7f07ecdfb380>, 'extend': <function LazyFrameList.extend at 0x7f07ecdfb420>, 'insert': <function LazyFrameList.insert at 0x7f07ecdfb4c0>, '__setitem__': <function LazyFrameList.__setitem__ at 0x7f07ecdfb560>, '__delitem__': <function LazyFrameList.__delitem__ at 0x7f07ecdfb600>, '__static_attributes__': ('_store', '_supplementary'), '__dict__': <attribute '__dict__' of 'LazyFrameList' objects>, '__weakref__': <attribute '__weakref__' of 'LazyFrameList' objects>})
class-attribute
¶
Read-only proxy of a mapping.
__doc__ = 'List-like proxy that materializes LabeledFrame objects on access.\n\nThis provides backward compatibility for code that accesses\nlabels.labeled_frames directly. Frames are created on-demand when\naccessed via indexing or iteration.\n\nMutations are blocked with helpful error messages suggesting to call\nlabels.materialize() first.\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__ = 856
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
__module__ = 'sleap_io.io.slp_lazy'
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__ = ('_store', '_supplementary')
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
__delitem__(idx)
¶
Block item deletion with helpful error.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
Always, with guidance to materialize first. |
__getitem__(idx)
¶
Get frame(s) by index or slice.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
idx
|
int | slice
|
Integer index or slice object. |
required |
Returns:
| Type | Description |
|---|---|
LabeledFrame | list[LabeledFrame]
|
A single LabeledFrame for integer indexing, or a list of LabeledFrames for slicing. |
Raises:
| Type | Description |
|---|---|
IndexError
|
If index is out of range. |
Source code in sleap_io/io/slp_lazy.py
def __getitem__(self, idx: int | slice) -> "LabeledFrame | list[LabeledFrame]":
"""Get frame(s) by index or slice.
Args:
idx: Integer index or slice object.
Returns:
A single LabeledFrame for integer indexing, or a list of
LabeledFrames for slicing.
Raises:
IndexError: If index is out of range.
"""
n = len(self)
n_store = len(self._store)
if isinstance(idx, slice):
# Handle slice
indices = range(*idx.indices(n))
results = []
for i in indices:
if i < n_store:
results.append(self._store.materialize_frame(i))
else:
results.append(self._supplementary[i - n_store])
return results
# Handle negative indexing
if idx < 0:
idx = n + idx
# Bounds check
if idx < 0 or idx >= n:
raise IndexError(f"Index {idx} out of range for {n} frames")
if idx < n_store:
return self._store.materialize_frame(idx)
return self._supplementary[idx - n_store]
__init__(store)
¶
Initialize with a LazyDataStore.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
store
|
LazyDataStore
|
The LazyDataStore containing raw frame data. |
required |
__iter__()
¶
__len__()
¶
__repr__()
¶
__setitem__(idx, value)
¶
Block item assignment with helpful error.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
Always, with guidance to materialize first. |
append(item)
¶
Block append with helpful error.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
Always, with guidance to materialize first. |
extend(items)
¶
Block extend with helpful error.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
Always, with guidance to materialize first. |
insert(idx, item)
¶
Block insert with helpful error.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
Always, with guidance to materialize first. |