sleap.nn.tracking

Tracking tools for linking grouped instances over time.

class sleap.nn.tracking.BaseTracker

Abstract base class for tracker.

class sleap.nn.tracking.FlowCandidateMaker(min_points: int = 0, img_scale: float = 1.0, of_window_size: int = 21, of_max_levels: int = 3, save_shifted_instances: bool = False, track_window: int = 5, shifted_instances: Dict[Tuple[int, int], List[ShiftedInstance]] = _Nothing.NOTHING)

Class for producing optical flow shift matching candidates.

min_points

Minimum number of points that must be detected in the new frame in order to generate a new shifted instance.

Type:

int

img_scale

Factor to scale the images by when computing optical flow. Decrease this to increase performance at the cost of finer accuracy. Sometimes decreasing the image scale can improve performance with fast movements.

Type:

float

of_window_size

Optical flow window size to consider at each pyramid scale level.

Type:

int

of_max_levels

Number of pyramid scale levels to consider. This is different from the scale parameter, which determines the initial image scaling.

Type:

int

save_shifted_instances

If True, save the shifted instances between elapsed frames.

Type:

bool

track_window

How many frames back to look for candidate instances to match instances in the current frame against.

Type:

int

static flow_shift_instances(ref_instances: List[InstanceType], ref_img: ndarray, new_img: ndarray, min_shifted_points: int = 0, scale: float = 1.0, window_size: int = 21, max_levels: int = 3) → List[ShiftedInstance]

Generates instances in a new frame by applying optical flow displacements.

Parameters:

• ref_instances – Reference instances in the previous frame.

• ref_img – Previous frame image as a numpy array.

• new_img – New frame image as a numpy array.

• min_shifted_points – Minimum number of points that must be detected in the new frame in order to generate a new shifted instance.

• scale – Factor to scale the images by when computing optical flow. Decrease this to increase performance at the cost of finer accuracy. Sometimes decreasing the image scale can improve performance with fast movements.

• window_size – Optical flow window size to consider at each pyramid scale level.

• max_levels – Number of pyramid scale levels to consider. This is different from the scale parameter, which determines the initial image scaling.

Returns:

A list of ShiftedInstances with the optical flow displacements applied to the reference instance points. Points that are not found will be represented as NaNs in the points array for each shifted instance.

Notes

This function relies on the Lucas-Kanade method for optical flow estimation.

get_shifted_instances(ref_instances: List[InstanceType], ref_img: ndarray, ref_t: int, img: ndarray, t: int) → List[ShiftedInstance]

Returns a list of shifted instances and save shifted instances if needed.

Parameters:

• ref_instances – Reference instances in the previous frame.

• ref_img – Previous frame image as a numpy array.

• ref_t – Previous frame time instance.

• img – Current frame image as a numpy array.

• t – Current time instance.

get_shifted_instances_from_earlier_time(ref_t: int, ref_img: ~numpy.ndarray, ref_instances: ~typing.List[~sleap.nn.tracker.components.InstanceType], t: int) -> (<class 'numpy.ndarray'>, typing.List[~InstanceType])

Generate shifted instances and corresponding image from earlier time.

Parameters:

• ref_instances – Reference instances in the previous frame.

• ref_img – Previous frame image as a numpy array.

• ref_t – Previous frame time instance.

• t – Current time instance.

prune_shifted_instances(t: int)

Prune the shifted instances older than self.track_window.

If self.save_shifted_instances is False, do nothing.

Args

t: reference instances from a frame number more than self.track_window before

the current frame t will be pruned from the self.shifted_instances dict.

class sleap.nn.tracking.FlowMaxTracker(max_tracks: ~typing.Optional[int] = None, track_window: int = 5, similarity_function: ~typing.Optional[~typing.Callable] = <function instance_similarity>, matching_function: ~typing.Callable = <function greedy_matching>, candidate_maker: object = _Nothing.NOTHING, max_tracking: bool = False, cleaner: ~typing.Optional[~typing.Callable] = None, target_instance_count: int = 0, pre_cull_function: ~typing.Optional[~typing.Callable] = None, post_connect_single_breaks: bool = False, robust_best_instance: float = 1.0, min_new_track_points: int = 0, track_matching_queue: ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstances] = _Nothing.NOTHING, track_matching_queue_dict: ~typing.Dict[~sleap.instance.Track, ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstance]] = _Nothing.NOTHING, spawned_tracks: ~typing.List[~sleap.instance.Track] = _Nothing.NOTHING, save_tracked_instances: bool = False, tracked_instances: ~typing.Dict[int, ~typing.List[~sleap.nn.tracker.components.InstanceType]] = _Nothing.NOTHING, last_matches: ~typing.Optional[~sleap.nn.tracker.components.FrameMatches] = None)

Pre-configured tracker to use optical flow shifted candidates with max tracks.

matching_function() → List[Tuple[int, int]]

Performs greedy bipartite matching.

similarity_function(query_instance: InstanceType) → float

Computes similarity between instances.

class sleap.nn.tracking.FlowMaxTracksCandidateMaker(min_points: int = 0, img_scale: float = 1.0, of_window_size: int = 21, of_max_levels: int = 3, save_shifted_instances: bool = False, track_window: int = 5, shifted_instances: Dict[Tuple[int, int], List[ShiftedInstance]] = _Nothing.NOTHING, max_tracks: Optional[int] = None)

Class for producing optical flow shift matching candidates with maximum tracks.

max_tracks

The maximum number of tracks to avoid redundant tracks.

Type:

int

static get_ref_instances(ref_t: int, ref_img: ndarray, track_matching_queue_dict: Dict[Track, Deque[MatchedFrameInstance]]) → List[InstanceType]

Generates a list of instances based on the reference time and image.

Parameters:

• ref_t – Previous frame time instance.

• ref_img – Previous frame image as a numpy array.

• track_matching_queue_dict – A dictionary of mapping between the tracks and the corresponding instances associated with the track.

class sleap.nn.tracking.FlowTracker(max_tracks: ~typing.Optional[int] = None, track_window: int = 5, max_tracking: bool = False, cleaner: ~typing.Optional[~typing.Callable] = None, target_instance_count: int = 0, pre_cull_function: ~typing.Optional[~typing.Callable] = None, post_connect_single_breaks: bool = False, robust_best_instance: float = 1.0, min_new_track_points: int = 0, track_matching_queue: ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstances] = _Nothing.NOTHING, track_matching_queue_dict: ~typing.Dict[~sleap.instance.Track, ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstance]] = _Nothing.NOTHING, spawned_tracks: ~typing.List[~sleap.instance.Track] = _Nothing.NOTHING, save_tracked_instances: bool = False, tracked_instances: ~typing.Dict[int, ~typing.List[~sleap.nn.tracker.components.InstanceType]] = _Nothing.NOTHING, last_matches: ~typing.Optional[~sleap.nn.tracker.components.FrameMatches] = None, similarity_function: ~typing.Callable = <function instance_similarity>, matching_function: ~typing.Callable = <function greedy_matching>, candidate_maker: object = _Nothing.NOTHING)

A Tracker pre-configured to use optical flow shifted candidates.

class sleap.nn.tracking.KalmanTracker(init_tracker: Optional[Tracker], init_set: KalmanInitSet, kalman_tracker: BareKalmanTracker, cull_function: Optional[Callable] = None, init_frame_count: int = 10, re_init_cooldown: int = 100, re_init_after: int = 20, init_done: bool = False, pre_tracked: bool = False, last_t: int = 0, last_init_t: int = 0)

Class for Kalman filter-based tracking pipeline.

Kalman filters need to be initialized with a certain number of already tracked instances.

Parameters:

• init_tracker – The regular Tracker we can use to track data needed for initializing Kalman filters. If not specified, then you can use pre-tracked data (i.e., track assignments already set on instances) if pre_tracked is True.

• init_set – Object to keep track of tracked “init” data and determine when we have enough good data to initialize filters.

• kalman_tracker – The object which handles the actual Kalman filter-based tracking.

• cull_function – If given, this is called to cull instances before tracking.

• init_frame_count – The target number of instances/identities per frame.

• re_init_cooldown – Number of frames to wait after initializing filters before checking if we need to re-init (because they aren’t successfully matching tracks).

• re_init_after – If there’s a gap of this many frames since filters have matched tracks (and we’ve also waited for cooldown frames), start using the regular tracker so that we can re-initialize Kalman filters.

• init_done – Keeps track of whether we’re initialized the filters yet.

• pre_tracked – Whether to use init_tracker or tracks already set on instances.

• last_t – The last frame index we’ve tracked.

• last_init_t – The last frame index on which Kalman filters were initialized; used to checking cooldown period.

property is_valid

Do we have everything we need to run tracking?

classmethod make_tracker(init_tracker: Optional[Tracker], node_indices: List[int], instance_count: int, instance_iou_threshold: float = 0.8, init_frame_count: int = 10)

Creates KalmanTracker object.

Parameters:

• init_tracker – The Kalman filters need to be initialized with data that’s already been tracked. This is a regular Tracker which can be used to generate this tracked data (when needed).

• node_indices – Which nodes to track using Kalman filters; these should be nodes that are reliably present in the predictions.

• instance_count – The target number of instances to track per frame. A distinct Kalman filter is created/initialized to track each distinct identity. We’ll also use this to cull the number of predicted instances before trying to track.

• instance_iou_threshold – This is the IOU threshold so that we first cull instances which have high overlap.

• init_frame_count – How many frames of tracked data to use when initializing Kalman filters.

track(untracked_instances: List[InstanceType], img: Optional[ndarray] = None, t: Optional[int] = None, **kwargs) → List[InstanceType]

Tracks individual frame, using Kalman filters if possible.

class sleap.nn.tracking.SimpleCandidateMaker(min_points: int = 0)

Class for producing list of matching candidates from prior frames.

class sleap.nn.tracking.SimpleMaxTracker(track_window: int = 5, cleaner: ~typing.Optional[~typing.Callable] = None, target_instance_count: int = 0, pre_cull_function: ~typing.Optional[~typing.Callable] = None, post_connect_single_breaks: bool = False, robust_best_instance: float = 1.0, min_new_track_points: int = 0, track_matching_queue: ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstances] = _Nothing.NOTHING, track_matching_queue_dict: ~typing.Dict[~sleap.instance.Track, ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstance]] = _Nothing.NOTHING, spawned_tracks: ~typing.List[~sleap.instance.Track] = _Nothing.NOTHING, save_tracked_instances: bool = False, tracked_instances: ~typing.Dict[int, ~typing.List[~sleap.nn.tracker.components.InstanceType]] = _Nothing.NOTHING, last_matches: ~typing.Optional[~sleap.nn.tracker.components.FrameMatches] = None, similarity_function: ~typing.Callable = <function instance_iou>, matching_function: ~typing.Callable = <function hungarian_matching>, candidate_maker: object = _Nothing.NOTHING, max_tracking: bool = True, *, max_tracks: int)

Pre-configured tracker to use simple, non-image-based candidates with max tracks.

class sleap.nn.tracking.SimpleMaxTracksCandidateMaker(min_points: int = 0, max_tracks: Optional[int] = None)

Class to generate instances with maximum number of tracks from prior frames.

class sleap.nn.tracking.SimpleTracker(max_tracks: ~typing.Optional[int] = None, track_window: int = 5, max_tracking: bool = False, cleaner: ~typing.Optional[~typing.Callable] = None, target_instance_count: int = 0, pre_cull_function: ~typing.Optional[~typing.Callable] = None, post_connect_single_breaks: bool = False, robust_best_instance: float = 1.0, min_new_track_points: int = 0, track_matching_queue: ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstances] = _Nothing.NOTHING, track_matching_queue_dict: ~typing.Dict[~sleap.instance.Track, ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstance]] = _Nothing.NOTHING, spawned_tracks: ~typing.List[~sleap.instance.Track] = _Nothing.NOTHING, save_tracked_instances: bool = False, tracked_instances: ~typing.Dict[int, ~typing.List[~sleap.nn.tracker.components.InstanceType]] = _Nothing.NOTHING, last_matches: ~typing.Optional[~sleap.nn.tracker.components.FrameMatches] = None, similarity_function: ~typing.Callable = <function instance_iou>, matching_function: ~typing.Callable = <function hungarian_matching>, candidate_maker: object = _Nothing.NOTHING)

A Tracker pre-configured to use simple, non-image-based candidates.

class sleap.nn.tracking.TrackCleaner(instance_count: int, iou_threshold: Optional[float] = None)

Class for merging breaks in the predicted tracks.

Method: 1. You specify how many instances there should be in each frame. 2. The lowest scoring instances beyond this limit are deleting from each frame. 3. Going frame by frame, any time there’s exactly one missing track and exactly

one new track, we merge the new track into the missing track.

You should review the results to check for “swaps”. This can be done using the velocity threshold suggestion method.

instance_count

The maximum number of instances we want per frame.

Type:

int

iou_threshold

Intersection over Union (IOU) threshold to use when removing overlapping instances over target count; if None, then only use score to determine which instances to remove.

Type:

Optional[float]

class sleap.nn.tracking.Tracker(max_tracks: ~typing.Optional[int] = None, track_window: int = 5, similarity_function: ~typing.Optional[~typing.Callable] = <function instance_similarity>, matching_function: ~typing.Callable = <function greedy_matching>, candidate_maker: object = _Nothing.NOTHING, max_tracking: bool = False, cleaner: ~typing.Optional[~typing.Callable] = None, target_instance_count: int = 0, pre_cull_function: ~typing.Optional[~typing.Callable] = None, post_connect_single_breaks: bool = False, robust_best_instance: float = 1.0, min_new_track_points: int = 0, track_matching_queue: ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstances] = _Nothing.NOTHING, track_matching_queue_dict: ~typing.Dict[~sleap.instance.Track, ~typing.Deque[~sleap.nn.tracking.MatchedFrameInstance]] = _Nothing.NOTHING, spawned_tracks: ~typing.List[~sleap.instance.Track] = _Nothing.NOTHING, save_tracked_instances: bool = False, tracked_instances: ~typing.Dict[int, ~typing.List[~sleap.nn.tracker.components.InstanceType]] = _Nothing.NOTHING, last_matches: ~typing.Optional[~sleap.nn.tracker.components.FrameMatches] = None)

Instance pose tracker.

Use by instantiated with the desired parameters and then calling the track method for each frame.

track_window

How many frames back to look for candidate instances to match instances in the current frame against.

Type:

int

similarity_function

A function that returns a numeric pairwise instance similarity value.

Type:

Optional[Callable]

matching_function

A function that takes a matrix of pairwise similarities and determines the matches to use.

Type:

Callable

candidate_maker

A class instance with a get_candidates method which returns a list of Instances-like objects which we can match the predicted instances in a frame against.

Type:

object

cleaner

A class with a run method which attempts to clean tracks after the other tracking has run for all frames.

Type:

Optional[Callable]

min_new_track_points

We won’t spawn a new track for an instance with fewer than this many points.

Type:

int

robust_best_instance

if the value is between 0 and 1 (excluded), use a robust quantile similarity score for the track. If the value is 1, use the max similarity (non-robust). For selecting a robust score, 0.95 is a good value.

Type:

float

max_tracking

Max tracking is incorporated when this is set to true.

Type:

bool

final_pass(frames: List[LabeledFrame])

Called after tracking has run on all frames to do any post-processing.

track(untracked_instances: List[InstanceType], img_hw: Tuple[int], img: Optional[ndarray] = None, t: Optional[int] = None) → List[InstanceType]

Performs a single step of tracking.

Parameters:

• untracked_instances – List of instances to assign to tracks.

• img_hw – (height, width) of the image used to normalize the keypoints.

• img – Image data of the current frame for flow shifting.

• t – Current timestep. If not provided, increments from the internal queue.

Returns:

A list of the instances that were tracked.

property unique_tracks_in_queue: List[Track]

Returns the unique tracks in the matching queue.

sleap.nn.tracking.run_tracker(frames: List[LabeledFrame], tracker: BaseTracker) → List[LabeledFrame]

Run a tracker on a set of labeled frames.

Parameters:

• frames – A list of labeled frames with instances.

• tracker – An initialized Tracker.

Returns:

The input frames with the new tracks assigned. If the frames already had tracks, they will be cleared if the tracker has been re-initialized.