Skip to content

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:

Basic merge example
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

  1. Skeleton matching: Identifies corresponding skeletons between datasets based on node names and structure
  2. Video matching: Determines which videos represent the same source data (handles different paths, formats)
  3. Track matching: Maps tracks (instance identities) between datasets
  4. Frame merging: For each frame, applies the chosen strategy to combine instances
  5. 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
Smart merge logic (simplified)
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
Using smart strategy
result = base_labels.merge(new_labels, frame_strategy="smart")

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
Using keep original strategy
result = base_labels.merge(new_labels, frame_strategy="keep_original")

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
Using keep new strategy
result = base_labels.merge(new_labels, frame_strategy="keep_new")

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
Using keep both strategy
result = base_labels.merge(new_labels, frame_strategy="keep_both")

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
from sleap_io.model.matching import VideoMatcher, VideoMatchMethod

matcher = VideoMatcher(method=VideoMatchMethod.AUTO)
result = base_labels.merge(new_labels, video_matcher=matcher)

Exact path matching with optional strict mode:

# Strict: paths must be identical
strict_matcher = VideoMatcher(method=VideoMatchMethod.PATH, strict=True)

# Lenient: normalizes paths before comparison
lenient_matcher = VideoMatcher(method=VideoMatchMethod.PATH, strict=False)

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:

# Useful when files are renamed but content is identical
content_matcher = VideoMatcher(method=VideoMatchMethod.CONTENT)

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:

# Only matches if instances belong to the same track
matcher = InstanceMatcher(method="identity")

Matches by bounding box overlap:

# Threshold is minimum Intersection over Union
matcher = InstanceMatcher(method="iou", threshold=0.5)

Skeleton matching

Controls how skeletons are matched and mapped between datasets.

Same nodes and edges, order doesn't matter:

from sleap_io.model.matching import STRUCTURE_SKELETON_MATCHER

result = base_labels.merge(new_labels, skeleton_matcher=STRUCTURE_SKELETON_MATCHER)

Nodes must be identical and in the same order:

matcher = SkeletonMatcher(method="exact")

Partial match based on Jaccard similarity:

# Requires 70% node overlap
matcher = SkeletonMatcher(method="overlap", min_overlap=0.7)

One skeleton's nodes must be a subset of the other:

from sleap_io.model.matching import SUBSET_SKELETON_MATCHER

# Useful for adding nodes to existing skeletons
result = base_labels.merge(new_labels, skeleton_matcher=SUBSET_SKELETON_MATCHER)

Common workflows

Human-in-the-loop (HITL) training

The most common workflow: merging model predictions back into your training data.

HITL workflow
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
Cross-platform merging
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:

Multi-annotator workflow
# 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:

Updating predictions
# 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.

Deduplicating image sequences
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:

Frame-level merge control
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:

Error handling strategies
# 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:

Accessing merge history
# 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:

  1. Videos don't match

    # Try different video matching strategies
    result = labels.merge(other, video_matcher=BASENAME_VIDEO_MATCHER)
    # or
    result = labels.merge(other, video_matcher=AUTO_VIDEO_MATCHER)
    

  2. Frame indices don't align

    # Check frame indices
    print(f"Base frames: {[f.frame_idx for f in base_labels]}")
    print(f"New frames: {[f.frame_idx for f in new_labels]}")
    

  3. Skeleton mismatch

    # Validate and use lenient matching
    result = labels.merge(
        other,
        validate=True,
        skeleton_matcher=OVERLAP_SKELETON_MATCHER
    )
    

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:

  1. Tighten spatial matching

    matcher = InstanceMatcher(method="spatial", threshold=2.0)  # Stricter
    result = labels.merge(other, instance_matcher=matcher)
    

  2. Use identity matching for tracked data

    matcher = InstanceMatcher(method="identity")
    result = labels.merge(other, instance_matcher=matcher)
    

  3. 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:

  1. Process in batches

    # Split by video
    for video in new_labels.videos:
        video_labels = new_labels.extract_video(video)
        result = base_labels.merge(video_labels)
    

  2. Use progress callback to monitor

    def progress(current, total, message):
        print(f"{current}/{total}: {message}")
        # Add memory monitoring here if needed
    
    result = labels.merge(other, progress_callback=progress)
    

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
class TrackMatchMethod(str, Enum):
    """Methods for matching tracks.

    Attributes:
        NAME: Match tracks by their name attribute.
        IDENTITY: Match tracks by object identity (same Python object).
    """

    NAME = "name"
    IDENTITY = "identity"

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)

Check if two tracks match according to the configured method.

Source code in sleap_io/model/matching.py
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)

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)

Check if two frames match.

Source code in sleap_io/model/matching.py
def match(self, frame1: LabeledFrame, frame2: LabeledFrame) -> bool:
    """Check if two frames match."""
    return frame1.matches(frame2, video_must_match=self.video_must_match)

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
@attrs.define
class MergeError(Exception):
    """Base exception for merge errors.

    Attributes:
        message: Human-readable error message.
        details: Dictionary containing additional error details and context.
    """

    message: str
    details: dict = attrs.field(factory=dict)

sleap_io.model.matching.SkeletonMismatchError

Bases: MergeError

Raised when skeletons don't match during merge.

Source code in sleap_io/model/matching.py
class SkeletonMismatchError(MergeError):
    """Raised when skeletons don't match during merge."""

    pass

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
class VideoNotFoundError(MergeError):
    """Raised when a video file cannot be found during merge."""

    pass

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__()

Enter the context manager.

Source code in sleap_io/model/matching.py
def __enter__(self):
    """Enter the context manager."""
    return self
__exit__(exc_type, exc_val, exc_tb)

Exit the context manager and close the progress bar.

Source code in sleap_io/model/matching.py
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()
__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
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
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()