linking

sleap_nn.tracking.mot.linking

Identity linking: object-level long-term memory for tracked videos.

:func:link_tracklets runs once per video AFTER tracking (any tracker, the classic one included). It treats every track as an identity made of tracklets and:

1. Corrects swaps at crossings (needs calibrated appearance): around each ambiguity event (two or more identities' boxes overlapping), it compares each identity's appearance gallery from before the event with the vectors right after it, and applies a permutation only when appearance prefers it by the calibrated margin.
2. Links tracklets across gaps (re-identification): an identity that ends and one that starts later, that never co-occur (cannot-link) and are spatially plausible given the gap, are merged when their appearance agrees (calibrated threshold and margin), or -- without usable appearance -- when the match is spatially unique within a short gap (an identity's own later fragments, reachable through the candidate, do not count as competitors).
3. Enforces a closed set when the number of animals N is known: the N best-supported identities that are present together become anchors (one per animal), and every other tracklet is assigned to a compatible anchor -- one it never co-occurs with, apart from frames where it merely duplicates the anchor's detection -- by spatial continuity and appearance. Only duplicate detections and tracklets that fit no anchor are untracked; a real fragment is never dropped because some other, short false-positive identity happened to co-occur with it.

Every vector already exists when tracking runs in sleap-nn (offline flow), so linking decides with full tracklets on both sides of every gap -- more evidence than an online re-ID decision at track confirmation could use.

Conventions used throughout:

• Identity: one distinct sio.Track object among the tracked detections.
• Tracklet: a maximal run of an identity's detections with no gap longer than track_buffer steps.
• Time is counted in steps, exactly as the trackers count them: the calibration's resolved time_base and stride through :func:~sleap_nn.tracking.mot.calibration.frame_steps (frame-index deltas divided by frame_step in the "frames" time base, one step per processed frame in "steps").
• Gated vector: the appearance vector of a detection that is confident (score >= high_threshold), mostly visible (at least half of its nodes) and isolated (padded-box IoU below 0.1 against every other detection of its frame). Only gated vectors enter galleries: crops of overlapping animals are contaminated.
• Linking statistic s_link(Q, G): the mean over query vectors q of the mean of the top-5 cosines of q against gallery G -- the statistic :func:~sleap_nn.tracking.mot.calibration.calibrate calibrates the linking rule on, for every query length (link_thresh_by_len): a decision on an n-vector query uses the threshold and margin of n vectors.
• Cannot-link: two identities whose temporal extents overlap are never linked by the open-set rules (an identity is linked only to one that starts after it ends); identities that share a frame are different animals for certain, which is what makes them the competitors that appearance margins are measured against.
• Duplicate: two same-frame detections whose raw boxes overlap by at least DUPLICATE_IOMIN of the smaller one, or whose centroids are closer than DUPLICATE_CENTROID_BODY body sizes -- one animal detected twice.

Scaling: the open set keeps a sparse set of candidate pairs per identity (every pair inside the spatial-only window plus the best appearance candidates), not dense identity x identity matrices, and the closed set is linear in the number of tracklets times N.

Classes:

Name	Description
LinkingResult	Outcome of :func:link_tracklets for one video.

Functions:

Name	Description
link_tracklets	Link a tracked video's identities across gaps and crossings, in place.

LinkingResult

Outcome of :func:link_tracklets for one video.

Attributes:

Name	Type	Description
n_identities_before	int	Identities (tracks) before linking.
n_identities	int	Identities after linking.
n_merges	int	Gap links applied (appearance or spatial).
n_forced_merges	int	Closed-set assignments of a tracklet to an anchor.
n_swaps	int	Swap corrections applied.
n_untracked	int	Detections set to untracked by the closed-set rule.
notes	List[str]	Human-readable decisions worth logging.

Methods:

Name	Description
as_dict	JSON-serializable dict.

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

@attrs.define
class LinkingResult:
    """Outcome of :func:`link_tracklets` for one video.

    Attributes:
        n_identities_before: Identities (tracks) before linking.
        n_identities: Identities after linking.
        n_merges: Gap links applied (appearance or spatial).
        n_forced_merges: Closed-set assignments of a tracklet to an anchor.
        n_swaps: Swap corrections applied.
        n_untracked: Detections set to untracked by the closed-set rule.
        notes: Human-readable decisions worth logging.
    """

    n_identities_before: int = 0
    n_identities: int = 0
    n_merges: int = 0
    n_forced_merges: int = 0
    n_swaps: int = 0
    n_untracked: int = 0
    notes: List[str] = attrs.field(factory=list)

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

as_dict()

JSON-serializable dict.

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

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

link_tracklets(frames, *, carrier='pose', calibration=None, max_tracks=None, track_buffer=30, swap_correction=True, spatial_gate_frames=300, box_padding=0.1, high_threshold=0.5, score_mode='auto', score_normalizer=None, time_base='auto', body_size=None, track_objects=None)

Link a tracked video's identities across gaps and crossings, in place.

Steps, in order (see the module docstring for the definitions):

1. Swap correction (when swap_correction and the calibration reports appearance_for_linking): for each ambiguity event over at most MAX_EVENT_IDENTITIES identities, the post-event vectors of each identity are scored against every identity's pre-event gallery; the best permutation is applied to the moved identities' detections after the event when its total gain is at least the calibrated margin per moved identity and every moved pair clears the calibrated threshold and beats its unswapped pair by the margin (the rule of each query's own length).
2. Gap linking, greedily best-first (appearance before spatial, score desc, gap asc, start asc, name asc) until nothing is accepted, between an identity that ends and one that starts later within the spatial gate body + 3 * s * (gap if gap <= 10 else sqrt(10 * gap)) with s the calibrated step_p95 floored at 0.05 body per step (no gate beyond spatial_gate_frames steps or when step_p95 is unknown). Pairs with gated vectors on both sides (and calibrated appearance) are decided by appearance (the threshold for the later identity's head length, then the margin over the competing candidates that provably are other animals); others are linked only when the gap is at most 4 track buffers and the pair is unique on that window's frontier (a candidate that starts after the nearest one ends can follow through it, so it is no competitor). A link that would skip over another candidate of the same identity waits for it (so that fragment is not stranded inside the merged identity's gap).
3. Closed set (max_tracks): while more identities remain, the max_tracks best-supported identities present together become anchors; every other tracklet, most constrained first (a single viable anchor before a choice; decisive choices before close calls), joins the anchor it never co-occurs with (frames where it only duplicates the anchor's detection are resolved: the higher-scoring detection stays), ranked by appearance agreement, then a gap-aware geometric cost of its junctions (gated first; past the gate only when appearance agrees, or for a tracklet longer than track_buffer that appearance does not contradict). A tracklet that fits no anchor is untracked.

Parameters:

Name	Type	Description	Default
frames	Sequence[Tuple[int, Sequence[Any]]]	(frame_idx, objects) per frame in temporal order: the tracked detections (.track set or None) of ONE video.	required
carrier	str	"pose" or "mask".	'pose'
calibration	Optional[VideoCalibration]	The video's calibration (computes one from frames if None, with time_base and body_size). Appearance is used only where it reports appearance_for_linking. Its resolved time_base and stride count the steps.	None
max_tracks	Optional[int]	Known number of animals: closed-set linking.	None
track_buffer	int	Gap (steps) below which a track's detections form one tracklet; also scales the spatial-only rule.	30
swap_correction	bool	Correct swaps at crossings (appearance only).	True
spatial_gate_frames	int	Gaps longer than this (steps) are not spatially gated.	300
box_padding	float	Box padding used for overlap tests.	0.1
high_threshold	float	Minimum score for a vector to enter a gallery.	0.5
score_mode	str	Detection score mode (see MOTParams.score_mode).	'auto'
score_normalizer	Optional[float]	The tracker's per-video score normalizer (MOTParams.normalize_scores): scores are divided by it and clipped to [0, 1] before the high_threshold gate, so the gate sees the scores the tracker saw. None: raw scores.	None
time_base	str	MOTParams.time_base, used when calibration is None (a given calibration carries its resolved time base).	'auto'
body_size	Optional[float]	A known body size (px, MOTParams.body_size): pseudo-box size, duplicate and gate scale (default: the calibration's).	None
track_objects	Optional[Dict[int, Track]]	The tracker's {public_id: sio.Track}; updated to drop merged-away / untracked identities when given.	None

Returns:

Name	Type	Description
A	LinkingResult	class:LinkingResult. Merged identities keep the EARLIER identity's sio.Track. Never raises on degenerate input.

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

def link_tracklets(
    frames: Sequence[Tuple[int, Sequence[Any]]],
    *,
    carrier: str = "pose",
    calibration: Optional[VideoCalibration] = None,
    max_tracks: Optional[int] = None,
    track_buffer: int = 30,
    swap_correction: bool = True,
    spatial_gate_frames: int = 300,
    box_padding: float = 0.1,
    high_threshold: float = 0.5,
    score_mode: str = "auto",
    score_normalizer: Optional[float] = None,
    time_base: str = "auto",
    body_size: Optional[float] = None,
    track_objects: Optional[Dict[int, sio.Track]] = None,
) -> LinkingResult:
    """Link a tracked video's identities across gaps and crossings, in place.

    Steps, in order (see the module docstring for the definitions):

    1. **Swap correction** (when ``swap_correction`` and the calibration reports
       ``appearance_for_linking``): for each ambiguity event over at most
       ``MAX_EVENT_IDENTITIES`` identities, the post-event vectors of each
       identity are scored against every identity's pre-event gallery; the best
       permutation is applied to the moved identities' detections after the event
       when its total gain is at least the calibrated margin per moved identity
       and every moved pair clears the calibrated threshold and beats its
       unswapped pair by the margin (the rule of each query's own length).
    2. **Gap linking**, greedily best-first (appearance before spatial, score
       desc, gap asc, start asc, name asc) until nothing is accepted, between an
       identity that ends and one that starts later within the spatial gate
       ``body + 3 * s * (gap if gap <= 10 else sqrt(10 * gap))`` with ``s`` the
       calibrated ``step_p95`` floored at 0.05 body per step (no gate beyond
       ``spatial_gate_frames`` steps or when ``step_p95`` is unknown). Pairs with
       gated vectors on both sides (and calibrated appearance) are decided by
       appearance (the threshold for the later identity's head length, then the
       margin over the competing candidates that provably are other animals);
       others are linked only when the gap is at most 4 track buffers and the
       pair is unique on that window's frontier (a candidate that starts after
       the nearest one ends can follow through it, so it is no competitor). A
       link that would skip over another candidate of the same identity
       waits for it (so that fragment is not stranded inside the merged
       identity's gap).
    3. **Closed set** (``max_tracks``): while more identities remain, the
       ``max_tracks`` best-supported identities present together become
       anchors; every other tracklet, most constrained first (a single viable
       anchor before a choice; decisive choices before close calls), joins the
       anchor it never co-occurs with (frames where it only duplicates the
       anchor's detection are resolved: the higher-scoring detection stays),
       ranked by appearance agreement, then a gap-aware geometric cost of its
       junctions (gated first; past the gate only when appearance agrees, or for
       a tracklet longer than ``track_buffer`` that appearance does not
       contradict). A tracklet that fits no anchor is untracked.

    Args:
        frames: ``(frame_idx, objects)`` per frame in temporal order: the tracked
            detections (``.track`` set or ``None``) of ONE video.
        carrier: ``"pose"`` or ``"mask"``.
        calibration: The video's calibration (computes one from ``frames`` if
            ``None``, with ``time_base`` and ``body_size``). Appearance is used
            only where it reports ``appearance_for_linking``. Its resolved
            ``time_base`` and stride count the steps.
        max_tracks: Known number of animals: closed-set linking.
        track_buffer: Gap (steps) below which a track's detections form one
            tracklet; also scales the spatial-only rule.
        swap_correction: Correct swaps at crossings (appearance only).
        spatial_gate_frames: Gaps longer than this (steps) are not spatially gated.
        box_padding: Box padding used for overlap tests.
        high_threshold: Minimum score for a vector to enter a gallery.
        score_mode: Detection score mode (see ``MOTParams.score_mode``).
        score_normalizer: The tracker's per-video score normalizer
            (``MOTParams.normalize_scores``): scores are divided by it and
            clipped to ``[0, 1]`` before the ``high_threshold`` gate, so the
            gate sees the scores the tracker saw. ``None``: raw scores.
        time_base: ``MOTParams.time_base``, used when ``calibration`` is
            ``None`` (a given calibration carries its resolved time base).
        body_size: A known body size (px, ``MOTParams.body_size``): pseudo-box
            size, duplicate and gate scale (default: the calibration's).
        track_objects: The tracker's ``{public_id: sio.Track}``; updated to drop
            merged-away / untracked identities when given.

    Returns:
        A :class:`LinkingResult`. Merged identities keep the EARLIER identity's
        ``sio.Track``. Never raises on degenerate input.
    """
    result = LinkingResult()
    frames = [(int(f), list(objs)) for f, objs in frames]
    if not any(
        getattr(o, "track", None) is not None for _, objs in frames for o in objs
    ):
        result.notes.append("no tracked detections: nothing to link")
        return result

    if body_size is not None and not (np.isfinite(body_size) and body_size > 0):
        body_size = None

    def extract(pseudo: Optional[float]) -> List[FrameDetections]:
        return [
            extract_detections(
                objs,
                f,
                carrier=carrier,
                box_padding=box_padding,
                pseudo_box_size=pseudo,
                score_mode=score_mode,
            )
            for f, objs in frames
        ]

    if calibration is None:
        # Raw geometry and scores, as the trackers' `prepare` calibrates them.
        dets = extract(None)
        calibration = calibrate(
            dets,
            track_buffer=track_buffer,
            time_base=time_base,
            body_size=body_size,
        )
        if carrier == "pose" and any(_has_degenerate_pose(d) for d in dets):
            # Degenerate poses (single node) get the calibrated body-size box, as
            # in the trackers, so overlap tests see a realistic extent.
            dets = extract(float(calibration.body_size))
    else:
        dets = extract(
            float(calibration.body_size) if body_size is None else float(body_size)
        )
    for d in dets:
        normalize_scores(d, score_normalizer)

    linker = _Linker(
        frames,
        dets,
        calibration,
        track_buffer=track_buffer,
        spatial_gate_frames=spatial_gate_frames,
        high_threshold=high_threshold,
        body_size=body_size,
    )
    n_before = len(linker.tracks)
    result.n_identities_before = n_before

    n_tracklets = 0
    for mem in linker.members:
        if len(mem):
            n_tracklets += 1 + int(
                (np.diff(linker.step[mem]) > max(int(track_buffer), 0)).sum()
            )
    linker.note(
        f"{n_before} identities in {n_tracklets} tracklets; appearance calibration "
        f"{calibration.status}, used for linking: {calibration.appearance_for_linking}",
        decision=False,
    )
    if time_base != "auto" and time_base != calibration.time_base:
        linker.note(
            f"time_base={time_base!r} differs from the calibration's "
            f"{calibration.time_base!r}: steps are counted in the calibration's",
            decision=False,
        )
    appearance = (
        bool(calibration.appearance_for_linking)
        and bool(np.isfinite(linker.thresh_n).any())
        and linker.vectors is not None
    )

    # 1. Swap correction at crossings.
    if swap_correction and appearance:
        result.n_swaps = linker.correct_swaps()
    elif swap_correction:
        linker.note(
            "swap correction skipped: appearance is not calibrated for linking",
            decision=False,
        )

    # 2. Open-set gap linking.
    linker.init_linking(appearance)
    while True:
        best = linker.best_gap_link()
        if best is None:
            break
        i, j, kind, score = best
        gap = linker.first_step[j] - linker.last_step[i]
        detail = f"score {score:.3f}" if kind == "appearance" else "unique"
        linker.note(
            f"linked {linker.names[j]} -> {linker.names[i]} ({kind}, {detail}, "
            f"gap {gap:g} steps)"
        )
        linker.merge(i, j)
        result.n_merges += 1
    n_ambiguous = linker.n_spatial_ambiguous()
    if n_ambiguous:
        # Kept with the summary notes (decision notes stay last).
        linker.notes.insert(
            1,
            f"{n_ambiguous} spatial-only junction(s) left unlinked: more than one "
            "feasible candidate within the window (no usable appearance)",
        )

    # 3. Closed set.
    if max_tracks is not None and int(max_tracks) < 1:
        linker.note(f"max_tracks={max_tracks} ignored (must be >= 1)", decision=False)
    elif max_tracks is not None and linker.n_alive() > int(max_tracks):
        result.n_forced_merges, result.n_untracked = linker.close_set(int(max_tracks))

    # Write back: each live identity's detections take its final Track.
    final = linker.final_tracks()
    for d, obj in enumerate(linker.objs):
        k = int(linker.label[d])
        if k < 0 or k not in final:
            obj.track = None
            obj.tracking_score = 0.0
            continue
        track = linker.tracks[final[k]]
        if obj.track is not track:
            obj.track = track

    kept = {id(linker.tracks[t]) for t in final.values()}
    removed = {id(t) for t in linker.tracks} - kept
    if track_objects is not None and removed:
        for key in [k for k, t in track_objects.items() if id(t) in removed]:
            del track_objects[key]

    result.n_identities = len(final)
    if linker._n_decision_notes > MAX_DECISION_NOTES:
        linker.notes.append(
            f"... {linker._n_decision_notes - MAX_DECISION_NOTES} more decision(s) "
            "not listed"
        )
    result.notes.extend(linker.notes)

    logger.info(
        f"Identity linking: {n_before} -> {result.n_identities} identities "
        f"({result.n_merges} gap links, {result.n_forced_merges} forced, "
        f"{result.n_swaps} swaps corrected, {result.n_untracked} detections "
        "untracked)."
    )
    return result