instances

Turning per-view center peaks into animals, and animals into tracks (ADR-0063).

The center channel says where animals are in each view but not which peak across views is one animal, and it has no memory. assign_centers triangulates every view pair's every peak pair as one fixed-shape batch and keeps what the most views agree with; link_step carries the slots onto the next frame by Hungarian assignment under a step gate, causally; center_windows draws each animal's fixed-size window. Fixed shapes keep the jitted triangulate and Rig.project from recompiling.

assign_centers

assign_centers(
    rig: Rig,
    cands: Float[ndarray, "V C 3"],
    *,
    min_views: int = 2,
    gate_px: float = 30.0,
    min_separation_px: float = 50.0,
    max_animals: int = 2,
) -> tuple[
    Float[np.ndarray, "N 3"], Int[np.ndarray, "V N"]
]

One frame's center peaks -> up to max_animals 3D centers and their peaks.

cands (V, C, 3) are (x, y, score) per view in video pixels, NaN-padded; the score is not read, a peak earning its place by the views that agree with it. A center needs min_views support, two of them far enough from parallel and from opposite to place it in depth (ADR-0130), a view supports a hypothesis within gate_px, and centers within min_separation_px in every view are one animal found twice. Returns (centers, assigned): (max_animals, 3) world centers, NaN where unfilled, and (V, max_animals) each view's peak index, -1 for none.

Source code in src/deeperfly/instances.py

def assign_centers(
    rig: Rig,
    cands: Float[np.ndarray, "V C 3"],
    *,
    min_views: int = 2,
    gate_px: float = 30.0,
    min_separation_px: float = 50.0,
    max_animals: int = 2,
) -> tuple[Float[np.ndarray, "N 3"], Int[np.ndarray, "V N"]]:
    """One frame's center peaks -> up to `max_animals` 3D centers and their peaks.

    `cands` `(V, C, 3)` are `(x, y, score)` per view in video pixels, NaN-padded; the
    score is not read, a peak earning its place by the views that agree with it. A
    center needs `min_views` support, two of them far enough from parallel and from
    opposite to place it in depth (ADR-0130), a view supports a hypothesis within
    `gate_px`, and centers within `min_separation_px` in every view are one animal
    found twice.
    Returns `(centers, assigned)`: `(max_animals, 3)` world centers, NaN where unfilled,
    and `(V, max_animals)` each view's peak index, -1 for none.
    """
    cands = np.asarray(cands, dtype=float)
    if cands.ndim != 3 or cands.shape[-1] != 3:
        raise ValueError(f"cands must be (V, C, 3), got {tuple(cands.shape)}")
    n_views, n_cand = cands.shape[0], cands.shape[1]
    xy = cands[..., :2]
    assigned = np.full((n_views, max_animals), -1, dtype=int)
    if n_views < 2 or n_cand == 0 or max_animals < 1:
        return np.full((max_animals, 3), np.nan), assigned

    # Every view pair times every pair of their peaks, as one fixed (V, H, 2) batch.
    left, right = _view_pairs(n_views)
    a_idx = np.repeat(np.arange(n_cand), n_cand)
    b_idx = np.tile(np.arange(n_cand), n_cand)
    obs = np.full((n_views, len(left), n_cand * n_cand, 2), np.nan)
    pair = np.arange(len(left))
    obs[left, pair] = xy[left][:, a_idx]
    obs[right, pair] = xy[right][:, b_idx]
    obs = obs.reshape(n_views, -1, 2)

    guess = np.asarray(triangulate(rig, obs))  # (H, 3)
    proj = np.asarray(rig.project(guess))  # (V, H, 2)
    d = np.linalg.norm(proj[:, :, None, :] - xy[:, None, :, :], axis=-1)  # (V, H, C)
    d = np.where(np.isfinite(d), d, np.inf)
    nearest = np.argmin(d, axis=-1)  # (V, H)
    closest = np.min(d, axis=-1)  # (V, H)
    supports = closest <= gate_px
    n_support = supports.sum(axis=0)
    residual = np.where(supports, closest, 0.0).sum(axis=0) / np.maximum(n_support, 1)
    # A pair looking along one line reprojects onto both its peaks at any depth, so
    # without a third view at an angle it would win every tie on its zero residual.
    usable = (n_support >= min_views) & placed_in_depth(rig, guess, supports)

    taken = np.zeros((n_views, n_cand), dtype=bool)
    picked: list[int] = []
    for hyp in np.lexsort((residual, -n_support)):
        if not usable[hyp]:
            continue
        # A view whose only peak is taken stops supporting a hypothesis instead of
        # vetoing it: the floor-and-ceiling case (ADR-0063).
        seen = np.flatnonzero(supports[:, hyp])
        seen = seen[~taken[seen, nearest[seen, hyp]]]
        if len(seen) < min_views:
            continue
        left_in = np.zeros(n_views, dtype=bool)
        left_in[seen] = True
        if not placed_in_depth(rig, guess[hyp], left_in):
            continue
        if any(_coincide(proj[:, hyp], proj[:, q], min_separation_px) for q in picked):
            continue
        slot = len(picked)
        picked.append(hyp)
        assigned[seen, slot] = nearest[seen, hyp]
        taken[seen, nearest[seen, hyp]] = True
        if len(picked) == max_animals:
            break

    # Re-solve each pick from every peak that supported it, not from the two the
    # hypothesis was built on.
    final = np.full((n_views, max_animals, 2), np.nan)
    rows, cols = np.nonzero(assigned >= 0)
    final[rows, cols] = xy[rows, assigned[rows, cols]]
    centers = np.asarray(triangulate(rig, final))
    return centers, assigned

center_windows

center_windows(
    rig: Rig,
    centers: Float[ndarray, "N 3"],
    sizes: Float[ndarray, "V 2"],
    letterbox: Float[ndarray, "V 4"],
) -> Float[np.ndarray, "N V 4"]

A crop box per animal and view: the view's fixed size around the center.

One size per view per recording (ADR-0063). A slot with no center gets the view's letterbox as a placeholder, which the caller skips. sizes is (V, 2) (width, height) and the result (N, V, 4) boxes, both in video pixels.

Source code in src/deeperfly/instances.py

def center_windows(
    rig: Rig,
    centers: Float[np.ndarray, "N 3"],
    sizes: Float[np.ndarray, "V 2"],
    letterbox: Float[np.ndarray, "V 4"],
) -> Float[np.ndarray, "N V 4"]:
    """A crop box per animal and view: the view's fixed size around the center.

    One size per view per recording (ADR-0063). A slot with no center gets the view's
    letterbox as a placeholder, which the caller skips. `sizes` is `(V, 2)`
    `(width, height)` and the result `(N, V, 4)` boxes, both in video pixels.
    """
    centers = np.asarray(centers, dtype=float)
    sizes = np.asarray(sizes, dtype=float)
    letterbox = np.asarray(letterbox, dtype=float)
    n, n_views = centers.shape[0], sizes.shape[0]
    proj = np.asarray(rig.project(centers)).transpose(1, 0, 2)  # (N, V, 2)
    half = sizes[None] / 2.0  # (1, V, 2)
    boxes = np.concatenate([proj - half, proj + half], axis=-1)  # (N, V, 4)
    missing = ~np.isfinite(boxes).all(axis=-1)
    boxes[missing] = np.broadcast_to(letterbox, (n, n_views, 4))[missing]
    return boxes

link_step

link_step(
    rig: Rig,
    last: Float[ndarray, "N 3"],
    age: Int[ndarray, N],
    detections: Float[ndarray, "M 3"],
    *,
    max_step_px: float = 60.0,
    hold_frames: int = 12,
) -> tuple[
    Float[np.ndarray, "N 3"],
    Int[np.ndarray, N],
    Float[np.ndarray, "N 3"],
    Int[np.ndarray, N],
]

Carry N track slots onto this frame's detections, causally.

Hungarian assignment on the largest reprojected distance over the views, no velocity model and no appearance (ADR-0063). last (N, 3) and age (N,) are each slot's last position and frames unseen; a slot held k frames may jump (1 + k) * max_step_px, and after hold_frames it is freed. Returns (centers, flags, last, age): the position to report (NaN when lost), uint8 flags (0 tracked, 1 held, 2 lost), and the carried state.

Source code in src/deeperfly/instances.py

def link_step(
    rig: Rig,
    last: Float[np.ndarray, "N 3"],
    age: Int[np.ndarray, "N"],
    detections: Float[np.ndarray, "M 3"],
    *,
    max_step_px: float = 60.0,
    hold_frames: int = 12,
) -> tuple[
    Float[np.ndarray, "N 3"],
    Int[np.ndarray, "N"],
    Float[np.ndarray, "N 3"],
    Int[np.ndarray, "N"],
]:
    """Carry `N` track slots onto this frame's detections, causally.

    Hungarian assignment on the largest reprojected distance over the views, no
    velocity model and no appearance (ADR-0063). `last` `(N, 3)` and `age` `(N,)` are
    each slot's last position and frames unseen; a slot held `k` frames may jump
    `(1 + k) * max_step_px`, and after `hold_frames` it is freed. Returns
    `(centers, flags, last, age)`: the position to report (NaN when lost), uint8 flags
    (0 tracked, 1 held, 2 lost), and the carried state.
    """
    from scipy.optimize import linear_sum_assignment

    last = np.asarray(last, dtype=float)
    age = np.asarray(age, dtype=int)
    detections = np.asarray(detections, dtype=float)
    n, m = last.shape[0], detections.shape[0]
    live = np.isfinite(last).all(axis=-1)
    found = np.isfinite(detections).all(axis=-1)

    centers = np.full((n, 3), np.nan)
    flags = np.full(n, LOST, dtype=np.uint8)
    out_last, out_age = last.copy(), age.copy()

    cost = np.full((n, m), np.inf)
    if n and m:
        proj = np.asarray(rig.project(np.concatenate([last, detections])))
        d = np.linalg.norm(proj[:, :n, None, :] - proj[:, None, n:, :], axis=-1)
        cost = np.where(np.isfinite(d), d, np.inf).max(axis=0)
        cost[~live] = np.inf
        cost[:, ~found] = np.inf
    matched_slots, matched_dets = np.empty(0, int), np.empty(0, int)
    if n and m and np.isfinite(cost).any():
        # Hungarian needs finite costs; the gate then rejects the padding it matched.
        rows, cols = linear_sum_assignment(np.where(np.isfinite(cost), cost, 1e12))
        # The gate widens with the slot's age, so an animal that reappears from behind
        # the other is not refused its own detection (ADR-0063).
        keep = cost[rows, cols] <= max_step_px * (1 + age[rows])
        matched_slots, matched_dets = rows[keep], cols[keep]

    centers[matched_slots] = detections[matched_dets]
    out_last[matched_slots] = detections[matched_dets]
    out_age[matched_slots] = 0
    flags[matched_slots] = TRACKED

    stale = live.copy()
    stale[matched_slots] = False
    for slot in np.flatnonzero(stale):
        out_age[slot] = age[slot] + 1
        if out_age[slot] <= hold_frames:
            centers[slot] = last[slot]
            flags[slot] = HELD
        else:
            out_last[slot] = np.nan
            out_age[slot] = 0

    leftover = [i for i in np.flatnonzero(found) if i not in set(matched_dets.tolist())]
    for det in leftover:
        free = np.flatnonzero(flags == LOST)
        if not free.size:
            break  # more animals in view than slots; the extras are dropped
        slot = free[0]
        centers[slot] = detections[det]
        out_last[slot] = detections[det]
        out_age[slot] = 0
        flags[slot] = TRACKED
    return centers, flags, out_last, out_age