codecs
sleap_io.codecs
¶
In-memory serialization codecs for SLEAP Labels objects.
This package provides flexible conversion between Labels objects and various in-memory representations:
- DataFrames: Multiple formats (multi_index, points, instances, frames) with pandas/polars support
- Dictionaries: JSON-serializable primitive dictionaries
- NumPy: Via Labels.numpy() and Labels.from_numpy() with enhanced flexibility
The codecs package is designed for in-memory serialization, separate from disk I/O
operations in the sleap_io.io package. This separation allows for:
- Reusability: Common serialization code shared across I/O backends
- Flexibility: Work with Labels in different formats without touching disk
- Composability: Chain codecs (e.g., Labels → DataFrame → CSV)
Examples:
Convert to DataFrame for analysis:
>>> from sleap_io import load_file
>>> from sleap_io.codecs import to_dataframe
>>> labels = load_file("predictions.slp")
>>> df = to_dataframe(labels, format="instances")
>>> df.groupby("track")["nose.x"].mean()
Round-trip through dict:
>>> from sleap_io.codecs import to_dict
>>> d = to_dict(labels)
>>> import json
>>> json.dumps(d) # Fully JSON-serializable!
Use with I/O backends:
Modules:
| Name | Description |
|---|---|
dataframe |
DataFrame codec for SLEAP Labels objects. |
dictionary |
Dictionary codec for SLEAP Labels objects. |
numpy |
NumPy array codec for SLEAP Labels objects. |
Classes:
| Name | Description |
|---|---|
DataFrameFormat |
Enumeration of supported DataFrame formats. |
Functions:
| Name | Description |
|---|---|
from_dataframe |
Create a Labels object from a DataFrame. |
from_dict |
Create a Labels object from a dictionary. |
from_numpy |
Create a new Labels object from a numpy array of tracks. |
to_dataframe |
Convert Labels to a DataFrame. |
to_dataframe_iter |
Iterate over Labels data, yielding DataFrames in chunks. |
to_dict |
Convert Labels to a primitive dictionary (JSON-serializable). |
to_numpy |
Convert Labels to a numpy array. |
__all__ = ['DataFrameFormat', 'to_dataframe', 'to_dataframe_iter', 'from_dataframe', 'to_dict', 'from_dict', 'to_numpy', 'from_numpy']
module-attribute
¶
Built-in mutable sequence.
If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.
__cached__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/codecs/__pycache__/__init__.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__ = 'In-memory serialization codecs for SLEAP Labels objects.\n\nThis package provides flexible conversion between Labels objects and various\nin-memory representations:\n\n- **DataFrames**: Multiple formats (multi_index, points, instances, frames) with\n pandas/polars support\n- **Dictionaries**: JSON-serializable primitive dictionaries\n- **NumPy**: Via Labels.numpy() and Labels.from_numpy() with enhanced flexibility\n\nThe codecs package is designed for in-memory serialization, separate from disk I/O\noperations in the `sleap_io.io` package. This separation allows for:\n\n1. **Reusability**: Common serialization code shared across I/O backends\n2. **Flexibility**: Work with Labels in different formats without touching disk\n3. **Composability**: Chain codecs (e.g., Labels → DataFrame → CSV)\n\nExamples:\n Convert to DataFrame for analysis:\n\n >>> from sleap_io import load_file\n >>> from sleap_io.codecs import to_dataframe\n >>> labels = load_file("predictions.slp")\n >>> df = to_dataframe(labels, format="instances")\n >>> df.groupby("track")["nose.x"].mean()\n\n Round-trip through dict:\n\n >>> from sleap_io.codecs import to_dict\n >>> d = to_dict(labels)\n >>> import json\n >>> json.dumps(d) # Fully JSON-serializable!\n\n Use with I/O backends:\n\n >>> df = to_dataframe(labels, format="points")\n >>> df.to_csv("predictions.csv")\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/codecs/__init__.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.codecs'
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.codecs'
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'.
__path__ = ['/home/runner/work/sleap-io/sleap-io/sleap_io/codecs']
module-attribute
¶
Built-in mutable sequence.
If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.
DataFrameFormat
¶
Bases: builtins.str, enum.Enum
Enumeration of supported DataFrame formats.
Attributes:
| Name | Type | Description |
|---|---|---|
FRAMES |
Enumeration of supported DataFrame formats. |
|
INSTANCES |
Enumeration of supported DataFrame formats. |
|
MULTI_INDEX |
Enumeration of supported DataFrame formats. |
|
POINTS |
Enumeration of supported DataFrame formats. |
|
__doc__ |
str(object='') -> str |
|
__module__ |
str(object='') -> str |
Source code in sleap_io/codecs/dataframe.py
class DataFrameFormat(str, Enum):
"""Enumeration of supported DataFrame formats."""
POINTS = "points"
"""One row per point (frame, instance, node). Most normalized format."""
INSTANCES = "instances"
"""One row per instance. Columns for each node's x/y coordinates."""
FRAMES = "frames"
"""One row per frame-track combination. For trajectory analysis."""
MULTI_INDEX = "multi_index"
"""Hierarchical column structure. Similar to NWB format."""
FRAMES = <DataFrameFormat.FRAMES: 'frames'>
class-attribute
¶
Enumeration of supported DataFrame formats.
INSTANCES = <DataFrameFormat.INSTANCES: 'instances'>
class-attribute
¶
Enumeration of supported DataFrame formats.
MULTI_INDEX = <DataFrameFormat.MULTI_INDEX: 'multi_index'>
class-attribute
¶
Enumeration of supported DataFrame formats.
POINTS = <DataFrameFormat.POINTS: 'points'>
class-attribute
¶
Enumeration of supported DataFrame formats.
__doc__ = 'Enumeration of supported DataFrame formats.'
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.codecs.dataframe'
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'.
from_dataframe(df, *, video=None, skeleton=None, format=<DataFrameFormat.POINTS: 'points'>)
¶
Create a Labels object from a DataFrame.
This function reconstructs a Labels object from a DataFrame created by
to_dataframe(). Supports all formats: points, instances, frames, multi_index.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
df
|
DataFrame
|
DataFrame created by to_dataframe() or compatible structure. |
required |
video
|
Video | None
|
Video object to associate with all frames. Required if the DataFrame does not have video information. |
None
|
skeleton
|
Skeleton | None
|
Skeleton object to use. Required if the DataFrame does not have skeleton information or if the skeleton needs to be provided explicitly. |
None
|
format
|
DataFrameFormat | str
|
The format of the input DataFrame. One of "points", "instances", "frames", "multi_index". |
<DataFrameFormat.POINTS: 'points'>
|
Returns:
| Type | Description |
|---|---|
Labels
|
A Labels object reconstructed from the DataFrame. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If required columns are missing or format is invalid. |
Examples:
>>> df = to_dataframe(labels, format="points")
>>> labels_restored = from_dataframe(df, video=video, skeleton=skeleton)
>>> df = to_dataframe(labels, format="instances")
>>> labels_restored = from_dataframe(df, format="instances", skeleton=skeleton)
Notes
- The DataFrame must have the expected structure for the specified format.
- If video information is not in the DataFrame, a Video must be provided.
- If skeleton is not provided, it will be inferred from column names where possible.
- Tracks are reconstructed from track/track_name columns if present.
Source code in sleap_io/codecs/dataframe.py
def from_dataframe(
df: pd.DataFrame,
*,
video: Video | None = None,
skeleton: "Skeleton | None" = None, # noqa: F821
format: DataFrameFormat | str = DataFrameFormat.POINTS,
) -> Labels:
"""Create a Labels object from a DataFrame.
This function reconstructs a Labels object from a DataFrame created by
`to_dataframe()`. Supports all formats: points, instances, frames, multi_index.
Args:
df: DataFrame created by to_dataframe() or compatible structure.
video: Video object to associate with all frames. Required if the DataFrame
does not have video information.
skeleton: Skeleton object to use. Required if the DataFrame does not have
skeleton information or if the skeleton needs to be provided explicitly.
format: The format of the input DataFrame. One of "points", "instances",
"frames", "multi_index".
Returns:
A Labels object reconstructed from the DataFrame.
Raises:
ValueError: If required columns are missing or format is invalid.
Examples:
>>> df = to_dataframe(labels, format="points")
>>> labels_restored = from_dataframe(df, video=video, skeleton=skeleton)
>>> df = to_dataframe(labels, format="instances")
>>> labels_restored = from_dataframe(df, format="instances", skeleton=skeleton)
Notes:
- The DataFrame must have the expected structure for the specified format.
- If video information is not in the DataFrame, a Video must be provided.
- If skeleton is not provided, it will be inferred from column names where
possible.
- Tracks are reconstructed from track/track_name columns if present.
"""
# Normalize format parameter
if isinstance(format, str):
try:
format = DataFrameFormat(format.lower())
except ValueError:
valid_formats = ", ".join([f.value for f in DataFrameFormat])
raise ValueError(
f"Invalid format '{format}'. Must be one of: {valid_formats}"
)
if format == DataFrameFormat.POINTS:
return _from_points_df(df, video=video, skeleton=skeleton)
elif format == DataFrameFormat.INSTANCES:
return _from_instances_df(df, video=video, skeleton=skeleton)
elif format == DataFrameFormat.FRAMES:
return _from_frames_df(df, video=video, skeleton=skeleton)
elif format == DataFrameFormat.MULTI_INDEX:
return _from_multi_index_df(df, video=video, skeleton=skeleton)
else:
raise ValueError(f"Unknown format: {format}")
from_dict(data)
¶
Create a Labels object from a dictionary.
This is the inverse of to_dict() and reconstructs a Labels object from
its dictionary representation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict[str, Any]
|
Dictionary in the format produced by |
required |
Returns:
| Type | Description |
|---|---|
Labels
|
A Labels object reconstructed from the dictionary. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the dictionary format is invalid or missing required keys. |
Examples:
>>> # Round-trip through JSON
>>> import json
>>> json_str = json.dumps(to_dict(labels))
>>> labels_restored = from_dict(json.loads(json_str))
Notes
- The
from_predictedrelationship cannot be fully restored since the dictionary only indicates its presence, not the actual reference. - Video backends are not restored; videos are created with filename only.
Source code in sleap_io/codecs/dictionary.py
def from_dict(data: dict[str, Any]) -> Labels:
"""Create a Labels object from a dictionary.
This is the inverse of `to_dict()` and reconstructs a Labels object from
its dictionary representation.
Args:
data: Dictionary in the format produced by `to_dict()`.
Returns:
A Labels object reconstructed from the dictionary.
Raises:
ValueError: If the dictionary format is invalid or missing required keys.
Examples:
>>> d = to_dict(labels)
>>> labels_restored = from_dict(d)
>>> # Round-trip through JSON
>>> import json
>>> json_str = json.dumps(to_dict(labels))
>>> labels_restored = from_dict(json.loads(json_str))
Notes:
- The `from_predicted` relationship cannot be fully restored since the
dictionary only indicates its presence, not the actual reference.
- Video backends are not restored; videos are created with filename only.
"""
import numpy as np
from sleap_io.model.labeled_frame import LabeledFrame
from sleap_io.model.skeleton import Edge, Node, Skeleton, Symmetry
from sleap_io.model.suggestions import SuggestionFrame
# Validate required keys
required_keys = ["skeletons", "videos", "labeled_frames"]
for key in required_keys:
if key not in data:
raise ValueError(f"Missing required key: {key}")
# Build skeletons
skeletons = []
for skel_dict in data["skeletons"]:
# Create nodes
nodes = [Node(name=name) for name in skel_dict["nodes"]]
# Create edges
edges = []
for src_idx, dst_idx in skel_dict.get("edges", []):
edges.append(Edge(source=nodes[src_idx], destination=nodes[dst_idx]))
# Create symmetries
symmetries = []
for sym_indices in skel_dict.get("symmetries", []):
symmetries.append(
Symmetry(nodes={nodes[sym_indices[0]], nodes[sym_indices[1]]})
)
skeleton = Skeleton(
nodes=nodes,
edges=edges,
symmetries=symmetries,
name=skel_dict.get("name", ""),
)
skeletons.append(skeleton)
# Build videos
videos = []
for vid_dict in data["videos"]:
video = Video(filename=vid_dict["filename"])
videos.append(video)
# Build tracks
tracks = []
for track_dict in data.get("tracks", []):
track = Track(name=track_dict["name"])
tracks.append(track)
# Build labeled frames
labeled_frames = []
for lf_dict in data["labeled_frames"]:
video = videos[lf_dict["video_idx"]]
frame_idx = lf_dict["frame_idx"]
instances = []
for inst_dict in lf_dict.get("instances", []):
skeleton = skeletons[inst_dict["skeleton_idx"]]
is_predicted = inst_dict["type"] == "predicted_instance"
# Build points array
points_list = inst_dict["points"]
n_nodes = len(points_list)
points_data = np.full((n_nodes, 2), np.nan, dtype="float64")
for node_idx, pt in enumerate(points_list):
if pt.get("visible", True):
points_data[node_idx, 0] = pt["x"]
points_data[node_idx, 1] = pt["y"]
# Get track if present
track = None
if "track_idx" in inst_dict:
track = tracks[inst_dict["track_idx"]]
# Get tracking score if present
tracking_score = inst_dict.get("tracking_score")
if is_predicted:
# Get instance score
score = inst_dict.get("score", 0.0)
# Create predicted instance
instance = PredictedInstance.from_numpy(
points_data=points_data,
skeleton=skeleton,
score=score,
track=track,
tracking_score=tracking_score,
)
else:
# Create user instance
instance = Instance.from_numpy(
points_data=points_data,
skeleton=skeleton,
track=track,
tracking_score=tracking_score,
)
instances.append(instance)
labeled_frame = LabeledFrame(
video=video,
frame_idx=frame_idx,
instances=instances,
is_negative=lf_dict.get("is_negative", False),
)
labeled_frames.append(labeled_frame)
# Build suggestions
suggestions = []
for sug_dict in data.get("suggestions", []):
video = videos[sug_dict["video_idx"]]
suggestion = SuggestionFrame(
video=video,
frame_idx=sug_dict["frame_idx"],
)
suggestions.append(suggestion)
# Build provenance
provenance = dict(data.get("provenance", {}))
# Create Labels object
labels = Labels(
labeled_frames=labeled_frames,
videos=videos,
skeletons=skeletons,
tracks=tracks,
suggestions=suggestions,
provenance=provenance,
)
return labels
from_numpy(tracks_array, *, videos=None, video=None, skeletons=None, skeleton=None, tracks=None, track_names=None, first_frame=0, return_confidence=False)
¶
Create a new Labels object from a numpy array of tracks.
This factory method creates a new Labels object with instances constructed from the provided numpy array. It is a more flexible version of Labels.from_numpy().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tracks_array
|
ndarray
|
A numpy array of tracks, with shape
|
required |
videos
|
list[Video] | None
|
List of Video objects to associate with the labels. At least one
video is required. Mutually exclusive with |
None
|
video
|
Video | None
|
Single Video object to associate with the labels. Mutually exclusive
with |
None
|
skeletons
|
list[Skeleton] | Skeleton | None
|
Skeleton or list of Skeleton objects to use for the instances.
At least one skeleton is required. Mutually exclusive with |
None
|
skeleton
|
Skeleton | None
|
Single Skeleton object to use. Mutually exclusive with |
None
|
tracks
|
list[Track] | None
|
List of Track objects corresponding to the second dimension of the
array. If not specified, new tracks will be created automatically using
|
None
|
track_names
|
list[str] | None
|
List of track names to use when auto-creating tracks. Only used
if |
None
|
first_frame
|
int
|
Frame index to start the labeled frames from. Default is 0. |
0
|
return_confidence
|
bool
|
Whether the tracks array contains confidence scores in the last dimension. If True, tracks.shape[-1] should be 3. If False or None, will be inferred from array shape. |
False
|
Returns:
| Type | Description |
|---|---|
Labels
|
A new Labels object with instances constructed from the numpy array. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the array dimensions are invalid, or if no videos or
skeletons are provided, or if both |
Examples:
>>> import numpy as np
>>> from sleap_io import Video, Skeleton
>>> from sleap_io.codecs import from_numpy
>>> # Create a simple tracking array for 2 frames, 1 track, 2 nodes
>>> arr = np.zeros((2, 1, 2, 2))
>>> arr[0, 0] = [[10, 20], [30, 40]] # Frame 0
>>> arr[1, 0] = [[15, 25], [35, 45]] # Frame 1
>>> # Create labels from the array
>>> video = Video(filename="example.mp4")
>>> skeleton = Skeleton(["head", "tail"])
>>> labels = from_numpy(arr, video=video, skeleton=skeleton)
>>> # With custom track names
>>> labels = from_numpy(arr, video=video, skeleton=skeleton,
... track_names=["mouse1"])
>>> # With confidence scores
>>> arr_with_conf = np.zeros((2, 1, 2, 3))
>>> arr_with_conf[0, 0] = [[10, 20, 0.95], [30, 40, 0.98]]
>>> labels = from_numpy(arr_with_conf, video=video, skeleton=skeleton,
... return_confidence=True)
Source code in sleap_io/codecs/numpy.py
def from_numpy(
tracks_array: np.ndarray,
*,
videos: list[Video] | None = None,
video: Video | None = None,
skeletons: list[Skeleton] | Skeleton | None = None,
skeleton: Skeleton | None = None,
tracks: list[Track] | None = None,
track_names: list[str] | None = None,
first_frame: int = 0,
return_confidence: bool = False,
) -> Labels:
"""Create a new Labels object from a numpy array of tracks.
This factory method creates a new Labels object with instances constructed from
the provided numpy array. It is a more flexible version of Labels.from_numpy().
Args:
tracks_array: A numpy array of tracks, with shape
`(n_frames, n_tracks, n_nodes, 2)` or `(n_frames, n_tracks, n_nodes, 3)`,
where the last dimension contains the x,y coordinates (and optionally
confidence scores).
videos: List of Video objects to associate with the labels. At least one
video is required. Mutually exclusive with `video`.
video: Single Video object to associate with the labels. Mutually exclusive
with `videos`.
skeletons: Skeleton or list of Skeleton objects to use for the instances.
At least one skeleton is required. Mutually exclusive with `skeleton`.
skeleton: Single Skeleton object to use. Mutually exclusive with `skeletons`.
tracks: List of Track objects corresponding to the second dimension of the
array. If not specified, new tracks will be created automatically using
`track_names` if provided, or default names.
track_names: List of track names to use when auto-creating tracks. Only used
if `tracks` is None.
first_frame: Frame index to start the labeled frames from. Default is 0.
return_confidence: Whether the tracks array contains confidence scores in the
last dimension. If True, tracks.shape[-1] should be 3. If False or None,
will be inferred from array shape.
Returns:
A new Labels object with instances constructed from the numpy array.
Raises:
ValueError: If the array dimensions are invalid, or if no videos or
skeletons are provided, or if both `videos` and `video` are provided.
Examples:
>>> import numpy as np
>>> from sleap_io import Video, Skeleton
>>> from sleap_io.codecs import from_numpy
>>> # Create a simple tracking array for 2 frames, 1 track, 2 nodes
>>> arr = np.zeros((2, 1, 2, 2))
>>> arr[0, 0] = [[10, 20], [30, 40]] # Frame 0
>>> arr[1, 0] = [[15, 25], [35, 45]] # Frame 1
>>> # Create labels from the array
>>> video = Video(filename="example.mp4")
>>> skeleton = Skeleton(["head", "tail"])
>>> labels = from_numpy(arr, video=video, skeleton=skeleton)
>>> # With custom track names
>>> labels = from_numpy(arr, video=video, skeleton=skeleton,
... track_names=["mouse1"])
>>> # With confidence scores
>>> arr_with_conf = np.zeros((2, 1, 2, 3))
>>> arr_with_conf[0, 0] = [[10, 20, 0.95], [30, 40, 0.98]]
>>> labels = from_numpy(arr_with_conf, video=video, skeleton=skeleton,
... return_confidence=True)
"""
# Check dimensions
if len(tracks_array.shape) != 4:
raise ValueError(
f"Array must have 4 dimensions (n_frames, n_tracks, n_nodes, 2 or 3), "
f"but got {tracks_array.shape}"
)
# Handle video/videos parameter
if video is not None and videos is not None:
raise ValueError("Cannot specify both 'video' and 'videos' parameters")
if video is not None:
videos = [video]
elif videos is None:
raise ValueError("At least one video must be provided via 'video' or 'videos'")
if not videos:
raise ValueError("At least one video must be provided")
video = videos[0] # Use the first video for creating labeled frames
# Handle skeleton/skeletons parameter
if skeleton is not None and skeletons is not None:
raise ValueError("Cannot specify both 'skeleton' and 'skeletons' parameters")
if skeleton is not None:
skeletons = [skeleton]
elif skeletons is None:
raise ValueError(
"At least one skeleton must be provided via 'skeleton' or 'skeletons'"
)
elif isinstance(skeletons, Skeleton):
skeletons = [skeletons]
elif not skeletons: # Check for empty list
raise ValueError("At least one skeleton must be provided")
skeleton = skeletons[0] # Use the first skeleton for creating instances
n_nodes = len(skeleton.nodes)
# Check if tracks_array contains confidence scores
has_confidence = tracks_array.shape[-1] == 3 or return_confidence
# Get dimensions
n_frames, n_tracks_arr, _ = tracks_array.shape[:3]
# Create or validate tracks
if tracks is None:
# Auto-create tracks
if track_names is not None:
if len(track_names) < n_tracks_arr:
# Extend with default names if needed
track_names = list(track_names) + [
f"track_{i}" for i in range(len(track_names), n_tracks_arr)
]
tracks = [Track(name=name) for name in track_names[:n_tracks_arr]]
else:
tracks = [Track(f"track_{i}") for i in range(n_tracks_arr)]
elif len(tracks) < n_tracks_arr:
# Add missing tracks if needed
original_len = len(tracks)
for i in range(n_tracks_arr - original_len):
tracks.append(Track(f"track_{i}"))
# Create a new empty Labels object
labels = Labels()
labels.videos = list(videos)
labels.skeletons = list(skeletons)
labels.tracks = list(tracks)
# Create labeled frames and instances from the array data
for i in range(n_frames):
frame_idx = i + first_frame
# Check if this frame has any valid data across all tracks
frame_has_valid_data = False
for j in range(n_tracks_arr):
track_data = tracks_array[i, j]
# Check if at least one node in this track has valid xy coordinates
if np.any(~np.isnan(track_data[:, 0])):
frame_has_valid_data = True
break
# Skip creating a frame if there's no valid data
if not frame_has_valid_data:
continue
# Create a new labeled frame
labeled_frame = LabeledFrame(video=video, frame_idx=frame_idx)
frame_has_valid_instances = False
# Process each track in this frame
for j in range(n_tracks_arr):
track = tracks[j]
track_data = tracks_array[i, j]
# Check if there's any valid data for this track at this frame
valid_points = ~np.isnan(track_data[:, 0])
if not np.any(valid_points):
continue
# Create points from numpy data
points = track_data[:, :2].copy()
# Create new instance
if has_confidence:
# Get confidence scores
if tracks_array.shape[-1] == 3:
scores = track_data[:, 2].copy()
else:
scores = np.ones(n_nodes)
# Fix NaN scores
scores = np.where(np.isnan(scores), 1.0, scores)
# Create instance with confidence scores
new_instance = PredictedInstance.from_numpy(
points_data=points,
skeleton=skeleton,
point_scores=scores,
score=1.0,
track=track,
)
else:
# Create instance with default scores
new_instance = PredictedInstance.from_numpy(
points_data=points,
skeleton=skeleton,
point_scores=np.ones(n_nodes),
score=1.0,
track=track,
)
# Add to frame
labeled_frame.instances.append(new_instance)
frame_has_valid_instances = True
# Only add frames that have instances
if frame_has_valid_instances:
labels.append(labeled_frame, update=False)
# Update internal references
labels.update()
return labels
to_dataframe(labels, format=<DataFrameFormat.POINTS: 'points'>, *, video=None, include_metadata=True, include_score=True, include_user_instances=True, include_predicted_instances=True, video_id='path', include_video=None, instance_id='index', untracked='error', backend='pandas', all_frames=False, start_frame=None, end_frame=None)
¶
Convert Labels to a DataFrame.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
labels
|
Labels
|
Labels object to convert. |
required |
format
|
DataFrameFormat | str
|
Output format. One of "points", "instances", "frames", "multi_index". |
<DataFrameFormat.POINTS: 'points'>
|
video
|
Video | int | None
|
Optional video filter. If specified, only frames from this video are included. Can be a Video object or integer index. |
None
|
include_metadata
|
bool
|
Include track, video information in columns. |
True
|
include_score
|
bool
|
Include confidence scores for predicted instances. |
True
|
include_user_instances
|
bool
|
Include user-labeled instances. |
True
|
include_predicted_instances
|
bool
|
Include predicted instances. |
True
|
video_id
|
Literal['path', 'index', 'name', 'object']
|
How to represent videos in the DataFrame. Options: - "path": Full filename/path (default). Works for all video types. - "index": Integer video index. Compact, requires video list for decoding. - "name": Just the video filename (no directory). May not be unique. - "object": Store Video object directly. Not serializable but preserves all video metadata (dataset for HDF5, frame paths for ImageVideo). |
'path'
|
include_video
|
bool | None
|
Whether to include video information. If None (default), automatically includes video info if there are multiple videos or if video metadata is needed. Set False to always omit, True to always include. |
None
|
instance_id
|
Literal['index', 'track']
|
How to name instance columns in "frames" and "multi_index" formats. - "index": Use inst0, inst1, inst2, etc. (default). - "track": Use track names as column prefixes (e.g., mouse1, mouse2). |
'index'
|
untracked
|
Literal['error', 'ignore']
|
Behavior for untracked instances with instance_id="track". - "error": Raise error if any instance lacks a track (default). - "ignore": Skip untracked instances silently. |
'error'
|
backend
|
Literal['pandas', 'polars']
|
"pandas" or "polars". Polars requires the polars package. When using polars, DataFrames are constructed natively without going through pandas, providing better performance for large datasets. |
'pandas'
|
all_frames
|
bool
|
If True, include rows for frames without instances (filled with NaN values). If False (default), only include frames that have instances. Only applies to "frames" and "instances" formats. |
False
|
start_frame
|
int | None
|
Start frame index (inclusive) for frame padding. If None, starts from 0 when all_frames=True, or from first labeled frame otherwise. |
None
|
end_frame
|
int | None
|
End frame index (exclusive) for frame padding. If None, ends at the full video length when known, otherwise at last labeled frame + 1. |
None
|
Returns:
| Type | Description |
|---|---|
DataFrame | DataFrame
|
DataFrame in the specified format. Type depends on backend parameter. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If an invalid format is specified or polars is requested but not installed. |
Examples:
Basic usage:
>>> labels = load_file("predictions.slp")
>>> df = to_dataframe(labels, format="points")
>>> df.head()
frame_idx video_path track node x y score
0 0 video.mp4 track0 nose 10.0 20.0 0.95
1 0 video.mp4 track0 tail 5.0 8.0 0.92
Wide format with instances multiplexed per frame:
>>> df = to_dataframe(labels, format="frames")
>>> df.columns # inst0.track, inst0.nose.x, inst0.nose.y, ...
Track-named columns (requires tracked instances):
>>> df = to_dataframe(labels, format="frames", instance_id="track")
>>> df.columns # mouse1.nose.x, mouse1.nose.y, mouse2.nose.x, ...
Native polars backend for better performance:
>>> df = to_dataframe(labels, format="points", backend="polars")
>>> type(df)
<class 'polars.dataframe.frame.DataFrame'>
Notes
The specific columns and structure depend on the format parameter. See the DataFrameFormat enum documentation for details on each format.
Column naming conventions: - Points: frame_idx, node, x, y, track, track_score, instance_score - Instances: frame_idx, track, track_score, score, {node}.x/y/score - Frames: frame_idx, {inst}.track, {inst}.track_score, {inst}.score, {inst}.{node}.x, {inst}.{node}.y, {inst}.{node}.score - Multi-index: Hierarchical columns (inst, node, coord) with frame idx For polars backend, multi-index columns are flattened to dot-separated names (e.g., "inst0.nose.x").
Source code in sleap_io/codecs/dataframe.py
def to_dataframe(
labels: Labels,
format: DataFrameFormat | str = DataFrameFormat.POINTS,
*,
video: Video | int | None = None,
include_metadata: bool = True,
include_score: bool = True,
include_user_instances: bool = True,
include_predicted_instances: bool = True,
video_id: Literal["path", "index", "name", "object"] = "path",
include_video: bool | None = None,
instance_id: Literal["index", "track"] = "index",
untracked: Literal["error", "ignore"] = "error",
backend: Literal["pandas", "polars"] = "pandas",
all_frames: bool = False,
start_frame: int | None = None,
end_frame: int | None = None,
) -> pd.DataFrame | "pl.DataFrame":
"""Convert Labels to a DataFrame.
Args:
labels: Labels object to convert.
format: Output format. One of "points", "instances", "frames", "multi_index".
video: Optional video filter. If specified, only frames from this video
are included. Can be a Video object or integer index.
include_metadata: Include track, video information in columns.
include_score: Include confidence scores for predicted instances.
include_user_instances: Include user-labeled instances.
include_predicted_instances: Include predicted instances.
video_id: How to represent videos in the DataFrame. Options:
- "path": Full filename/path (default). Works for all video types.
- "index": Integer video index. Compact, requires video list for decoding.
- "name": Just the video filename (no directory). May not be unique.
- "object": Store Video object directly. Not serializable but preserves
all video metadata (dataset for HDF5, frame paths for ImageVideo).
include_video: Whether to include video information. If None (default),
automatically includes video info if there are multiple videos or if
video metadata is needed. Set False to always omit, True to always include.
instance_id: How to name instance columns in "frames" and "multi_index" formats.
- "index": Use inst0, inst1, inst2, etc. (default).
- "track": Use track names as column prefixes (e.g., mouse1, mouse2).
untracked: Behavior for untracked instances with instance_id="track".
- "error": Raise error if any instance lacks a track (default).
- "ignore": Skip untracked instances silently.
backend: "pandas" or "polars". Polars requires the polars package.
When using polars, DataFrames are constructed natively without
going through pandas, providing better performance for large datasets.
all_frames: If True, include rows for frames without instances (filled with
NaN values). If False (default), only include frames that have instances.
Only applies to "frames" and "instances" formats.
start_frame: Start frame index (inclusive) for frame padding. If None, starts
from 0 when all_frames=True, or from first labeled frame otherwise.
end_frame: End frame index (exclusive) for frame padding. If None, ends at
the full video length when known, otherwise at last labeled frame + 1.
Returns:
DataFrame in the specified format. Type depends on backend parameter.
Raises:
ValueError: If an invalid format is specified or polars is requested but
not installed.
Examples:
Basic usage:
>>> labels = load_file("predictions.slp")
>>> df = to_dataframe(labels, format="points")
>>> df.head()
frame_idx video_path track node x y score
0 0 video.mp4 track0 nose 10.0 20.0 0.95
1 0 video.mp4 track0 tail 5.0 8.0 0.92
Wide format with instances multiplexed per frame:
>>> df = to_dataframe(labels, format="frames")
>>> df.columns # inst0.track, inst0.nose.x, inst0.nose.y, ...
Track-named columns (requires tracked instances):
>>> df = to_dataframe(labels, format="frames", instance_id="track")
>>> df.columns # mouse1.nose.x, mouse1.nose.y, mouse2.nose.x, ...
Native polars backend for better performance:
>>> df = to_dataframe(labels, format="points", backend="polars")
>>> type(df)
<class 'polars.dataframe.frame.DataFrame'>
Notes:
The specific columns and structure depend on the format parameter.
See the DataFrameFormat enum documentation for details on each format.
Column naming conventions:
- Points: frame_idx, node, x, y, track, track_score, instance_score
- Instances: frame_idx, track, track_score, score, {node}.x/y/score
- Frames: frame_idx, {inst}.track, {inst}.track_score, {inst}.score,
{inst}.{node}.x, {inst}.{node}.y, {inst}.{node}.score
- Multi-index: Hierarchical columns (inst, node, coord) with frame idx
For polars backend, multi-index columns are flattened to dot-separated
names (e.g., "inst0.nose.x").
"""
# Validate backend
if backend == "polars" and not HAS_POLARS:
raise ValueError(
"Polars backend requested but polars is not installed. "
"Install with: pip install polars"
)
# Normalize format parameter
if isinstance(format, str):
try:
format = DataFrameFormat(format.lower())
except ValueError:
valid_formats = ", ".join([f.value for f in DataFrameFormat])
raise ValueError(
f"Invalid format '{format}'. Must be one of: {valid_formats}"
)
# Convert video parameter to index for fast path filtering
video_filter_idx: int | None = None
if video is not None:
if isinstance(video, int):
video_filter_idx = video
video = labels.videos[video]
else:
video_filter_idx = labels.videos.index(video)
# Determine whether to include video info
if include_video is None:
# Auto-detect: include if multiple videos, unless explicitly omitted
include_video = len(labels.videos) > 1
# Use lazy fast path when available (for POINTS and INSTANCES formats)
if labels.is_lazy and format in (DataFrameFormat.POINTS, DataFrameFormat.INSTANCES):
store = labels.labeled_frames._store
if format == DataFrameFormat.POINTS:
return _to_points_df_lazy(
store,
labels,
video_filter=video_filter_idx,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
backend=backend,
)
else: # INSTANCES
return _to_instances_df_lazy(
store,
labels,
video_filter=video_filter_idx,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
backend=backend,
)
# Eager path: filter labeled frames
if video is not None:
labeled_frames = [lf for lf in labels.labeled_frames if lf.video == video]
else:
labeled_frames = labels.labeled_frames
# Route to appropriate converter based on format
if format == DataFrameFormat.POINTS:
df = _to_points_df(
labels,
labeled_frames,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
backend=backend,
)
elif format == DataFrameFormat.INSTANCES:
df = _to_instances_df(
labels,
labeled_frames,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
backend=backend,
video_filter_idx=video_filter_idx,
all_frames=all_frames,
start_frame=start_frame,
end_frame=end_frame,
)
elif format == DataFrameFormat.FRAMES:
df = _to_frames_df(
labels,
labeled_frames,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
instance_id=instance_id,
untracked=untracked,
backend=backend,
video_filter_idx=video_filter_idx,
all_frames=all_frames,
start_frame=start_frame,
end_frame=end_frame,
)
elif format == DataFrameFormat.MULTI_INDEX:
df = _to_multi_index_df(
labels,
labeled_frames,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
instance_id=instance_id,
untracked=untracked,
backend=backend,
)
else:
raise ValueError(f"Unknown format: {format}")
return df
to_dataframe_iter(labels, format=<DataFrameFormat.POINTS: 'points'>, *, chunk_size=None, video=None, include_metadata=True, include_score=True, include_user_instances=True, include_predicted_instances=True, video_id='path', include_video=None, instance_id='index', untracked='error', backend='pandas')
¶
Iterate over Labels data, yielding DataFrames in chunks.
This is a memory-efficient alternative to to_dataframe() for large datasets.
Instead of materializing the entire DataFrame at once, it yields smaller
DataFrames (chunks) that can be processed incrementally.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
labels
|
Labels
|
Labels object to convert. |
required |
format
|
DataFrameFormat | str
|
Output format. One of "points", "instances", "frames", "multi_index". |
<DataFrameFormat.POINTS: 'points'>
|
chunk_size
|
int | None
|
Number of rows per chunk. If None (default), yields the entire
DataFrame in a single chunk (equivalent to |
None
|
video
|
Video | int | None
|
Optional video filter. If specified, only frames from this video are included. Can be a Video object or integer index. |
None
|
include_metadata
|
bool
|
Include track, video information in columns. |
True
|
include_score
|
bool
|
Include confidence scores for predicted instances. |
True
|
include_user_instances
|
bool
|
Include user-labeled instances. |
True
|
include_predicted_instances
|
bool
|
Include predicted instances. |
True
|
video_id
|
Literal['path', 'index', 'name', 'object']
|
How to represent videos in the DataFrame. Options: - "path": Full filename/path (default). - "index": Integer video index. - "name": Just the video filename. - "object": Store Video object directly. |
'path'
|
include_video
|
bool | None
|
Whether to include video information. |
None
|
instance_id
|
Literal['index', 'track']
|
How to name instance columns in "frames" and "multi_index" formats. - "index": Use inst0, inst1, inst2, etc. (default). - "track": Use track names as column prefixes. |
'index'
|
untracked
|
Literal['error', 'ignore']
|
Behavior for untracked instances with instance_id="track". - "error": Raise error if any instance lacks a track (default). - "ignore": Skip untracked instances silently. |
'error'
|
backend
|
Literal['pandas', 'polars']
|
"pandas" or "polars". Polars requires the polars package. When using polars, DataFrames are constructed natively without going through pandas, providing better performance for large datasets. |
'pandas'
|
Yields:
| Type | Description |
|---|---|
DataFrame | DataFrame
|
DataFrames, each containing up to |
Examples:
Process large datasets in chunks:
>>> for df_chunk in to_dataframe_iter(labels, chunk_size=10000):
... df_chunk.to_parquet("output.parquet", append=True)
Concatenate chunks to get full DataFrame (equivalent to to_dataframe):
Memory-efficient per-video processing:
>>> for video in labels.videos:
... for chunk in to_dataframe_iter(labels, video=video, chunk_size=5000):
... process_chunk(chunk)
Source code in sleap_io/codecs/dataframe.py
def to_dataframe_iter(
labels: Labels,
format: DataFrameFormat | str = DataFrameFormat.POINTS,
*,
chunk_size: int | None = None,
video: Video | int | None = None,
include_metadata: bool = True,
include_score: bool = True,
include_user_instances: bool = True,
include_predicted_instances: bool = True,
video_id: Literal["path", "index", "name", "object"] = "path",
include_video: bool | None = None,
instance_id: Literal["index", "track"] = "index",
untracked: Literal["error", "ignore"] = "error",
backend: Literal["pandas", "polars"] = "pandas",
) -> Iterator[pd.DataFrame | "pl.DataFrame"]:
"""Iterate over Labels data, yielding DataFrames in chunks.
This is a memory-efficient alternative to `to_dataframe()` for large datasets.
Instead of materializing the entire DataFrame at once, it yields smaller
DataFrames (chunks) that can be processed incrementally.
Args:
labels: Labels object to convert.
format: Output format. One of "points", "instances", "frames", "multi_index".
chunk_size: Number of rows per chunk. If None (default), yields the entire
DataFrame in a single chunk (equivalent to `to_dataframe()`).
The meaning of "row" depends on the format:
- points: One point (node) per row
- instances: One instance per row
- frames: One frame per row
- multi_index: One frame per row
video: Optional video filter. If specified, only frames from this video
are included. Can be a Video object or integer index.
include_metadata: Include track, video information in columns.
include_score: Include confidence scores for predicted instances.
include_user_instances: Include user-labeled instances.
include_predicted_instances: Include predicted instances.
video_id: How to represent videos in the DataFrame. Options:
- "path": Full filename/path (default).
- "index": Integer video index.
- "name": Just the video filename.
- "object": Store Video object directly.
include_video: Whether to include video information.
instance_id: How to name instance columns in "frames" and "multi_index" formats.
- "index": Use inst0, inst1, inst2, etc. (default).
- "track": Use track names as column prefixes.
untracked: Behavior for untracked instances with instance_id="track".
- "error": Raise error if any instance lacks a track (default).
- "ignore": Skip untracked instances silently.
backend: "pandas" or "polars". Polars requires the polars package.
When using polars, DataFrames are constructed natively without
going through pandas, providing better performance for large datasets.
Yields:
DataFrames, each containing up to `chunk_size` rows.
Examples:
Process large datasets in chunks:
>>> for df_chunk in to_dataframe_iter(labels, chunk_size=10000):
... df_chunk.to_parquet("output.parquet", append=True)
Concatenate chunks to get full DataFrame (equivalent to to_dataframe):
>>> import pandas as pd
>>> df = pd.concat(list(to_dataframe_iter(labels, chunk_size=1000)))
Memory-efficient per-video processing:
>>> for video in labels.videos:
... for chunk in to_dataframe_iter(labels, video=video, chunk_size=5000):
... process_chunk(chunk)
"""
# Validate backend
if backend == "polars" and not HAS_POLARS:
raise ValueError(
"Polars backend requested but polars is not installed. "
"Install with: pip install polars"
)
# Normalize format parameter
if isinstance(format, str):
try:
format = DataFrameFormat(format.lower())
except ValueError:
valid_formats = ", ".join([f.value for f in DataFrameFormat])
raise ValueError(
f"Invalid format '{format}'. Must be one of: {valid_formats}"
)
# Filter to specific video if requested
if video is not None:
if isinstance(video, int):
video = labels.videos[video]
labeled_frames = [lf for lf in labels.labeled_frames if lf.video == video]
else:
labeled_frames = labels.labeled_frames
# Determine whether to include video info
if include_video is None:
include_video = len(labels.videos) > 1
# If no chunk_size specified, yield entire DataFrame at once
if chunk_size is None:
df = to_dataframe(
labels,
format=format,
video=video,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
video_id=video_id,
include_video=include_video,
instance_id=instance_id,
untracked=untracked,
backend=backend,
)
yield df
return
# Get the appropriate row iterator and DataFrame builder
if format == DataFrameFormat.POINTS:
row_iter = _iter_points_rows(
labels,
labeled_frames,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
)
elif format == DataFrameFormat.INSTANCES:
row_iter = _iter_instances_rows(
labels,
labeled_frames,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
)
elif format == DataFrameFormat.FRAMES:
# For frames format, we need to pre-scan for max_instances and tracks
max_instances, all_tracks, skeleton = _prescan_for_frames(
labels,
labeled_frames,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
instance_id=instance_id,
untracked=untracked,
)
row_iter = _iter_frames_rows(
labels,
labeled_frames,
include_metadata=include_metadata,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
instance_id=instance_id,
untracked=untracked,
max_instances=max_instances,
all_tracks=all_tracks,
skeleton=skeleton,
)
elif format == DataFrameFormat.MULTI_INDEX:
# For multi_index format, we also need to pre-scan
max_instances, all_tracks, skeleton = _prescan_for_frames(
labels,
labeled_frames,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
instance_id=instance_id,
untracked=untracked,
)
row_iter = _iter_multi_index_rows(
labels,
labeled_frames,
include_score=include_score,
include_user_instances=include_user_instances,
include_predicted_instances=include_predicted_instances,
include_video=include_video,
video_id=video_id,
instance_id=instance_id,
untracked=untracked,
max_instances=max_instances,
all_tracks=all_tracks,
skeleton=skeleton,
)
else:
raise ValueError(f"Unknown format: {format}")
# Buffer rows and yield DataFrames
buffer: list[dict] = []
yielded_any = False
for row in row_iter:
buffer.append(row)
if len(buffer) >= chunk_size:
# For multi_index with polars, flatten tuple keys
if format == DataFrameFormat.MULTI_INDEX and backend == "polars":
buffer = _flatten_tuple_keys(buffer)
df = _create_dataframe_from_rows(buffer, backend)
yield df
yielded_any = True
buffer = []
# Yield remaining rows (or empty DataFrame if no data)
if buffer or not yielded_any:
# For multi_index with polars, flatten tuple keys
if format == DataFrameFormat.MULTI_INDEX and backend == "polars":
buffer = _flatten_tuple_keys(buffer)
df = _create_dataframe_from_rows(buffer, backend)
yield df
to_dict(labels, *, video=None, skip_empty_frames=False)
¶
Convert Labels to a primitive dictionary (JSON-serializable).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
labels
|
Labels
|
Labels object to convert. |
required |
video
|
Video | int | None
|
Optional video filter. If specified, only frames from this video are included. Can be a Video object or integer index. |
None
|
skip_empty_frames
|
bool
|
If True, exclude frames with no instances. |
False
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Dictionary with structure: { "version": "1.0.0", "skeletons": [...], "videos": [...], "tracks": [...], "labeled_frames": [...], "suggestions": [...], "provenance": {...} } All values are JSON-serializable primitives (str, int, float, bool, None, list, dict). No numpy arrays or custom objects. |
Examples:
>>> labels = load_file("predictions.slp")
>>> d = to_dict(labels)
>>> import json
>>> json.dumps(d) # Fully serializable!
Notes
- Uses index-based references (e.g., skeleton_idx, video_idx) for compactness
- Preserves all metadata including provenance
- Points are stored as list of dicts with x, y, visible, complete fields
- The output is fully compatible with JSON, YAML, or any format that handles Python primitives
Source code in sleap_io/codecs/dictionary.py
def to_dict(
labels: Labels,
*,
video: Video | int | None = None,
skip_empty_frames: bool = False,
) -> dict[str, Any]:
"""Convert Labels to a primitive dictionary (JSON-serializable).
Args:
labels: Labels object to convert.
video: Optional video filter. If specified, only frames from this video
are included. Can be a Video object or integer index.
skip_empty_frames: If True, exclude frames with no instances.
Returns:
Dictionary with structure:
{
"version": "1.0.0",
"skeletons": [...],
"videos": [...],
"tracks": [...],
"labeled_frames": [...],
"suggestions": [...],
"provenance": {...}
}
All values are JSON-serializable primitives (str, int, float, bool, None,
list, dict). No numpy arrays or custom objects.
Examples:
>>> labels = load_file("predictions.slp")
>>> d = to_dict(labels)
>>> import json
>>> json.dumps(d) # Fully serializable!
>>> # Filter to specific video
>>> d = to_dict(labels, video=0)
>>> # Exclude empty frames
>>> d = to_dict(labels, skip_empty_frames=True)
Notes:
- Uses index-based references (e.g., skeleton_idx, video_idx) for compactness
- Preserves all metadata including provenance
- Points are stored as list of dicts with x, y, visible, complete fields
- The output is fully compatible with JSON, YAML, or any format that handles
Python primitives
"""
# Convert video parameter to index for fast path filtering
video_filter_idx: int | None = None
if video is not None:
if isinstance(video, int):
video_filter_idx = video
video = labels.videos[video]
else:
video_filter_idx = labels.videos.index(video)
# Build skeleton list
skeletons_list = []
for skeleton in labels.skeletons:
skeleton_dict = {
"name": skeleton.name,
"nodes": [node.name for node in skeleton.nodes],
"edges": [
[
skeleton.nodes.index(edge.source),
skeleton.nodes.index(edge.destination),
]
for edge in skeleton.edges
],
}
# Add symmetries if present
if skeleton.symmetries:
symmetries_list = []
for symmetry in skeleton.symmetries:
# Convert set to list for indexing and get indices
nodes_list = list(symmetry.nodes)
indices = [
skeleton.nodes.index(nodes_list[0]),
skeleton.nodes.index(nodes_list[1]),
]
# Sort indices for consistent ordering
indices.sort()
symmetries_list.append(indices)
skeleton_dict["symmetries"] = symmetries_list
skeletons_list.append(skeleton_dict)
# Build video list
videos_list = []
for vid in labels.videos:
video_dict = {
"filename": vid.filename,
}
# Add shape if available (Video.shape handles exceptions internally)
if vid.shape is not None:
video_dict["shape"] = list(vid.shape)
# Add backend info if available
if vid.backend is not None:
video_dict["backend"] = {"type": type(vid.backend).__name__}
videos_list.append(video_dict)
# Build track list
tracks_list = [{"name": track.name} for track in labels.tracks]
# Build labeled frames list - use fast path for lazy Labels
if labels.is_lazy:
labeled_frames_list = _build_labeled_frames_lazy(
labels.labeled_frames._store,
video_filter=video_filter_idx,
skip_empty_frames=skip_empty_frames,
)
else:
# Eager path: filter labeled frames
if video is not None:
labeled_frames = [lf for lf in labels.labeled_frames if lf.video == video]
else:
labeled_frames = labels.labeled_frames
# Skip empty frames if requested (preserve negative frames)
if skip_empty_frames:
labeled_frames = [
lf for lf in labeled_frames if len(lf.instances) > 0 or lf.is_negative
]
labeled_frames_list = []
for lf in labeled_frames:
instances_list = []
for instance in lf.instances:
# Determine instance type
is_predicted = isinstance(instance, PredictedInstance)
# Build points list
points_list = []
for point in instance.points:
point_dict = {
"x": float(point["xy"][0]),
"y": float(point["xy"][1]),
"visible": bool(point["visible"]),
"complete": bool(point["complete"]),
}
points_list.append(point_dict)
# Build instance dict
instance_dict = {
"type": "predicted_instance" if is_predicted else "instance",
"skeleton_idx": labels.skeletons.index(instance.skeleton),
"points": points_list,
}
# Add track if present
if instance.track is not None:
instance_dict["track_idx"] = labels.tracks.index(instance.track)
# Add tracking score if present
if (
hasattr(instance, "tracking_score")
and instance.tracking_score is not None
):
instance_dict["tracking_score"] = float(instance.tracking_score)
# Add score for predicted instances
if is_predicted:
instance_dict["score"] = float(instance.score)
# Add from_predicted if present
if (
hasattr(instance, "from_predicted")
and instance.from_predicted is not None
):
# Note: We can't serialize the reference, just indicate it exists
instance_dict["has_from_predicted"] = True
instances_list.append(instance_dict)
frame_dict = {
"frame_idx": int(lf.frame_idx),
"video_idx": labels.videos.index(lf.video),
"instances": instances_list,
}
if lf.is_negative:
frame_dict["is_negative"] = True
labeled_frames_list.append(frame_dict)
# Build suggestions list (if filtering by video, also filter suggestions)
if video is not None:
suggestions_to_include = [sf for sf in labels.suggestions if sf.video == video]
else:
suggestions_to_include = labels.suggestions
suggestions_list = []
for sf in suggestions_to_include:
suggestion_dict = {
"frame_idx": int(sf.frame_idx),
"video_idx": labels.videos.index(sf.video),
}
suggestions_list.append(suggestion_dict)
# Build complete dictionary
result = {
"version": "1.0.0",
"skeletons": skeletons_list,
"videos": videos_list,
"tracks": tracks_list,
"labeled_frames": labeled_frames_list,
"suggestions": suggestions_list,
"provenance": dict(labels.provenance), # Copy to ensure it's a plain dict
}
return result
to_numpy(labels, *, video=None, untracked=False, return_confidence=False, user_instances=True, predicted_instances=True)
¶
Convert Labels to a numpy array.
This is a more flexible version of Labels.numpy() with enhanced options.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
labels
|
Labels
|
Labels object to convert. |
required |
video
|
Video | int | None
|
Video or video index to convert to numpy arrays. If None (the default), uses the first video. |
None
|
untracked
|
bool
|
If False (the default), include only instances that have a track assignment. If True, includes all instances in each frame in arbitrary order. |
False
|
return_confidence
|
bool
|
If False (the default), only return points of nodes. If True, return the points and scores of nodes. |
False
|
user_instances
|
bool
|
If True (the default), include user instances when available, preferring them over predicted instances with the same track. If False, only include predicted instances. |
True
|
predicted_instances
|
bool
|
If True (the default), include predicted instances. If False, only include user instances. |
True
|
Returns:
| Type | Description |
|---|---|
ndarray
|
An array of tracks of shape Missing data will be replaced with If this is a single instance project, a track does not need to be assigned. When |
Notes
This method assumes that instances have tracks assigned and is intended to function primarily for single-video prediction results.
This function contains the core logic for numpy conversion. The Labels.numpy() method delegates to this function.
Examples:
>>> arr = to_numpy(labels, video=0, return_confidence=True)
>>> arr.shape
(100, 2, 5, 3) # 100 frames, 2 tracks, 5 nodes, (x, y, score)
>>> # Get only user instances
>>> arr = to_numpy(labels, user_instances=True, predicted_instances=False)
Source code in sleap_io/codecs/numpy.py
def to_numpy(
labels: Labels,
*,
video: Video | int | None = None,
untracked: bool = False,
return_confidence: bool = False,
user_instances: bool = True,
predicted_instances: bool = True,
) -> np.ndarray:
"""Convert Labels to a numpy array.
This is a more flexible version of Labels.numpy() with enhanced options.
Args:
labels: Labels object to convert.
video: Video or video index to convert to numpy arrays. If None (the default),
uses the first video.
untracked: If False (the default), include only instances that have a track
assignment. If True, includes all instances in each frame in arbitrary
order.
return_confidence: If False (the default), only return points of nodes. If
True, return the points and scores of nodes.
user_instances: If True (the default), include user instances when available,
preferring them over predicted instances with the same track. If False,
only include predicted instances.
predicted_instances: If True (the default), include predicted instances.
If False, only include user instances.
Returns:
An array of tracks of shape `(n_frames, n_tracks, n_nodes, 2)` if
`return_confidence` is False. Otherwise returned shape is
`(n_frames, n_tracks, n_nodes, 3)` if `return_confidence` is True.
Missing data will be replaced with `np.nan`.
If this is a single instance project, a track does not need to be assigned.
When `user_instances=False`, only predicted instances will be returned.
When `user_instances=True`, user instances will be preferred over predicted
instances with the same track or if linked via `from_predicted`.
Notes:
This method assumes that instances have tracks assigned and is intended to
function primarily for single-video prediction results.
This function contains the core logic for numpy conversion. The
Labels.numpy() method delegates to this function.
Examples:
>>> arr = to_numpy(labels, video=0, return_confidence=True)
>>> arr.shape
(100, 2, 5, 3) # 100 frames, 2 tracks, 5 nodes, (x, y, score)
>>> # Get only user instances
>>> arr = to_numpy(labels, user_instances=True, predicted_instances=False)
>>> # Include untracked instances
>>> arr = to_numpy(labels, untracked=True)
"""
# Convert video parameter to Video object
if video is None:
video = labels.videos[0] if labels.videos else None
elif type(video) is int:
video = labels.videos[video]
# Use lazy fast path when available
if labels.is_lazy:
store = labels.labeled_frames._store
return store.to_numpy(
video=video,
untracked=untracked,
return_confidence=return_confidence,
user_instances=user_instances,
)
# Eager path: filter labeled frames by video
lfs = [lf for lf in labels.labeled_frames if lf.video == video]
# Figure out frame index range.
first_frame, last_frame = 0, 0
for lf in lfs:
first_frame = min(first_frame, lf.frame_idx)
last_frame = max(last_frame, lf.frame_idx)
# Use video length when available so output spans the full video duration.
video_length = len(video) if video is not None else 0
if video_length > 0:
last_frame = max(last_frame, video_length - 1)
# Figure out the number of tracks based on number of instances in each frame.
n_instances = _max_instances_per_frame(
lfs, user_instances=user_instances, predicted_instances=predicted_instances
)
# Case 1: We don't care about order because there's only 1 instance per frame,
# or we're considering untracked instances.
is_single_instance = n_instances == 1
untracked = untracked or is_single_instance
if untracked:
n_tracks = n_instances
else:
# Case 2: We're considering only tracked instances.
n_tracks = len(labels.tracks)
n_frames = int(last_frame - first_frame + 1)
skeleton = labels.skeletons[-1] # Assume project only uses last skeleton
n_nodes = len(skeleton.nodes)
if return_confidence:
tracks = np.full((n_frames, n_tracks, n_nodes, 3), np.nan, dtype="float32")
else:
tracks = np.full((n_frames, n_tracks, n_nodes, 2), np.nan, dtype="float32")
for lf in lfs:
i = int(lf.frame_idx - first_frame)
if untracked:
# For untracked instances, fill them in arbitrary order.
frame_instances = _untracked_frame_instances(
lf,
is_single_instance=is_single_instance,
user_instances=user_instances,
predicted_instances=predicted_instances,
)
for j, inst in enumerate(frame_instances):
if j < n_tracks:
if return_confidence:
if isinstance(inst, PredictedInstance):
tracks[i, j] = inst.numpy(scores=True)
else:
# For user instances, set confidence to 1.0
points_data = inst.numpy()
confidence = np.ones(
(points_data.shape[0], 1), dtype="float32"
)
tracks[i, j] = np.hstack((points_data, confidence))
else:
tracks[i, j] = inst.numpy()
else: # untracked is False
# For tracked instances, organize by track ID.
track_to_instance = _tracked_frame_instances(
lf,
user_instances=user_instances,
predicted_instances=predicted_instances,
)
for track, inst in track_to_instance.items():
j = labels.tracks.index(track)
if type(inst) is PredictedInstance:
tracks[i, j] = inst.numpy(scores=return_confidence)
elif type(inst) is Instance:
tracks[i, j, :, :2] = inst.numpy()
# If return_confidence is True, add dummy confidence scores
if return_confidence:
tracks[i, j, :, 2] = 1.0
return tracks