mot

sleap_nn.tracking.mot

Modern tracking-by-detection MOT trackers for animal pose and mask detections.

Native implementations (numpy/scipy) of SORT, DeepSORT, StrongSORT, ByteTrack, BoT-SORT, OC-SORT and Deep OC-SORT, adapted to sleap-io pose instances and segmentation masks, plus per-video self-calibration of appearance (re-ID) thresholds and offline identity linking (long-term object memory, closed-set re-identification). See the tracking guide for usage.

Modules:

Name	Description
appearance	Per-track appearance memory for the MOT trackers.
association	Similarity matrices, gates and assignment helpers for the MOT trackers.
base	Shared machinery of the MOT trackers.
bytetrack	ByteTrack (Zhang et al., 2022) and BoT-SORT (Aharon et al., 2022).
calibration	Per-video self-calibration for the MOT trackers.
deepsort	DeepSORT (Wojke et al., 2017) and StrongSORT (Du et al., 2023).
detections	Per-frame detection arrays for the MOT trackers.
kalman	Kalman filters for the MOT trackers (numpy only).
linking	Identity linking: object-level long-term memory for tracked videos.
ocsort	OC-SORT (Cao et al., 2023) and Deep OC-SORT (Maggiolino et al., 2023).
params	Resolved parameters and presets for the modern MOT trackers.
sort	SORT (Bewley et al., 2016) for animal pose/mask detections.
track	Track state for the MOT trackers.

Classes:

Name	Description
AssociationResult	What a tracker's :meth:BaseMOTTracker._associate decided for one frame.
BaseMOTTracker	Base class of the MOT trackers (see module docstring for the contract).
BoTSORTTracker	BoT-SORT tracker (see :mod:sleap_nn.tracking.mot.bytetrack).
ByteTracker	ByteTrack tracker (see :mod:sleap_nn.tracking.mot.bytetrack).
DeepOCSORTTracker	Deep OC-SORT tracker (see :mod:sleap_nn.tracking.mot.ocsort).
DeepSORTTracker	DeepSORT tracker (see :mod:sleap_nn.tracking.mot.deepsort).
MOTParams	Resolved parameters for one MOT tracker run.
OCSORTTracker	OC-SORT tracker (see :mod:sleap_nn.tracking.mot.ocsort).
SORTTracker	SORT tracker (see :mod:sleap_nn.tracking.mot.sort).
StrongSORTTracker	StrongSORT tracker (see :mod:sleap_nn.tracking.mot.deepsort).

Functions:

Name	Description
build_mot_tracker	Instantiate the tracker named by params.tracker.

AssociationResult

What a tracker's :meth:BaseMOTTracker._associate decided for one frame.

Attributes:

Name	Type	Description
matches	List[Tuple[MOTTrack, int, float]]	(track, det_index, geometric_similarity) per matched pair. A LOST track in a match is re-activated (same id). The similarity must be the pair's GEOMETRIC similarity (_geometric_similarity or _last_observation_similarity), not a fused / momentum / appearance score: it becomes the detection's tracking_score. Matches to invalid detections or repeated tracks/detections are ignored.
no_spawn	Set[int]	Detection indices that must not start a track even if unmatched (e.g. low-score detections consumed by a second stage).
removed	List[MOTTrack]	Tracks the algorithm removed this frame (e.g. ByteTrack's unmatched tentative tracks, duplicate removal). A removed track's detection matched this frame stays untracked.

Source code in sleap_nn/tracking/mot/base.py

@attrs.define
class AssociationResult:
    """What a tracker's :meth:`BaseMOTTracker._associate` decided for one frame.

    Attributes:
        matches: ``(track, det_index, geometric_similarity)`` per matched pair.
            A LOST track in a match is re-activated (same id). The similarity
            must be the pair's GEOMETRIC similarity (``_geometric_similarity`` or
            ``_last_observation_similarity``), not a fused / momentum /
            appearance score: it becomes the detection's ``tracking_score``.
            Matches to invalid detections or repeated tracks/detections are
            ignored.
        no_spawn: Detection indices that must not start a track even if
            unmatched (e.g. low-score detections consumed by a second stage).
        removed: Tracks the algorithm removed this frame (e.g. ByteTrack's
            unmatched tentative tracks, duplicate removal). A removed track's
            detection matched this frame stays untracked.
    """

    matches: List[Tuple[MOTTrack, int, float]] = attrs.field(factory=list)
    no_spawn: Set[int] = attrs.field(factory=set)
    removed: List[MOTTrack] = attrs.field(factory=list)

BaseMOTTracker

Base class of the MOT trackers (see module docstring for the contract).

Class attributes (set by subclasses): name: Tracker name (registry key). default_kalman: Filter used when params.kalman == "auto". uses_appearance: The algorithm associates by appearance vectors. requires_vectors: Appearance is the algorithm's primary cue. apply_tracking (and the CLI, before inference) refuses a run whose tracked carrier holds no usable vector (:data:~sleap_nn.tracking.mot.VECTOR_REQUIRED_TRACKERS); the tracker itself never raises on missing vectors -- detections without one are matched on geometry alone. tentative_dies_on_miss: A tentative track is removed at its first miss (DeepSORT/ByteTrack/BoT-SORT) instead of coasting until track_buffer (SORT/OC-SORT/Deep OC-SORT). hits_start: hits/hit_streak of a newly spawned track (SORT-family reference: 0, so min_hits=3 needs 4 consecutive detections). confirm_on_streak: Confirm on hit_streak >= min_hits (SORT family) instead of hits >= min_hits. use_oru: Freeze the state at the first miss and apply OC-SORT's observation-centric re-update when re-matched. keeps_observation_history: Maintain OC-SORT's observation history on every match: observations (the last 2 * delta_t + 1) and the velocity direction its OCM term reads (OC-SORT family). Off for the other trackers, which never read them; last_observation is kept by every tracker. lost_zero_velocity: State indices zeroed before predicting a track that is not CONFIRMED (ByteTrack: (7,); BoT-SORT: (6, 7)). Precisely: a track that was not matched at the previous step (the references' "not Tracked"); a live tentative track of those families always was.

Attributes:

Name	Type	Description
params		Resolved :class:MOTParams.
carrier		"pose" or "mask".
n_nodes		Skeleton size (pose), None if unknown.
kf		The Kalman filter in use.
kalman_kind		Its name ("xywh", "xyah" or "xysr").
calibration	Optional[VideoCalibration]	Per-video :class:VideoCalibration (None before prepare).
tracks	List[MOTTrack]	Live (not removed) tracks.
n_removed		Number of tracks removed so far. Removed tracks are dropped (identity linking re-reads the processed frames, not the tracks), so memory does not grow with track turnover.
track_name_offset	int	Added to public ids when naming sio.Tracks.
resolved	Dict[str, Any]	JSON-safe record of what ran (resolved params, time base, calibration, appearance decisions, linking result) for provenance.

Methods:

Name	Description
__init__	Create a tracker for one video.
finalize	End-of-video pass: identity linking when enabled. Idempotent.
prepare	Optional per-video pre-pass over every frame that will be tracked.
track	Track one frame (see the module contract). Returns every input object.

Source code in sleap_nn/tracking/mot/base.py

class BaseMOTTracker:
    """Base class of the MOT trackers (see module docstring for the contract).

    Class attributes (set by subclasses):
        name: Tracker name (registry key).
        default_kalman: Filter used when ``params.kalman == "auto"``.
        uses_appearance: The algorithm associates by appearance vectors.
        requires_vectors: Appearance is the algorithm's primary cue.
            ``apply_tracking`` (and the CLI, before inference) refuses a run whose
            tracked carrier holds no usable vector
            (:data:`~sleap_nn.tracking.mot.VECTOR_REQUIRED_TRACKERS`); the tracker
            itself never raises on missing vectors -- detections without one are
            matched on geometry alone.
        tentative_dies_on_miss: A tentative track is removed at its first miss
            (DeepSORT/ByteTrack/BoT-SORT) instead of coasting until
            ``track_buffer`` (SORT/OC-SORT/Deep OC-SORT).
        hits_start: ``hits``/``hit_streak`` of a newly spawned track (SORT-family
            reference: 0, so ``min_hits=3`` needs 4 consecutive detections).
        confirm_on_streak: Confirm on ``hit_streak >= min_hits`` (SORT family)
            instead of ``hits >= min_hits``.
        use_oru: Freeze the state at the first miss and apply OC-SORT's
            observation-centric re-update when re-matched.
        keeps_observation_history: Maintain OC-SORT's observation history on
            every match: ``observations`` (the last ``2 * delta_t + 1``) and the
            ``velocity`` direction its OCM term reads (OC-SORT family). Off for
            the other trackers, which never read them; ``last_observation`` is
            kept by every tracker.
        lost_zero_velocity: State indices zeroed before predicting a track that is
            not ``CONFIRMED`` (ByteTrack: ``(7,)``; BoT-SORT: ``(6, 7)``). Precisely:
            a track that was not matched at the previous step (the references'
            "not Tracked"); a live tentative track of those families always was.

    Attributes:
        params: Resolved :class:`MOTParams`.
        carrier: ``"pose"`` or ``"mask"``.
        n_nodes: Skeleton size (pose), ``None`` if unknown.
        kf: The Kalman filter in use.
        kalman_kind: Its name (``"xywh"``, ``"xyah"`` or ``"xysr"``).
        calibration: Per-video :class:`VideoCalibration` (``None`` before
            ``prepare``).
        tracks: Live (not removed) tracks.
        n_removed: Number of tracks removed so far. Removed tracks are dropped
            (identity linking re-reads the processed frames, not the tracks), so
            memory does not grow with track turnover.
        track_name_offset: Added to public ids when naming ``sio.Track``s.
        resolved: JSON-safe record of what ran (resolved params, time base,
            calibration, appearance decisions, linking result) for provenance.
    """

    name: ClassVar[str] = ""
    default_kalman: ClassVar[str] = "xywh"
    uses_appearance: ClassVar[bool] = False
    requires_vectors: ClassVar[bool] = False
    tentative_dies_on_miss: ClassVar[bool] = True
    hits_start: ClassVar[int] = 1
    confirm_on_streak: ClassVar[bool] = False
    use_oru: ClassVar[bool] = False
    keeps_observation_history: ClassVar[bool] = False
    lost_zero_velocity: ClassVar[Tuple[int, ...]] = ()

    def __init__(
        self,
        params: MOTParams,
        carrier: str = "pose",
        n_nodes: Optional[int] = None,
    ):
        """Create a tracker for one video.

        Args:
            params: Resolved parameters (see :meth:`MOTParams.for_tracker`).
            carrier: ``"pose"`` or ``"mask"``.
            n_nodes: Skeleton size for pose detections (``None`` if unknown; then
                taken from the detections).

        Raises:
            ValueError: On an unknown carrier, or parameters resolved for another
                tracker.
        """
        if carrier not in CARRIERS:
            raise ValueError(f"carrier must be one of {CARRIERS}, got {carrier!r}.")
        if self.name and params.tracker != self.name:
            raise ValueError(
                f"{type(self).__name__} needs parameters resolved for "
                f"{self.name!r}, got parameters for {params.tracker!r}."
            )
        self.params = params
        self.carrier = carrier
        self.n_nodes = None if n_nodes is None else int(n_nodes)
        self.kalman_kind = (
            params.kalman if params.kalman != "auto" else self.default_kalman
        )
        self.kf = make_kalman_filter(self.kalman_kind, nsa=params.nsa)

        self.calibration: Optional[VideoCalibration] = None
        self.tracks: List[MOTTrack] = []
        self.n_removed = 0
        self.track_name_offset = 0
        self._track_objects: Dict[int, sio.Track] = {}

        self._next_tid = 0
        self._next_public_id = 0
        # Step index of the latest processed frame (-1: nothing processed yet).
        self._step = -1
        self._last_frame_idx: Optional[int] = None
        # Frame-index delta to the previous processed frame (None at the first).
        self._last_delta: Optional[int] = None
        # Until `prepare` (or without it), the stride and time base are inferred
        # online from the positive frame deltas seen so far (`_infer_stride`).
        self._time_base = resolve_time_base(params.time_base, True)
        self._frame_step = 1
        self._delta_counts: Dict[int, int] = {}

        self._similarity_kind: Optional[str] = None
        if carrier == "mask" or self.n_nodes is not None:
            self._similarity_kind = self._resolve_similarity(self.n_nodes)

        # Running body-size evidence without calibration: raw-box diagonals,
        # else (extent-less detections) same-frame spacings and per-step lengths
        # of unambiguous frame-to-frame links, with the previous frame for them.
        self._diagonals: deque = deque(maxlen=BODY_SIZE_HISTORY)
        self._spacings: deque = deque(maxlen=BODY_SIZE_HISTORY)
        self._step_lengths: deque = deque(maxlen=BODY_SIZE_HISTORY)
        self._prev_dets: Optional[FrameDetections] = None
        self._embedding_dim: Optional[int] = None
        self._vectors_seen = False
        # Divisor of every detection score (`normalize_scores`; set by `prepare`).
        self._score_normalizer: Optional[float] = None
        self._n_valid_seen = 0
        self._warned: Set[str] = set()

        # `prepare` cache: frame_idx -> FrameDetections (validated by identity);
        # each entry is dropped when its frame is tracked.
        self._cache: Dict[int, FrameDetections] = {}
        # Every processed frame's objects, for identity linking in `finalize`
        # (kept only when linking runs).
        self._seen_frames: List[Tuple[int, list]] = []
        self._n_frames_seen = 0
        self._finalized = False
        self.resolved: Dict[str, Any] = {}
        self._refresh_resolved()

    # ------------------------------------------------------------------ public
    track_name_offset: int = 0
    _track_objects: Dict[int, sio.Track]

    @property
    def next_track_name_offset(self) -> int:
        """``track_name_offset`` for the NEXT video's tracker.

        Every name this tracker minted (``track_{public_id + offset}``) is below
        it, so per-video trackers chained with it never reuse a name.
        """
        return int(self.track_name_offset) + int(self._next_public_id)

    def prepare(self, frames: Sequence[Tuple[int, Sequence[Any]]]) -> None:
        """Optional per-video pre-pass over every frame that will be tracked.

        Extracts (and caches, keyed by frame index and object identities) each
        frame's :class:`FrameDetections`, calibrates the video
        (:func:`~sleap_nn.tracking.mot.calibration.calibrate`), resolves the
        time base, the body size (pseudo-box size of degenerate geometry and the
        ``centroid`` similarity scale) and, with ``normalize_scores``, the score
        normalizer applied to every detection from then on; records everything
        in ``resolved``. ``prepare([])`` is a no-op.

        Args:
            frames: ``(frame_idx, objects)`` per frame of ONE video, in
                increasing frame order, holding exactly the objects ``track()``
                will receive. Repeated or backwards frame indices are logged
                (WARNING): the calibration would measure motion across them.
        """
        frames = [(int(fi), list(objs)) for fi, objs in frames]
        if not frames:
            return
        p = self.params
        frame_idx = np.array([fi for fi, _ in frames], dtype=np.int64)
        n_bad = int(np.count_nonzero(np.diff(frame_idx) <= 0))
        if n_bad:
            logger.warning(
                f"{self.name}: prepare() got {n_bad} repeated or backwards frame "
                "index step(s). A MOT tracker tracks ONE video in increasing frame "
                "order: build one tracker per video and sort its frames (as "
                "apply_tracking does). The calibration measures motion across "
                "these jumps and tracks may continue across them."
            )

        # First pass: raw scores (the calibration measures them) and geometry
        # without a pseudo box (it measures the body size on non-degenerate
        # detections only).
        self._score_normalizer = None
        extracted = [
            self._extract(objs, fi, pseudo_box_size=None) for fi, objs in frames
        ]
        if self._similarity_kind is None and self.carrier == "pose":
            n_nodes = [d.n_nodes for d in extracted if len(d) > 0]
            if n_nodes:
                self._similarity_kind = self._resolve_similarity(max(n_nodes))

        # The g = 1 trials (gallery_cos_gate) use the gallery length the
        # tracker's memory actually keeps (DeepSORT: gallery_budget); the
        # appearance part is skipped when neither association nor linking
        # reads it.
        cal = calibrate(
            extracted,
            track_buffer=p.track_buffer,
            gallery_size=self._make_appearance_memory().gallery_size,
            ema_alpha=p.ema_alpha,
            time_base=p.time_base,
            body_size=p.body_size,
            appearance=self.uses_appearance or p.link,
        )
        self.calibration = cal

        # Scores: normalized per video, or a warning when the raw scores of
        # most detections cannot start a track.
        q95 = cal.score_quantiles[-1] if cal.score_quantiles else None
        if p.normalize_scores:
            self._score_normalizer = score_normalizer(cal)
        if self._score_normalizer is not None and self._score_normalizer != 1.0:
            logger.info(
                f"{self.name}: detection scores divided by the video's score q95 "
                f"({self._score_normalizer:.3f}) and clipped to [0, 1] "
                "(normalize_scores)."
            )
        self._check_score_cuts(q95)

        # Degenerate detections (single node, sub-pixel extent) get the body
        # size as their box; every padded box is floored at MIN_BOX_FRACTION of
        # it. Scores are normalized exactly once (`_extract` does it for the
        # re-extracted frames).
        body = self._body_size()
        for i, dets in enumerate(extracted):
            if self._has_degenerate(dets):
                extracted[i] = self._extract(
                    dets.sources, dets.frame_idx, pseudo_box_size=body
                )
            else:
                widen_boxes(extracted[i], MIN_BOX_FRACTION * body)
                normalize_scores(extracted[i], self._score_normalizer)
        self._cache = {dets.frame_idx: dets for dets in extracted}

        # Time base: the calibration's (it resolved `p.time_base` with the same
        # rule), and its stride in the frames time base.
        self._time_base = cal.time_base
        if p.time_base == "auto" and self._time_base == "steps":
            logger.warning(
                "The frames to track are irregularly spaced (the most common "
                f"frame step, {cal.frame_step}, covers less than half of the "
                "gaps): counting one step per processed frame (time_base="
                "'steps'). Kalman prediction, track_buffer and min_hits then "
                "count processed frames, not video frames."
            )
        self._frame_step = (
            max(1, int(cal.frame_step)) if self._time_base == "frames" else 1
        )

        if self.uses_appearance:
            self._unusable_vector_notice()
            if not self._vectors_seen:
                logger.warning(
                    f"{self.name}: the tracked detections carry no appearance "
                    "vectors; associating on geometry alone."
                )
            elif p.appearance_threshold == "auto" and self._appearance_gate() is None:
                logger.warning(
                    f"{self.name}: appearance is disabled for association by the "
                    f"video's calibration (status={cal.status!r}, assoc_rank1="
                    f"{cal.assoc_rank1}); associating on geometry alone."
                )
        self._refresh_resolved()

    def _score_cuts(self) -> List[Tuple[str, bool, str]]:
        """The score cuts of this tracker, for the ``prepare()`` score-gate warning.

        Each entry is ``(MOTParams field, strict, consequence)``: a detection
        passes the cut when its score is above (``strict``) or at least the
        field's value. Every tracker starts tracks only from detections scoring
        at least ``new_track_threshold``; a tracker with a stricter cut
        (ByteTrack's and OC-SORT's ``D_hi`` split, the ``track_low_threshold``
        below which SORT and DeepSORT ignore a detection) adds it.
        """
        return [
            (
                "new_track_threshold",
                False,
                "most detections can never start a track; lower "
                "--new_track_threshold"
                + (
                    ""
                    if self.params.normalize_scores
                    else " or use the animal preset (normalize_scores=True)"
                ),
            )
        ]

    def _check_score_cuts(self, q95: Optional[float]) -> None:
        """Warn once when 95% of the video's scores fail one of :meth:`_score_cuts`.

        The scores are the tracker's: divided by the per-video normalizer when
        ``normalize_scores`` is on. Only the first failing cut is reported.

        Args:
            q95: The calibration's 95th-percentile raw score (``None``/NaN: no
                check).
        """
        if q95 is None or not np.isfinite(q95):
            return
        q95 = float(q95)
        normalizer = self._score_normalizer
        normalized = normalizer is not None and np.isfinite(normalizer)
        if normalized and normalizer > 0:
            q95 = min(max(q95 / normalizer, 0.0), 1.0)
        for name, strict, consequence in self._score_cuts():
            value = float(getattr(self.params, name))
            if (q95 <= value) if strict else (q95 < value):
                logger.warning(
                    f"{self.name}: 95% of this video's detections score "
                    f"{'at or below' if strict else 'below'} {name}={value:g} "
                    f"(score q95 {q95:.3f}"
                    f"{', normalized' if self.params.normalize_scores else ''}): "
                    f"{consequence}."
                )
                return

    def _unusable_vector_notice(self) -> None:
        """Warn about valid detections whose vector is not usable.

        Such a detection holds an ``identity_embedding`` (so a check for its mere
        presence passes) but fails the shared usable-vector rule
        (``FrameDetections.has_embedding``: finite, non-zero, of the video's
        dimension), so every appearance decision treats it as vector-less.
        Called by :meth:`prepare` for every appearance tracker.
        """
        n_valid = n_usable = n_unusable = 0
        for dets in self._cache.values():
            valid = np.asarray(dets.valid, dtype=bool)
            usable = valid & np.asarray(dets.has_embedding, dtype=bool)
            n_valid += int(valid.sum())
            n_usable += int(usable.sum())
            n_unusable += sum(
                _has_any_vector(dets.sources[j])
                for j in np.flatnonzero(valid & ~usable)
            )
        if not n_unusable:
            return
        dim = f" ({self._embedding_dim}-D)" if self._embedding_dim is not None else ""
        logger.warning(
            f"{self.name}: {n_unusable} of {n_valid} detection(s) carry an "
            "appearance vector that cannot be used (non-finite, all zeros, or of "
            f"another dimension than the video's vectors{dim}); they count as "
            "having no vector"
            + (
                ", and no detection has a usable one: associating on geometry "
                "alone. Re-run the embedding model."
                if n_usable == 0
                else "."
            )
        )

    def track(
        self, untracked_instances: Sequence[Any], frame_idx: int, image: Any = None
    ) -> list:
        """Track one frame (see the module contract). Returns every input object."""
        objs = list(untracked_instances)
        frame_idx = int(frame_idx)
        for obj in objs:
            _set_label(obj, None, 0.0)
        self._n_frames_seen += 1
        if self.params.link:
            self._seen_frames.append((frame_idx, objs))

        dets = self._detections(objs, frame_idx)
        n_steps = self._advance(frame_idx)
        if self.calibration is None:
            # Before association, so this frame already uses the evidence.
            self._observe_uncalibrated(dets, n_steps)
        self._n_valid_seen += int(np.count_nonzero(dets.valid))

        if len(dets) == 0:
            # Nothing to match; the bookkeeping below still ages every track.
            result = AssociationResult()
        else:
            result = self._associate(dets)

        pool = list(self.tracks)
        removed_tids = {t.tid for t in result.removed}
        gate_ctx = self._gate_context(pool, dets, result)

        # Matches: Kalman update, counters, history, members, appearance.
        applied: List[Tuple[MOTTrack, int, float]] = []
        used_tids: Set[int] = set()
        used_dets: Set[int] = set()
        for track, det_idx, sim in result.matches:
            det_idx = int(det_idx)
            if (
                track.tid in used_tids
                or det_idx in used_dets
                or not track.is_live
                or not 0 <= det_idx < len(dets)
                or not dets.valid[det_idx]
            ):
                logger.debug(
                    f"Ignoring an invalid match (track {track.tid}, detection "
                    f"{det_idx}) at frame {frame_idx}."
                )
                continue
            used_tids.add(track.tid)
            used_dets.add(det_idx)
            sim = _clip_score(sim)
            self._apply_match(track, dets, det_idx, sim, gate_ctx)
            applied.append((track, det_idx, sim))

        # Unmatched live tracks: freeze for ORU, CONFIRMED -> LOST, tentative
        # tracks of the DeepSORT/ByteTrack families die.
        for track in pool:
            if track.tid in used_tids or track.tid in removed_tids:
                continue
            self._mark_missed(track)
        for track in result.removed:
            self._remove(track)

        # Spawn tentative tracks from unmatched, valid, confident detections.
        for det_idx in range(len(dets)):
            if (
                det_idx in used_dets
                or det_idx in result.no_spawn
                or not dets.valid[det_idx]
                or dets.scores[det_idx] < self.params.new_track_threshold
            ):
                continue
            self._spawn(dets, det_idx)

        # Confirm tentative tracks that reached `min_hits` (retro-labels members).
        for track in self.tracks:
            if track.state == TrackState.TENTATIVE and track.time_since_update == 0:
                count = track.hit_streak if self.confirm_on_streak else track.hits
                if count >= self.params.min_hits:
                    self._confirm(track)

        # Remove tracks unmatched for longer than the buffer (any state).
        for track in list(self.tracks):
            if track.time_since_update > self.params.track_buffer:
                self._remove(track)

        # Label this frame's detections of confirmed tracks.
        for track, det_idx, sim in applied:
            if track.state == TrackState.CONFIRMED and track.public_id is not None:
                _set_label(
                    dets.sources[det_idx], self._track_objects[track.public_id], sim
                )
        return list(objs)

    def finalize(self) -> None:
        """End-of-video pass: identity linking when enabled. Idempotent.

        Logs a WARNING when valid detections were tracked but no track was ever
        confirmed (every detection came back untracked).
        """
        if self._finalized:
            return
        self._finalized = True
        p = self.params
        if self._n_valid_seen and self._next_public_id == 0:
            logger.warning(
                f"{self.name}: {self._n_valid_seen} valid detection(s) over "
                f"{self._n_frames_seen} frame(s), but no track was ever "
                "confirmed: every detection is untracked. Check the score "
                f"thresholds (new_track_threshold={p.new_track_threshold}, "
                f"track_low_threshold={p.track_low_threshold}; normalize_scores="
                f"{p.normalize_scores}), min_hits={p.min_hits} and the frame "
                "order / time base"
                + (
                    "; without prepare() scores are not normalized and nothing is "
                    "calibrated."
                    if self.calibration is None
                    else "."
                )
            )
        if p.link:
            result = link_tracklets(
                self._seen_frames,
                carrier=self.carrier,
                calibration=self.calibration,
                max_tracks=p.max_tracks,
                track_buffer=p.track_buffer,
                swap_correction=p.link_swap_correction,
                spatial_gate_frames=p.link_spatial_gate_frames,
                box_padding=p.box_padding,
                high_threshold=p.track_high_threshold,
                score_mode=p.score_mode,
                score_normalizer=self._score_normalizer,
                time_base=p.time_base,
                body_size=p.body_size,
                track_objects=self._track_objects,
            )
            self.resolved["linking"] = result.as_dict()
        self._refresh_resolved()

    # --------------------------------------------------------- subclass hook
    def _associate(self, dets: FrameDetections) -> AssociationResult:
        """Match the live tracks (already predicted to this step) to ``dets``."""
        raise NotImplementedError

    # ------------------------------------------------------ helpers (shared)
    def _live(self, *states: TrackState) -> List[MOTTrack]:
        """Live tracks, optionally restricted to ``states``, in ``tid`` order."""
        if not states:
            return list(self.tracks)
        return [t for t in self.tracks if t.state in states]

    def _geometric_similarity(
        self, tracks: Sequence[MOTTrack], dets: FrameDetections
    ) -> np.ndarray:
        """``(n_tracks, n_dets)`` similarity of the tracks' PREDICTIONS to ``dets``.

        Kind from ``params.similarity`` (``"auto"`` resolved per carrier and
        skeleton): ``iou`` on padded boxes, ``oks`` against the predicted pose,
        ``centroid`` on box centers with scale ``body_size / 2``. Invalid
        detections score 0.
        """
        tracks = list(tracks)
        n, m = len(tracks), len(dets)
        if n == 0 or m == 0:
            return np.zeros((n, m), dtype=np.float64)
        means = np.stack([t.mean for t in tracks])
        kind = self._similarity(dets)
        if kind == "oks" and dets.keypoints is not None:
            pred_kp, areas = self._predicted_poses(tracks, means)
            sim = self._oks(dets.keypoints, pred_kp, areas)
        elif kind == "centroid":
            centers = np.asarray(means[:, :2], dtype=np.float64)
            sim = centroid_similarity(centers, dets.centroids, 0.5 * self._body_size())
        else:
            sim = iou_matrix(self.kf.xyxy_from_mean(means), dets.boxes)
        return self._finish_similarity(sim, dets)

    def _last_observation_similarity(
        self, tracks: Sequence[MOTTrack], dets: FrameDetections
    ) -> np.ndarray:
        """Similarity of each track's LAST OBSERVATION to ``dets`` (OC-SORT OCR).

        Same kinds and conventions as :meth:`_geometric_similarity`. A track
        never updated since its spawn has no observation (OC-SORT reference: the
        spawn box is not an observation) and scores 0.
        """
        tracks = list(tracks)
        n, m = len(tracks), len(dets)
        if n == 0 or m == 0:
            return np.zeros((n, m), dtype=np.float64)
        observed = np.array([t.last_observation is not None for t in tracks])
        boxes = np.full((n, 4), np.nan, dtype=np.float64)
        for i, t in enumerate(tracks):
            if observed[i]:
                boxes[i] = t.last_observation
        kind = self._similarity(dets)
        if kind == "oks" and dets.keypoints is not None:
            kp = self._stack_keypoints(tracks, dets.keypoints.shape[1])
            kp[~observed] = np.nan
            areas = _box_area(np.stack([t.last_raw_box for t in tracks]))
            sim = self._oks(dets.keypoints, kp, areas)
        elif kind == "centroid":
            centers = 0.5 * (boxes[:, :2] + boxes[:, 2:])
            sim = centroid_similarity(centers, dets.centroids, 0.5 * self._body_size())
        else:
            sim = iou_matrix(boxes, dets.boxes)
        return self._finish_similarity(sim, dets)

    def _appearance_cos(
        self, tracks: Sequence[MOTTrack], dets: FrameDetections, statistic: str = "ema"
    ) -> np.ndarray:
        """``(n_tracks, n_dets)`` appearance cosines; NaN where unavailable.

        ``statistic`` is ``"ema"`` (cosine to the track's EMA) or
        ``"gallery_max"`` (max over its gallery, EMA if empty). All NaN when
        appearance is disabled for association (see :meth:`_appearance_gate`).
        """
        tracks = list(tracks)
        out = np.full((len(tracks), len(dets)), np.nan, dtype=np.float64)
        if (
            out.size == 0
            or dets.embeddings is None
            or self._appearance_gate(statistic) is None
        ):
            return out
        if statistic == "ema":
            return self._ema_cos(tracks, dets)
        q_idx = np.flatnonzero(dets.has_embedding)
        # Normalized once for every track (the memory would redo it per track).
        rows, ok = unit_rows(np.asarray(dets.embeddings[q_idx], dtype=np.float64))
        for i, t in enumerate(tracks):
            mem = t.appearance
            if mem is None or mem.is_empty:
                continue
            ema = np.asarray(mem.ema).reshape(-1)
            if ema.shape[0] != rows.shape[1]:
                continue  # A vector of another model's dimension: no evidence.
            out[i, q_idx] = mem.gallery_max_cos_unit(rows, ok)
        return out

    def _appearance_gate(self, statistic: str = "ema") -> Optional[float]:
        """Minimum cosine for appearance to support a match, or ``None`` if disabled.

        ``appearance_threshold="auto"``: the calibrated gate for ``statistic``
        when ``calibration.appearance_for_association``, else ``None``. A float:
        that value (when the video has vectors).
        """
        if statistic not in APPEARANCE_STATISTICS:
            raise ValueError(
                f"statistic must be one of {APPEARANCE_STATISTICS}, got {statistic!r}."
            )
        if not self.uses_appearance or not self._vectors_seen:
            return None
        threshold = self.params.appearance_threshold
        if threshold != "auto":
            return float(threshold)
        cal = self.calibration
        if cal is None or not cal.appearance_for_association:
            return None
        gate = cal.ema_cos_gate if statistic == "ema" else cal.gallery_cos_gate
        if gate is None or not np.isfinite(gate):
            return None
        return float(gate)

    def _appearance_margin(self) -> float:
        """Minimum row/column cosine margin for appearance to decide a pair.

        Calibrated ``m_assoc`` when available, else 0.
        """
        cal = self.calibration
        if cal is None or cal.m_assoc is None or not np.isfinite(cal.m_assoc):
            return 0.0
        return float(cal.m_assoc)

    # --------------------------------------------------------- private: time
    def _advance(self, frame_idx: int) -> int:
        """Advance the clock to ``frame_idx`` and predict every live track.

        The first processed frame is step 0. Later frames advance by ``n`` steps
        (``1`` in the ``"steps"`` time base; :func:`steps_between` of the frame
        delta and the stride in the ``"frames"`` time base) and the gap's
        intermediate steps are treated as empty frames (see the module
        docstring). Uncalibrated, the stride and time base are first updated
        with this delta (:meth:`_infer_stride`). A repeated or backwards frame
        index counts one step and logs a WARNING once.

        Returns:
            The number of steps advanced (0 at the first processed frame).
        """
        if self._step < 0:
            self._step = 0
            self._last_frame_idx = frame_idx
            self._last_delta = None
            return 0
        delta = int(frame_idx - self._last_frame_idx)
        self._last_delta = delta
        if delta <= 0:
            self._warn_once(
                "order",
                f"{self.name}: frame {frame_idx} does not come after frame "
                f"{self._last_frame_idx} (repeated or backwards frame index). A MOT "
                "tracker tracks ONE video in increasing frame order: build one "
                "tracker per video and sort its frames (as apply_tracking does). "
                "Counting it as one step; live tracks continue across the jump.",
            )
        elif self.calibration is None:
            self._infer_stride(delta)
        n_steps = 1
        if self._time_base == "frames":
            n_steps = steps_between(delta, self._frame_step)
        first_step = self._step + 1
        self._step += n_steps
        self._last_frame_idx = frame_idx
        if not self.tracks:
            return n_steps

        # Tracks that the gap's empty steps would already have removed.
        for track in list(self.tracks):
            expired = track.time_since_update + n_steps - 1 > self.params.track_buffer
            died = (
                n_steps > 1
                and self.tentative_dies_on_miss
                and track.state == TrackState.TENTATIVE
            )
            if expired or died:
                self._remove(track)
        tracks = self.tracks
        if not tracks:
            return n_steps

        means = np.stack([t.mean for t in tracks])
        covs = np.stack([t.cov for t in tracks])
        tsu = np.array([t.time_since_update for t in tracks])
        zero_idx = list(self.lost_zero_velocity)
        for k in range(n_steps):
            if zero_idx:
                # Not matched at the previous step (the references' "not Tracked").
                missed = tsu + k > 0
                if missed.any():
                    means[np.ix_(missed, zero_idx)] = 0.0
            means, covs = self.kf.multi_predict(means, covs)
            if k == 0 and n_steps > 1 and self.use_oru:
                # The gap's first empty step is these tracks' first miss.
                for i, track in enumerate(tracks):
                    if (
                        tsu[i] == 0
                        and track.frozen is None
                        and track.last_observation is not None
                    ):
                        track.frozen = (means[i].copy(), covs[i].copy(), first_step)

        for i, track in enumerate(tracks):
            track.mean = means[i]
            track.cov = covs[i]
            track.time_since_update += n_steps
            track.age += n_steps
            if track.time_since_update > 1:
                # Missed at least one step: the streak restarts (SORT semantics).
                track.hit_streak = 0
            if n_steps > 1 and track.state == TrackState.CONFIRMED:
                track.state = TrackState.LOST
        return n_steps

    def _infer_stride(self, delta: int) -> None:
        """Uncalibrated: update the stride and time base with a positive delta.

        Same rule as the calibration over the deltas seen so far
        (:func:`stride_from_counts`, :func:`resolve_time_base`), applied BEFORE
        this frame's steps are counted, so a strided stream counts one step per
        processed frame from its first gap on, a dense stream with dropped frames
        keeps stride 1 (genuine gaps still count) and, under ``"auto"``,
        irregular input falls back to one step per processed frame (WARNING,
        once). Recorded in ``resolved``.
        """
        self._delta_counts[delta] = self._delta_counts.get(delta, 0) + 1
        frame_step, regular = stride_from_counts(self._delta_counts)
        time_base = resolve_time_base(self.params.time_base, regular)
        n_deltas = sum(self._delta_counts.values())
        if (
            self.params.time_base == "auto"
            and time_base == "steps"
            and n_deltas >= MIN_DELTAS_TO_WARN
        ):
            self._warn_once(
                "irregular",
                f"{self.name}: the frames tracked so far are irregularly spaced "
                f"(the most common frame step, {frame_step}, covers less than half "
                "of the gaps): counting one step per processed frame (time_base="
                "'steps') while that holds. prepare() decides this once over the "
                "whole video.",
            )
        self._time_base = time_base
        self._frame_step = frame_step if time_base == "frames" else 1
        self.resolved["time_base"] = self._time_base
        self.resolved["frame_step"] = int(self._frame_step)

    def _warn_once(self, key: str, message: str) -> None:
        """Log ``message`` as a WARNING the first time ``key`` is seen."""
        if key not in self._warned:
            self._warned.add(key)
            logger.warning(message)

    # ---------------------------------------------------- private: lifecycle
    def _apply_match(
        self,
        track: MOTTrack,
        dets: FrameDetections,
        det_idx: int,
        sim: float,
        gate_ctx: Optional[dict],
    ) -> None:
        """Update ``track`` with detection ``det_idx`` (already validated)."""
        box = np.array(dets.boxes[det_idx], dtype=np.float64)
        score = float(dets.scores[det_idx])
        mean, cov = track.mean, track.cov
        if (
            self.use_oru
            and track.frozen is not None
            and track.last_observation is not None
        ):
            frozen_mean, frozen_cov, frozen_step = track.frozen
            mean, cov = oru_replay(
                self.kf,
                frozen_mean,
                frozen_cov,
                track.last_observation,
                box,
                n_gap=self._step - track.step,
                conf=score,
                frozen_steps=frozen_step - track.step,
            )
        track.frozen = None
        track.mean, track.cov = self.kf.update(mean, cov, box, conf=score)

        track.hits += 1
        track.hit_streak += 1
        track.time_since_update = 0
        track.step = self._step
        track.last_frame = dets.frame_idx
        track.last_box = box
        track.last_raw_box = np.array(dets.raw_boxes[det_idx], dtype=np.float64)
        if dets.keypoints is not None:
            track.last_keypoints = np.array(dets.keypoints[det_idx], dtype=np.float64)
        track.last_score = score
        track.kf_center_at_update = np.array(self.kf.center(track.mean), dtype=float)
        if track.state == TrackState.LOST:
            track.state = TrackState.CONFIRMED

        if self.keeps_observation_history:
            # OC-SORT observation history and velocity direction: from the
            # observation `delta_t` steps back (the oldest within that window),
            # else the latest observation, to this one.
            if track.last_observation is not None:
                previous = None
                for dt in range(self.params.delta_t, 0, -1):
                    previous = track.observations.get(self._step - dt)
                    if previous is not None:
                        break
                if previous is None:
                    previous = track.last_observation
                track.velocity = np.asarray(speed_direction(previous, box), dtype=float)
            track.observations[self._step] = box
            while len(track.observations) > 2 * self.params.delta_t + 1:
                del track.observations[min(track.observations)]
        track.last_observation = box

        if track.public_id is None:
            # Labeled when the track confirms; a confirmed track's detections
            # are labeled as they are matched (`track`), so none are kept.
            track.members.append((dets.frame_idx, dets.sources[det_idx], sim))
        self._update_appearance(track, dets, det_idx, gate_ctx)

    def _mark_missed(self, track: MOTTrack) -> None:
        """Handle a live track left unmatched at this step."""
        if (
            self.use_oru
            and track.frozen is None
            and track.last_observation is not None
            and track.time_since_update > 0
        ):
            # First miss since the latest observation: save the predicted state.
            track.frozen = (track.mean.copy(), track.cov.copy(), self._step)
        if track.state == TrackState.CONFIRMED:
            track.state = TrackState.LOST
        elif track.state == TrackState.TENTATIVE and self.tentative_dies_on_miss:
            self._remove(track)

    def _spawn(self, dets: FrameDetections, det_idx: int) -> MOTTrack:
        """Start a tentative track from detection ``det_idx``."""
        box = np.array(dets.boxes[det_idx], dtype=np.float64)
        mean, cov = self.kf.initiate(box)
        track = MOTTrack(
            tid=self._next_tid,
            state=TrackState.TENTATIVE,
            mean=mean,
            cov=cov,
            start_frame=dets.frame_idx,
            last_frame=dets.frame_idx,
            step=self._step,
            last_box=box,
            last_raw_box=np.array(dets.raw_boxes[det_idx], dtype=np.float64),
            kf_center_at_update=np.array(self.kf.center(mean), dtype=np.float64),
            last_keypoints=(
                None
                if dets.keypoints is None
                else np.array(dets.keypoints[det_idx], dtype=np.float64)
            ),
            last_score=float(dets.scores[det_idx]),
            hits=self.hits_start,
            hit_streak=self.hits_start,
            age=0,
            time_since_update=0,
        )
        self._next_tid += 1
        # The spawn detection is a member (labeled if the track confirms) but, as
        # in OC-SORT, not an "observation" for velocity / OCR / ORU.
        track.members.append((dets.frame_idx, dets.sources[det_idx], 1.0))
        if self.uses_appearance:
            track.appearance = self._make_appearance_memory()
            if dets.embeddings is not None and dets.has_embedding[det_idx]:
                track.appearance.seed(np.asarray(dets.embeddings[det_idx]))
        self.tracks.append(track)
        return track

    def _confirm(self, track: MOTTrack) -> None:
        """Confirm ``track``: mint its public id and label all its detections."""
        track.state = TrackState.CONFIRMED
        track.public_id = self._next_public_id
        self._next_public_id += 1
        track_obj = sio.Track(name=f"track_{track.public_id + self.track_name_offset}")
        self._track_objects[track.public_id] = track_obj
        for _, obj, sim in track.members:
            _set_label(obj, track_obj, sim)
        track.members.clear()

    def _remove(self, track: MOTTrack) -> None:
        """Mark ``track`` removed and drop it (counted in ``n_removed``)."""
        if track.state == TrackState.REMOVED:
            return
        track.state = TrackState.REMOVED
        self.tracks = [t for t in self.tracks if t is not track]
        self.n_removed += 1

    # --------------------------------------------------- private: appearance
    def _make_appearance_memory(self) -> AppearanceMemory:
        """A new track's appearance memory (subclasses may override)."""
        p = self.params
        return AppearanceMemory(
            ema_alpha=p.ema_alpha,
            gallery_size=p.gallery_size,
            adaptive=p.adaptive_ema,
            score_floor=p.track_high_threshold,
        )

    def _ema_cos(self, tracks: Sequence[MOTTrack], dets: FrameDetections) -> np.ndarray:
        """Cosine of each track's EMA to each detection vector (NaN if missing)."""
        n, m = len(tracks), len(dets)
        if n == 0 or m == 0 or dets.embeddings is None:
            return np.full((n, m), np.nan, dtype=np.float64)
        dim = dets.embeddings.shape[1]
        emas = np.zeros((n, dim), dtype=np.float64)
        ok = np.zeros(n, dtype=bool)
        for i, t in enumerate(tracks):
            mem = t.appearance
            if mem is None or mem.ema is None:
                continue
            ema = np.asarray(mem.ema, dtype=np.float64).reshape(-1)
            if ema.shape[0] == dim:
                emas[i] = ema
                ok[i] = True
        return cosine_matrix(emas, ok, dets.embeddings, dets.has_embedding)

    def _gate_context(
        self,
        pool: List[MOTTrack],
        dets: FrameDetections,
        result: AssociationResult,
    ) -> Optional[dict]:
        """Everything the appearance quality gate needs, computed BEFORE updates.

        ``None`` when no matched detection carries a vector (or the tracker does
        not use appearance).
        """
        if (
            not self.uses_appearance
            or dets.embeddings is None
            or not result.matches
            or not pool
        ):
            return None
        boxes = dets.boxes
        det_iou = iou_matrix(boxes, boxes)
        np.fill_diagonal(det_iou, 0.0)
        pred_boxes = self.kf.xyxy_from_mean(np.stack([t.mean for t in pool]))
        neighbor = np.array(
            [
                t.state == TrackState.CONFIRMED
                or (
                    t.state == TrackState.LOST
                    and t.time_since_update <= GATE_LOST_MAX_STEPS
                )
                for t in pool
            ]
        )
        return {
            "row": {t.tid: i for i, t in enumerate(pool)},
            "neighbor": neighbor,
            "det_iou": det_iou,
            "trk_iou": iou_matrix(pred_boxes, boxes),
            "geo": self._geometric_similarity(pool, dets),
            "cos": self._ema_cos(pool, dets),
        }

    def _update_appearance(
        self,
        track: MOTTrack,
        dets: FrameDetections,
        det_idx: int,
        gate_ctx: Optional[dict],
    ) -> None:
        """Quality-gated appearance update of a matched track."""
        if (
            not self.uses_appearance
            or dets.embeddings is None
            or not dets.has_embedding[det_idx]
        ):
            return
        vec = np.asarray(dets.embeddings[det_idx])
        if track.appearance is None:
            track.appearance = self._make_appearance_memory()
        if track.appearance.is_empty:
            track.appearance.seed(vec)
            return
        gated = False
        row = None if gate_ctx is None else gate_ctx["row"].get(track.tid)
        if row is not None:
            others = np.arange(len(gate_ctx["neighbor"])) != row
            iou_det = gate_ctx["det_iou"][det_idx]
            iou_trk = gate_ctx["trk_iou"][others & gate_ctx["neighbor"], det_idx]
            max_iou = float(max(iou_det.max(initial=0.0), iou_trk.max(initial=0.0)))
            cos = gate_ctx["cos"][:, det_idx]
            competitors = cos[others & (gate_ctx["geo"][:, det_idx] > 0)]
            competitors = competitors[np.isfinite(competitors)]
            own = cos[row]
            if not np.isfinite(own) or len(competitors) == 0:
                margin = float("inf")
            else:
                margin = float(own - competitors.max())
            gated = passes_quality_gate(
                float(dets.scores[det_idx]),
                int(dets.n_visible[det_idx]),
                int(dets.n_nodes),
                max_iou,
                margin,
                high_threshold=self.params.track_high_threshold,
                min_margin=self._appearance_margin(),
                gallery_len=len(track.appearance.gallery),
            )
        track.appearance.update(vec, score=float(dets.scores[det_idx]), gated=gated)

    # ----------------------------------------------------- private: geometry
    def _resolve_similarity(self, n_nodes: Optional[int]) -> str:
        """Resolve ``params.similarity`` for the carrier and skeleton size."""
        kind = self.params.similarity
        if self.carrier == "mask":
            if kind == "oks":
                logger.warning(
                    "similarity='oks' needs keypoints; using 'iou' on masks."
                )
                return "iou"
            return "iou" if kind == "auto" else kind
        if kind == "auto":
            return "centroid" if n_nodes == 1 else "iou"
        if kind == "oks" and n_nodes == 1:
            logger.warning(
                "similarity='oks' on a single-node skeleton is a centroid distance; "
                "using 'centroid'."
            )
            return "centroid"
        return kind

    def _similarity(self, dets: Optional[FrameDetections] = None) -> str:
        """The resolved similarity kind (resolving it from ``dets`` if needed)."""
        if self._similarity_kind is None and dets is not None and len(dets) > 0:
            self._similarity_kind = self._resolve_similarity(dets.n_nodes)
            self.resolved["similarity"] = self._similarity_kind
        if self._similarity_kind is None:
            # Unknown skeleton yet: the multi-node default.
            return "iou" if self.params.similarity == "auto" else self.params.similarity
        return self._similarity_kind

    def _finish_similarity(self, sim: np.ndarray, dets: FrameDetections) -> np.ndarray:
        """NaN-free, ``[0, 1]``-clipped similarity with invalid detections at 0."""
        sim = np.clip(np.nan_to_num(sim, nan=0.0, posinf=0.0, neginf=0.0), 0.0, 1.0)
        sim[:, ~np.asarray(dets.valid, dtype=bool)] = 0.0
        return sim

    def _stack_keypoints(self, tracks: Sequence[MOTTrack], n_nodes: int) -> np.ndarray:
        """``(n, K, 2)`` last observed keypoints (NaN where missing), padded to K."""
        k = max(
            [n_nodes]
            + [
                t.last_keypoints.shape[0]
                for t in tracks
                if t.last_keypoints is not None
            ]
        )
        out = np.full((len(tracks), k, 2), np.nan, dtype=np.float64)
        for i, t in enumerate(tracks):
            if t.last_keypoints is not None:
                kp = np.asarray(t.last_keypoints, dtype=np.float64).reshape(-1, 2)
                out[i, : kp.shape[0]] = kp
        return out

    def _predicted_poses(
        self, tracks: Sequence[MOTTrack], means: np.ndarray
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Predicted poses and their raw-extent areas for OKS.

        The pose is the last observed pose shifted by the Kalman center's motion
        since the last update (both centers are KF posteriors/predictions of the
        same reference point, the box center). The area is the last raw extent's
        scaled by the predicted / last-observed padded-box area ratio.
        """
        kp = self._stack_keypoints(tracks, 1)
        shift = np.asarray(means[:, :2], dtype=np.float64) - np.stack(
            [t.kf_center_at_update for t in tracks]
        )
        kp = kp + shift[:, None, :]
        last_area = _box_area(np.stack([t.last_box for t in tracks]))
        pred_area = _box_area(self.kf.xyxy_from_mean(means))
        ratio = np.ones(len(tracks), dtype=np.float64)
        np.divide(pred_area, last_area, out=ratio, where=last_area > 0)
        areas = _box_area(np.stack([t.last_raw_box for t in tracks])) * ratio
        return kp, areas

    def _oks(
        self, det_kp: np.ndarray, trk_kp: np.ndarray, areas: np.ndarray
    ) -> np.ndarray:
        """``(n_tracks, n_dets)`` OKS with the tracks' poses as the scale side."""
        k = max(det_kp.shape[1], trk_kp.shape[1])
        sim = oks_matrix(
            _pad_nodes(np.asarray(det_kp, dtype=np.float64), k),
            _pad_nodes(trk_kp, k),
            areas,
            self.params.oks_stddev,
        )
        return sim.T

    def _body_size(self) -> float:
        """Body size (px); see :meth:`_body_size_info`."""
        return self._body_size_info()[0]

    def _body_size_info(self) -> Tuple[float, str]:
        """Body size (px) and its source.

        ``params.body_size`` (``"override"``), else the calibration's, else
        (uncalibrated) calibration's fallback chain
        (:func:`~sleap_nn.tracking.mot.calibration.resolve_body_size`) on the
        frames seen so far: the median raw-box diagonal, else 8x the per-step
        p99 of unambiguous links (single-node detections), else the median
        same-frame spacing, else the chain's default (50 px).
        """
        if self.params.body_size is not None:
            return float(self.params.body_size), "override"
        cal = self.calibration
        if cal is not None and np.isfinite(cal.body_size) and cal.body_size > 0:
            return float(cal.body_size), str(cal.body_size_source)
        measured = (
            float(np.median(np.asarray(self._diagonals)))
            if self._diagonals
            else float("nan")
        )
        step_p99 = (
            float(np.percentile(np.asarray(self._step_lengths), 99))
            if self._step_lengths
            else float("nan")
        )
        spacing = (
            float(np.median(np.asarray(self._spacings)))
            if self._spacings
            else float("nan")
        )
        body, source = resolve_body_size(measured, step_p99, spacing)
        return float(body), source

    # -------------------------------------------------- private: detections
    def _extract(
        self, objs: Sequence[Any], frame_idx: int, pseudo_box_size: Optional[float]
    ) -> FrameDetections:
        """Extract one frame's detections with the tracker's settings."""
        p = self.params
        # Thin poses (nodes nearly collinear) would get sliver boxes whose IoU
        # collapses under tiny shifts; floor the padded box at a fraction of the
        # body size (only known once a body size is: calibrated or observed).
        min_box = (
            MIN_BOX_FRACTION * pseudo_box_size if pseudo_box_size is not None else None
        )
        dets = extract_detections(
            objs,
            frame_idx,
            carrier=self.carrier,
            box_padding=p.box_padding,
            pseudo_box_size=pseudo_box_size,
            score_mode=p.score_mode,
            embedding_dim=self._embedding_dim,
            min_box_size=min_box,
        )
        if dets.embeddings is not None:
            self._vectors_seen = True
            if self._embedding_dim is None:
                self._embedding_dim = int(dets.embeddings.shape[1])
        # Per-video score normalization (set by `prepare`; exactly once here).
        normalize_scores(dets, self._score_normalizer)
        return dets

    def _detections(self, objs: List[Any], frame_idx: int) -> FrameDetections:
        """This frame's detections: from the ``prepare`` cache, else extracted."""
        cached = self._cache.pop(frame_idx, None)
        if (
            cached is not None
            and len(cached.sources) == len(objs)
            and all(a is b for a, b in zip(cached.sources, objs))
        ):
            return cached
        return self._extract(objs, frame_idx, pseudo_box_size=self._body_size())

    def _observe_uncalibrated(self, dets: FrameDetections, n_steps: int) -> None:
        """Uncalibrated: add this frame's body-size evidence (see `_body_size_info`).

        Extents feed the running diagonal median. While no extent has been seen
        (single-node detections), the frame's same-frame spacings and the
        per-step lengths of its unambiguous links to the previous processed
        frame (:func:`~sleap_nn.tracking.mot.calibration.pair_geometry`; not
        across a repeated or backwards frame) are remembered instead.
        """
        self._remember_diagonals(dets)
        if self._diagonals or self.params.body_size is not None:
            self._prev_dets = None
            return
        prev = self._prev_dets
        if self._last_delta is not None and self._last_delta <= 0:
            prev = None
        max_dist = (
            0.5 * float(np.median(np.asarray(self._spacings)))
            if self._spacings
            else None
        )
        spacings, links = pair_geometry(prev, dets, max_dist=max_dist)
        self._spacings.extend(spacings.tolist())
        if n_steps > 0 and len(links):
            self._step_lengths.extend((links / n_steps).tolist())
        self._prev_dets = dets

    def _remember_diagonals(self, dets: FrameDetections) -> None:
        """Add the frame's non-degenerate extents to the running body size."""
        if len(dets) == 0:
            return
        if dets.keypoints is not None:
            # Keypoint extents (a pseudo box would feed the estimate back).
            kp = dets.keypoints[dets.valid & (dets.n_visible >= 2)]
            vis = np.isfinite(kp).all(axis=-1)
            lo = np.where(vis[..., None], kp, np.inf).min(axis=1, initial=np.inf)
            hi = np.where(vis[..., None], kp, -np.inf).max(axis=1, initial=-np.inf)
            wh = hi - lo
        else:
            raw = dets.raw_boxes[dets.valid]
            wh = raw[:, 2:] - raw[:, :2]
        keep = np.isfinite(wh).all(axis=1) & (wh >= 1.0).all(axis=1)
        self._diagonals.extend(np.hypot(wh[keep, 0], wh[keep, 1]).tolist())

    def _has_degenerate(self, dets: FrameDetections) -> bool:
        """Whether any valid detection got a fallback (degenerate) box."""
        if len(dets) == 0 or not dets.valid.any():
            return False
        raw = dets.raw_boxes[dets.valid]
        thin = ((raw[:, 2:] - raw[:, :2]) <= 1.0).any(axis=1)
        single = dets.keypoints is not None and (dets.n_visible[dets.valid] == 1).any()
        return bool(thin.any() or single)

    # ---------------------------------------------------- private: provenance
    def _appearance_summary(self) -> Dict[str, Any]:
        """JSON-safe record of the appearance decisions."""
        if not self.uses_appearance:
            return {"enabled": False, "reason": "the tracker does not use appearance"}
        threshold = self.params.appearance_threshold
        ema_gate = self._appearance_gate("ema")
        gallery_gate = self._appearance_gate("gallery_max")
        cal = self.calibration
        if not self._vectors_seen:
            reason = "no appearance vectors on the tracked detections"
        elif threshold != "auto":
            reason = "explicit appearance_threshold"
        elif cal is None:
            reason = "uncalibrated (no prepare): 'auto' disables appearance"
        elif not cal.appearance_for_association:
            reason = (
                f"calibration status={cal.status!r}, assoc_rank1={cal.assoc_rank1}: "
                "appearance not discriminative enough for association"
            )
        else:
            reason = "calibrated gates"
        return {
            "enabled": ema_gate is not None or gallery_gate is not None,
            "threshold": threshold,
            "ema_cos_gate": ema_gate,
            "gallery_cos_gate": gallery_gate,
            "margin": self._appearance_margin(),
            "reason": reason,
        }

    def _refresh_resolved(self) -> None:
        """Rebuild ``resolved`` (keeps the linking result once recorded)."""
        linking = self.resolved.get("linking") if self.resolved else None
        body, body_source = self._body_size_info()
        self.resolved = {
            "tracker": self.name,
            "params": self.params.as_dict(),
            "kalman": self.kalman_kind,
            "time_base": self._time_base,
            "frame_step": int(self._frame_step),
            "similarity": self._similarity(),
            "body_size": float(body),
            "body_size_source": body_source,
            "score_normalizer": (
                None
                if self._score_normalizer is None
                else float(self._score_normalizer)
            ),
            "calibration": (
                None if self.calibration is None else self.calibration.as_dict()
            ),
            "appearance": self._appearance_summary(),
        }
        if linking is not None:
            self.resolved["linking"] = linking

next_track_name_offset property

track_name_offset for the NEXT video's tracker.

Every name this tracker minted (track_{public_id + offset}) is below it, so per-video trackers chained with it never reuse a name.

__init__(params, carrier='pose', n_nodes=None)

Create a tracker for one video.

Parameters:

Name	Type	Description	Default
params	MOTParams	Resolved parameters (see :meth:MOTParams.for_tracker).	required
carrier	str	"pose" or "mask".	'pose'
n_nodes	Optional[int]	Skeleton size for pose detections (None if unknown; then taken from the detections).	None

Raises:

Type	Description
ValueError	On an unknown carrier, or parameters resolved for another tracker.

Source code in sleap_nn/tracking/mot/base.py

def __init__(
    self,
    params: MOTParams,
    carrier: str = "pose",
    n_nodes: Optional[int] = None,
):
    """Create a tracker for one video.

    Args:
        params: Resolved parameters (see :meth:`MOTParams.for_tracker`).
        carrier: ``"pose"`` or ``"mask"``.
        n_nodes: Skeleton size for pose detections (``None`` if unknown; then
            taken from the detections).

    Raises:
        ValueError: On an unknown carrier, or parameters resolved for another
            tracker.
    """
    if carrier not in CARRIERS:
        raise ValueError(f"carrier must be one of {CARRIERS}, got {carrier!r}.")
    if self.name and params.tracker != self.name:
        raise ValueError(
            f"{type(self).__name__} needs parameters resolved for "
            f"{self.name!r}, got parameters for {params.tracker!r}."
        )
    self.params = params
    self.carrier = carrier
    self.n_nodes = None if n_nodes is None else int(n_nodes)
    self.kalman_kind = (
        params.kalman if params.kalman != "auto" else self.default_kalman
    )
    self.kf = make_kalman_filter(self.kalman_kind, nsa=params.nsa)

    self.calibration: Optional[VideoCalibration] = None
    self.tracks: List[MOTTrack] = []
    self.n_removed = 0
    self.track_name_offset = 0
    self._track_objects: Dict[int, sio.Track] = {}

    self._next_tid = 0
    self._next_public_id = 0
    # Step index of the latest processed frame (-1: nothing processed yet).
    self._step = -1
    self._last_frame_idx: Optional[int] = None
    # Frame-index delta to the previous processed frame (None at the first).
    self._last_delta: Optional[int] = None
    # Until `prepare` (or without it), the stride and time base are inferred
    # online from the positive frame deltas seen so far (`_infer_stride`).
    self._time_base = resolve_time_base(params.time_base, True)
    self._frame_step = 1
    self._delta_counts: Dict[int, int] = {}

    self._similarity_kind: Optional[str] = None
    if carrier == "mask" or self.n_nodes is not None:
        self._similarity_kind = self._resolve_similarity(self.n_nodes)

    # Running body-size evidence without calibration: raw-box diagonals,
    # else (extent-less detections) same-frame spacings and per-step lengths
    # of unambiguous frame-to-frame links, with the previous frame for them.
    self._diagonals: deque = deque(maxlen=BODY_SIZE_HISTORY)
    self._spacings: deque = deque(maxlen=BODY_SIZE_HISTORY)
    self._step_lengths: deque = deque(maxlen=BODY_SIZE_HISTORY)
    self._prev_dets: Optional[FrameDetections] = None
    self._embedding_dim: Optional[int] = None
    self._vectors_seen = False
    # Divisor of every detection score (`normalize_scores`; set by `prepare`).
    self._score_normalizer: Optional[float] = None
    self._n_valid_seen = 0
    self._warned: Set[str] = set()

    # `prepare` cache: frame_idx -> FrameDetections (validated by identity);
    # each entry is dropped when its frame is tracked.
    self._cache: Dict[int, FrameDetections] = {}
    # Every processed frame's objects, for identity linking in `finalize`
    # (kept only when linking runs).
    self._seen_frames: List[Tuple[int, list]] = []
    self._n_frames_seen = 0
    self._finalized = False
    self.resolved: Dict[str, Any] = {}
    self._refresh_resolved()

finalize()

End-of-video pass: identity linking when enabled. Idempotent.

Logs a WARNING when valid detections were tracked but no track was ever confirmed (every detection came back untracked).

Source code in sleap_nn/tracking/mot/base.py

def finalize(self) -> None:
    """End-of-video pass: identity linking when enabled. Idempotent.

    Logs a WARNING when valid detections were tracked but no track was ever
    confirmed (every detection came back untracked).
    """
    if self._finalized:
        return
    self._finalized = True
    p = self.params
    if self._n_valid_seen and self._next_public_id == 0:
        logger.warning(
            f"{self.name}: {self._n_valid_seen} valid detection(s) over "
            f"{self._n_frames_seen} frame(s), but no track was ever "
            "confirmed: every detection is untracked. Check the score "
            f"thresholds (new_track_threshold={p.new_track_threshold}, "
            f"track_low_threshold={p.track_low_threshold}; normalize_scores="
            f"{p.normalize_scores}), min_hits={p.min_hits} and the frame "
            "order / time base"
            + (
                "; without prepare() scores are not normalized and nothing is "
                "calibrated."
                if self.calibration is None
                else "."
            )
        )
    if p.link:
        result = link_tracklets(
            self._seen_frames,
            carrier=self.carrier,
            calibration=self.calibration,
            max_tracks=p.max_tracks,
            track_buffer=p.track_buffer,
            swap_correction=p.link_swap_correction,
            spatial_gate_frames=p.link_spatial_gate_frames,
            box_padding=p.box_padding,
            high_threshold=p.track_high_threshold,
            score_mode=p.score_mode,
            score_normalizer=self._score_normalizer,
            time_base=p.time_base,
            body_size=p.body_size,
            track_objects=self._track_objects,
        )
        self.resolved["linking"] = result.as_dict()
    self._refresh_resolved()

prepare(frames)

Optional per-video pre-pass over every frame that will be tracked.

Extracts (and caches, keyed by frame index and object identities) each frame's :class:FrameDetections, calibrates the video (:func:~sleap_nn.tracking.mot.calibration.calibrate), resolves the time base, the body size (pseudo-box size of degenerate geometry and the centroid similarity scale) and, with normalize_scores, the score normalizer applied to every detection from then on; records everything in resolved. prepare([]) is a no-op.

Parameters:

Name	Type	Description	Default
frames	Sequence[Tuple[int, Sequence[Any]]]	(frame_idx, objects) per frame of ONE video, in increasing frame order, holding exactly the objects track() will receive. Repeated or backwards frame indices are logged (WARNING): the calibration would measure motion across them.	required

Source code in sleap_nn/tracking/mot/base.py

def prepare(self, frames: Sequence[Tuple[int, Sequence[Any]]]) -> None:
    """Optional per-video pre-pass over every frame that will be tracked.

    Extracts (and caches, keyed by frame index and object identities) each
    frame's :class:`FrameDetections`, calibrates the video
    (:func:`~sleap_nn.tracking.mot.calibration.calibrate`), resolves the
    time base, the body size (pseudo-box size of degenerate geometry and the
    ``centroid`` similarity scale) and, with ``normalize_scores``, the score
    normalizer applied to every detection from then on; records everything
    in ``resolved``. ``prepare([])`` is a no-op.

    Args:
        frames: ``(frame_idx, objects)`` per frame of ONE video, in
            increasing frame order, holding exactly the objects ``track()``
            will receive. Repeated or backwards frame indices are logged
            (WARNING): the calibration would measure motion across them.
    """
    frames = [(int(fi), list(objs)) for fi, objs in frames]
    if not frames:
        return
    p = self.params
    frame_idx = np.array([fi for fi, _ in frames], dtype=np.int64)
    n_bad = int(np.count_nonzero(np.diff(frame_idx) <= 0))
    if n_bad:
        logger.warning(
            f"{self.name}: prepare() got {n_bad} repeated or backwards frame "
            "index step(s). A MOT tracker tracks ONE video in increasing frame "
            "order: build one tracker per video and sort its frames (as "
            "apply_tracking does). The calibration measures motion across "
            "these jumps and tracks may continue across them."
        )

    # First pass: raw scores (the calibration measures them) and geometry
    # without a pseudo box (it measures the body size on non-degenerate
    # detections only).
    self._score_normalizer = None
    extracted = [
        self._extract(objs, fi, pseudo_box_size=None) for fi, objs in frames
    ]
    if self._similarity_kind is None and self.carrier == "pose":
        n_nodes = [d.n_nodes for d in extracted if len(d) > 0]
        if n_nodes:
            self._similarity_kind = self._resolve_similarity(max(n_nodes))

    # The g = 1 trials (gallery_cos_gate) use the gallery length the
    # tracker's memory actually keeps (DeepSORT: gallery_budget); the
    # appearance part is skipped when neither association nor linking
    # reads it.
    cal = calibrate(
        extracted,
        track_buffer=p.track_buffer,
        gallery_size=self._make_appearance_memory().gallery_size,
        ema_alpha=p.ema_alpha,
        time_base=p.time_base,
        body_size=p.body_size,
        appearance=self.uses_appearance or p.link,
    )
    self.calibration = cal

    # Scores: normalized per video, or a warning when the raw scores of
    # most detections cannot start a track.
    q95 = cal.score_quantiles[-1] if cal.score_quantiles else None
    if p.normalize_scores:
        self._score_normalizer = score_normalizer(cal)
    if self._score_normalizer is not None and self._score_normalizer != 1.0:
        logger.info(
            f"{self.name}: detection scores divided by the video's score q95 "
            f"({self._score_normalizer:.3f}) and clipped to [0, 1] "
            "(normalize_scores)."
        )
    self._check_score_cuts(q95)

    # Degenerate detections (single node, sub-pixel extent) get the body
    # size as their box; every padded box is floored at MIN_BOX_FRACTION of
    # it. Scores are normalized exactly once (`_extract` does it for the
    # re-extracted frames).
    body = self._body_size()
    for i, dets in enumerate(extracted):
        if self._has_degenerate(dets):
            extracted[i] = self._extract(
                dets.sources, dets.frame_idx, pseudo_box_size=body
            )
        else:
            widen_boxes(extracted[i], MIN_BOX_FRACTION * body)
            normalize_scores(extracted[i], self._score_normalizer)
    self._cache = {dets.frame_idx: dets for dets in extracted}

    # Time base: the calibration's (it resolved `p.time_base` with the same
    # rule), and its stride in the frames time base.
    self._time_base = cal.time_base
    if p.time_base == "auto" and self._time_base == "steps":
        logger.warning(
            "The frames to track are irregularly spaced (the most common "
            f"frame step, {cal.frame_step}, covers less than half of the "
            "gaps): counting one step per processed frame (time_base="
            "'steps'). Kalman prediction, track_buffer and min_hits then "
            "count processed frames, not video frames."
        )
    self._frame_step = (
        max(1, int(cal.frame_step)) if self._time_base == "frames" else 1
    )

    if self.uses_appearance:
        self._unusable_vector_notice()
        if not self._vectors_seen:
            logger.warning(
                f"{self.name}: the tracked detections carry no appearance "
                "vectors; associating on geometry alone."
            )
        elif p.appearance_threshold == "auto" and self._appearance_gate() is None:
            logger.warning(
                f"{self.name}: appearance is disabled for association by the "
                f"video's calibration (status={cal.status!r}, assoc_rank1="
                f"{cal.assoc_rank1}); associating on geometry alone."
            )
    self._refresh_resolved()

track(untracked_instances, frame_idx, image=None)

Track one frame (see the module contract). Returns every input object.

Source code in sleap_nn/tracking/mot/base.py

def track(
    self, untracked_instances: Sequence[Any], frame_idx: int, image: Any = None
) -> list:
    """Track one frame (see the module contract). Returns every input object."""
    objs = list(untracked_instances)
    frame_idx = int(frame_idx)
    for obj in objs:
        _set_label(obj, None, 0.0)
    self._n_frames_seen += 1
    if self.params.link:
        self._seen_frames.append((frame_idx, objs))

    dets = self._detections(objs, frame_idx)
    n_steps = self._advance(frame_idx)
    if self.calibration is None:
        # Before association, so this frame already uses the evidence.
        self._observe_uncalibrated(dets, n_steps)
    self._n_valid_seen += int(np.count_nonzero(dets.valid))

    if len(dets) == 0:
        # Nothing to match; the bookkeeping below still ages every track.
        result = AssociationResult()
    else:
        result = self._associate(dets)

    pool = list(self.tracks)
    removed_tids = {t.tid for t in result.removed}
    gate_ctx = self._gate_context(pool, dets, result)

    # Matches: Kalman update, counters, history, members, appearance.
    applied: List[Tuple[MOTTrack, int, float]] = []
    used_tids: Set[int] = set()
    used_dets: Set[int] = set()
    for track, det_idx, sim in result.matches:
        det_idx = int(det_idx)
        if (
            track.tid in used_tids
            or det_idx in used_dets
            or not track.is_live
            or not 0 <= det_idx < len(dets)
            or not dets.valid[det_idx]
        ):
            logger.debug(
                f"Ignoring an invalid match (track {track.tid}, detection "
                f"{det_idx}) at frame {frame_idx}."
            )
            continue
        used_tids.add(track.tid)
        used_dets.add(det_idx)
        sim = _clip_score(sim)
        self._apply_match(track, dets, det_idx, sim, gate_ctx)
        applied.append((track, det_idx, sim))

    # Unmatched live tracks: freeze for ORU, CONFIRMED -> LOST, tentative
    # tracks of the DeepSORT/ByteTrack families die.
    for track in pool:
        if track.tid in used_tids or track.tid in removed_tids:
            continue
        self._mark_missed(track)
    for track in result.removed:
        self._remove(track)

    # Spawn tentative tracks from unmatched, valid, confident detections.
    for det_idx in range(len(dets)):
        if (
            det_idx in used_dets
            or det_idx in result.no_spawn
            or not dets.valid[det_idx]
            or dets.scores[det_idx] < self.params.new_track_threshold
        ):
            continue
        self._spawn(dets, det_idx)

    # Confirm tentative tracks that reached `min_hits` (retro-labels members).
    for track in self.tracks:
        if track.state == TrackState.TENTATIVE and track.time_since_update == 0:
            count = track.hit_streak if self.confirm_on_streak else track.hits
            if count >= self.params.min_hits:
                self._confirm(track)

    # Remove tracks unmatched for longer than the buffer (any state).
    for track in list(self.tracks):
        if track.time_since_update > self.params.track_buffer:
            self._remove(track)

    # Label this frame's detections of confirmed tracks.
    for track, det_idx, sim in applied:
        if track.state == TrackState.CONFIRMED and track.public_id is not None:
            _set_label(
                dets.sources[det_idx], self._track_objects[track.public_id], sim
            )
    return list(objs)

BoTSORTTracker

Bases: ByteTracker

BoT-SORT tracker (see :mod:sleap_nn.tracking.mot.bytetrack).

Methods:

Name	Description
__init__	Create a tracker for one video.

Source code in sleap_nn/tracking/mot/bytetrack.py

class BoTSORTTracker(ByteTracker):
    """BoT-SORT tracker (see :mod:`sleap_nn.tracking.mot.bytetrack`)."""

    name = "botsort"
    default_kalman = "xywh"
    uses_appearance = True
    lost_zero_velocity = (6, 7)
    # BoT-SORT zeroes the width and height velocities (aspect and height for
    # `xyah`; the area for `xysr`, whose aspect has no velocity).
    size_velocity: ClassVar[Dict[str, Tuple[int, ...]]] = {
        "xywh": (6, 7),
        "xyah": (6, 7),
        "xysr": (6,),
    }

    def __init__(
        self,
        params: MOTParams,
        carrier: str = "pose",
        n_nodes: Optional[int] = None,
    ):
        """Create a tracker for one video.

        Args:
            params: Resolved :class:`~sleap_nn.tracking.mot.params.MOTParams`.
            carrier: ``"pose"`` or ``"mask"``.
            n_nodes: Skeleton size for pose detections (``None`` if unknown).
        """
        super().__init__(params, carrier=carrier, n_nodes=n_nodes)
        self._warned_no_vectors = False

    def _associate(self, dets: FrameDetections) -> AssociationResult:
        """BYTE association with BoT-SORT's gated appearance cost (stages 1 and 3).

        Args:
            dets: This frame's detections.

        Returns:
            See :meth:`ByteTracker._associate`.
        """
        if (
            self.calibration is None
            and not self._vectors_seen
            and not self._warned_no_vectors
            and bool(np.any(dets.valid))
        ):
            self._warned_no_vectors = True
            logger.warning(
                "botsort: the detections carry no usable appearance vectors; "
                "associating on geometry alone until vectors appear."
            )
        return super()._associate(dets)

    def _fused_cost(
        self,
        tracks: Sequence[MOTTrack],
        det_idx: np.ndarray,
        dets: FrameDetections,
        sim: np.ndarray,
    ) -> np.ndarray:
        """Stage-1/3 cost: geometric cost, lowered by gated decisive appearance.

        Under the ``paper`` preset every in-gate cell takes the appearance cost
        (published min-fusion); otherwise only :func:`decisive_cells` with the
        calibrated margin ``m_assoc`` do.

        Args:
            tracks: The stage's tracks (rows).
            det_idx: The stage's detection indices (columns).
            dets: This frame's detections.
            sim: ``(len(tracks), len(det_idx))`` geometric similarity (unfused).

        Returns:
            The ``(len(tracks), len(det_idx))`` cost matrix.
        """
        d_geo = super()._fused_cost(tracks, det_idx, dets, sim)
        gate = self._appearance_gate("ema")
        if gate is None or dets.embeddings is None:
            return d_geo
        cos = self._appearance_cos(tracks, dets, "ema")[:, det_idx]
        # Appearance only inside the geometric (proximity) gate.
        near = sim >= self.params.proximity_threshold
        in_gate = np.isfinite(cos) & (cos >= gate) & near
        if not in_gate.any():
            return d_geo
        if self.params.preset == "paper":
            decisive = in_gate
        else:
            # ...and only where it discriminates: the cell is the top in-gate
            # cosine of its row and of its column, by the calibrated margin.
            decisive = decisive_cells(cos, in_gate, self._appearance_margin())
        d_emb = np.where(decisive, 0.5 * (1.0 - np.nan_to_num(cos)), 1.0)
        return np.where(
            decisive,
            np.minimum(d_geo, d_emb) + GEOMETRY_TIE_BREAK * d_geo,
            d_geo,
        )

__init__(params, carrier='pose', n_nodes=None)

Create a tracker for one video.

Parameters:

Name	Type	Description	Default
params	MOTParams	Resolved :class:~sleap_nn.tracking.mot.params.MOTParams.	required
carrier	str	"pose" or "mask".	'pose'
n_nodes	Optional[int]	Skeleton size for pose detections (None if unknown).	None

Source code in sleap_nn/tracking/mot/bytetrack.py

def __init__(
    self,
    params: MOTParams,
    carrier: str = "pose",
    n_nodes: Optional[int] = None,
):
    """Create a tracker for one video.

    Args:
        params: Resolved :class:`~sleap_nn.tracking.mot.params.MOTParams`.
        carrier: ``"pose"`` or ``"mask"``.
        n_nodes: Skeleton size for pose detections (``None`` if unknown).
    """
    super().__init__(params, carrier=carrier, n_nodes=n_nodes)
    self._warned_no_vectors = False

ByteTracker

Bases: BaseMOTTracker

ByteTrack tracker (see :mod:sleap_nn.tracking.mot.bytetrack).

Class attributes

size_velocity: Kalman state indices zeroed before predicting a track that was not matched at the previous step, per filter (ByteTrack: the height velocity; the area velocity for xysr). Selects lost_zero_velocity for the filter in use.

Methods:

Name	Description
__init__	Create a tracker for one video.

Source code in sleap_nn/tracking/mot/bytetrack.py

class ByteTracker(BaseMOTTracker):
    """ByteTrack tracker (see :mod:`sleap_nn.tracking.mot.bytetrack`).

    Class attributes:
        size_velocity: Kalman state indices zeroed before predicting a track that
            was not matched at the previous step, per filter (ByteTrack: the
            height velocity; the area velocity for ``xysr``). Selects
            ``lost_zero_velocity`` for the filter in use.
    """

    name = "bytetrack"
    default_kalman = "xyah"
    tentative_dies_on_miss = True
    hits_start = 1
    lost_zero_velocity = (7,)
    size_velocity: ClassVar[Dict[str, Tuple[int, ...]]] = {
        "xyah": (7,),
        "xywh": (7,),
        "xysr": (6,),
    }

    def __init__(
        self,
        params: MOTParams,
        carrier: str = "pose",
        n_nodes: Optional[int] = None,
    ):
        """Create a tracker for one video.

        Args:
            params: Resolved :class:`~sleap_nn.tracking.mot.params.MOTParams`.
            carrier: ``"pose"`` or ``"mask"``.
            n_nodes: Skeleton size for pose detections (``None`` if unknown).
        """
        super().__init__(params, carrier=carrier, n_nodes=n_nodes)
        # Zero the velocities that exist in the filter actually in use (a
        # `kalman` override changes the state layout).
        self.lost_zero_velocity = self.size_velocity[self.kalman_kind]

    def _score_cuts(self) -> List[Tuple[str, bool, str]]:
        """Base cuts, plus the STRICT high-score split.

        Only ``s > track_high_threshold`` (``D_hi``) may start a track.
        """
        return super()._score_cuts() + [
            (
                "track_high_threshold",
                True,
                "only detections scoring above it can start a track, so most "
                "detections can never start one. Lower --track_high_threshold",
            )
        ]

    def _associate(self, dets: FrameDetections) -> AssociationResult:
        """BYTE association: high, low and unconfirmed stages, then de-duplication.

        Args:
            dets: This frame's detections.

        Returns:
            The matches (with geometric similarities), the detections that must
            not start a track and the tracks removed as duplicates.
        """
        p = self.params
        scores = np.asarray(dets.scores, dtype=np.float64)
        valid = np.asarray(dets.valid, dtype=bool)
        high = valid & (scores > p.track_high_threshold)
        low = (
            valid & (scores > p.track_low_threshold) & (scores < p.track_high_threshold)
        )
        # Only high-score detections may start a track, and only from the
        # new-track threshold on.
        spawnable = high & (scores >= p.new_track_threshold)
        result = AssociationResult(
            no_spawn={int(j) for j in np.flatnonzero(~spawnable)}
        )

        tracks = self._live()
        if not tracks:
            return result
        sim = self._geometric_similarity(tracks, dets)
        rows = {t.tid: i for i, t in enumerate(tracks)}

        # Stage 1: confirmed and lost tracks vs high-score detections.
        pool = [t for t in tracks if t.state != TrackState.TENTATIVE]
        left_pool, left_high = self._assign(
            pool,
            np.flatnonzero(high),
            dets,
            sim,
            rows,
            max_cost=1.0 - p.min_match_similarity,
            result=result,
            fused=True,
        )

        # Stage 2: tracks matched at the previous step vs low-score detections
        # (geometry only, never fused). Unmatched ones become LOST in the base.
        tracked = [t for t in left_pool if t.state == TrackState.CONFIRMED]
        self._assign(
            tracked,
            np.flatnonzero(low),
            dets,
            sim,
            rows,
            max_cost=1.0 - p.low_stage_min_similarity,
            result=result,
            fused=False,
        )

        # Stage 3: tentative tracks vs the high-score detections left over.
        tentative = [t for t in tracks if t.state == TrackState.TENTATIVE]
        _, left_high = self._assign(
            tentative,
            left_high,
            dets,
            sim,
            rows,
            max_cost=1.0 - p.unconfirmed_min_similarity,
            result=result,
            fused=True,
        )

        self._remove_duplicates(tracks, sim, rows, dets, left_high, spawnable, result)
        return result

    # ------------------------------------------------------------- helpers
    def _geometric_similarity(
        self, tracks: Sequence[MOTTrack], dets: FrameDetections
    ) -> np.ndarray:
        """Base similarity; pose IoU also scores the shared visible nodes.

        For the ``iou`` similarity on the pose carrier, each pair scores the
        larger of the Kalman-box IoU and :func:`shared_node_iou` of the track's
        predicted pose (last pose shifted by the Kalman center's motion) and the
        detection, so a partly occluded detection is compared with the same part
        of the track (see the module docstring). Other kinds are unchanged.

        Args:
            tracks: Live tracks (rows), already predicted to this step.
            dets: This frame's detections.

        Returns:
            ``(len(tracks), len(dets))`` similarity in ``[0, 1]``.
        """
        tracks = list(tracks)
        sim = super()._geometric_similarity(tracks, dets)
        if (
            sim.size == 0
            or self.carrier != "pose"
            or dets.keypoints is None
            or self._similarity(dets) != "iou"
        ):
            return sim
        means = np.stack([t.mean for t in tracks])
        predicted, _ = self._predicted_poses(tracks, means)
        shared = shared_node_iou(
            predicted,
            dets.keypoints,
            box_padding=self.params.box_padding,
            min_box_size=MIN_BOX_FRACTION * self._body_size(),
        )
        return self._finish_similarity(np.fmax(sim, shared), dets)

    def _assign(
        self,
        tracks: Sequence[MOTTrack],
        det_idx: np.ndarray,
        dets: FrameDetections,
        sim: np.ndarray,
        rows: Dict[int, int],
        *,
        max_cost: float,
        result: AssociationResult,
        fused: bool,
    ) -> Tuple[List[MOTTrack], np.ndarray]:
        """One assignment stage; appends its matches to ``result``.

        Args:
            tracks: The stage's tracks.
            det_idx: The stage's detection indices.
            dets: This frame's detections.
            sim: ``(n_live, n_dets)`` geometric similarity of every live track.
            rows: Live track ``tid`` -> row of ``sim``.
            max_cost: Cost cap of a match.
            result: Receives the matches.
            fused: Use the stage-1/3 cost (:meth:`_fused_cost`) instead of
                ``1 - sim``.

        Returns:
            The unmatched tracks and the unmatched detection indices.
        """
        tracks = list(tracks)
        det_idx = np.asarray(det_idx, dtype=np.int64)
        if not tracks or len(det_idx) == 0:
            return tracks, det_idx
        sub = sim[np.ix_([rows[t.tid] for t in tracks], det_idx)]
        cost = self._fused_cost(tracks, det_idx, dets, sub) if fused else 1.0 - sub
        pairs, left_rows, left_cols = linear_assignment(cost, max_cost=max_cost)
        for r, c in pairs:
            result.matches.append((tracks[r], int(det_idx[c]), float(sub[r, c])))
        return [tracks[r] for r in left_rows], det_idx[left_cols]

    def _fused_cost(
        self,
        tracks: Sequence[MOTTrack],
        det_idx: np.ndarray,
        dets: FrameDetections,
        sim: np.ndarray,
    ) -> np.ndarray:
        """Stage-1/3 cost: ``1 - sim * score`` (``fuse_score``) or ``1 - sim``.

        Args:
            tracks: The stage's tracks (rows).
            det_idx: The stage's detection indices (columns).
            dets: This frame's detections.
            sim: ``(len(tracks), len(det_idx))`` geometric similarity.

        Returns:
            The ``(len(tracks), len(det_idx))`` cost matrix.
        """
        if self.params.fuse_score:
            sim = fuse_score(sim, np.asarray(dets.scores, dtype=np.float64)[det_idx])
        return 1.0 - sim

    def _duplicate_similarity(
        self,
        lost: Sequence[MOTTrack],
        cols: Sequence[int],
        sim: np.ndarray,
        rows: Dict[int, int],
        dets: FrameDetections,
    ) -> np.ndarray:
        """``(len(lost), len(cols))`` overlap compared with ``duplicate_similarity``.

        The IoU of each lost track's predicted (Kalman) box and the detection's
        padded box -- the reference test -- for the ``iou`` and ``centroid``
        similarities (single-node detections carry body-sized pseudo boxes); the
        association OKS for ``oks``.
        """
        if self._similarity(dets) == "oks":
            return sim[np.ix_([rows[t.tid] for t in lost], list(cols))]
        boxes = self.kf.xyxy_from_mean(np.stack([t.mean for t in lost]))
        return iou_matrix(boxes, np.asarray(dets.boxes, dtype=np.float64)[list(cols)])

    def _remove_duplicates(
        self,
        tracks: Sequence[MOTTrack],
        sim: np.ndarray,
        rows: Dict[int, int],
        dets: FrameDetections,
        left_high: np.ndarray,
        spawnable: np.ndarray,
        result: AssociationResult,
    ) -> None:
        """Tracked-vs-lost duplicate removal (shorter history loses), in place.

        The tracked side holds this frame's matched tracks (box: the matched
        detection) and the detections about to start a track (empty history); the
        lost side holds the live tracks left unmatched (box: their prediction)
        that survive this step. A pair whose overlap
        (:meth:`_duplicate_similarity`) exceeds ``duplicate_similarity`` drops the
        one with the shorter history; ties drop the tracked one (reference
        semantics). Every pair is decided independently.

        Args:
            tracks: Every live track (rows of ``sim``).
            sim: ``(n_live, n_dets)`` geometric similarity.
            rows: Live track ``tid`` -> row of ``sim``.
            dets: This frame's detections.
            left_high: High-score detections left unmatched by every stage.
            spawnable: ``(n_dets,)`` detections allowed to start a track.
            result: Updated in place (matches, ``no_spawn``, ``removed``).
        """
        p = self.params
        matched_tids = {t.tid for t, _, _ in result.matches}
        lost = [
            t
            for t in tracks
            if t.state != TrackState.TENTATIVE
            and t.tid not in matched_tids
            and t.time_since_update <= p.track_buffer
        ]
        if not lost:
            return
        # Tracked side: (matched track or None for a would-be new track, det, history).
        tracked: List[Tuple[Optional[MOTTrack], int, int]] = [
            (t, int(j), int(dets.frame_idx) - int(t.start_frame))
            for t, j, _ in result.matches
        ]
        tracked += [(None, int(j), 0) for j in left_high if spawnable[j]]
        if not tracked:
            return

        dup = self._duplicate_similarity(
            lost, [j for _, j, _ in tracked], sim, rows, dets
        )
        drop_tracked, drop_lost = set(), set()
        for q, k in np.argwhere(dup > p.duplicate_similarity):
            lost_history = int(lost[q].last_frame) - int(lost[q].start_frame)
            if tracked[k][2] > lost_history:
                drop_lost.add(int(q))
            else:
                drop_tracked.add(int(k))
        if not drop_tracked and not drop_lost:
            return

        removed = {t.tid: t for t in result.removed}
        for k in sorted(drop_tracked):
            track, det_idx, _ = tracked[k]
            # The detection stays untracked: no match, and no new track from it.
            result.no_spawn.add(det_idx)
            if track is not None:
                removed.setdefault(track.tid, track)
        for q in sorted(drop_lost):
            removed.setdefault(lost[q].tid, lost[q])
        result.matches = [m for m in result.matches if m[0].tid not in removed]
        result.removed = list(removed.values())
        logger.debug(
            f"{self.name}: frame {dets.frame_idx}: removed {len(removed)} duplicate "
            f"track(s) (tids {sorted(removed)})."
        )

__init__(params, carrier='pose', n_nodes=None)

Create a tracker for one video.

Parameters:

Name	Type	Description	Default
params	MOTParams	Resolved :class:~sleap_nn.tracking.mot.params.MOTParams.	required
carrier	str	"pose" or "mask".	'pose'
n_nodes	Optional[int]	Skeleton size for pose detections (None if unknown).	None

Source code in sleap_nn/tracking/mot/bytetrack.py

def __init__(
    self,
    params: MOTParams,
    carrier: str = "pose",
    n_nodes: Optional[int] = None,
):
    """Create a tracker for one video.

    Args:
        params: Resolved :class:`~sleap_nn.tracking.mot.params.MOTParams`.
        carrier: ``"pose"`` or ``"mask"``.
        n_nodes: Skeleton size for pose detections (``None`` if unknown).
    """
    super().__init__(params, carrier=carrier, n_nodes=n_nodes)
    # Zero the velocities that exist in the filter actually in use (a
    # `kalman` override changes the state layout).
    self.lost_zero_velocity = self.size_velocity[self.kalman_kind]

DeepOCSORTTracker

Bases: OCSORTTracker

Deep OC-SORT tracker (see :mod:sleap_nn.tracking.mot.ocsort).

Methods:

Name	Description
__init__	Create a tracker for one video.
prepare	Per-video pre-pass (see :meth:BaseMOTTracker.prepare).

Source code in sleap_nn/tracking/mot/ocsort.py

class DeepOCSORTTracker(OCSORTTracker):
    """Deep OC-SORT tracker (see :mod:`sleap_nn.tracking.mot.ocsort`)."""

    name = "deepocsort"
    default_kalman = "xywh"
    uses_appearance = True

    # Deep OC-SORT associates high-score detections only (no BYTE stage).
    _supports_byte = False

    def __init__(
        self,
        params: MOTParams,
        carrier: str = "pose",
        n_nodes: Optional[int] = None,
    ):
        """Create a tracker for one video.

        Args:
            params: Resolved parameters (see :meth:`MOTParams.for_tracker`).
            carrier: ``"pose"`` or ``"mask"``.
            n_nodes: Skeleton size for pose detections (``None`` if unknown; then
                taken from the detections).

        Raises:
            ValueError: On an unknown carrier, or parameters resolved for another
                tracker.
        """
        super().__init__(params, carrier=carrier, n_nodes=n_nodes)
        # The reference stops the box-size velocities of a frozen (missed) track
        # on its [cx, cy, w, h] filter; other filters keep their velocities.
        self.lost_zero_velocity = (
            _XYWH_SIZE_VELOCITY if self.kalman_kind == "xywh" else ()
        )
        if params.use_byte:
            logger.warning(
                "deepocsort has no BYTE (low-score) stage; use_byte is ignored."
            )
        self._appearance_noticed = False
        self._raw_term_noticed = False

    def prepare(self, frames: Sequence[Tuple[int, Sequence[Any]]]) -> None:
        """Per-video pre-pass (see :meth:`BaseMOTTracker.prepare`).

        Without any usable appearance vector (the shared rule:
        :func:`~sleap_nn.tracking.mot.detections.has_usable_embedding`; a NaN or
        all-zero vector is missing) Deep OC-SORT is OC-SORT on a ``[cx, cy, w,
        h]`` filter: with ``appearance_threshold="auto"`` this is logged and the
        video is tracked on geometry alone; an explicit (float)
        ``appearance_threshold`` asks for appearance that does not exist and
        raises. An explicit threshold on a video whose calibration cannot
        standardize the cosines (too few trials) logs that the raw EMA cosine
        is used instead of ``E'``.

        Args:
            frames: ``(frame_idx, objects)`` per frame, in temporal order, holding
                exactly the objects ``track()`` will receive.

        Raises:
            ValueError: An explicit ``appearance_threshold`` was given but no
                detection carries a usable appearance vector.
        """
        frames = [(int(fi), list(objs)) for fi, objs in frames]
        n_objects = sum(len(objs) for _, objs in frames)
        if (
            n_objects > 0
            and self.params.appearance_threshold != "auto"
            and not any(has_usable_embedding(obj) for _, objs in frames for obj in objs)
        ):
            raise ValueError(
                "deepocsort was given an explicit appearance_threshold "
                f"({self.params.appearance_threshold}) but none of the "
                f"{n_objects} detections to track carries a usable appearance "
                "vector (present, finite and non-zero). "
                "Run an embedding model first, or use --tracker ocsort (the same "
                "algorithm without appearance)."
            )
        super().prepare(frames)
        self._raw_term_notice()

    # ------------------------------------------------------------ stages
    def _first_stage(
        self,
        tracks: Sequence[MOTTrack],
        dets: FrameDetections,
        cols: np.ndarray,
        sim: np.ndarray,
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Stage 1 with the standardized appearance term and the rescue rule.

        Args:
            tracks: The live tracks (rows).
            dets: This frame's detections.
            cols: Detection indices of the columns (``D_hi``).
            sim: ``(n_tracks, len(cols))`` geometric similarity of the
                predictions.

        Returns:
            ``(valid, score)``: ``sim >= min_match_similarity`` extended by
            decisive-appearance rescues, and ``sim + OCM + W * E'`` (``E'``: the
            standardized term, or the raw EMA cosine when the video has no
            usable calibrated spread), where a negative ``W * E'`` on a pair
            that is the only admissible candidate of both its row and its
            column is clamped to 0.
        """
        valid, score = super()._first_stage(tracks, dets, cols, sim)
        cos = self._appearance_cos(tracks, dets, "ema")[:, cols]
        if not np.isfinite(cos).any():
            return valid, score  # No appearance evidence: plain OC-SORT.

        valid = valid | self._rescue(cos, sim)
        appearance = self._association_appearance(cos)
        # As in the reference, appearance only counts where geometry overlaps.
        appearance[sim <= 0.0] = 0.0
        if appearance.any():
            p = self.params
            if p.aw_off:
                weight = np.full(appearance.shape, float(p.w_assoc_emb))
            else:
                weight = adaptive_weight(
                    appearance, aw_param=p.aw_param, base=p.w_assoc_emb
                )
            term = weight * appearance
            # Appearance orders competing candidates; it must not veto an
            # uncontested one. The masked assignment maximizes over the WHOLE
            # matrix, so a lone admissible pair scored below an inadmissible,
            # zero-evidence cell (a spare false positive, a lost track) would be
            # traded for it and dropped -- but only when some unrelated pair in
            # the frame stops the one-to-one shortcut. Such a pair keeps at least
            # its OC-SORT score.
            lone = (
                valid
                & (valid.sum(axis=1, keepdims=True) == 1)
                & (valid.sum(axis=0, keepdims=True) == 1)
            )
            score = score + np.where(lone, np.maximum(term, 0.0), term)
        return valid, score

    def _association_term(self) -> str:
        """Which appearance term stage 1 adds to its objective.

        Returns:
            ``"off"`` when appearance is disabled for association (no vectors,
            or ``"auto"`` without a calibration enabling it); ``"standardized"``
            (``E'``) when the calibration has a usable ``mu_pos`` / ``mu_neg``
            spread; otherwise ``"raw_cosine"``: the reference's raw EMA cosine
            (e.g. an explicit ``appearance_threshold`` on a video that is
            uncalibrated or has too few trials to calibrate).
        """
        if self._appearance_gate("ema") is None:
            return _TERM_OFF
        if self._appearance_spread() is None:
            return _TERM_RAW_COSINE
        return _TERM_STANDARDIZED

    def _association_appearance(self, cos: np.ndarray) -> np.ndarray:
        """The appearance values of the stage-1 term (see :meth:`_association_term`).

        Args:
            cos: ``(n, m)`` EMA cosines, NaN where unavailable.

        Returns:
            ``(n, m)`` values in ``[-1, 1]``: the standardized ``E'``, or the raw
            cosine without a usable calibrated spread; 0 where a cosine is
            missing and everywhere when appearance is off.
        """
        term = self._association_term()
        if term == _TERM_STANDARDIZED:
            return self._standardized_appearance(cos)
        out = np.zeros(cos.shape, dtype=np.float64)
        if term == _TERM_RAW_COSINE:
            ok = np.isfinite(cos)
            out[ok] = np.clip(cos[ok], -1.0, 1.0)
        return out

    def _standardized_appearance(self, cos: np.ndarray) -> np.ndarray:
        """``E' = clip(2 (cos - mu_neg) / (mu_pos - mu_neg) - 1, 0, 1)``.

        ``mu_pos`` / ``mu_neg`` are the video's calibrated median true and
        competitor EMA cosines (one step apart). Missing cosines (no vector on
        either side) give 0, the neutral value; so does every pair when the video
        has no usable calibration (uncalibrated, or a degenerate spread).

        Negative evidence is dropped (clipped at 0): appearance can add support to
        a geometric match but never subtract it. With animal re-ID embedders that
        are right ~90-99% of the time per crop, a negative term lets one bad crop
        veto a valid, contested geometric match; clipping cut ID switches 70 -> 54
        on a 6-seed crossing stress test without affecting resolved crossings.

        Args:
            cos: ``(n, m)`` EMA cosines, NaN where unavailable.

        Returns:
            ``(n, m)`` standardized appearance in ``[0, 1]``.
        """
        out = np.zeros(cos.shape, dtype=np.float64)
        spread = self._appearance_spread()
        if spread is None:
            return out
        mu_neg, width = spread
        ok = np.isfinite(cos)
        out[ok] = np.clip(2.0 * (cos[ok] - mu_neg) / width - 1.0, 0.0, 1.0)
        return out

    def _appearance_spread(self) -> Optional[Tuple[float, float]]:
        """``(mu_neg, mu_pos - mu_neg)`` from the calibration, or ``None``."""
        cal = self.calibration
        if cal is None or cal.mu_pos is None or cal.mu_neg is None:
            return None
        mu_pos, mu_neg = float(cal.mu_pos), float(cal.mu_neg)
        if not (np.isfinite(mu_pos) and np.isfinite(mu_neg)):
            return None
        if mu_pos - mu_neg < _MIN_APPEARANCE_SPREAD:
            return None
        return mu_neg, mu_pos - mu_neg

    def _rescue(self, cos: np.ndarray, sim: np.ndarray) -> np.ndarray:
        """Pairs admissible through decisive appearance despite weak geometry.

        A pair is rescued when ``rescue`` is on, its geometric similarity is at
        least ``rescue_min_similarity``, its EMA cosine clears the appearance gate
        (``"auto"``: the calibrated EMA gate; a float: that value) and it is the
        strict top of its row and of its column -- among the pairs geometrically
        plausible enough to be matched at all -- by at least the calibrated
        margin ``m_assoc``.

        Args:
            cos: ``(n, m)`` EMA cosines (NaN where unavailable).
            sim: ``(n, m)`` geometric similarities.

        Returns:
            ``(n, m)`` bool mask of rescued pairs (all False when disabled).
        """
        p = self.params
        rescued = np.zeros(cos.shape, dtype=bool)
        gate = self._appearance_gate("ema")
        if not p.rescue or gate is None or cos.size == 0:
            return rescued
        floor = min(p.rescue_min_similarity, p.min_match_similarity)
        plausible = (sim >= floor) & np.isfinite(cos)
        if not plausible.any():
            return rescued
        decisive = decisive_cells(cos, plausible, self._appearance_margin())
        return decisive & (sim >= p.rescue_min_similarity) & (cos >= gate)

    def _appearance_notice(self, dets: FrameDetections) -> None:
        """Log once how a run associates when the standardized term is unavailable.

        Covers the two prepare-less cases where appearance is unavailable: no
        appearance vectors (so far), or ``appearance_threshold="auto"`` without a
        calibration (with ``prepare()`` the base class reports the decision);
        and, once, any run whose stage-1 term falls back to the raw cosine.
        """
        if len(dets) == 0:
            return
        if not self._appearance_noticed and self.calibration is None:
            self._appearance_noticed = True
            if not self._vectors_seen:
                logger.warning(
                    "deepocsort: the tracked detections carry no appearance "
                    "vectors; associating on geometry alone (OC-SORT on a [cx, cy, "
                    "w, h] filter)."
                )
            elif self._appearance_gate("ema") is None:
                logger.warning(
                    "deepocsort: appearance_threshold='auto' needs the per-video "
                    "calibration of prepare(); without it appearance is disabled "
                    "and association uses geometry alone. Call prepare() (the "
                    "calibrated, standardized appearance term) or pass an explicit "
                    "appearance_threshold (the reference's raw-cosine term)."
                )
        self._raw_term_notice()

    def _raw_term_notice(self) -> None:
        """Log once when stage 1 uses the raw EMA cosine instead of ``E'``."""
        if self._raw_term_noticed or self._association_term() != _TERM_RAW_COSINE:
            return
        self._raw_term_noticed = True
        logger.warning(
            f"deepocsort: {self._raw_term_reason()}, so appearance cosines cannot "
            "be standardized; the stage-1 appearance term is the raw EMA cosine "
            "(as in the reference Deep OC-SORT) instead of the standardized E'."
        )

    def _raw_term_reason(self) -> str:
        """Why the calibration gives no usable ``mu_pos`` / ``mu_neg`` spread."""
        cal = self.calibration
        if cal is None:
            return "the video is uncalibrated (prepare() was not called)"
        detail = f"status={cal.status!r}, {cal.n_trials} trials"
        if cal.mu_pos is not None and cal.mu_neg is not None:
            detail += f", mu_pos={cal.mu_pos:.3g}, mu_neg={cal.mu_neg:.3g}"
        return (
            f"the video's calibration has no usable true/competitor cosine spread "
            f"({detail})"
        )

    def _appearance_summary(self) -> Dict[str, Any]:
        """JSON-safe record of the appearance decisions, with the stage-1 term.

        Adds ``association_term`` (``"standardized"``, ``"raw_cosine"`` or
        ``"off"``, see :meth:`_association_term`) to the base record and, for the
        raw-cosine fallback, why the calibrated term is unavailable.
        """
        summary = super()._appearance_summary()
        term = self._association_term()
        summary["association_term"] = term
        if term == _TERM_RAW_COSINE:
            summary["reason"] = (
                f"{summary['reason']}; {self._raw_term_reason()}: the stage-1 "
                "term is the raw EMA cosine"
            )
        return summary

__init__(params, carrier='pose', n_nodes=None)

Create a tracker for one video.

Parameters:

Name	Type	Description	Default
params	MOTParams	Resolved parameters (see :meth:MOTParams.for_tracker).	required
carrier	str	"pose" or "mask".	'pose'
n_nodes	Optional[int]	Skeleton size for pose detections (None if unknown; then taken from the detections).	None

Raises:

Type	Description
ValueError	On an unknown carrier, or parameters resolved for another tracker.

Source code in sleap_nn/tracking/mot/ocsort.py

def __init__(
    self,
    params: MOTParams,
    carrier: str = "pose",
    n_nodes: Optional[int] = None,
):
    """Create a tracker for one video.

    Args:
        params: Resolved parameters (see :meth:`MOTParams.for_tracker`).
        carrier: ``"pose"`` or ``"mask"``.
        n_nodes: Skeleton size for pose detections (``None`` if unknown; then
            taken from the detections).

    Raises:
        ValueError: On an unknown carrier, or parameters resolved for another
            tracker.
    """
    super().__init__(params, carrier=carrier, n_nodes=n_nodes)
    # The reference stops the box-size velocities of a frozen (missed) track
    # on its [cx, cy, w, h] filter; other filters keep their velocities.
    self.lost_zero_velocity = (
        _XYWH_SIZE_VELOCITY if self.kalman_kind == "xywh" else ()
    )
    if params.use_byte:
        logger.warning(
            "deepocsort has no BYTE (low-score) stage; use_byte is ignored."
        )
    self._appearance_noticed = False
    self._raw_term_noticed = False

prepare(frames)

Per-video pre-pass (see :meth:BaseMOTTracker.prepare).

Without any usable appearance vector (the shared rule: :func:~sleap_nn.tracking.mot.detections.has_usable_embedding; a NaN or all-zero vector is missing) Deep OC-SORT is OC-SORT on a [cx, cy, w, h] filter: with appearance_threshold="auto" this is logged and the video is tracked on geometry alone; an explicit (float) appearance_threshold asks for appearance that does not exist and raises. An explicit threshold on a video whose calibration cannot standardize the cosines (too few trials) logs that the raw EMA cosine is used instead of E'.

Parameters:

Name	Type	Description	Default
frames	Sequence[Tuple[int, Sequence[Any]]]	(frame_idx, objects) per frame, in temporal order, holding exactly the objects track() will receive.	required

Raises:

Type	Description
ValueError	An explicit appearance_threshold was given but no detection carries a usable appearance vector.

Source code in sleap_nn/tracking/mot/ocsort.py

def prepare(self, frames: Sequence[Tuple[int, Sequence[Any]]]) -> None:
    """Per-video pre-pass (see :meth:`BaseMOTTracker.prepare`).

    Without any usable appearance vector (the shared rule:
    :func:`~sleap_nn.tracking.mot.detections.has_usable_embedding`; a NaN or
    all-zero vector is missing) Deep OC-SORT is OC-SORT on a ``[cx, cy, w,
    h]`` filter: with ``appearance_threshold="auto"`` this is logged and the
    video is tracked on geometry alone; an explicit (float)
    ``appearance_threshold`` asks for appearance that does not exist and
    raises. An explicit threshold on a video whose calibration cannot
    standardize the cosines (too few trials) logs that the raw EMA cosine
    is used instead of ``E'``.

    Args:
        frames: ``(frame_idx, objects)`` per frame, in temporal order, holding
            exactly the objects ``track()`` will receive.

    Raises:
        ValueError: An explicit ``appearance_threshold`` was given but no
            detection carries a usable appearance vector.
    """
    frames = [(int(fi), list(objs)) for fi, objs in frames]
    n_objects = sum(len(objs) for _, objs in frames)
    if (
        n_objects > 0
        and self.params.appearance_threshold != "auto"
        and not any(has_usable_embedding(obj) for _, objs in frames for obj in objs)
    ):
        raise ValueError(
            "deepocsort was given an explicit appearance_threshold "
            f"({self.params.appearance_threshold}) but none of the "
            f"{n_objects} detections to track carries a usable appearance "
            "vector (present, finite and non-zero). "
            "Run an embedding model first, or use --tracker ocsort (the same "
            "algorithm without appearance)."
        )
    super().prepare(frames)
    self._raw_term_notice()

DeepSORTTracker

Bases: BaseMOTTracker

DeepSORT tracker (see :mod:sleap_nn.tracking.mot.deepsort).

Attributes:

Name	Type	Description
appearance_statistic	str	Appearance statistic thresholded by the appearance stage ("gallery_max": DeepSORT's nearest neighbor in the gallery; "ema": StrongSORT's EMA feature). It also selects the calibrated gate under appearance_threshold="auto".

Source code in sleap_nn/tracking/mot/deepsort.py

class DeepSORTTracker(BaseMOTTracker):
    """DeepSORT tracker (see :mod:`sleap_nn.tracking.mot.deepsort`).

    Attributes:
        appearance_statistic: Appearance statistic thresholded by the
            appearance stage (``"gallery_max"``: DeepSORT's nearest neighbor in
            the gallery; ``"ema"``: StrongSORT's EMA feature). It also selects
            the calibrated gate under ``appearance_threshold="auto"``.
    """

    name = "deepsort"
    default_kalman = "xyah"
    uses_appearance = True
    requires_vectors = True
    tentative_dies_on_miss = True
    hits_start = 1

    appearance_statistic: ClassVar[str] = "gallery_max"

    # Set once the "appearance disabled" INFO message has been logged.
    _degraded_logged: bool = False

    def _score_cuts(self) -> List[Tuple[str, bool, str]]:
        """Base cuts, plus ``track_low_threshold``.

        A detection scoring at or below it is ignored (never matched, never
        starts a track).
        """
        return super()._score_cuts() + [
            (
                "track_low_threshold",
                True,
                "they are ignored (never matched, never start a track). Lower "
                "--track_low_threshold",
            )
        ]

    def _associate(self, dets: FrameDetections) -> AssociationResult:
        """Appearance stage over confirmed tracks, then a geometric stage.

        Args:
            dets: This frame's detections.

        Returns:
            The matches (with their geometric similarity); low-score detections
            are excluded from spawning.
        """
        p = self.params
        # Low-score detections are invisible (never matched, never spawn).
        ignored = np.asarray(dets.scores) <= p.track_low_threshold
        result = AssociationResult(no_spawn={int(j) for j in np.flatnonzero(ignored)})
        cols = np.flatnonzero(np.asarray(dets.valid, dtype=bool) & ~ignored)
        tracks = self._live()
        if not tracks or len(cols) == 0:
            return result

        # Geometric similarity of every live track's prediction to the candidates:
        # the geometric stage's score and every match's tracking score.
        sim = self._geometric_similarity(tracks, dets)[:, cols]
        row_of = {t.tid: i for i, t in enumerate(tracks)}
        free = np.ones(len(cols), dtype=bool)

        tentative = [t for t in tracks if t.state == TrackState.TENTATIVE]
        confirmed = [
            t for t in tracks if t.state in (TrackState.CONFIRMED, TrackState.LOST)
        ]
        gate = self._appearance_gate(self.appearance_statistic)
        if gate is None:
            self._log_degraded()
            pool, coasting = [], confirmed
        else:
            pool = [t for t in confirmed if _has_appearance(t)]
            coasting = [t for t in confirmed if not _has_appearance(t)]

        # Stage 1: appearance (cascade / global assignment) over confirmed tracks.
        matched = self._match_appearance(pool, dets, cols, gate, free) if pool else []
        matched_tids = {t.tid for t, _ in matched}

        # Stage 2: geometry over tentative tracks, the appearance stage's
        # unmatched tracks updated at the previous step and, when appearance
        # cannot decide for them, every coasting confirmed track.
        geo_pool = (
            tentative
            + [
                t
                for t in pool
                if t.tid not in matched_tids and t.time_since_update == 1
            ]
            + [t for t in coasting if t.time_since_update <= p.track_buffer]
        )
        geo_pool.sort(key=lambda t: t.tid)
        free_idx = np.flatnonzero(free)
        if geo_pool and len(free_idx):
            sub = sim[np.ix_([row_of[t.tid] for t in geo_pool], free_idx)]
            min_sim = p.min_match_similarity
            cost = np.where(sub >= min_sim, 1.0 - sub, np.inf)
            pairs, _, _ = linear_assignment(cost, max_cost=1.0 - min_sim)
            for r, c in pairs:
                matched.append((geo_pool[int(r)], int(free_idx[int(c)])))
                free[free_idx[int(c)]] = False

        for track, c in matched:
            result.matches.append(
                (track, int(cols[c]), float(sim[row_of[track.tid], c]))
            )
        return result

    # ------------------------------------------------------- appearance stage
    def _match_appearance(
        self,
        pool: List[MOTTrack],
        dets: FrameDetections,
        cols: np.ndarray,
        gate: float,
        free: np.ndarray,
    ) -> List[Tuple[MOTTrack, int]]:
        """DeepSORT matching cascade (most recently updated tracks first).

        Args:
            pool: Confirmed tracks with an appearance memory, in ``tid`` order.
            dets: This frame's detections.
            cols: Candidate detection indices (into ``dets``).
            gate: Minimum gallery-max cosine of a match.
            free: ``(len(cols),)`` availability of the candidates; matched
                candidates are set to False in place.

        Returns:
            ``(track, candidate position)`` per match (positions index ``cols``).
        """
        cost = self._appearance_cost(pool, dets, cols)
        max_cost = 1.0 - gate
        tsu = np.array([t.time_since_update for t in pool])
        matches: List[Tuple[MOTTrack, int]] = []
        # Levels 1..track_buffer; only the levels that hold tracks do anything.
        for level in np.unique(tsu):
            if not 1 <= level <= self.params.track_buffer:
                continue
            free_idx = np.flatnonzero(free)
            if len(free_idx) == 0:
                break
            rows = np.flatnonzero(tsu == level)
            pairs, _, _ = linear_assignment(
                cost[np.ix_(rows, free_idx)], max_cost=max_cost
            )
            for r, c in pairs:
                matches.append((pool[int(rows[int(r)])], int(free_idx[int(c)])))
                free[free_idx[int(c)]] = False
        return matches

    def _appearance_cost(
        self, pool: Sequence[MOTTrack], dets: FrameDetections, cols: np.ndarray
    ) -> np.ndarray:
        """Gated cosine distance ``1 - cos`` of ``pool`` to the candidates.

        A missing vector (either side) costs 1; a pair outside the Kalman gate
        costs ``INFEASIBLE_COST``.

        Args:
            pool: Tracks (rows).
            dets: This frame's detections.
            cols: Candidate detection indices (columns).

        Returns:
            ``(len(pool), len(cols))`` costs.
        """
        cos = self._appearance_cos(pool, dets, self.appearance_statistic)[:, cols]
        cost = np.where(np.isfinite(cos), 1.0 - cos, 1.0)
        cost[self._mahalanobis(pool, dets, cols) > GATING_THRESHOLD] = INFEASIBLE_COST
        return cost

    def _mahalanobis(
        self, pool: Sequence[MOTTrack], dets: FrameDetections, cols: np.ndarray
    ) -> np.ndarray:
        """Squared position-only Mahalanobis distances (un-scaled measurement noise).

        Args:
            pool: Tracks (rows), already predicted to this step.
            dets: This frame's detections.
            cols: Candidate detection indices (columns).

        Returns:
            ``(len(pool), len(cols))`` squared distances (``inf`` for non-finite
            boxes or states).
        """
        boxes = np.asarray(dets.boxes, dtype=np.float64)[cols]
        return self.kf.gating_distances(
            [t.mean for t in pool], [t.cov for t in pool], boxes, only_position=True
        )

    # ------------------------------------------------------------- lifecycle
    def _make_appearance_memory(self) -> AppearanceMemory:
        """A track memory whose gallery holds ``gallery_budget`` vectors (``nn_budget``)."""
        p = self.params
        return AppearanceMemory(
            ema_alpha=p.ema_alpha,
            gallery_size=p.gallery_budget,
            adaptive=p.adaptive_ema,
            score_floor=p.track_high_threshold,
        )

    def _log_degraded(self) -> None:
        """Log once that the appearance stage is off (geometric matching only)."""
        if self._degraded_logged:
            return
        self._degraded_logged = True
        reason = self._appearance_summary()["reason"]
        logger.info(
            f"{self.name}: appearance is disabled for association ({reason}); "
            "skipping the appearance stage and matching every confirmed track "
            "by geometry."
        )

MOTParams

Resolved parameters for one MOT tracker run.

Attributes:

Name	Type	Description
tracker	str	Tracker name, one of :data:MOT_TRACKER_NAMES.
preset	str	Preset the defaults came from ("animal" or "paper").
similarity	str	Geometric similarity between a track's prediction and a detection: "iou" (padded boxes), "oks" (keypoints vs the track's predicted pose), "centroid" (Gaussian on the box-center distance, scale body_size / 2) or "auto" (masks -> iou, single-node skeletons -> centroid, otherwise iou).
body_size	Optional[float]	Animal size in px (box diagonal), or None to measure it per video. It scales the centroid similarity (body_size / 2), sizes the pseudo boxes of single-node detections and floors thin boxes. None: the video's calibration (prepare()): the median box diagonal of multi-node poses / masks; single-node poses have no extent, so 8 x the per-step motion p99 (at least 16 px, at most the nearest-neighbor spacing), else the spacing, else 50 px; without prepare(), the same chain on the frames seen so far. Set it for centroid-only models whose calibration cannot size the animal.
oks_stddev	float	OKS per-node standard deviation for similarity="oks" (a finite number > 0; NaN and inf are refused). The MOT default (0.1) is 4x the evaluation default: tracking compares two noisy observations a frame apart, and at 0.025 an OKS threshold is ~7x stricter than the same IoU threshold.
box_padding	float	Padding added to each side of a keypoint/mask box, as a fraction of max(w, h).
score_mode	str	Detection confidence: "auto" (mean visible point score for predicted poses, score for masks, 1.0 for user instances), "instance" (the object's score) or "none" (all 1.0).
normalize_scores	bool	Divide every detection score by the video's 95th-percentile score (max(q95, 1e-6) of the calibration's raw score_quantiles) and clip to [0, 1] before any score threshold applies. SLEAP mean point scores depend on the model (q95 ~0.56 for one mouse model, ~0.97 for a fly model), so fixed thresholds tuned for pedestrian detectors (0.5-0.7) can silently block every track start; normalized, a typical detection of the video scores ~0.85-1.0 and a 1.0-scored (user) detection is unchanged. Needs prepare() (the calibration): without it scores are used as they are. animal preset: on; paper: off (the published thresholds on raw scores).
kalman	str	Kalman filter state parameterization; "auto" uses the tracker's own (xysr SORT/OC-SORT, xyah DeepSORT/StrongSORT/ ByteTrack, xywh BoT-SORT/Deep OC-SORT). The animal preset sets xywh for DeepSORT/StrongSORT/ByteTrack: xyah assumes a slowly varying aspect ratio, which a turning animal's box breaks.
time_base	str	How time advances between processed frames: "frames" (frame-index deltas, normalized by the video's stride: the most common delta), "steps" (one step per processed frame) or "auto" (frames when the processed frames are regularly spaced, else steps). prepare() resolves it over the whole video; without it the stride and regularity are inferred online from the deltas seen so far (same rule), so a strided stream counts one step per processed frame from its first gap on.
track_buffer	int	Steps a lost track is kept alive for re-association (max_age; 1 in SORT's paper preset, as published).
min_hits	int	Matches required before a track is confirmed (n_init). Detections of a track that is later confirmed are labeled retroactively.
track_high_threshold	float	Detections with a score above this are high-confidence (ByteTrack's track_thresh / OC-SORT's det_thresh); also the minimum score for appearance-memory updates.
track_low_threshold	float	Detections with a score at or below this are ignored.
new_track_threshold	float	Minimum score for an unmatched detection to start a track.
min_match_similarity	float	Minimum geometric similarity for a first-stage match (cost cap 1 - min_match_similarity).
low_stage_min_similarity	float	ByteTrack/BoT-SORT second (low-score) stage.
unconfirmed_min_similarity	float	ByteTrack/BoT-SORT unconfirmed-track stage.
duplicate_similarity	float	ByteTrack/BoT-SORT tracked-vs-lost duplicate removal.
fuse_score	bool	Multiply the geometric similarity by the detection score (ByteTrack fuse_score).
appearance_threshold	Union[float, str]	Minimum cosine similarity for appearance to support a match, or "auto" to calibrate per video (see :mod:sleap_nn.tracking.mot.calibration).
proximity_threshold	float	BoT-SORT: minimum geometric similarity for the appearance term to apply at all.
ema_alpha	float	Appearance EMA momentum (weight on the old feature).
adaptive_ema	bool	Deep OC-SORT score-adaptive EMA.
gallery_size	int	Per-track gallery of recent confident vectors.
gallery_budget	int	DeepSORT gallery budget (nn_budget).
mc_lambda	float	StrongSORT motion-cost weight on appearance (lambda).
nsa	bool	StrongSORT NSA Kalman (measurement noise scaled by detection score).
delta_t	int	OC-SORT observation lookback (steps) for velocity direction.
inertia	float	OC-SORT weight of the observation-centric momentum term.
use_byte	bool	OC-SORT optional BYTE second stage.
w_assoc_emb	float	Deep OC-SORT base appearance weight.
aw_param	float	Deep OC-SORT adaptive-weighting cap.
aw_off	bool	Disable Deep OC-SORT adaptive weighting (constant w_assoc_emb).
rescue	bool	Deep OC-SORT: accept a weak-geometry match when appearance is decisive (BoostTrack rule).
rescue_min_similarity	float	Minimum geometric similarity for a rescue.
link_tracklets	bool	Run identity linking (long-term memory) at the end of each video.
max_tracks	Optional[int]	Known number of animals; closed-set identity linking.
link_swap_correction	bool	Correct identity swaps at crossings during linking.
link_spatial_gate_frames	int	Gap (steps) above which linking applies no spatial gate (long absences can end anywhere).

Methods:

Name	Description
__attrs_post_init__	Validate cross-field constraints.
as_dict	Return the parameters as a JSON-serializable dict.
field_names	Names of every overridable field.
for_tracker	Resolve a tracker's preset and apply user overrides.

Source code in sleap_nn/tracking/mot/params.py

@attrs.frozen
class MOTParams:
    """Resolved parameters for one MOT tracker run.

    Attributes:
        tracker: Tracker name, one of :data:`MOT_TRACKER_NAMES`.
        preset: Preset the defaults came from (``"animal"`` or ``"paper"``).
        similarity: Geometric similarity between a track's prediction and a
            detection: ``"iou"`` (padded boxes), ``"oks"`` (keypoints vs the
            track's predicted pose), ``"centroid"`` (Gaussian on the box-center
            distance, scale ``body_size / 2``) or ``"auto"`` (masks -> iou,
            single-node skeletons -> centroid, otherwise iou).
        body_size: Animal size in px (box diagonal), or ``None`` to measure it
            per video. It scales the ``centroid`` similarity (``body_size / 2``),
            sizes the pseudo boxes of single-node detections and floors thin
            boxes. ``None``: the video's calibration (``prepare()``): the median
            box diagonal of multi-node poses / masks; single-node poses have no
            extent, so ``8 x`` the per-step motion p99 (at least 16 px, at most
            the nearest-neighbor spacing), else the spacing, else 50 px; without
            ``prepare()``, the same chain on the frames seen so far. Set it for
            centroid-only models whose calibration cannot size the animal.
        oks_stddev: OKS per-node standard deviation for ``similarity="oks"``
            (a finite number > 0; NaN and inf are refused). The
            MOT default (0.1) is 4x the evaluation default: tracking compares two
            noisy observations a frame apart, and at 0.025 an OKS threshold is
            ~7x stricter than the same IoU threshold.
        box_padding: Padding added to each side of a keypoint/mask box, as a
            fraction of ``max(w, h)``.
        score_mode: Detection confidence: ``"auto"`` (mean visible point score
            for predicted poses, ``score`` for masks, 1.0 for user instances),
            ``"instance"`` (the object's ``score``) or ``"none"`` (all 1.0).
        normalize_scores: Divide every detection score by the video's
            95th-percentile score (``max(q95, 1e-6)`` of the calibration's raw
            ``score_quantiles``) and clip to ``[0, 1]`` before any score
            threshold applies. SLEAP mean point scores depend on the model (q95
            ~0.56 for one mouse model, ~0.97 for a fly model), so fixed
            thresholds tuned for pedestrian detectors (0.5-0.7) can silently
            block every track start; normalized, a typical detection of the
            video scores ~0.85-1.0 and a 1.0-scored (user) detection is
            unchanged. Needs ``prepare()`` (the calibration): without it scores
            are used as they are. ``animal`` preset: on; ``paper``: off (the
            published thresholds on raw scores).
        kalman: Kalman filter state parameterization; ``"auto"`` uses the
            tracker's own (``xysr`` SORT/OC-SORT, ``xyah`` DeepSORT/StrongSORT/
            ByteTrack, ``xywh`` BoT-SORT/Deep OC-SORT). The ``animal`` preset
            sets ``xywh`` for DeepSORT/StrongSORT/ByteTrack: ``xyah`` assumes a
            slowly varying aspect ratio, which a turning animal's box breaks.
        time_base: How time advances between processed frames: ``"frames"``
            (frame-index deltas, normalized by the video's stride: the most
            common delta), ``"steps"`` (one step per processed frame) or
            ``"auto"`` (frames when the processed frames are regularly spaced,
            else steps). ``prepare()`` resolves it over the whole video; without
            it the stride and regularity are inferred online from the deltas
            seen so far (same rule), so a strided stream counts one step per
            processed frame from its first gap on.
        track_buffer: Steps a lost track is kept alive for re-association
            (``max_age``; 1 in SORT's ``paper`` preset, as published).
        min_hits: Matches required before a track is confirmed (``n_init``).
            Detections of a track that is later confirmed are labeled
            retroactively.
        track_high_threshold: Detections with a score above this are
            high-confidence (ByteTrack's ``track_thresh`` / OC-SORT's
            ``det_thresh``); also the minimum score for appearance-memory updates.
        track_low_threshold: Detections with a score at or below this are ignored.
        new_track_threshold: Minimum score for an unmatched detection to start a
            track.
        min_match_similarity: Minimum geometric similarity for a first-stage
            match (cost cap ``1 - min_match_similarity``).
        low_stage_min_similarity: ByteTrack/BoT-SORT second (low-score) stage.
        unconfirmed_min_similarity: ByteTrack/BoT-SORT unconfirmed-track stage.
        duplicate_similarity: ByteTrack/BoT-SORT tracked-vs-lost duplicate removal.
        fuse_score: Multiply the geometric similarity by the detection score
            (ByteTrack ``fuse_score``).
        appearance_threshold: Minimum cosine similarity for appearance to support
            a match, or ``"auto"`` to calibrate per video (see
            :mod:`sleap_nn.tracking.mot.calibration`).
        proximity_threshold: BoT-SORT: minimum geometric similarity for the
            appearance term to apply at all.
        ema_alpha: Appearance EMA momentum (weight on the old feature).
        adaptive_ema: Deep OC-SORT score-adaptive EMA.
        gallery_size: Per-track gallery of recent confident vectors.
        gallery_budget: DeepSORT gallery budget (``nn_budget``).
        mc_lambda: StrongSORT motion-cost weight on appearance (``lambda``).
        nsa: StrongSORT NSA Kalman (measurement noise scaled by detection score).
        delta_t: OC-SORT observation lookback (steps) for velocity direction.
        inertia: OC-SORT weight of the observation-centric momentum term.
        use_byte: OC-SORT optional BYTE second stage.
        w_assoc_emb: Deep OC-SORT base appearance weight.
        aw_param: Deep OC-SORT adaptive-weighting cap.
        aw_off: Disable Deep OC-SORT adaptive weighting (constant ``w_assoc_emb``).
        rescue: Deep OC-SORT: accept a weak-geometry match when appearance is
            decisive (BoostTrack rule).
        rescue_min_similarity: Minimum geometric similarity for a rescue.
        link_tracklets: Run identity linking (long-term memory) at the end of
            each video.
        max_tracks: Known number of animals; closed-set identity linking.
        link_swap_correction: Correct identity swaps at crossings during linking.
        link_spatial_gate_frames: Gap (steps) above which linking applies no
            spatial gate (long absences can end anywhere).
    """

    tracker: str = attrs.field(validator=_check_choice(MOT_TRACKER_NAMES))
    preset: str = attrs.field(default="animal", validator=_check_choice(PRESETS))
    similarity: str = attrs.field(
        default="auto", validator=_check_choice(SIMILARITY_KINDS)
    )
    body_size: Optional[float] = attrs.field(
        default=None, converter=_optional_float, validator=_check_positive_or_none
    )
    oks_stddev: float = 0.1
    box_padding: float = 0.1
    score_mode: str = attrs.field(default="auto", validator=_check_choice(SCORE_MODES))
    normalize_scores: bool = False
    kalman: str = attrs.field(default="auto", validator=_check_choice(KALMAN_KINDS))
    time_base: str = attrs.field(default="auto", validator=_check_choice(TIME_BASES))
    track_buffer: int = 30
    min_hits: int = 3
    track_high_threshold: float = attrs.field(
        default=0.5, validator=_check_unit_interval
    )
    track_low_threshold: float = attrs.field(
        default=0.1, validator=_check_unit_interval
    )
    new_track_threshold: float = attrs.field(
        default=0.6, validator=_check_unit_interval
    )
    min_match_similarity: float = attrs.field(
        default=0.3, validator=_check_unit_interval
    )
    low_stage_min_similarity: float = attrs.field(
        default=0.5, validator=_check_unit_interval
    )
    unconfirmed_min_similarity: float = attrs.field(
        default=0.3, validator=_check_unit_interval
    )
    duplicate_similarity: float = attrs.field(
        default=0.85, validator=_check_unit_interval
    )
    fuse_score: bool = False
    appearance_threshold: Union[float, str] = attrs.field(
        default="auto", converter=_convert_appearance_threshold
    )
    proximity_threshold: float = attrs.field(
        default=0.5, validator=_check_unit_interval
    )
    ema_alpha: float = attrs.field(default=0.8, validator=_check_unit_interval)
    adaptive_ema: bool = False
    gallery_size: int = 30
    gallery_budget: int = 100
    mc_lambda: float = attrs.field(default=0.98, validator=_check_unit_interval)
    nsa: bool = False
    delta_t: int = 3
    inertia: float = 0.2
    use_byte: bool = False
    w_assoc_emb: float = 0.75
    aw_param: float = 0.5
    aw_off: bool = False
    rescue: bool = True
    rescue_min_similarity: float = attrs.field(
        default=0.15, validator=_check_unit_interval
    )
    link_tracklets: bool = False
    max_tracks: Optional[int] = None
    link_swap_correction: bool = True
    link_spatial_gate_frames: int = 300

    def __attrs_post_init__(self):
        """Validate cross-field constraints."""
        if self.track_buffer < 0:
            raise ValueError(f"track_buffer must be >= 0, got {self.track_buffer}.")
        if self.min_hits < 1:
            raise ValueError(f"min_hits must be >= 1, got {self.min_hits}.")
        try:
            oks_ok = math.isfinite(self.oks_stddev) and self.oks_stddev > 0
        except TypeError:
            oks_ok = False
        if not oks_ok:
            raise ValueError(
                f"oks_stddev must be a finite number > 0, got {self.oks_stddev!r}."
            )
        if self.gallery_size < 1 or self.gallery_budget < 1:
            raise ValueError("gallery_size and gallery_budget must be >= 1.")
        if self.delta_t < 1:
            raise ValueError(f"delta_t must be >= 1, got {self.delta_t}.")
        if self.max_tracks is not None and self.max_tracks < 1:
            raise ValueError(f"max_tracks must be >= 1, got {self.max_tracks}.")
        if self.track_low_threshold > self.track_high_threshold:
            raise ValueError(
                "track_low_threshold must be <= track_high_threshold, got "
                f"{self.track_low_threshold} > {self.track_high_threshold}."
            )

    @property
    def link(self) -> bool:
        """Whether identity linking runs at the end of each video.

        Linking runs when requested explicitly, and always when the animal count
        is known: ``max_tracks`` is enforced by closed-set linking, never by
        refusing to start tracks online.
        """
        return self.link_tracklets or self.max_tracks is not None

    @classmethod
    def field_names(cls) -> tuple:
        """Names of every overridable field."""
        return tuple(a.name for a in attrs.fields(cls))

    @classmethod
    def for_tracker(
        cls, tracker: str, preset: str = "animal", **overrides: Any
    ) -> "MOTParams":
        """Resolve a tracker's preset and apply user overrides.

        Args:
            tracker: One of :data:`MOT_TRACKER_NAMES`.
            preset: ``"animal"`` (default) or ``"paper"``.
            **overrides: Field values that replace the preset's. ``None`` values
                are ignored (they mean "use the preset"), so callers can pass
                optional config fields straight through.

        Returns:
            The resolved, validated :class:`MOTParams`.

        Raises:
            ValueError: Unknown tracker, preset or override key, or an invalid
                value.
        """
        if tracker not in MOT_TRACKER_NAMES:
            raise ValueError(
                f"Unknown MOT tracker {tracker!r}; expected one of "
                f"{list(MOT_TRACKER_NAMES)}."
            )
        if preset not in PRESETS:
            raise ValueError(
                f"Unknown preset {preset!r}; expected one of {list(PRESETS)}."
            )
        valid = set(cls.field_names()) - {"tracker", "preset"}
        unknown = sorted(set(overrides) - valid)
        if unknown:
            raise ValueError(
                f"Unknown MOT parameter(s) {unknown}; valid parameters are "
                f"{sorted(valid)}."
            )
        values: Dict[str, Any] = dict(_PAPER_COMMON)
        values.update(_PAPER[tracker])
        if preset == "animal":
            values.update(_ANIMAL_COMMON)
            values.update(_ANIMAL.get(tracker, {}))
        values.update({k: v for k, v in overrides.items() if v is not None})
        return cls(tracker=tracker, preset=preset, **values)

    def as_dict(self) -> Dict[str, Any]:
        """Return the parameters as a JSON-serializable dict."""
        return attrs.asdict(self)

link property

Whether identity linking runs at the end of each video.

Linking runs when requested explicitly, and always when the animal count is known: max_tracks is enforced by closed-set linking, never by refusing to start tracks online.

__attrs_post_init__()

Validate cross-field constraints.

Source code in sleap_nn/tracking/mot/params.py

def __attrs_post_init__(self):
    """Validate cross-field constraints."""
    if self.track_buffer < 0:
        raise ValueError(f"track_buffer must be >= 0, got {self.track_buffer}.")
    if self.min_hits < 1:
        raise ValueError(f"min_hits must be >= 1, got {self.min_hits}.")
    try:
        oks_ok = math.isfinite(self.oks_stddev) and self.oks_stddev > 0
    except TypeError:
        oks_ok = False
    if not oks_ok:
        raise ValueError(
            f"oks_stddev must be a finite number > 0, got {self.oks_stddev!r}."
        )
    if self.gallery_size < 1 or self.gallery_budget < 1:
        raise ValueError("gallery_size and gallery_budget must be >= 1.")
    if self.delta_t < 1:
        raise ValueError(f"delta_t must be >= 1, got {self.delta_t}.")
    if self.max_tracks is not None and self.max_tracks < 1:
        raise ValueError(f"max_tracks must be >= 1, got {self.max_tracks}.")
    if self.track_low_threshold > self.track_high_threshold:
        raise ValueError(
            "track_low_threshold must be <= track_high_threshold, got "
            f"{self.track_low_threshold} > {self.track_high_threshold}."
        )

as_dict()

Return the parameters as a JSON-serializable dict.

Source code in sleap_nn/tracking/mot/params.py

def as_dict(self) -> Dict[str, Any]:
    """Return the parameters as a JSON-serializable dict."""
    return attrs.asdict(self)

field_names() classmethod

Names of every overridable field.

Source code in sleap_nn/tracking/mot/params.py

@classmethod
def field_names(cls) -> tuple:
    """Names of every overridable field."""
    return tuple(a.name for a in attrs.fields(cls))

for_tracker(tracker, preset='animal', **overrides) classmethod

Resolve a tracker's preset and apply user overrides.

Parameters:

Name	Type	Description	Default
tracker	str	One of :data:MOT_TRACKER_NAMES.	required
preset	str	"animal" (default) or "paper".	'animal'
**overrides	Any	Field values that replace the preset's. None values are ignored (they mean "use the preset"), so callers can pass optional config fields straight through.	{}

Returns:

Type	Description
MOTParams	The resolved, validated :class:MOTParams.

Raises:

Type	Description
ValueError	Unknown tracker, preset or override key, or an invalid value.

Source code in sleap_nn/tracking/mot/params.py

@classmethod
def for_tracker(
    cls, tracker: str, preset: str = "animal", **overrides: Any
) -> "MOTParams":
    """Resolve a tracker's preset and apply user overrides.

    Args:
        tracker: One of :data:`MOT_TRACKER_NAMES`.
        preset: ``"animal"`` (default) or ``"paper"``.
        **overrides: Field values that replace the preset's. ``None`` values
            are ignored (they mean "use the preset"), so callers can pass
            optional config fields straight through.

    Returns:
        The resolved, validated :class:`MOTParams`.

    Raises:
        ValueError: Unknown tracker, preset or override key, or an invalid
            value.
    """
    if tracker not in MOT_TRACKER_NAMES:
        raise ValueError(
            f"Unknown MOT tracker {tracker!r}; expected one of "
            f"{list(MOT_TRACKER_NAMES)}."
        )
    if preset not in PRESETS:
        raise ValueError(
            f"Unknown preset {preset!r}; expected one of {list(PRESETS)}."
        )
    valid = set(cls.field_names()) - {"tracker", "preset"}
    unknown = sorted(set(overrides) - valid)
    if unknown:
        raise ValueError(
            f"Unknown MOT parameter(s) {unknown}; valid parameters are "
            f"{sorted(valid)}."
        )
    values: Dict[str, Any] = dict(_PAPER_COMMON)
    values.update(_PAPER[tracker])
    if preset == "animal":
        values.update(_ANIMAL_COMMON)
        values.update(_ANIMAL.get(tracker, {}))
    values.update({k: v for k, v in overrides.items() if v is not None})
    return cls(tracker=tracker, preset=preset, **values)

OCSORTTracker

Bases: BaseMOTTracker

OC-SORT tracker (see :mod:sleap_nn.tracking.mot.ocsort).

Source code in sleap_nn/tracking/mot/ocsort.py

class OCSORTTracker(BaseMOTTracker):
    """OC-SORT tracker (see :mod:`sleap_nn.tracking.mot.ocsort`)."""

    name = "ocsort"
    default_kalman = "xysr"
    tentative_dies_on_miss = False
    hits_start = 0
    confirm_on_streak = True
    use_oru = True
    # OCM reads `velocity` / `observations` (the base keeps them on request).
    keeps_observation_history = True

    # Whether the optional BYTE (low-score) stage is available to this family.
    _supports_byte = True

    def _score_cuts(self) -> List[Tuple[str, bool, str]]:
        """Base cuts, plus the STRICT high-score split.

        Only ``s > track_high_threshold`` (``D_hi``) may start a track (and,
        without the optional BYTE stage, be matched at all).
        """
        return super()._score_cuts() + [
            (
                "track_high_threshold",
                True,
                "only detections scoring above it can start a track, so most "
                "detections can never start one. Lower --track_high_threshold",
            )
        ]

    def _associate(self, dets: FrameDetections) -> AssociationResult:
        """Stage 1 (geometry-gated, OCM-ordered), optional BYTE, then OCR.

        Args:
            dets: This frame's detections.

        Returns:
            The matches (each with its geometric similarity); every detection
            outside ``D_hi`` is excluded from spawning.
        """
        p = self.params
        scores = np.asarray(dets.scores, dtype=np.float64)
        usable = np.asarray(dets.valid, dtype=bool) & np.isfinite(scores)
        high = usable & (scores > p.track_high_threshold)
        low = (
            usable
            & (scores > p.track_low_threshold)
            & (scores <= p.track_high_threshold)
        )
        result = AssociationResult(
            no_spawn={int(j) for j in np.flatnonzero(~high)},
        )
        self._appearance_notice(dets)

        tracks = self._live()
        hi_idx = np.flatnonzero(high)
        if not tracks or (len(hi_idx) == 0 and not self._byte_enabled()):
            return result
        sim_all = self._geometric_similarity(tracks, dets)

        # Stage 1: every live track vs D_hi.
        free_rows = np.arange(len(tracks))
        free_hi = hi_idx
        if len(hi_idx):
            sim = sim_all[:, hi_idx]
            valid, score = self._first_stage(tracks, dets, hi_idx, sim)
            pairs = one_to_one_pattern(valid)
            if pairs is None:
                pairs, _, _ = linear_assignment_masked(score, valid)
            free_rows, free_hi = self._record(
                result, tracks, free_rows, hi_idx, pairs, sim
            )

        # BYTE: remaining tracks' predictions vs D_lo (before OCR).
        lo_idx = np.flatnonzero(low)
        if self._byte_enabled() and len(free_rows) and len(lo_idx):
            sim = sim_all[np.ix_(free_rows, lo_idx)]
            pairs, _, _ = linear_assignment_masked(sim, sim >= p.min_match_similarity)
            free_rows, _ = self._record(result, tracks, free_rows, lo_idx, pairs, sim)

        # OCR: remaining D_hi vs remaining tracks' last observations.
        if len(free_rows) and len(free_hi):
            remaining = [tracks[int(i)] for i in free_rows]
            sim = self._last_observation_similarity(remaining, dets)[:, free_hi]
            pairs, _, _ = linear_assignment_masked(sim, sim >= p.min_match_similarity)
            self._record(result, tracks, free_rows, free_hi, pairs, sim)
        return result

    # ------------------------------------------------------------ stages
    def _byte_enabled(self) -> bool:
        """Whether the BYTE (low-score) stage runs."""
        return self._supports_byte and bool(self.params.use_byte)

    def _first_stage(
        self,
        tracks: Sequence[MOTTrack],
        dets: FrameDetections,
        cols: np.ndarray,
        sim: np.ndarray,
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Admissibility and objective of stage 1.

        Args:
            tracks: The live tracks (rows).
            dets: This frame's detections.
            cols: Detection indices of the columns (``D_hi``).
            sim: ``(n_tracks, len(cols))`` geometric similarity of the
                predictions.

        Returns:
            ``(valid, score)``: admissibility on geometry alone and the
            objective ``sim + OCM`` the assignment maximizes.
        """
        valid = sim >= self.params.min_match_similarity
        return valid, sim + self._momentum(tracks, dets, cols)

    def _momentum(
        self, tracks: Sequence[MOTTrack], dets: FrameDetections, cols: np.ndarray
    ) -> np.ndarray:
        """``(n_tracks, len(cols))`` observation-centric momentum (OCM) term.

        A track contributes only once it has a velocity direction (two
        observations) and an earlier observation to measure the direction to
        each detection from.
        """
        n = len(tracks)
        dirs = np.zeros((n, 2), dtype=np.float64)
        prev = np.full((n, 4), np.nan, dtype=np.float64)
        has_dir = np.zeros(n, dtype=bool)
        for i, track in enumerate(tracks):
            if track.velocity is None:
                continue
            obs = self._previous_observation(track)
            if obs is None:
                continue
            dirs[i] = np.asarray(track.velocity, dtype=np.float64).reshape(2)
            prev[i] = np.asarray(obs, dtype=np.float64).reshape(4)
            has_dir[i] = True
        return ocm_matrix(
            dirs,
            prev,
            has_dir,
            np.asarray(dets.boxes)[cols],
            np.asarray(dets.scores)[cols],
            self.params.inertia,
        )

    def _previous_observation(self, track: MOTTrack) -> Optional[np.ndarray]:
        """The track's observation ``delta_t`` steps back, for the OCM direction.

        The oldest observation within the last ``delta_t`` steps, else the latest
        observation (``None`` if the track has none), as in the reference.
        """
        for dt in range(self.params.delta_t, 0, -1):
            obs = track.observations.get(self._step - dt)
            if obs is not None:
                return obs
        return track.last_observation

    @staticmethod
    def _record(
        result: AssociationResult,
        tracks: Sequence[MOTTrack],
        rows: np.ndarray,
        cols: np.ndarray,
        pairs: np.ndarray,
        sim: np.ndarray,
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Append a stage's matches and return the still-unmatched rows/columns.

        Args:
            result: The frame's association result (appended to).
            tracks: Every live track; ``rows`` index into it.
            rows: Track indices of the stage's rows.
            cols: Detection indices of the stage's columns.
            pairs: ``(k, 2)`` matched ``(row, col)`` positions in the stage.
            sim: The stage's geometric similarity (the tracking score).

        Returns:
            ``(free_rows, free_cols)``: the unmatched track and detection indices.
        """
        pairs = np.asarray(pairs, dtype=np.int64).reshape(-1, 2)
        for r, c in pairs:
            result.matches.append(
                (tracks[int(rows[r])], int(cols[c]), float(sim[r, c]))
            )
        free_rows = np.setdiff1d(np.arange(len(rows)), pairs[:, 0])
        free_cols = np.setdiff1d(np.arange(len(cols)), pairs[:, 1])
        return np.asarray(rows)[free_rows], np.asarray(cols)[free_cols]

    def _appearance_notice(self, dets: FrameDetections) -> None:
        """Hook for appearance trackers to log once (OC-SORT: nothing)."""

SORTTracker

Bases: BaseMOTTracker

SORT tracker (see :mod:sleap_nn.tracking.mot.sort).

Source code in sleap_nn/tracking/mot/sort.py

class SORTTracker(BaseMOTTracker):
    """SORT tracker (see :mod:`sleap_nn.tracking.mot.sort`)."""

    name = "sort"
    default_kalman = "xysr"
    tentative_dies_on_miss = False
    hits_start = 0
    confirm_on_streak = True

    def _score_cuts(self) -> List[Tuple[str, bool, str]]:
        """Base cuts, plus ``track_low_threshold``.

        A detection scoring at or below it is ignored (never matched, never
        starts a track).
        """
        return super()._score_cuts() + [
            (
                "track_low_threshold",
                True,
                "they are ignored (never matched, never start a track). Lower "
                "--track_low_threshold",
            )
        ]

    def _associate(self, dets: FrameDetections) -> AssociationResult:
        """Single-pool, single-stage assignment on geometric similarity.

        Args:
            dets: This frame's detections.

        Returns:
            The matches; low-score detections are excluded from spawning.
        """
        p = self.params
        # Low-score detections are invisible to SORT (never matched, never spawn).
        ignored = np.asarray(dets.scores) <= p.track_low_threshold
        result = AssociationResult(no_spawn={int(j) for j in np.flatnonzero(ignored)})
        candidates = np.flatnonzero(np.asarray(dets.valid, dtype=bool) & ~ignored)
        tracks = self._live()
        if not tracks or len(candidates) == 0:
            return result

        sim = self._geometric_similarity(tracks, dets)[:, candidates]
        valid = sim >= p.min_match_similarity
        pairs = one_to_one_pattern(valid)
        if pairs is None:
            # `1 - sim` is monotone, so the admissible set is exactly `valid`.
            cost = np.where(valid, 1.0 - sim, np.inf)
            pairs, _, _ = linear_assignment(cost, max_cost=1.0 - p.min_match_similarity)
        for row, col in pairs:
            result.matches.append(
                (tracks[int(row)], int(candidates[int(col)]), float(sim[row, col]))
            )
        return result

StrongSORTTracker

Bases: DeepSORTTracker

StrongSORT tracker (see :mod:sleap_nn.tracking.mot.deepsort).

Source code in sleap_nn/tracking/mot/deepsort.py

class StrongSORTTracker(DeepSORTTracker):
    """StrongSORT tracker (see :mod:`sleap_nn.tracking.mot.deepsort`)."""

    name = "strongsort"
    appearance_statistic: ClassVar[str] = "ema"

    def _match_appearance(
        self,
        pool: List[MOTTrack],
        dets: FrameDetections,
        cols: np.ndarray,
        gate: float,
        free: np.ndarray,
    ) -> List[Tuple[MOTTrack, int]]:
        """One global assignment on the motion-blended (MC) cost.

        Args:
            pool: Confirmed tracks with an appearance memory, in ``tid`` order.
            dets: This frame's detections.
            cols: Candidate detection indices (into ``dets``).
            gate: Minimum EMA cosine of a match (before the MC allowance).
            free: ``(len(cols),)`` availability of the candidates; matched
                candidates are set to False in place.

        Returns:
            ``(track, candidate position)`` per match (positions index ``cols``).
        """
        free_idx = np.flatnonzero(free)
        if len(free_idx) == 0:
            return []
        lam = float(self.params.mc_lambda)
        max_cost = 1.0 - gate + (MC_COST_ALLOWANCE if lam < 1.0 else 0.0)
        cost = self._mc_cost(pool, dets, cols[free_idx])
        pairs, _, _ = linear_assignment(cost, max_cost=max_cost)
        matches = []
        for r, c in pairs:
            matches.append((pool[int(r)], int(free_idx[int(c)])))
            free[free_idx[int(c)]] = False
        return matches

    def _mc_cost(
        self, pool: Sequence[MOTTrack], dets: FrameDetections, cols: np.ndarray
    ) -> np.ndarray:
        """StrongSORT motion-blended cost ``lambda * d_app + (1 - lambda) * d^2``.

        ``d_app = 1 - EMA cosine`` (1 when a vector is missing) and ``d^2`` is the
        raw squared position-only Mahalanobis distance. Pairs with ``d^2`` above
        the gate get ``d_app = INFEASIBLE_COST`` BEFORE the blend.

        Args:
            pool: Tracks (rows).
            dets: This frame's detections.
            cols: Candidate detection indices (columns).

        Returns:
            ``(len(pool), len(cols))`` costs.
        """
        cos = self._appearance_cos(pool, dets, self.appearance_statistic)[:, cols]
        d_app = np.where(np.isfinite(cos), 1.0 - cos, 1.0)
        d2 = self._mahalanobis(pool, dets, cols)
        # A non-finite distance (invalid state) is simply out of the gate.
        d2 = np.where(np.isfinite(d2), d2, INFEASIBLE_COST)
        d_app[d2 > GATING_THRESHOLD] = INFEASIBLE_COST
        lam = float(self.params.mc_lambda)
        return lam * d_app + (1.0 - lam) * d2

    def _make_appearance_memory(self) -> AppearanceMemory:
        """The base memory (EMA statistic; the gallery only feeds the quality gate)."""
        return BaseMOTTracker._make_appearance_memory(self)

build_mot_tracker(params, carrier='pose', n_nodes=None)

Instantiate the tracker named by params.tracker.

Parameters:

Name	Type	Description	Default
params	MOTParams	Resolved parameters (see :meth:MOTParams.for_tracker).	required
carrier	str	"pose" or "mask".	'pose'
n_nodes	Optional[int]	Skeleton size for pose detections (None if unknown).	None

Returns:

Type	Description
BaseMOTTracker	A fresh tracker for one video.

Source code in sleap_nn/tracking/mot/__init__.py

def build_mot_tracker(
    params: MOTParams, carrier: str = "pose", n_nodes: Optional[int] = None
) -> BaseMOTTracker:
    """Instantiate the tracker named by ``params.tracker``.

    Args:
        params: Resolved parameters (see :meth:`MOTParams.for_tracker`).
        carrier: ``"pose"`` or ``"mask"``.
        n_nodes: Skeleton size for pose detections (``None`` if unknown).

    Returns:
        A fresh tracker for one video.
    """
    return MOT_TRACKERS[params.tracker](params, carrier=carrier, n_nodes=n_nodes)