Merging annotations¶
The merging system in sleap-io provides powerful tools for combining multiple annotation files while intelligently handling conflicts, duplicates, and different data sources. This is essential for collaborative annotation, human-in-the-loop workflows, and consolidating predictions from multiple models.
Key concepts
- Merging: Combining annotations from multiple sources into a single dataset
- Matching: Determining which objects (skeletons, videos, tracks, instances) correspond between datasets
- Conflict resolution: Handling overlapping or contradicting annotations using configurable strategies
- Provenance tracking: Automatic recording of merge history and data sources
Quick start¶
The simplest and most common use case is merging predictions back into a manually annotated project:
import sleap_io as sio
# Load your manual annotations and new predictions
base_labels = sio.load_file("manual_annotations.slp")
predictions = sio.load_file("model_predictions.slp")
# Merge predictions into base project
# Default 'smart' strategy preserves manual labels
result = base_labels.merge(predictions)
# Check what happened
print(f"Frames merged: {result.frames_merged}")
print(f"Instances added: {result.instances_added}")
# Save the merged project
base_labels.save("merged_project.slp")
Tip
The default smart
strategy automatically preserves your manual annotations while adding new predictions, making it ideal for human-in-the-loop workflows.
How merging works¶
The merge process follows a systematic approach to combine data structures while maintaining consistency:
graph TD
A[Start Merge] --> B[Match Skeletons]
B --> C[Match Videos]
C --> D[Match Tracks]
D --> E[For Each Frame]
E --> F{Frame Exists?}
F -->|No| G[Add New Frame]
F -->|Yes| H[Match Instances]
H --> I[Apply Strategy]
I --> J[Resolve Conflicts]
G --> K[Next Frame]
J --> K
K --> E
K --> L[Return MergeResult]
Step-by-step process¶
- Skeleton matching: Identifies corresponding skeletons between datasets based on node names and structure
- Video matching: Determines which videos represent the same source data (handles different paths, formats)
- Track matching: Maps tracks (instance identities) between datasets
- Frame merging: For each frame, applies the chosen strategy to combine instances
- Conflict resolution: Handles overlapping instances according to the selected strategy
Merging strategies¶
The merge strategy determines how instances are combined when frames overlap. Each strategy is designed for specific workflows and requirements.
Smart strategy (default)¶
Intelligently preserves manual annotations while updating predictions. This is the recommended strategy for most use cases.
Algorithm details
def smart_merge(base_frame, new_frame):
merged = []
# Step 1: Always keep user labels from base
for instance in base_frame:
if isinstance(instance, Instance): # User label
merged.append(instance)
# Step 2: Find spatial matches between frames
matches = find_spatial_matches(base_frame, new_frame)
# Step 3: Process each match
for base_inst, new_inst in matches:
if both_are_user_labels:
keep(base_inst) # Preserve original
elif base_is_user_label:
keep(base_inst) # User labels always win
elif new_is_user_label:
keep(new_inst) # New user label replaces prediction
else: # Both are predictions
keep(higher_score_instance)
# Step 4: Add unmatched instances from new frame
merged.extend(unmatched_new_instances)
return merged
When to use
- Merging model predictions into manually annotated projects
- Updating predictions while preserving validated annotations
- Human-in-the-loop training workflows
Behavior matrix:
Base frame | New frame | Result | Reasoning |
---|---|---|---|
User label | Prediction | User label | Manual annotations are preserved |
User label | User label | Base user label | Avoid duplicates, keep original |
Prediction | User label | User label | Manual corrections override predictions |
Prediction (score: 0.8) | Prediction (score: 0.9) | Prediction (0.9) | Higher confidence wins |
Empty | Any | New instance | Add missing annotations |
Keep original strategy¶
Preserves all instances from the base dataset, ignoring new instances at matching positions.
When to use
- Protecting validated datasets from modification
- Reviewing changes without applying them
- Maintaining data integrity for published datasets
Keep new strategy¶
Replaces matched instances with new ones, effectively updating the base dataset.
When to use
- Completely replacing old predictions with new ones
- Applying corrections from a revised annotation pass
- Updating with higher quality annotations
Keep both strategy¶
Retains all instances from both datasets without filtering duplicates.
Warning
This strategy can create overlapping instances at the same locations. Use with caution and consider post-processing to remove duplicates.
When to use
- Combining annotations from multiple annotators
- Preserving all data for later review
- Creating training data with augmented annotations
Matching configuration¶
Matching determines which objects correspond between datasets. Different matchers handle various scenarios and data inconsistencies.
Video matching¶
Controls how videos are matched between projects, essential for cross-platform workflows and different file organizations.
Tries multiple strategies in sequence until a match is found:
graph LR
A[Try PATH match] -->|Fail| B[Try BASENAME match]
B -->|Fail| C[Try CONTENT match]
C -->|Fail| D[No match]
A -->|Success| E[Match found]
B -->|Success| E
C -->|Success| E
Exact path matching with optional strict mode:
Matches by filename only, ignoring directories:
from sleap_io.model.matching import BASENAME_VIDEO_MATCHER
# Matches "video.mp4" regardless of path
result = base_labels.merge(new_labels, video_matcher=BASENAME_VIDEO_MATCHER)
Tip
Perfect for cross-platform collaboration where the same files exist in different locations.
Matches videos by shape and backend type:
For image sequences, removes duplicate images:
from sleap_io.model.matching import IMAGE_DEDUP_VIDEO_MATCHER
# Automatically deduplicates overlapping images
result = labels1.merge(labels2, video_matcher=IMAGE_DEDUP_VIDEO_MATCHER)
Note
Only works with ImageVideo
backends (image sequences from COCO/CVAT exports).
Instance matching¶
Determines when two instances represent the same annotation.
Matches instances by Euclidean distance between corresponding points:
from sleap_io.model.matching import InstanceMatcher
# Threshold is maximum pixel distance
matcher = InstanceMatcher(method="spatial", threshold=5.0)
How it works
Computes average distance between visible points. Instances match if distance < threshold.
Matches instances by track identity:
Skeleton matching¶
Controls how skeletons are matched and mapped between datasets.
Same nodes and edges, order doesn't matter:
Partial match based on Jaccard similarity:
Common workflows¶
Human-in-the-loop (HITL) training¶
The most common workflow: merging model predictions back into your training data.
import sleap_io as sio
# Load manual annotations and predictions
manual_labels = sio.load_file("manual_annotations.slp")
predictions = sio.load_file("predictions.slp")
# Merge with smart strategy (default)
# This preserves all manual labels while adding predictions
result = manual_labels.merge(predictions)
# Review what was merged
print(f"Frames merged: {result.frames_merged}")
print(f"New instances: {result.instances_added}")
print(f"Conflicts resolved: {len(result.conflicts)}")
# Check for any errors
if result.errors:
for error in result.errors:
print(f"Error: {error.message}")
# Save the updated project
manual_labels.save("updated_project.slp")
Cross-platform collaboration¶
When working across different operating systems or file structures:
Scenario: Windows and Linux collaboration
Your team uses different operating systems with different file paths:
- Windows:
C:\research\videos\session_001.mp4
- Linux:
/home/lab/data/videos/session_001.mp4
from sleap_io.model.matching import BASENAME_VIDEO_MATCHER
# Load annotations from different systems
windows_annotations = sio.load_file("windows_annotations.slp")
linux_predictions = sio.load_file("linux_predictions.slp")
# Use basename matching to ignore path differences
result = windows_annotations.merge(
linux_predictions,
video_matcher=BASENAME_VIDEO_MATCHER # Matches by filename only
)
print(f"Successfully matched {len(result.videos_merged)} videos across platforms")
Combining multiple annotators¶
Merge annotations from different team members:
# Load annotations from multiple sources
annotator1 = sio.load_file("annotator1.slp")
annotator2 = sio.load_file("annotator2.slp")
annotator3 = sio.load_file("annotator3.slp")
# Start with first annotator
merged = annotator1
# Merge others, keeping all annotations for review
for labels in [annotator2, annotator3]:
result = merged.merge(labels, frame_strategy="keep_both")
print(f"Added {result.instances_added} instances from {labels.filename}")
# Review overlapping annotations
for frame in merged:
instances = frame.instances
if len(instances) > expected_count:
print(f"Frame {frame.frame_idx}: {len(instances)} instances (review needed)")
Updating predictions¶
Replace old predictions with improved ones:
# Load existing project
project = sio.load_file("project_with_old_predictions.slp")
# Load new predictions from improved model
new_predictions = sio.load_file("better_model_predictions.slp")
# Configure matching to be more lenient for predictions
from sleap_io.model.matching import InstanceMatcher
instance_matcher = InstanceMatcher(
method="spatial",
threshold=10.0 # More lenient for predictions
)
# Merge with smart strategy to preserve manual labels
result = project.merge(
new_predictions,
frame_strategy="smart",
instance_matcher=instance_matcher
)
print(f"Updated {result.frames_merged} frames with new predictions")
print(f"Preserved {sum(1 for f in project for i in f.instances if type(i).__name__ == 'Instance')} manual labels")
Advanced topics¶
Image sequence deduplication¶
When working with COCO or CVAT exports that contain overlapping images:
Background
CVAT and COCO exports often use image sequences (ImageVideo) rather than video files. When merging multiple exports, you may have duplicate images that need to be handled intelligently.
from sleap_io.model.matching import IMAGE_DEDUP_VIDEO_MATCHER, SHAPE_VIDEO_MATCHER
# Load CVAT exports with potential overlaps
batch1 = sio.load_file("cvat_batch1.json") # 1000 images
batch2 = sio.load_file("cvat_batch2.json") # 1000 images, 22 overlapping
# Option 1: Remove duplicates, keep videos separate
result = batch1.merge(batch2, video_matcher=IMAGE_DEDUP_VIDEO_MATCHER)
print(f"Result: {sum(len(v.filename) for v in batch1.videos)} unique images")
# Option 2: Merge same-shaped videos into one
result = batch1.merge(batch2, video_matcher=SHAPE_VIDEO_MATCHER)
print(f"Result: Single video with {len(batch1.videos[0].filename)} images")
Frame-level merging¶
For fine-grained control over individual frames:
from sleap_io.model.matching import InstanceMatcher
# Get specific frames
frame1 = base_labels.labeled_frames[0]
frame2 = new_labels.labeled_frames[0]
# Configure matching
matcher = InstanceMatcher(method="spatial", threshold=5.0)
# Merge at frame level
merged_instances, conflicts = frame1.merge(
frame2,
instance_matcher=matcher,
strategy="smart"
)
# Apply merged instances
frame1.instances = merged_instances
# Review conflicts
for original, new, resolution in conflicts:
print(f"Conflict resolved: {resolution}")
Error handling¶
Configure how merge errors are handled:
# Strict mode: Stop on first error
try:
result = labels.merge(other, error_mode="strict")
except Exception as e:
print(f"Merge failed: {e}")
# Handle error...
# Continue mode: Collect errors but continue (default)
result = labels.merge(other, error_mode="continue")
if result.errors:
for error in result.errors:
print(f"Error: {error.message}")
# Warning mode: Print warnings but continue
result = labels.merge(other, error_mode="warn")
Provenance tracking¶
All merge operations are automatically tracked:
# After merging
result = base_labels.merge(new_labels)
# Check merge history
merge_history = base_labels.provenance.get("merge_history", [])
for merge in merge_history:
print(f"Merged at: {merge['timestamp']}")
print(f"Source: {merge['source_labels']['filename']}")
print(f"Frames merged: {merge['result']['frames_merged']}")
print(f"Instances added: {merge['result']['instances_added']}")
Troubleshooting¶
Common issues and solutions when merging datasets.
No frames merged
Symptom: result.frames_merged == 0
Possible causes and solutions:
-
Videos don't match
-
Frame indices don't align
-
Skeleton mismatch
Manual annotations lost
Symptom: User labels disappear after merging
Solution: Ensure you're using the smart
strategy (default):
# Explicitly set smart strategy
result = labels.merge(other, frame_strategy="smart")
# Verify manual labels are preserved
manual_count_before = sum(1 for f in labels for i in f.instances
if type(i).__name__ == 'Instance')
result = labels.merge(other)
manual_count_after = sum(1 for f in labels for i in f.instances
if type(i).__name__ == 'Instance')
assert manual_count_after >= manual_count_before
Duplicate instances
Symptom: Multiple instances at the same location
Solutions:
-
Tighten spatial matching
-
Use identity matching for tracked data
-
Post-process to remove duplicates
from sleap_io.model.instance import PredictedInstance for frame in labels: # Remove duplicate predictions within threshold cleaned = [] for inst in frame.instances: is_duplicate = False for other in cleaned: if inst.numpy().mean(axis=0).dist(other.numpy().mean(axis=0)) < 5: is_duplicate = True break if not is_duplicate: cleaned.append(inst) frame.instances = cleaned
Memory issues with large datasets
Symptom: Out of memory errors during merge
Solutions:
-
Process in batches
-
Use progress callback to monitor
API reference¶
Labels merge method¶
sleap_io.model.labels.Labels.merge(other, instance_matcher=None, skeleton_matcher=None, video_matcher=None, track_matcher=None, frame_strategy='smart', validate=True, progress_callback=None, error_mode='continue')
¶
Merge another Labels object into this one.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
other
|
'Labels'
|
Another Labels object to merge into this one. |
required |
instance_matcher
|
Optional['InstanceMatcher']
|
Matcher for comparing instances. If None, uses default spatial matching with 5px tolerance. |
None
|
skeleton_matcher
|
Optional['SkeletonMatcher']
|
Matcher for comparing skeletons. If None, uses structure matching. |
None
|
video_matcher
|
Optional['VideoMatcher']
|
Matcher for comparing videos. If None, uses auto matching. |
None
|
track_matcher
|
Optional['TrackMatcher']
|
Matcher for comparing tracks. If None, uses name matching. |
None
|
frame_strategy
|
str
|
Strategy for merging frames: - "smart": Keep user labels, update predictions - "keep_original": Keep original frames - "keep_new": Replace with new frames - "keep_both": Keep all frames |
'smart'
|
validate
|
bool
|
If True, validate for conflicts before merging. |
True
|
progress_callback
|
Optional[Callable]
|
Optional callback for progress updates. Should accept (current, total, message) arguments. |
None
|
error_mode
|
str
|
How to handle errors: - "continue": Log errors but continue - "strict": Raise exception on first error - "warn": Print warnings but continue |
'continue'
|
Returns:
Type | Description |
---|---|
'MergeResult'
|
MergeResult object with statistics and any errors/conflicts. |
Notes
This method modifies the Labels object in place. The merge is designed to handle common workflows like merging predictions back into a project.
Source code in sleap_io/model/labels.py
def merge(
self,
other: "Labels",
instance_matcher: Optional["InstanceMatcher"] = None,
skeleton_matcher: Optional["SkeletonMatcher"] = None,
video_matcher: Optional["VideoMatcher"] = None,
track_matcher: Optional["TrackMatcher"] = None,
frame_strategy: str = "smart",
validate: bool = True,
progress_callback: Optional[Callable] = None,
error_mode: str = "continue",
) -> "MergeResult":
"""Merge another Labels object into this one.
Args:
other: Another Labels object to merge into this one.
instance_matcher: Matcher for comparing instances. If None, uses default
spatial matching with 5px tolerance.
skeleton_matcher: Matcher for comparing skeletons. If None, uses structure
matching.
video_matcher: Matcher for comparing videos. If None, uses auto matching.
track_matcher: Matcher for comparing tracks. If None, uses name matching.
frame_strategy: Strategy for merging frames:
- "smart": Keep user labels, update predictions
- "keep_original": Keep original frames
- "keep_new": Replace with new frames
- "keep_both": Keep all frames
validate: If True, validate for conflicts before merging.
progress_callback: Optional callback for progress updates.
Should accept (current, total, message) arguments.
error_mode: How to handle errors:
- "continue": Log errors but continue
- "strict": Raise exception on first error
- "warn": Print warnings but continue
Returns:
MergeResult object with statistics and any errors/conflicts.
Notes:
This method modifies the Labels object in place. The merge is designed to
handle common workflows like merging predictions back into a project.
"""
from datetime import datetime
from pathlib import Path
from sleap_io.model.matching import (
ConflictResolution,
ErrorMode,
InstanceMatcher,
MergeError,
MergeResult,
SkeletonMatcher,
SkeletonMatchMethod,
SkeletonMismatchError,
TrackMatcher,
VideoMatcher,
VideoMatchMethod,
)
# Initialize matchers with defaults if not provided
if instance_matcher is None:
instance_matcher = InstanceMatcher()
if skeleton_matcher is None:
skeleton_matcher = SkeletonMatcher(method=SkeletonMatchMethod.STRUCTURE)
if video_matcher is None:
video_matcher = VideoMatcher()
if track_matcher is None:
track_matcher = TrackMatcher()
# Parse error mode
error_mode_enum = ErrorMode(error_mode)
# Initialize result
result = MergeResult(successful=True)
# Track merge history in provenance
if "merge_history" not in self.provenance:
self.provenance["merge_history"] = []
merge_record = {
"timestamp": datetime.now().isoformat(),
"source_labels": {
"n_frames": len(other.labeled_frames),
"n_videos": len(other.videos),
"n_skeletons": len(other.skeletons),
"n_tracks": len(other.tracks),
},
"strategy": frame_strategy,
}
try:
# Step 1: Match and merge skeletons
skeleton_map = {}
for other_skel in other.skeletons:
matched = False
for self_skel in self.skeletons:
if skeleton_matcher.match(self_skel, other_skel):
skeleton_map[other_skel] = self_skel
matched = True
break
if not matched:
if validate and error_mode_enum == ErrorMode.STRICT:
raise SkeletonMismatchError(
message=f"No matching skeleton found for {other_skel.name}",
details={"skeleton": other_skel},
)
elif error_mode_enum == ErrorMode.WARN:
print(f"Warning: No matching skeleton for {other_skel.name}")
# Add new skeleton if no match
self.skeletons.append(other_skel)
skeleton_map[other_skel] = other_skel
# Step 2: Match and merge videos
video_map = {}
frame_idx_map = {} # Maps (old_video, old_idx) -> (new_video, new_idx)
for other_video in other.videos:
matched = False
for self_video in self.videos:
if video_matcher.match(self_video, other_video):
# Special handling for different match methods
if video_matcher.method == VideoMatchMethod.IMAGE_DEDUP:
# Deduplicate images from other_video
deduped_video = other_video.deduplicate_with(self_video)
if deduped_video is None:
# All images were duplicates, map to existing video
video_map[other_video] = self_video
# Build frame index mapping for deduplicated frames
if isinstance(
other_video.filename, list
) and isinstance(self_video.filename, list):
other_basenames = [
Path(f).name for f in other_video.filename
]
self_basenames = [
Path(f).name for f in self_video.filename
]
for old_idx, basename in enumerate(other_basenames):
if basename in self_basenames:
new_idx = self_basenames.index(basename)
frame_idx_map[(other_video, old_idx)] = (
self_video,
new_idx,
)
else:
# Add deduplicated video as new
self.videos.append(deduped_video)
video_map[other_video] = deduped_video
# Build frame index mapping for remaining frames
if isinstance(
other_video.filename, list
) and isinstance(deduped_video.filename, list):
other_basenames = [
Path(f).name for f in other_video.filename
]
deduped_basenames = [
Path(f).name for f in deduped_video.filename
]
for old_idx, basename in enumerate(other_basenames):
if basename in deduped_basenames:
new_idx = deduped_basenames.index(basename)
frame_idx_map[(other_video, old_idx)] = (
deduped_video,
new_idx,
)
elif video_matcher.method == VideoMatchMethod.SHAPE:
# Merge videos with same shape
merged_video = self_video.merge_with(other_video)
# Replace self_video with merged version
self_video_idx = self.videos.index(self_video)
self.videos[self_video_idx] = merged_video
video_map[other_video] = merged_video
video_map[self_video] = (
merged_video # Update mapping for self too
)
# Build frame index mapping
if isinstance(other_video.filename, list) and isinstance(
merged_video.filename, list
):
other_basenames = [
Path(f).name for f in other_video.filename
]
merged_basenames = [
Path(f).name for f in merged_video.filename
]
for old_idx, basename in enumerate(other_basenames):
if basename in merged_basenames:
new_idx = merged_basenames.index(basename)
frame_idx_map[(other_video, old_idx)] = (
merged_video,
new_idx,
)
else:
# Regular matching, no special handling
video_map[other_video] = self_video
matched = True
break
if not matched:
# Add new video if no match
self.videos.append(other_video)
video_map[other_video] = other_video
# Step 3: Match and merge tracks
track_map = {}
for other_track in other.tracks:
matched = False
for self_track in self.tracks:
if track_matcher.match(self_track, other_track):
track_map[other_track] = self_track
matched = True
break
if not matched:
# Add new track if no match
self.tracks.append(other_track)
track_map[other_track] = other_track
# Step 4: Merge frames
total_frames = len(other.labeled_frames)
for frame_idx, other_frame in enumerate(other.labeled_frames):
if progress_callback:
progress_callback(
frame_idx,
total_frames,
f"Merging frame {frame_idx + 1}/{total_frames}",
)
# Check if frame index needs remapping (for deduplicated/merged videos)
if (other_frame.video, other_frame.frame_idx) in frame_idx_map:
mapped_video, mapped_frame_idx = frame_idx_map[
(other_frame.video, other_frame.frame_idx)
]
else:
# Map video to self
mapped_video = video_map.get(other_frame.video, other_frame.video)
mapped_frame_idx = other_frame.frame_idx
# Find matching frame in self
matching_frames = self.find(mapped_video, mapped_frame_idx)
if len(matching_frames) == 0:
# No matching frame, create new one
new_frame = LabeledFrame(
video=mapped_video,
frame_idx=mapped_frame_idx,
instances=[],
)
# Map instances to new skeleton/track
for inst in other_frame.instances:
new_inst = self._map_instance(inst, skeleton_map, track_map)
new_frame.instances.append(new_inst)
result.instances_added += 1
self.append(new_frame)
result.frames_merged += 1
else:
# Merge into existing frame
self_frame = matching_frames[0]
# Merge instances using frame-level merge
merged_instances, conflicts = self_frame.merge(
other_frame,
instance_matcher=instance_matcher,
strategy=frame_strategy,
)
# Remap skeleton and track references for instances from other frame
remapped_instances = []
for inst in merged_instances:
# Check if instance needs remapping (from other_frame)
if inst.skeleton in skeleton_map:
# Instance needs remapping
remapped_inst = self._map_instance(
inst, skeleton_map, track_map
)
remapped_instances.append(remapped_inst)
else:
# Instance already has correct skeleton (from self_frame)
remapped_instances.append(inst)
merged_instances = remapped_instances
# Count changes
n_before = len(self_frame.instances)
n_after = len(merged_instances)
result.instances_added += max(0, n_after - n_before)
# Record conflicts
for orig, new, resolution in conflicts:
result.conflicts.append(
ConflictResolution(
frame=self_frame,
conflict_type="instance_conflict",
original_data=orig,
new_data=new,
resolution=resolution,
)
)
# Update frame instances
self_frame.instances = merged_instances
result.frames_merged += 1
# Step 5: Merge suggestions
for other_suggestion in other.suggestions:
mapped_video = video_map.get(
other_suggestion.video, other_suggestion.video
)
# Check if suggestion already exists
exists = False
for self_suggestion in self.suggestions:
if (
self_suggestion.video == mapped_video
and self_suggestion.frame_idx == other_suggestion.frame_idx
):
exists = True
break
if not exists:
# Create new suggestion with mapped video
new_suggestion = SuggestionFrame(
video=mapped_video, frame_idx=other_suggestion.frame_idx
)
self.suggestions.append(new_suggestion)
# Update merge record
merge_record["result"] = {
"frames_merged": result.frames_merged,
"instances_added": result.instances_added,
"conflicts": len(result.conflicts),
}
self.provenance["merge_history"].append(merge_record)
except MergeError as e:
result.successful = False
result.errors.append(e)
if error_mode_enum == ErrorMode.STRICT:
raise
except Exception as e:
result.successful = False
result.errors.append(
MergeError(message=str(e), details={"exception": type(e).__name__})
)
if error_mode_enum == ErrorMode.STRICT:
raise
if progress_callback:
progress_callback(total_frames, total_frames, "Merge complete")
return result
Enums¶
sleap_io.model.matching.SkeletonMatchMethod
¶
Bases: str
, Enum
Methods for matching skeletons.
Attributes:
Name | Type | Description |
---|---|---|
EXACT |
Exact match requiring same nodes in the same order. |
|
STRUCTURE |
Match requiring same nodes and edges, but order doesn't matter. |
|
OVERLAP |
Partial match based on overlapping nodes (uses Jaccard similarity). |
|
SUBSET |
Match if one skeleton's nodes are a subset of another's. |
Source code in sleap_io/model/matching.py
class SkeletonMatchMethod(str, Enum):
"""Methods for matching skeletons.
Attributes:
EXACT: Exact match requiring same nodes in the same order.
STRUCTURE: Match requiring same nodes and edges, but order doesn't matter.
OVERLAP: Partial match based on overlapping nodes (uses Jaccard similarity).
SUBSET: Match if one skeleton's nodes are a subset of another's.
"""
EXACT = "exact"
STRUCTURE = "structure"
OVERLAP = "overlap"
SUBSET = "subset"
sleap_io.model.matching.InstanceMatchMethod
¶
Bases: str
, Enum
Methods for matching instances.
Attributes:
Name | Type | Description |
---|---|---|
SPATIAL |
Match instances by spatial proximity using Euclidean distance. |
|
IDENTITY |
Match instances by track identity (same track object). |
|
IOU |
Match instances by bounding box Intersection over Union. |
Source code in sleap_io/model/matching.py
class InstanceMatchMethod(str, Enum):
"""Methods for matching instances.
Attributes:
SPATIAL: Match instances by spatial proximity using Euclidean distance.
IDENTITY: Match instances by track identity (same track object).
IOU: Match instances by bounding box Intersection over Union.
"""
SPATIAL = "spatial"
IDENTITY = "identity"
IOU = "iou"
sleap_io.model.matching.TrackMatchMethod
¶
Bases: str
, Enum
Methods for matching tracks.
Attributes:
Name | Type | Description |
---|---|---|
NAME |
Match tracks by their name attribute. |
|
IDENTITY |
Match tracks by object identity (same Python object). |
Source code in sleap_io/model/matching.py
sleap_io.model.matching.VideoMatchMethod
¶
Bases: str
, Enum
Methods for matching videos.
Attributes:
Name | Type | Description |
---|---|---|
PATH |
Match by exact file path (strict or lenient based on VideoMatcher.strict setting). |
|
BASENAME |
Match by filename only, ignoring directory paths. |
|
CONTENT |
Match by video shape (frames, height, width, channels) and backend type. |
|
AUTO |
Automatic matching - tries BASENAME first, then falls back to CONTENT. |
|
IMAGE_DEDUP |
(ImageVideo only) Match ImageVideo instances with overlapping image files. Used to deduplicate individual images when merging datasets where videos are image sequences. |
|
SHAPE |
Match videos by shape only (height, width, channels), ignoring filenames and frame count. Commonly used with ImageVideo to merge same-shaped image sequences. |
Source code in sleap_io/model/matching.py
class VideoMatchMethod(str, Enum):
"""Methods for matching videos.
Attributes:
PATH: Match by exact file path (strict or lenient based on
VideoMatcher.strict setting).
BASENAME: Match by filename only, ignoring directory paths.
CONTENT: Match by video shape (frames, height, width, channels) and
backend type.
AUTO: Automatic matching - tries BASENAME first, then falls back to CONTENT.
IMAGE_DEDUP: (ImageVideo only) Match ImageVideo instances with overlapping
image files. Used to deduplicate individual images when merging datasets
where videos are image sequences.
SHAPE: Match videos by shape only (height, width, channels), ignoring
filenames and frame count. Commonly used with ImageVideo to merge
same-shaped image sequences.
"""
PATH = "path"
BASENAME = "basename"
CONTENT = "content"
AUTO = "auto"
IMAGE_DEDUP = "image_dedup"
SHAPE = "shape"
sleap_io.model.matching.FrameStrategy
¶
Bases: str
, Enum
Strategies for handling frame merging.
Attributes:
Name | Type | Description |
---|---|---|
SMART |
Smart merging that preserves user labels over predictions when they overlap. |
|
KEEP_ORIGINAL |
Always keep instances from the original (base) frame. |
|
KEEP_NEW |
Always keep instances from the new (incoming) frame. |
|
KEEP_BOTH |
Keep all instances from both frames without filtering. |
Source code in sleap_io/model/matching.py
class FrameStrategy(str, Enum):
"""Strategies for handling frame merging.
Attributes:
SMART: Smart merging that preserves user labels over predictions when
they overlap.
KEEP_ORIGINAL: Always keep instances from the original (base) frame.
KEEP_NEW: Always keep instances from the new (incoming) frame.
KEEP_BOTH: Keep all instances from both frames without filtering.
"""
SMART = "smart"
KEEP_ORIGINAL = "keep_original"
KEEP_NEW = "keep_new"
KEEP_BOTH = "keep_both"
sleap_io.model.matching.ErrorMode
¶
Bases: str
, Enum
Error handling modes for merge operations.
Attributes:
Name | Type | Description |
---|---|---|
CONTINUE |
Continue merging on errors, collecting them in the result. |
|
STRICT |
Raise an exception on the first error encountered. |
|
WARN |
Issue warnings about errors but continue merging. |
Source code in sleap_io/model/matching.py
class ErrorMode(str, Enum):
"""Error handling modes for merge operations.
Attributes:
CONTINUE: Continue merging on errors, collecting them in the result.
STRICT: Raise an exception on the first error encountered.
WARN: Issue warnings about errors but continue merging.
"""
CONTINUE = "continue"
STRICT = "strict"
WARN = "warn"
Matcher classes¶
sleap_io.model.matching.SkeletonMatcher
¶
Matcher for comparing and matching skeletons.
Attributes:
Name | Type | Description |
---|---|---|
method |
Union[SkeletonMatchMethod, str]
|
The matching method to use. Can be a SkeletonMatchMethod enum value or a string that will be converted to the enum. Default is STRUCTURE. |
require_same_order |
bool
|
Whether to require nodes in the same order for STRUCTURE matching. Only used when method is STRUCTURE. Default is False. |
min_overlap |
float
|
Minimum Jaccard similarity required for OVERLAP matching. Only used when method is OVERLAP. Default is 0.5. |
Methods:
Name | Description |
---|---|
match |
Check if two skeletons match according to the configured method. |
Source code in sleap_io/model/matching.py
@attrs.define
class SkeletonMatcher:
"""Matcher for comparing and matching skeletons.
Attributes:
method: The matching method to use. Can be a SkeletonMatchMethod enum value
or a string that will be converted to the enum. Default is STRUCTURE.
require_same_order: Whether to require nodes in the same order for STRUCTURE
matching. Only used when method is STRUCTURE. Default is False.
min_overlap: Minimum Jaccard similarity required for OVERLAP matching.
Only used when method is OVERLAP. Default is 0.5.
"""
method: Union[SkeletonMatchMethod, str] = attrs.field(
default=SkeletonMatchMethod.STRUCTURE,
converter=lambda x: SkeletonMatchMethod(x) if isinstance(x, str) else x,
)
require_same_order: bool = False
min_overlap: float = 0.5
def match(self, skeleton1: Skeleton, skeleton2: Skeleton) -> bool:
"""Check if two skeletons match according to the configured method."""
if self.method == SkeletonMatchMethod.EXACT:
return skeleton1.matches(skeleton2, require_same_order=True)
elif self.method == SkeletonMatchMethod.STRUCTURE:
return skeleton1.matches(
skeleton2, require_same_order=self.require_same_order
)
elif self.method == SkeletonMatchMethod.OVERLAP:
metrics = skeleton1.node_similarities(skeleton2)
return metrics["jaccard"] >= self.min_overlap
elif self.method == SkeletonMatchMethod.SUBSET:
# Check if skeleton1 nodes are subset of skeleton2
nodes1 = set(skeleton1.node_names)
nodes2 = set(skeleton2.node_names)
return nodes1.issubset(nodes2)
else:
raise ValueError(f"Unknown skeleton match method: {self.method}")
match(skeleton1, skeleton2)
¶
Check if two skeletons match according to the configured method.
Source code in sleap_io/model/matching.py
def match(self, skeleton1: Skeleton, skeleton2: Skeleton) -> bool:
"""Check if two skeletons match according to the configured method."""
if self.method == SkeletonMatchMethod.EXACT:
return skeleton1.matches(skeleton2, require_same_order=True)
elif self.method == SkeletonMatchMethod.STRUCTURE:
return skeleton1.matches(
skeleton2, require_same_order=self.require_same_order
)
elif self.method == SkeletonMatchMethod.OVERLAP:
metrics = skeleton1.node_similarities(skeleton2)
return metrics["jaccard"] >= self.min_overlap
elif self.method == SkeletonMatchMethod.SUBSET:
# Check if skeleton1 nodes are subset of skeleton2
nodes1 = set(skeleton1.node_names)
nodes2 = set(skeleton2.node_names)
return nodes1.issubset(nodes2)
else:
raise ValueError(f"Unknown skeleton match method: {self.method}")
sleap_io.model.matching.InstanceMatcher
¶
Matcher for comparing and matching instances.
Attributes:
Name | Type | Description |
---|---|---|
method |
Union[InstanceMatchMethod, str]
|
The matching method to use. Can be an InstanceMatchMethod enum value or a string that will be converted to the enum. Default is SPATIAL. |
threshold |
float
|
The threshold value used for matching. For SPATIAL method, this is the maximum pixel distance. For IOU method, this is the minimum IoU value. Not used for IDENTITY method. Default is 5.0. |
Methods:
Name | Description |
---|---|
find_matches |
Find all matching instances between two lists. |
match |
Check if two instances match according to the configured method. |
Source code in sleap_io/model/matching.py
@attrs.define
class InstanceMatcher:
"""Matcher for comparing and matching instances.
Attributes:
method: The matching method to use. Can be an InstanceMatchMethod enum value
or a string that will be converted to the enum. Default is SPATIAL.
threshold: The threshold value used for matching. For SPATIAL method, this is
the maximum pixel distance. For IOU method, this is the minimum IoU value.
Not used for IDENTITY method. Default is 5.0.
"""
method: Union[InstanceMatchMethod, str] = attrs.field(
default=InstanceMatchMethod.SPATIAL,
converter=lambda x: InstanceMatchMethod(x) if isinstance(x, str) else x,
)
threshold: float = 5.0
def match(self, instance1: Instance, instance2: Instance) -> bool:
"""Check if two instances match according to the configured method."""
if self.method == InstanceMatchMethod.SPATIAL:
return instance1.same_pose_as(instance2, tolerance=self.threshold)
elif self.method == InstanceMatchMethod.IDENTITY:
return instance1.same_identity_as(instance2)
elif self.method == InstanceMatchMethod.IOU:
return instance1.overlaps_with(instance2, iou_threshold=self.threshold)
else:
raise ValueError(f"Unknown instance match method: {self.method}")
def find_matches(
self, instances1: list[Instance], instances2: list[Instance]
) -> list[tuple[int, int, float]]:
"""Find all matching instances between two lists.
Returns:
List of (idx1, idx2, score) tuples for matching instances.
"""
matches = []
for i, inst1 in enumerate(instances1):
for j, inst2 in enumerate(instances2):
if self.match(inst1, inst2):
# Calculate match score based on method
if self.method == InstanceMatchMethod.SPATIAL:
# Use inverse distance as score
pts1 = inst1.numpy()
pts2 = inst2.numpy()
valid = ~(np.isnan(pts1[:, 0]) | np.isnan(pts2[:, 0]))
if valid.any():
distances = np.linalg.norm(
pts1[valid] - pts2[valid], axis=1
)
score = 1.0 / (1.0 + np.mean(distances))
else:
score = 0.0
elif self.method == InstanceMatchMethod.IOU:
# Calculate actual IoU as score
bbox1 = inst1.bounding_box()
bbox2 = inst2.bounding_box()
if bbox1 is not None and bbox2 is not None:
# Calculate IoU
intersection_min = np.maximum(bbox1[0], bbox2[0])
intersection_max = np.minimum(bbox1[1], bbox2[1])
if np.all(intersection_min < intersection_max):
intersection_area = np.prod(
intersection_max - intersection_min
)
area1 = np.prod(bbox1[1] - bbox1[0])
area2 = np.prod(bbox2[1] - bbox2[0])
union_area = area1 + area2 - intersection_area
score = (
intersection_area / union_area
if union_area > 0
else 0
)
else:
score = 0.0
else:
score = 0.0
else:
score = 1.0 # Binary match for identity
matches.append((i, j, score))
return matches
find_matches(instances1, instances2)
¶
Find all matching instances between two lists.
Returns:
Type | Description |
---|---|
list[tuple[int, int, float]]
|
List of (idx1, idx2, score) tuples for matching instances. |
Source code in sleap_io/model/matching.py
def find_matches(
self, instances1: list[Instance], instances2: list[Instance]
) -> list[tuple[int, int, float]]:
"""Find all matching instances between two lists.
Returns:
List of (idx1, idx2, score) tuples for matching instances.
"""
matches = []
for i, inst1 in enumerate(instances1):
for j, inst2 in enumerate(instances2):
if self.match(inst1, inst2):
# Calculate match score based on method
if self.method == InstanceMatchMethod.SPATIAL:
# Use inverse distance as score
pts1 = inst1.numpy()
pts2 = inst2.numpy()
valid = ~(np.isnan(pts1[:, 0]) | np.isnan(pts2[:, 0]))
if valid.any():
distances = np.linalg.norm(
pts1[valid] - pts2[valid], axis=1
)
score = 1.0 / (1.0 + np.mean(distances))
else:
score = 0.0
elif self.method == InstanceMatchMethod.IOU:
# Calculate actual IoU as score
bbox1 = inst1.bounding_box()
bbox2 = inst2.bounding_box()
if bbox1 is not None and bbox2 is not None:
# Calculate IoU
intersection_min = np.maximum(bbox1[0], bbox2[0])
intersection_max = np.minimum(bbox1[1], bbox2[1])
if np.all(intersection_min < intersection_max):
intersection_area = np.prod(
intersection_max - intersection_min
)
area1 = np.prod(bbox1[1] - bbox1[0])
area2 = np.prod(bbox2[1] - bbox2[0])
union_area = area1 + area2 - intersection_area
score = (
intersection_area / union_area
if union_area > 0
else 0
)
else:
score = 0.0
else:
score = 0.0
else:
score = 1.0 # Binary match for identity
matches.append((i, j, score))
return matches
match(instance1, instance2)
¶
Check if two instances match according to the configured method.
Source code in sleap_io/model/matching.py
def match(self, instance1: Instance, instance2: Instance) -> bool:
"""Check if two instances match according to the configured method."""
if self.method == InstanceMatchMethod.SPATIAL:
return instance1.same_pose_as(instance2, tolerance=self.threshold)
elif self.method == InstanceMatchMethod.IDENTITY:
return instance1.same_identity_as(instance2)
elif self.method == InstanceMatchMethod.IOU:
return instance1.overlaps_with(instance2, iou_threshold=self.threshold)
else:
raise ValueError(f"Unknown instance match method: {self.method}")
sleap_io.model.matching.TrackMatcher
¶
Matcher for comparing and matching tracks.
Attributes:
Name | Type | Description |
---|---|---|
method |
Union[TrackMatchMethod, str]
|
The matching method to use. Can be a TrackMatchMethod enum value or a string that will be converted to the enum. Default is NAME. |
Methods:
Name | Description |
---|---|
match |
Check if two tracks match according to the configured method. |
Source code in sleap_io/model/matching.py
@attrs.define
class TrackMatcher:
"""Matcher for comparing and matching tracks.
Attributes:
method: The matching method to use. Can be a TrackMatchMethod enum value
or a string that will be converted to the enum. Default is NAME.
"""
method: Union[TrackMatchMethod, str] = attrs.field(
default=TrackMatchMethod.NAME,
converter=lambda x: TrackMatchMethod(x) if isinstance(x, str) else x,
)
def match(self, track1: Track, track2: Track) -> bool:
"""Check if two tracks match according to the configured method."""
return track1.matches(track2, method=self.method.value)
match(track1, track2)
¶
sleap_io.model.matching.VideoMatcher
¶
Matcher for comparing and matching videos.
Attributes:
Name | Type | Description |
---|---|---|
method |
Union[VideoMatchMethod, str]
|
The matching method to use. Can be a VideoMatchMethod enum value or a string that will be converted to the enum. Default is AUTO. |
strict |
bool
|
Whether to use strict path matching for the PATH method. When True, paths must be exactly identical. When False, paths are normalized before comparison. Only used when method is PATH. Default is False. |
Methods:
Name | Description |
---|---|
match |
Check if two videos match according to the configured method. |
Source code in sleap_io/model/matching.py
@attrs.define
class VideoMatcher:
"""Matcher for comparing and matching videos.
Attributes:
method: The matching method to use. Can be a VideoMatchMethod enum value
or a string that will be converted to the enum. Default is AUTO.
strict: Whether to use strict path matching for the PATH method.
When True, paths must be exactly identical. When False, paths
are normalized before comparison. Only used when method is PATH.
Default is False.
"""
method: Union[VideoMatchMethod, str] = attrs.field(
default=VideoMatchMethod.AUTO,
converter=lambda x: VideoMatchMethod(x) if isinstance(x, str) else x,
)
strict: bool = False
def match(self, video1: Video, video2: Video) -> bool:
"""Check if two videos match according to the configured method."""
if self.method == VideoMatchMethod.AUTO:
# Try different methods in order (identity check is redundant)
if video1.matches_path(video2, strict=False):
return True
if video1.matches_content(video2):
return True
return False
elif self.method == VideoMatchMethod.PATH:
return video1.matches_path(video2, strict=self.strict)
elif self.method == VideoMatchMethod.BASENAME:
return video1.matches_path(video2, strict=False)
elif self.method == VideoMatchMethod.CONTENT:
return video1.matches_content(video2)
elif self.method == VideoMatchMethod.IMAGE_DEDUP:
# Match ImageVideo instances with overlapping images (ImageVideo only)
return video1.has_overlapping_images(video2)
elif self.method == VideoMatchMethod.SHAPE:
# Match videos by shape only (height, width, channels)
return video1.matches_shape(video2)
else:
raise ValueError(f"Unknown video match method: {self.method}")
match(video1, video2)
¶
Check if two videos match according to the configured method.
Source code in sleap_io/model/matching.py
def match(self, video1: Video, video2: Video) -> bool:
"""Check if two videos match according to the configured method."""
if self.method == VideoMatchMethod.AUTO:
# Try different methods in order (identity check is redundant)
if video1.matches_path(video2, strict=False):
return True
if video1.matches_content(video2):
return True
return False
elif self.method == VideoMatchMethod.PATH:
return video1.matches_path(video2, strict=self.strict)
elif self.method == VideoMatchMethod.BASENAME:
return video1.matches_path(video2, strict=False)
elif self.method == VideoMatchMethod.CONTENT:
return video1.matches_content(video2)
elif self.method == VideoMatchMethod.IMAGE_DEDUP:
# Match ImageVideo instances with overlapping images (ImageVideo only)
return video1.has_overlapping_images(video2)
elif self.method == VideoMatchMethod.SHAPE:
# Match videos by shape only (height, width, channels)
return video1.matches_shape(video2)
else:
raise ValueError(f"Unknown video match method: {self.method}")
sleap_io.model.matching.FrameMatcher
¶
Matcher for comparing and matching labeled frames.
Attributes:
Name | Type | Description |
---|---|---|
video_must_match |
bool
|
Whether frames must belong to the same video to be considered a match. Default is True. |
Methods:
Name | Description |
---|---|
match |
Check if two frames match. |
Source code in sleap_io/model/matching.py
@attrs.define
class FrameMatcher:
"""Matcher for comparing and matching labeled frames.
Attributes:
video_must_match: Whether frames must belong to the same video to be
considered a match. Default is True.
"""
video_must_match: bool = True
def match(self, frame1: LabeledFrame, frame2: LabeledFrame) -> bool:
"""Check if two frames match."""
return frame1.matches(frame2, video_must_match=self.video_must_match)
match(frame1, frame2)
¶
Pre-configured matchers¶
sleap_io.model.matching.STRUCTURE_SKELETON_MATCHER = SkeletonMatcher(method=(SkeletonMatchMethod.STRUCTURE))
module-attribute
¶
sleap_io.model.matching.SUBSET_SKELETON_MATCHER = SkeletonMatcher(method=(SkeletonMatchMethod.SUBSET))
module-attribute
¶
sleap_io.model.matching.OVERLAP_SKELETON_MATCHER = SkeletonMatcher(method=(SkeletonMatchMethod.OVERLAP), min_overlap=0.7)
module-attribute
¶
sleap_io.model.matching.DUPLICATE_MATCHER = InstanceMatcher(method=(InstanceMatchMethod.SPATIAL), threshold=5.0)
module-attribute
¶
sleap_io.model.matching.IOU_MATCHER = InstanceMatcher(method=(InstanceMatchMethod.IOU), threshold=0.5)
module-attribute
¶
sleap_io.model.matching.IDENTITY_INSTANCE_MATCHER = InstanceMatcher(method=(InstanceMatchMethod.IDENTITY))
module-attribute
¶
sleap_io.model.matching.NAME_TRACK_MATCHER = TrackMatcher(method=(TrackMatchMethod.NAME))
module-attribute
¶
sleap_io.model.matching.IDENTITY_TRACK_MATCHER = TrackMatcher(method=(TrackMatchMethod.IDENTITY))
module-attribute
¶
sleap_io.model.matching.AUTO_VIDEO_MATCHER = VideoMatcher(method=(VideoMatchMethod.AUTO))
module-attribute
¶
sleap_io.model.matching.SOURCE_VIDEO_MATCHER = VideoMatcher(method=(VideoMatchMethod.BASENAME))
module-attribute
¶
sleap_io.model.matching.PATH_VIDEO_MATCHER = VideoMatcher(method=(VideoMatchMethod.PATH), strict=True)
module-attribute
¶
sleap_io.model.matching.BASENAME_VIDEO_MATCHER = VideoMatcher(method=(VideoMatchMethod.BASENAME))
module-attribute
¶
sleap_io.model.matching.IMAGE_DEDUP_VIDEO_MATCHER = VideoMatcher(method=(VideoMatchMethod.IMAGE_DEDUP))
module-attribute
¶
sleap_io.model.matching.SHAPE_VIDEO_MATCHER = VideoMatcher(method=(VideoMatchMethod.SHAPE))
module-attribute
¶
Result classes¶
sleap_io.model.matching.MergeResult
¶
Result of a merge operation.
Attributes:
Name | Type | Description |
---|---|---|
successful |
bool
|
Whether the merge completed successfully. |
frames_merged |
int
|
Number of frames that were merged. |
instances_added |
int
|
Number of new instances added. |
instances_updated |
int
|
Number of existing instances that were updated. |
instances_skipped |
int
|
Number of instances that were skipped. |
conflicts |
list[ConflictResolution]
|
List of conflicts that were resolved during merging. |
errors |
list[MergeError]
|
List of errors encountered during merging. |
Methods:
Name | Description |
---|---|
summary |
Generate a human-readable summary of the merge result. |
Source code in sleap_io/model/matching.py
@attrs.define
class MergeResult:
"""Result of a merge operation.
Attributes:
successful: Whether the merge completed successfully.
frames_merged: Number of frames that were merged.
instances_added: Number of new instances added.
instances_updated: Number of existing instances that were updated.
instances_skipped: Number of instances that were skipped.
conflicts: List of conflicts that were resolved during merging.
errors: List of errors encountered during merging.
"""
successful: bool
frames_merged: int = 0
instances_added: int = 0
instances_updated: int = 0
instances_skipped: int = 0
conflicts: list[ConflictResolution] = attrs.field(factory=list)
errors: list[MergeError] = attrs.field(factory=list)
def summary(self) -> str:
"""Generate a human-readable summary of the merge result."""
lines = []
if self.successful:
lines.append("✓ Merge completed successfully")
else:
lines.append("✗ Merge completed with errors")
lines.append(f" Frames merged: {self.frames_merged}")
lines.append(f" Instances added: {self.instances_added}")
if self.instances_updated:
lines.append(f" Instances updated: {self.instances_updated}")
if self.instances_skipped:
lines.append(f" Instances skipped: {self.instances_skipped}")
if self.conflicts:
lines.append(f" Conflicts resolved: {len(self.conflicts)}")
if self.errors:
lines.append(f" Errors encountered: {len(self.errors)}")
for error in self.errors[:5]: # Show first 5 errors
lines.append(f" - {error.message}")
if len(self.errors) > 5:
lines.append(f" ... and {len(self.errors) - 5} more")
return "\n".join(lines)
summary()
¶
Generate a human-readable summary of the merge result.
Source code in sleap_io/model/matching.py
def summary(self) -> str:
"""Generate a human-readable summary of the merge result."""
lines = []
if self.successful:
lines.append("✓ Merge completed successfully")
else:
lines.append("✗ Merge completed with errors")
lines.append(f" Frames merged: {self.frames_merged}")
lines.append(f" Instances added: {self.instances_added}")
if self.instances_updated:
lines.append(f" Instances updated: {self.instances_updated}")
if self.instances_skipped:
lines.append(f" Instances skipped: {self.instances_skipped}")
if self.conflicts:
lines.append(f" Conflicts resolved: {len(self.conflicts)}")
if self.errors:
lines.append(f" Errors encountered: {len(self.errors)}")
for error in self.errors[:5]: # Show first 5 errors
lines.append(f" - {error.message}")
if len(self.errors) > 5:
lines.append(f" ... and {len(self.errors) - 5} more")
return "\n".join(lines)
sleap_io.model.matching.ConflictResolution
¶
Information about a conflict that was resolved during merging.
Attributes:
Name | Type | Description |
---|---|---|
frame |
LabeledFrame
|
The labeled frame where the conflict occurred. |
conflict_type |
str
|
Type of conflict (e.g., "duplicate_instance", "skeleton_mismatch"). |
original_data |
Any
|
The original data before resolution. |
new_data |
Any
|
The new/incoming data that caused the conflict. |
resolution |
str
|
Description of how the conflict was resolved. |
Source code in sleap_io/model/matching.py
@attrs.define
class ConflictResolution:
"""Information about a conflict that was resolved during merging.
Attributes:
frame: The labeled frame where the conflict occurred.
conflict_type: Type of conflict (e.g., "duplicate_instance",
"skeleton_mismatch").
original_data: The original data before resolution.
new_data: The new/incoming data that caused the conflict.
resolution: Description of how the conflict was resolved.
"""
frame: LabeledFrame
conflict_type: str
original_data: Any
new_data: Any
resolution: str
sleap_io.model.matching.MergeError
¶
Bases: Exception
Base exception for merge errors.
Attributes:
Name | Type | Description |
---|---|---|
message |
str
|
Human-readable error message. |
details |
dict
|
Dictionary containing additional error details and context. |
Source code in sleap_io/model/matching.py
sleap_io.model.matching.SkeletonMismatchError
¶
Bases: MergeError
Raised when skeletons don't match during merge.
Source code in sleap_io/model/matching.py
sleap_io.model.matching.VideoNotFoundError
¶
Bases: MergeError
Raised when a video file cannot be found during merge.
Source code in sleap_io/model/matching.py
Progress tracking¶
sleap_io.model.matching.MergeProgressBar
¶
Context manager for merge progress tracking using tqdm.
This provides a clean interface for tracking merge progress with visual feedback.
Example
with MergeProgressBar("Merging predictions") as progress: result = labels.merge(predictions, progress_callback=progress.callback)
Methods:
Name | Description |
---|---|
__enter__ |
Enter the context manager. |
__exit__ |
Exit the context manager and close the progress bar. |
__init__ |
Initialize the progress bar. |
callback |
Progress callback for merge operations. |
Source code in sleap_io/model/matching.py
class MergeProgressBar:
"""Context manager for merge progress tracking using tqdm.
This provides a clean interface for tracking merge progress with visual feedback.
Example:
with MergeProgressBar("Merging predictions") as progress:
result = labels.merge(predictions, progress_callback=progress.callback)
"""
def __init__(self, desc: str = "Merging", leave: bool = True):
"""Initialize the progress bar.
Args:
desc: Description to show in the progress bar.
leave: Whether to leave the progress bar on screen after completion.
"""
self.desc = desc
self.leave = leave
self.pbar = None
def __enter__(self):
"""Enter the context manager."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Exit the context manager and close the progress bar."""
if self.pbar is not None:
self.pbar.close()
def callback(self, current: int, total: int, message: str = ""):
"""Progress callback for merge operations.
Args:
current: Current progress value.
total: Total items to process.
message: Optional message to display.
"""
from tqdm import tqdm
if self.pbar is None and total:
self.pbar = tqdm(total=total, desc=self.desc, leave=self.leave)
if self.pbar:
if message:
self.pbar.set_description(f"{self.desc}: {message}")
else:
self.pbar.set_description(self.desc)
self.pbar.n = current
self.pbar.refresh()
__enter__()
¶
__exit__(exc_type, exc_val, exc_tb)
¶
__init__(desc='Merging', leave=True)
¶
Initialize the progress bar.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
desc
|
str
|
Description to show in the progress bar. |
'Merging'
|
leave
|
bool
|
Whether to leave the progress bar on screen after completion. |
True
|
Source code in sleap_io/model/matching.py
callback(current, total, message='')
¶
Progress callback for merge operations.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
current
|
int
|
Current progress value. |
required |
total
|
int
|
Total items to process. |
required |
message
|
str
|
Optional message to display. |
''
|
Source code in sleap_io/model/matching.py
def callback(self, current: int, total: int, message: str = ""):
"""Progress callback for merge operations.
Args:
current: Current progress value.
total: Total items to process.
message: Optional message to display.
"""
from tqdm import tqdm
if self.pbar is None and total:
self.pbar = tqdm(total=total, desc=self.desc, leave=self.leave)
if self.pbar:
if message:
self.pbar.set_description(f"{self.desc}: {message}")
else:
self.pbar.set_description(self.desc)
self.pbar.n = current
self.pbar.refresh()