labels

Labels: a project's hand-labeled 2D keypoints, their files, and the labeling aids.

Below deeperfly.project: a Labels set is built from view and keypoint names and a frame count alone. store is the sparse label set and its file, and files where one project's files live. The aids sit above them and are not imported here: detections merges a project's runs into the detector 2D (ADR-0083), identity tells which of a run's animals each labeled one is (ADR-0128), check is the labeling solve and its checks, suggest ranks the joints a label would most likely fix.

LabelFiles dataclass

One project's label files, keyed to the project they are read against.

Attributes:

Name	Type	Description
directory	Path	The project folder holding the files, whether or not it exists yet.
view_names	tuple[str, ...]	The project's views, in order.
keypoint_names	tuple[str, ...]	The skeleton's points, in order.
n_frames	int	The project's frame count.

Source code in src/deeperfly/labels/files.py

@dataclass(frozen=True)
class LabelFiles:
    """One project's label files, keyed to the project they are read against.

    Attributes:
        directory: The project folder holding the files, whether or not it exists yet.
        view_names: The project's views, in order.
        keypoint_names: The skeleton's points, in order.
        n_frames: The project's frame count.
    """

    directory: Path
    view_names: tuple[str, ...]
    keypoint_names: tuple[str, ...]
    n_frames: int

    @property
    def labels_path(self) -> Path:
        """Where the project's `labels.h5` lives."""
        return self.directory / LABELS_FILENAME

    @property
    def recovery_path(self) -> Path:
        """Where the crash-recovery autosave sidecar lives."""
        return self.directory / LABELS_RECOVERY_FILENAME

    def load(self) -> Labels:
        """The saved labels, or an empty set; a reordered skeleton is remapped."""
        return self._read(self.labels_path)

    def open(self) -> tuple[Labels, bool]:
        """The labels for editing and whether they came from the recovery sidecar.

        Recovered labels load dirty, so the operator saves or discards them.
        """
        path, recovered = labels_source(self.directory)
        if recovered and path is not None:
            labels = self._read(path)
            labels.dirty = True
            return labels, True
        return self.load(), False

    @property
    def discarded_path(self) -> Path:
        """Where the labels the last discard threw away are kept (ADR-0120)."""
        return self.directory / LABELS_DISCARDED_FILENAME

    def save(self, labels: Labels) -> None:
        """Persist the labels to `labels.h5`, clearing the sidecar and any discard."""
        self.directory.mkdir(parents=True, exist_ok=True)
        save_labels(labels, self.labels_path)
        self.recovery_path.unlink(missing_ok=True)
        self.discarded_path.unlink(missing_ok=True)

    def autosave(self, labels: Labels) -> None:
        """Write in-progress labels to the recovery sidecar, keeping them dirty.

        A discard kept from before is stale now: restoring it would drop this edit.
        """
        self.directory.mkdir(parents=True, exist_ok=True)
        save_labels(labels, self.recovery_path, clear_dirty=False)
        self.discarded_path.unlink(missing_ok=True)

    def discard_unsaved(self, labels: Labels | None = None) -> Labels:
        """Set the unsaved labels aside and reload the saved ones.

        `labels` are the unsaved ones, an open session's; without them the sidecar is
        what is set aside. Either way it lands at `discarded_path`, for
        `restore_discarded`.
        """
        if labels is not None and labels.dirty:
            save_labels(labels, self.discarded_path, clear_dirty=False)
            self.recovery_path.unlink(missing_ok=True)
        elif self.recovery_path.exists():
            os.replace(self.recovery_path, self.discarded_path)
        return self.load()

    def restore_discarded(self) -> bool:
        """Put what the last discard set aside back as the sidecar; whether there was.

        It is touched as it lands, so the labels load from it again, dirty.
        """
        if not self.discarded_path.exists():
            return False
        os.replace(self.discarded_path, self.recovery_path)
        os.utime(self.recovery_path)
        return True

    def _read(self, path: Path) -> Labels:
        if path.exists():
            return load_labels(
                path,
                view_names=self.view_names,
                keypoint_names=self.keypoint_names,
                n_frames=self.n_frames,
            )
        return Labels(self.view_names, self.keypoint_names, self.n_frames)

discarded_path property

discarded_path: Path

Where the labels the last discard threw away are kept (ADR-0120).

labels_path property

labels_path: Path

Where the project's labels.h5 lives.

recovery_path property

recovery_path: Path

Where the crash-recovery autosave sidecar lives.

autosave

autosave(labels: Labels) -> None

Write in-progress labels to the recovery sidecar, keeping them dirty.

A discard kept from before is stale now: restoring it would drop this edit.

Source code in src/deeperfly/labels/files.py

def autosave(self, labels: Labels) -> None:
    """Write in-progress labels to the recovery sidecar, keeping them dirty.

    A discard kept from before is stale now: restoring it would drop this edit.
    """
    self.directory.mkdir(parents=True, exist_ok=True)
    save_labels(labels, self.recovery_path, clear_dirty=False)
    self.discarded_path.unlink(missing_ok=True)

discard_unsaved

discard_unsaved(labels: Labels | None = None) -> Labels

Set the unsaved labels aside and reload the saved ones.

labels are the unsaved ones, an open session's; without them the sidecar is what is set aside. Either way it lands at discarded_path, for restore_discarded.

Source code in src/deeperfly/labels/files.py

def discard_unsaved(self, labels: Labels | None = None) -> Labels:
    """Set the unsaved labels aside and reload the saved ones.

    `labels` are the unsaved ones, an open session's; without them the sidecar is
    what is set aside. Either way it lands at `discarded_path`, for
    `restore_discarded`.
    """
    if labels is not None and labels.dirty:
        save_labels(labels, self.discarded_path, clear_dirty=False)
        self.recovery_path.unlink(missing_ok=True)
    elif self.recovery_path.exists():
        os.replace(self.recovery_path, self.discarded_path)
    return self.load()

load

load() -> Labels

The saved labels, or an empty set; a reordered skeleton is remapped.

Source code in src/deeperfly/labels/files.py

def load(self) -> Labels:
    """The saved labels, or an empty set; a reordered skeleton is remapped."""
    return self._read(self.labels_path)

open

open() -> tuple[Labels, bool]

The labels for editing and whether they came from the recovery sidecar.

Recovered labels load dirty, so the operator saves or discards them.

Source code in src/deeperfly/labels/files.py

def open(self) -> tuple[Labels, bool]:
    """The labels for editing and whether they came from the recovery sidecar.

    Recovered labels load dirty, so the operator saves or discards them.
    """
    path, recovered = labels_source(self.directory)
    if recovered and path is not None:
        labels = self._read(path)
        labels.dirty = True
        return labels, True
    return self.load(), False

restore_discarded

restore_discarded() -> bool

Put what the last discard set aside back as the sidecar; whether there was.

It is touched as it lands, so the labels load from it again, dirty.

Source code in src/deeperfly/labels/files.py

def restore_discarded(self) -> bool:
    """Put what the last discard set aside back as the sidecar; whether there was.

    It is touched as it lands, so the labels load from it again, dirty.
    """
    if not self.discarded_path.exists():
        return False
    os.replace(self.discarded_path, self.recovery_path)
    os.utime(self.recovery_path)
    return True

save

save(labels: Labels) -> None

Persist the labels to labels.h5, clearing the sidecar and any discard.

Source code in src/deeperfly/labels/files.py

def save(self, labels: Labels) -> None:
    """Persist the labels to `labels.h5`, clearing the sidecar and any discard."""
    self.directory.mkdir(parents=True, exist_ok=True)
    save_labels(labels, self.labels_path)
    self.recovery_path.unlink(missing_ok=True)
    self.discarded_path.unlink(missing_ok=True)

Labels

Sparse per-(view, frame, keypoint) 2D ground truth for one project.

The V and K axes are positional, in the rig's and skeleton's order; an unlabeled cell is NaN, and dirty tracks unsaved edits. Every accessor takes a keyword-only instance, slot 0 by default.

Source code in src/deeperfly/labels/store.py

class Labels:
    """Sparse per-`(view, frame, keypoint)` 2D ground truth for one project.

    The `V` and `K` axes are positional, in the rig's and skeleton's order; an
    unlabeled cell is NaN, and `dirty` tracks unsaved edits. Every accessor takes a
    keyword-only `instance`, slot 0 by default.
    """

    def __init__(
        self,
        view_names: tuple[str, ...],
        keypoint_names: tuple[str, ...],
        n_frames: int,
    ) -> None:
        """An empty label set over the given views, keypoints and frame count."""
        self.view_names = tuple(view_names)
        self.keypoint_names = tuple(keypoint_names)
        self.n_frames = int(n_frames)
        # Keyed by (frame, instance): an instance slot is one animal of the
        # project, slot 0 being the only animal a single-fly project has.
        self._frames: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K, 2)
        # `hidden` only excludes a cell from the loss: it never moves, hides or
        # triangulates a point, and all four (gt, hidden) combinations are legal.
        self._hidden: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K) bool
        # `absent` is per (frame, keypoint), so it can vary over the frames (autotomy);
        # it masks a point out without destroying the pixels, so un-declaring restores.
        self._absent: dict[int, np.ndarray] = {}  # instance -> (T, K) bool, lazy
        self.dirty = False

    # -- sizes ---------------------------------------------------------------

    @property
    def n_views(self) -> int:
        """The number of views."""
        return len(self.view_names)

    @property
    def n_keypoints(self) -> int:
        """The number of keypoints."""
        return len(self.keypoint_names)

    # -- access --------------------------------------------------------------

    def frame(self, t: int, *, instance: int = 0) -> Float[np.ndarray, "V K 2"]:
        """The `(V, K, 2)` labels of one instance of frame `t`, all NaN if none."""
        self._check_frame(t)
        arr = self._frames.get((t, instance))
        if arr is not None:
            return arr.copy()
        return np.full((self.n_views, self.n_keypoints, 2), np.nan)

    def labeled_frames(self) -> list[int]:
        """Frame indices that carry a labeled point, sorted, each once."""
        return sorted(
            {t for (t, _), arr in self._frames.items() if np.isfinite(arr).any()}
        )

    def labeled_instances(self, t: int) -> list[int]:
        """The instance slots labeled in frame `t`, sorted (sparse track ids)."""
        self._check_frame(t)
        return sorted(
            i
            for (f, i), arr in self._frames.items()
            if f == t and np.isfinite(arr).any()
        )

    def frame_instances(self, t: int) -> list[int]:
        """Every instance slot that holds anything in frame `t`, sorted.

        A pixel, a hidden flag or an absent declaration counts, unlike
        `labeled_instances`, which takes pixels only.
        """
        self._check_frame(t)
        slots = {i for f, i in self._frames.keys() | self._hidden.keys() if f == t}
        slots.update(i for i, mask in self._absent.items() if mask[t].any())
        return sorted(slots)

    def label_counts(self) -> list[int]:
        """How many `(view, keypoint)` cells each of `labeled_frames` carries."""
        counts: dict[int, int] = {}
        for (t, _), arr in self._frames.items():
            counts[t] = counts.get(t, 0) + int(np.isfinite(arr).all(axis=-1).sum())
        return [counts[t] for t in self.labeled_frames()]

    def open_keypoints(self, min_views: int = MIN_CLICKED_VIEWS) -> list[int]:
        """How many keypoints each of `labeled_frames` still lacks to be complete.

        A frame is complete (0) when every non-absent keypoint of each labeled
        instance has at least `min_views` clicked views (ADR-0102). Counted per
        instance, so a second animal with nothing clicked is not owed; one with a
        single click owes its other keypoints.
        """
        owed: dict[int, int] = {}
        for (t, i), arr in self._frames.items():
            if not np.isfinite(arr).any():
                continue
            clicked = np.isfinite(arr).all(axis=-1).sum(axis=0)  # (K,)
            absent = self._absent.get(i)
            short = clicked < min_views
            if absent is not None:
                short &= ~absent[t]
            owed[t] = owed.get(t, 0) + int(short.sum())
        return [owed[t] for t in self.labeled_frames()]

    def frame_effective(
        self, t: int, *, instance: int = 0
    ) -> Float[np.ndarray, "V K 2"]:
        """The frame's GT with absent keypoints masked to NaN -- what the solve sees."""
        arr = self.frame(t, instance=instance)
        absent = self._absent.get(instance)
        if absent is not None:
            arr[:, absent[t]] = np.nan
        return arr

    def hidden_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
        """The `(V, K)` hidden mask for frame `t` (all False if none)."""
        self._check_frame(t)
        mask = self._hidden.get((t, instance))
        if mask is not None:
            return mask.copy()
        return np.zeros((self.n_views, self.n_keypoints), dtype=bool)

    def absent_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
        """The `(K,)` absent mask for frame `t` (all False if none)."""
        self._check_frame(t)
        absent = self._absent.get(instance)
        if absent is None:
            return np.zeros(self.n_keypoints, dtype=bool)
        return absent[t].copy()

    def is_absent(self, frame: int, keypoint: int, *, instance: int = 0) -> bool:
        """Whether a keypoint is declared not-on-this-animal in frame `t`."""
        self._check_frame(frame)
        absent = self._absent.get(instance)
        return bool(absent[frame, keypoint]) if absent is not None else False

    def absent_keypoint(
        self, keypoint: int, *, instance: int = 0
    ) -> Bool[np.ndarray, " T"]:
        """One keypoint's per-frame absent mask over every frame."""
        absent = self._absent.get(instance)
        if absent is None:
            return np.zeros(self.n_frames, dtype=bool)
        return absent[:, keypoint].copy()

    # -- mutation ------------------------------------------------------------

    def set_point(
        self, view: int, frame: int, keypoint: int, xy, *, instance: int = 0
    ) -> None:
        """Set the labeled pixel of one `(view, frame, keypoint)` cell."""
        self._check_frame(frame)
        arr = self._frames.get((frame, instance))
        if arr is None:
            arr = np.full((self.n_views, self.n_keypoints, 2), np.nan)
            self._frames[(frame, instance)] = arr
        arr[view, keypoint] = np.asarray(xy, dtype=float)
        self.dirty = True

    def clear_point(
        self, view: int, frame: int, keypoint: int, *, instance: int = 0
    ) -> None:
        """Remove the label of one `(view, frame, keypoint)` cell, if any."""
        self._check_frame(frame)
        arr = self._frames.get((frame, instance))
        if arr is None:
            return
        arr[view, keypoint] = np.nan
        if not np.isfinite(arr).any():
            del self._frames[(frame, instance)]
        self.dirty = True

    def set_hidden(
        self,
        view: int,
        frame: int,
        keypoint: int,
        value: bool = True,
        *,
        instance: int = 0,
    ) -> None:
        """Set or clear the hidden (exclude-from-loss) flag of one cell."""
        self._check_frame(frame)
        mask = self._hidden.get((frame, instance))
        if mask is None:
            if not value:
                return
            mask = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
            self._hidden[(frame, instance)] = mask
        mask[view, keypoint] = value
        if not mask.any():
            del self._hidden[(frame, instance)]
        self.dirty = True

    def set_absent(
        self, frame: int, keypoint: int, value: bool = True, *, instance: int = 0
    ) -> None:
        """Declare a keypoint absent (or present again) in one frame."""
        self._check_frame(frame)
        absent = self._absent.get(instance)
        if absent is None:
            if not value:
                return
            absent = self._absent_mask(instance)
        absent[frame, keypoint] = value
        self.dirty = True

    def set_absent_every_frame(
        self, keypoint: int, value: bool = True, *, instance: int = 0
    ) -> None:
        """Declare a keypoint absent (or present) in every frame."""
        self.set_absent_keypoint(
            keypoint, np.full(self.n_frames, value), instance=instance
        )

    def set_absent_keypoint(self, keypoint: int, values, *, instance: int = 0) -> None:
        """Replace one keypoint's per-frame absent flags with a whole `(T,)` mask.

        The general form of `set_absent_every_frame`, so an undo can put back any
        pattern.
        """
        values = np.asarray(values, dtype=bool)
        absent = self._absent.get(instance)
        if absent is None:
            if not values.any():
                return
            absent = self._absent_mask(instance)
        absent[:, keypoint] = values
        self.dirty = True

    def clear_instance(self, frame: int, instance: int) -> None:
        """Forget what one instance holds in one frame: points, flags, absences."""
        self._check_frame(frame)
        self._frames.pop((frame, instance), None)
        self._hidden.pop((frame, instance), None)
        absent = self._absent.get(instance)
        if absent is not None:
            absent[frame] = False
        self.dirty = True

    def changed_frames(self, other: "Labels") -> list[int]:
        """The frames whose authored state differs from `other`'s, sorted.

        A pixel, a hidden flag or an absent declaration of any instance counts; an
        empty cell and a cleared one are the same. The two sets share their shape.
        """
        changed: set[int] = set()
        nan = np.full((self.n_views, self.n_keypoints, 2), np.nan)
        for key in self._frames.keys() | other._frames.keys():
            mine, theirs = self._frames.get(key, nan), other._frames.get(key, nan)
            if not np.array_equal(mine, theirs, equal_nan=True):
                changed.add(key[0])
        none = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
        for key in self._hidden.keys() | other._hidden.keys():
            if not np.array_equal(
                self._hidden.get(key, none), other._hidden.get(key, none)
            ):
                changed.add(key[0])
        for instance in self._absent.keys() | other._absent.keys():
            mine, theirs = self._absent.get(instance), other._absent.get(instance)
            if mine is None:
                mine = np.zeros_like(theirs)
            if theirs is None:
                theirs = np.zeros_like(mine)
            changed.update(np.flatnonzero((mine != theirs).any(axis=1)).tolist())
        return sorted(changed)

    def _absent_mask(self, instance: int) -> np.ndarray:
        """One instance's `(T, K)` absent mask, allocated on first use."""
        mask = self._absent.get(instance)
        if mask is None:
            mask = np.zeros((self.n_frames, self.n_keypoints), dtype=bool)
            self._absent[instance] = mask
        return mask

    # -- COO (for persistence) -----------------------------------------------

    def to_coo(self) -> dict[str, np.ndarray]:
        """The authored state as flat COO arrays for persistence.

        Positions ride `view`/`frame`/`keypoint`/`xy`, hidden cells the `hidden_*`
        triples, and absence half-open spans per keypoint
        (`absent_keypoint`/`absent_start`/`absent_end`); each carries its instance
        slot, which `save_labels` leaves out when only slot 0 is named. Rows are in
        `(frame, instance)` order.
        """
        views, frames, kps, xys, insts = [], [], [], [], []
        for t, i in sorted(self._frames):
            arr = self._frames[(t, i)]
            vs, ks = np.nonzero(np.isfinite(arr).all(axis=-1))
            for v, k in zip(vs, ks, strict=True):
                views.append(v)
                frames.append(t)
                kps.append(k)
                xys.append(arr[v, k])
                insts.append(i)
        hv, hf, hk, hi = [], [], [], []
        for t, i in sorted(self._hidden):
            vs, ks = np.nonzero(self._hidden[(t, i)])
            for v, k in zip(vs, ks, strict=True):
                hv.append(v)
                hf.append(t)
                hk.append(k)
                hi.append(i)
        akp, astart, aend, ai = [], [], [], []
        for i in sorted(self._absent):
            absent = self._absent[i]
            for k in range(self.n_keypoints):
                diff = np.diff(np.concatenate(([0], absent[:, k].view(np.int8), [0])))
                for s, e in zip(
                    np.flatnonzero(diff == 1), np.flatnonzero(diff == -1), strict=True
                ):
                    akp.append(k)
                    astart.append(s)
                    aend.append(e)
                    ai.append(i)
        return {
            "view": np.asarray(views, dtype=np.int32),
            "frame": np.asarray(frames, dtype=np.int32),
            "keypoint": np.asarray(kps, dtype=np.int32),
            "xy": np.asarray(xys, dtype=np.float64).reshape(-1, 2),
            "hidden_view": np.asarray(hv, dtype=np.int32),
            "hidden_frame": np.asarray(hf, dtype=np.int32),
            "hidden_keypoint": np.asarray(hk, dtype=np.int32),
            "absent_keypoint": np.asarray(akp, dtype=np.int32),
            "absent_start": np.asarray(astart, dtype=np.int32),
            "absent_end": np.asarray(aend, dtype=np.int32),
            "instance": np.asarray(insts, dtype=np.int32),
            "hidden_instance": np.asarray(hi, dtype=np.int32),
            "absent_instance": np.asarray(ai, dtype=np.int32),
        }

    def _apply_coo(self, coo: Mapping[str, np.ndarray]) -> None:
        """Load the flat COO arrays of `to_coo` back into an empty store.

        Every column must be there; `load_labels` supplies zeros for one a file
        predates, so the strict zips stay a real check.
        """
        for v, t, k, i, p in zip(
            coo["view"],
            coo["frame"],
            coo["keypoint"],
            coo["instance"],
            coo["xy"],
            strict=True,
        ):
            self.set_point(int(v), int(t), int(k), p, instance=int(i))
        for v, t, k, i in zip(
            coo["hidden_view"],
            coo["hidden_frame"],
            coo["hidden_keypoint"],
            coo["hidden_instance"],
            strict=True,
        ):
            self.set_hidden(int(v), int(t), int(k), instance=int(i))
        for k, s, e, i in zip(
            coo["absent_keypoint"],
            coo["absent_start"],
            coo["absent_end"],
            coo["absent_instance"],
            strict=True,
        ):
            self._absent_mask(int(i))[int(s) : int(e), int(k)] = True
        self.dirty = False

    def _check_frame(self, t: int) -> None:
        if not 0 <= t < self.n_frames:
            raise IndexError(f"frame {t} out of range [0, {self.n_frames})")

n_keypoints property

n_keypoints: int

The number of keypoints.

n_views property

n_views: int

The number of views.

__init__

__init__(
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> None

An empty label set over the given views, keypoints and frame count.

Source code in src/deeperfly/labels/store.py

def __init__(
    self,
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> None:
    """An empty label set over the given views, keypoints and frame count."""
    self.view_names = tuple(view_names)
    self.keypoint_names = tuple(keypoint_names)
    self.n_frames = int(n_frames)
    # Keyed by (frame, instance): an instance slot is one animal of the
    # project, slot 0 being the only animal a single-fly project has.
    self._frames: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K, 2)
    # `hidden` only excludes a cell from the loss: it never moves, hides or
    # triangulates a point, and all four (gt, hidden) combinations are legal.
    self._hidden: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K) bool
    # `absent` is per (frame, keypoint), so it can vary over the frames (autotomy);
    # it masks a point out without destroying the pixels, so un-declaring restores.
    self._absent: dict[int, np.ndarray] = {}  # instance -> (T, K) bool, lazy
    self.dirty = False

absent_frame

absent_frame(t: int, *, instance: int = 0) -> np.ndarray

The (K,) absent mask for frame t (all False if none).

Source code in src/deeperfly/labels/store.py

def absent_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
    """The `(K,)` absent mask for frame `t` (all False if none)."""
    self._check_frame(t)
    absent = self._absent.get(instance)
    if absent is None:
        return np.zeros(self.n_keypoints, dtype=bool)
    return absent[t].copy()

absent_keypoint

absent_keypoint(
    keypoint: int, *, instance: int = 0
) -> Bool[np.ndarray, " T"]

One keypoint's per-frame absent mask over every frame.

Source code in src/deeperfly/labels/store.py

def absent_keypoint(
    self, keypoint: int, *, instance: int = 0
) -> Bool[np.ndarray, " T"]:
    """One keypoint's per-frame absent mask over every frame."""
    absent = self._absent.get(instance)
    if absent is None:
        return np.zeros(self.n_frames, dtype=bool)
    return absent[:, keypoint].copy()

changed_frames

changed_frames(other: Labels) -> list[int]

The frames whose authored state differs from other's, sorted.

A pixel, a hidden flag or an absent declaration of any instance counts; an empty cell and a cleared one are the same. The two sets share their shape.

Source code in src/deeperfly/labels/store.py

def changed_frames(self, other: "Labels") -> list[int]:
    """The frames whose authored state differs from `other`'s, sorted.

    A pixel, a hidden flag or an absent declaration of any instance counts; an
    empty cell and a cleared one are the same. The two sets share their shape.
    """
    changed: set[int] = set()
    nan = np.full((self.n_views, self.n_keypoints, 2), np.nan)
    for key in self._frames.keys() | other._frames.keys():
        mine, theirs = self._frames.get(key, nan), other._frames.get(key, nan)
        if not np.array_equal(mine, theirs, equal_nan=True):
            changed.add(key[0])
    none = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
    for key in self._hidden.keys() | other._hidden.keys():
        if not np.array_equal(
            self._hidden.get(key, none), other._hidden.get(key, none)
        ):
            changed.add(key[0])
    for instance in self._absent.keys() | other._absent.keys():
        mine, theirs = self._absent.get(instance), other._absent.get(instance)
        if mine is None:
            mine = np.zeros_like(theirs)
        if theirs is None:
            theirs = np.zeros_like(mine)
        changed.update(np.flatnonzero((mine != theirs).any(axis=1)).tolist())
    return sorted(changed)

clear_instance

clear_instance(frame: int, instance: int) -> None

Forget what one instance holds in one frame: points, flags, absences.

Source code in src/deeperfly/labels/store.py

def clear_instance(self, frame: int, instance: int) -> None:
    """Forget what one instance holds in one frame: points, flags, absences."""
    self._check_frame(frame)
    self._frames.pop((frame, instance), None)
    self._hidden.pop((frame, instance), None)
    absent = self._absent.get(instance)
    if absent is not None:
        absent[frame] = False
    self.dirty = True

clear_point

clear_point(
    view: int,
    frame: int,
    keypoint: int,
    *,
    instance: int = 0,
) -> None

Remove the label of one (view, frame, keypoint) cell, if any.

Source code in src/deeperfly/labels/store.py

def clear_point(
    self, view: int, frame: int, keypoint: int, *, instance: int = 0
) -> None:
    """Remove the label of one `(view, frame, keypoint)` cell, if any."""
    self._check_frame(frame)
    arr = self._frames.get((frame, instance))
    if arr is None:
        return
    arr[view, keypoint] = np.nan
    if not np.isfinite(arr).any():
        del self._frames[(frame, instance)]
    self.dirty = True

frame

frame(
    t: int, *, instance: int = 0
) -> Float[np.ndarray, "V K 2"]

The (V, K, 2) labels of one instance of frame t, all NaN if none.

Source code in src/deeperfly/labels/store.py

def frame(self, t: int, *, instance: int = 0) -> Float[np.ndarray, "V K 2"]:
    """The `(V, K, 2)` labels of one instance of frame `t`, all NaN if none."""
    self._check_frame(t)
    arr = self._frames.get((t, instance))
    if arr is not None:
        return arr.copy()
    return np.full((self.n_views, self.n_keypoints, 2), np.nan)

frame_effective

frame_effective(
    t: int, *, instance: int = 0
) -> Float[np.ndarray, "V K 2"]

The frame's GT with absent keypoints masked to NaN -- what the solve sees.

Source code in src/deeperfly/labels/store.py

def frame_effective(
    self, t: int, *, instance: int = 0
) -> Float[np.ndarray, "V K 2"]:
    """The frame's GT with absent keypoints masked to NaN -- what the solve sees."""
    arr = self.frame(t, instance=instance)
    absent = self._absent.get(instance)
    if absent is not None:
        arr[:, absent[t]] = np.nan
    return arr

frame_instances

frame_instances(t: int) -> list[int]

Every instance slot that holds anything in frame t, sorted.

A pixel, a hidden flag or an absent declaration counts, unlike labeled_instances, which takes pixels only.

Source code in src/deeperfly/labels/store.py

def frame_instances(self, t: int) -> list[int]:
    """Every instance slot that holds anything in frame `t`, sorted.

    A pixel, a hidden flag or an absent declaration counts, unlike
    `labeled_instances`, which takes pixels only.
    """
    self._check_frame(t)
    slots = {i for f, i in self._frames.keys() | self._hidden.keys() if f == t}
    slots.update(i for i, mask in self._absent.items() if mask[t].any())
    return sorted(slots)

hidden_frame

hidden_frame(t: int, *, instance: int = 0) -> np.ndarray

The (V, K) hidden mask for frame t (all False if none).

Source code in src/deeperfly/labels/store.py

def hidden_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
    """The `(V, K)` hidden mask for frame `t` (all False if none)."""
    self._check_frame(t)
    mask = self._hidden.get((t, instance))
    if mask is not None:
        return mask.copy()
    return np.zeros((self.n_views, self.n_keypoints), dtype=bool)

is_absent

is_absent(
    frame: int, keypoint: int, *, instance: int = 0
) -> bool

Whether a keypoint is declared not-on-this-animal in frame t.

Source code in src/deeperfly/labels/store.py

def is_absent(self, frame: int, keypoint: int, *, instance: int = 0) -> bool:
    """Whether a keypoint is declared not-on-this-animal in frame `t`."""
    self._check_frame(frame)
    absent = self._absent.get(instance)
    return bool(absent[frame, keypoint]) if absent is not None else False

label_counts

label_counts() -> list[int]

How many (view, keypoint) cells each of labeled_frames carries.

Source code in src/deeperfly/labels/store.py

def label_counts(self) -> list[int]:
    """How many `(view, keypoint)` cells each of `labeled_frames` carries."""
    counts: dict[int, int] = {}
    for (t, _), arr in self._frames.items():
        counts[t] = counts.get(t, 0) + int(np.isfinite(arr).all(axis=-1).sum())
    return [counts[t] for t in self.labeled_frames()]

labeled_frames

labeled_frames() -> list[int]

Frame indices that carry a labeled point, sorted, each once.

Source code in src/deeperfly/labels/store.py

def labeled_frames(self) -> list[int]:
    """Frame indices that carry a labeled point, sorted, each once."""
    return sorted(
        {t for (t, _), arr in self._frames.items() if np.isfinite(arr).any()}
    )

labeled_instances

labeled_instances(t: int) -> list[int]

The instance slots labeled in frame t, sorted (sparse track ids).

Source code in src/deeperfly/labels/store.py

def labeled_instances(self, t: int) -> list[int]:
    """The instance slots labeled in frame `t`, sorted (sparse track ids)."""
    self._check_frame(t)
    return sorted(
        i
        for (f, i), arr in self._frames.items()
        if f == t and np.isfinite(arr).any()
    )

open_keypoints

open_keypoints(
    min_views: int = MIN_CLICKED_VIEWS,
) -> list[int]

How many keypoints each of labeled_frames still lacks to be complete.

A frame is complete (0) when every non-absent keypoint of each labeled instance has at least min_views clicked views (ADR-0102). Counted per instance, so a second animal with nothing clicked is not owed; one with a single click owes its other keypoints.

Source code in src/deeperfly/labels/store.py

def open_keypoints(self, min_views: int = MIN_CLICKED_VIEWS) -> list[int]:
    """How many keypoints each of `labeled_frames` still lacks to be complete.

    A frame is complete (0) when every non-absent keypoint of each labeled
    instance has at least `min_views` clicked views (ADR-0102). Counted per
    instance, so a second animal with nothing clicked is not owed; one with a
    single click owes its other keypoints.
    """
    owed: dict[int, int] = {}
    for (t, i), arr in self._frames.items():
        if not np.isfinite(arr).any():
            continue
        clicked = np.isfinite(arr).all(axis=-1).sum(axis=0)  # (K,)
        absent = self._absent.get(i)
        short = clicked < min_views
        if absent is not None:
            short &= ~absent[t]
        owed[t] = owed.get(t, 0) + int(short.sum())
    return [owed[t] for t in self.labeled_frames()]

set_absent

set_absent(
    frame: int,
    keypoint: int,
    value: bool = True,
    *,
    instance: int = 0,
) -> None

Declare a keypoint absent (or present again) in one frame.

Source code in src/deeperfly/labels/store.py

def set_absent(
    self, frame: int, keypoint: int, value: bool = True, *, instance: int = 0
) -> None:
    """Declare a keypoint absent (or present again) in one frame."""
    self._check_frame(frame)
    absent = self._absent.get(instance)
    if absent is None:
        if not value:
            return
        absent = self._absent_mask(instance)
    absent[frame, keypoint] = value
    self.dirty = True

set_absent_every_frame

set_absent_every_frame(
    keypoint: int, value: bool = True, *, instance: int = 0
) -> None

Declare a keypoint absent (or present) in every frame.

Source code in src/deeperfly/labels/store.py

def set_absent_every_frame(
    self, keypoint: int, value: bool = True, *, instance: int = 0
) -> None:
    """Declare a keypoint absent (or present) in every frame."""
    self.set_absent_keypoint(
        keypoint, np.full(self.n_frames, value), instance=instance
    )

set_absent_keypoint

set_absent_keypoint(
    keypoint: int, values, *, instance: int = 0
) -> None

Replace one keypoint's per-frame absent flags with a whole (T,) mask.

The general form of set_absent_every_frame, so an undo can put back any pattern.

Source code in src/deeperfly/labels/store.py

def set_absent_keypoint(self, keypoint: int, values, *, instance: int = 0) -> None:
    """Replace one keypoint's per-frame absent flags with a whole `(T,)` mask.

    The general form of `set_absent_every_frame`, so an undo can put back any
    pattern.
    """
    values = np.asarray(values, dtype=bool)
    absent = self._absent.get(instance)
    if absent is None:
        if not values.any():
            return
        absent = self._absent_mask(instance)
    absent[:, keypoint] = values
    self.dirty = True

set_hidden

set_hidden(
    view: int,
    frame: int,
    keypoint: int,
    value: bool = True,
    *,
    instance: int = 0,
) -> None

Set or clear the hidden (exclude-from-loss) flag of one cell.

Source code in src/deeperfly/labels/store.py

def set_hidden(
    self,
    view: int,
    frame: int,
    keypoint: int,
    value: bool = True,
    *,
    instance: int = 0,
) -> None:
    """Set or clear the hidden (exclude-from-loss) flag of one cell."""
    self._check_frame(frame)
    mask = self._hidden.get((frame, instance))
    if mask is None:
        if not value:
            return
        mask = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
        self._hidden[(frame, instance)] = mask
    mask[view, keypoint] = value
    if not mask.any():
        del self._hidden[(frame, instance)]
    self.dirty = True

set_point

set_point(
    view: int,
    frame: int,
    keypoint: int,
    xy,
    *,
    instance: int = 0,
) -> None

Set the labeled pixel of one (view, frame, keypoint) cell.

Source code in src/deeperfly/labels/store.py

def set_point(
    self, view: int, frame: int, keypoint: int, xy, *, instance: int = 0
) -> None:
    """Set the labeled pixel of one `(view, frame, keypoint)` cell."""
    self._check_frame(frame)
    arr = self._frames.get((frame, instance))
    if arr is None:
        arr = np.full((self.n_views, self.n_keypoints, 2), np.nan)
        self._frames[(frame, instance)] = arr
    arr[view, keypoint] = np.asarray(xy, dtype=float)
    self.dirty = True

to_coo

to_coo() -> dict[str, np.ndarray]

The authored state as flat COO arrays for persistence.

Positions ride view/frame/keypoint/xy, hidden cells the hidden_* triples, and absence half-open spans per keypoint (absent_keypoint/absent_start/absent_end); each carries its instance slot, which save_labels leaves out when only slot 0 is named. Rows are in (frame, instance) order.

Source code in src/deeperfly/labels/store.py

def to_coo(self) -> dict[str, np.ndarray]:
    """The authored state as flat COO arrays for persistence.

    Positions ride `view`/`frame`/`keypoint`/`xy`, hidden cells the `hidden_*`
    triples, and absence half-open spans per keypoint
    (`absent_keypoint`/`absent_start`/`absent_end`); each carries its instance
    slot, which `save_labels` leaves out when only slot 0 is named. Rows are in
    `(frame, instance)` order.
    """
    views, frames, kps, xys, insts = [], [], [], [], []
    for t, i in sorted(self._frames):
        arr = self._frames[(t, i)]
        vs, ks = np.nonzero(np.isfinite(arr).all(axis=-1))
        for v, k in zip(vs, ks, strict=True):
            views.append(v)
            frames.append(t)
            kps.append(k)
            xys.append(arr[v, k])
            insts.append(i)
    hv, hf, hk, hi = [], [], [], []
    for t, i in sorted(self._hidden):
        vs, ks = np.nonzero(self._hidden[(t, i)])
        for v, k in zip(vs, ks, strict=True):
            hv.append(v)
            hf.append(t)
            hk.append(k)
            hi.append(i)
    akp, astart, aend, ai = [], [], [], []
    for i in sorted(self._absent):
        absent = self._absent[i]
        for k in range(self.n_keypoints):
            diff = np.diff(np.concatenate(([0], absent[:, k].view(np.int8), [0])))
            for s, e in zip(
                np.flatnonzero(diff == 1), np.flatnonzero(diff == -1), strict=True
            ):
                akp.append(k)
                astart.append(s)
                aend.append(e)
                ai.append(i)
    return {
        "view": np.asarray(views, dtype=np.int32),
        "frame": np.asarray(frames, dtype=np.int32),
        "keypoint": np.asarray(kps, dtype=np.int32),
        "xy": np.asarray(xys, dtype=np.float64).reshape(-1, 2),
        "hidden_view": np.asarray(hv, dtype=np.int32),
        "hidden_frame": np.asarray(hf, dtype=np.int32),
        "hidden_keypoint": np.asarray(hk, dtype=np.int32),
        "absent_keypoint": np.asarray(akp, dtype=np.int32),
        "absent_start": np.asarray(astart, dtype=np.int32),
        "absent_end": np.asarray(aend, dtype=np.int32),
        "instance": np.asarray(insts, dtype=np.int32),
        "hidden_instance": np.asarray(hi, dtype=np.int32),
        "absent_instance": np.asarray(ai, dtype=np.int32),
    }

count_complete_frames

count_complete_frames(
    path: str | Path, min_views: int = MIN_CLICKED_VIEWS
) -> int

How many frames of a labels.h5 are complete (ADR-0102), from its columns.

Source code in src/deeperfly/labels/store.py

def count_complete_frames(path: str | Path, min_views: int = MIN_CLICKED_VIEWS) -> int:
    """How many frames of a `labels.h5` are complete (ADR-0102), from its columns."""
    return count_frames(path, min_views)[1]

count_labeled_frames

count_labeled_frames(path: str | Path) -> int

How many frames a labels.h5 labels, reading only its frame index.

Nothing is checked against a project, so every project of a collection can be summarized at once.

Source code in src/deeperfly/labels/store.py

def count_labeled_frames(path: str | Path) -> int:
    """How many frames a `labels.h5` labels, reading only its frame index.

    Nothing is checked against a project, so every project of a collection can be
    summarized at once.
    """
    with h5py.File(path, "r") as f:
        return int(np.unique(f["frame"][:]).size)

label_summary

label_summary(
    directory: str | Path,
) -> tuple[int, int, bool]

A project's labeled and complete (ADR-0102) frame counts, and its unsaved.

From one open of the file open would read; unsaved says it is the sidecar.

Source code in src/deeperfly/labels/files.py

def label_summary(directory: str | Path) -> tuple[int, int, bool]:
    """A project's labeled and complete (ADR-0102) frame counts, and its `unsaved`.

    From one open of the file `open` would read; `unsaved` says it is the sidecar.
    """
    path, unsaved = labels_source(directory)
    if path is None:
        return 0, 0, False
    return *count_frames(path), unsaved

labeled_frame_count

labeled_frame_count(directory: str | Path) -> int

How many frames a project has labels for, from the file open would read.

Source code in src/deeperfly/labels/files.py

def labeled_frame_count(directory: str | Path) -> int:
    """How many frames a project has labels for, from the file `open` would read."""
    path, _ = labels_source(directory)
    return count_labeled_frames(path) if path is not None else 0

labels_source

labels_source(
    directory: str | Path,
) -> tuple[Path | None, bool]

The file a project's labels come from, and whether it is the sidecar.

Source code in src/deeperfly/labels/files.py

def labels_source(directory: str | Path) -> tuple[Path | None, bool]:
    """The file a project's labels come from, and whether it is the sidecar."""
    saved = Path(directory) / LABELS_FILENAME
    recovery = Path(directory) / LABELS_RECOVERY_FILENAME
    if recovery.exists() and (
        not saved.exists() or recovery.stat().st_mtime >= saved.stat().st_mtime
    ):
        return recovery, True
    return (saved, False) if saved.exists() else (None, False)

load_labels

load_labels(
    path: str | Path,
    *,
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> Labels

Load a labels.h5, checked and remapped against the current project.

A pure reordering of the views or keypoints is absorbed.

Raises:

Type	Description
ValueError	If a stored view or keypoint no longer exists, the frame count shrank below a labeled frame, or the format version is unknown.

Source code in src/deeperfly/labels/store.py

def load_labels(
    path: str | Path,
    *,
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> Labels:
    """Load a `labels.h5`, checked and remapped against the current project.

    A pure reordering of the views or keypoints is absorbed.

    Raises:
        ValueError: If a stored view or keypoint no longer exists, the frame count
            shrank below a labeled frame, or the format version is unknown.
    """
    columns = (
        *("view", "frame", "keypoint", "xy"),
        *("hidden_view", "hidden_frame", "hidden_keypoint"),
        *("absent_keypoint", "absent_start", "absent_end"),
    )
    with h5py.File(path, "r") as f:
        version = int(f.attrs["format_version"])
        if version != LABELS_FORMAT_VERSION:
            why = CORNER_ERA if version < 3 else ""
            raise ValueError(
                f"{path} is labels format {version}, expected {LABELS_FORMAT_VERSION}"
                f"{why}"
            )
        stored_views = tuple(json.loads(f.attrs["view_names"]))
        stored_kps = tuple(json.loads(f.attrs["keypoint_names"]))
        coo = {k: f[k][()] for k in columns}
        # A file naming only slot 0 has no instance column: zeros as long as the column
        # beside it, so `_apply_coo` still zips strictly.
        for name, beside in _INSTANCE_COLUMNS_BESIDE:
            coo[name] = f[name][()] if name in f else np.zeros_like(coo[beside])

    view_map = name_positions(stored_views, view_names, "view")
    kp_map = name_positions(stored_kps, keypoint_names, "keypoint")
    if coo["frame"].size and int(coo["frame"].max()) >= n_frames:
        raise ValueError(
            f"{path} labels a frame beyond the recording's {n_frames} frames"
        )

    def remap(m: np.ndarray, a: np.ndarray) -> np.ndarray:
        return m[a] if a.size else a

    labels = Labels(view_names, keypoint_names, n_frames)
    labels._apply_coo(
        {
            **coo,
            "view": remap(view_map, coo["view"]),
            "keypoint": remap(kp_map, coo["keypoint"]),
            "xy": coo["xy"].reshape(-1, 2),
            "hidden_view": remap(view_map, coo["hidden_view"]),
            "hidden_keypoint": remap(kp_map, coo["hidden_keypoint"]),
            "absent_keypoint": remap(kp_map, coo["absent_keypoint"]),
        }
    )
    return labels

save_labels

save_labels(
    labels: Labels,
    path: str | Path,
    *,
    clear_dirty: bool = True,
) -> None

Write a label set to a labels.h5, clearing its dirty flag unless told not to.

The file is replaced whole, so a reader never sees it half-written (ADR-0090). An autosave passes clear_dirty=False, since labels.h5 is still behind.

Source code in src/deeperfly/labels/store.py

def save_labels(labels: Labels, path: str | Path, *, clear_dirty: bool = True) -> None:
    """Write a label set to a `labels.h5`, clearing its dirty flag unless told not to.

    The file is replaced whole, so a reader never sees it half-written (ADR-0090). An
    autosave passes `clear_dirty=False`, since `labels.h5` is still behind.
    """
    coo = labels.to_coo()
    if not any(coo[name].any() for name in _INSTANCE_COLUMNS):
        # A project with one animal writes the file it always has.
        coo = {k: v for k, v in coo.items() if k not in _INSTANCE_COLUMNS}
    with atomic_write(path) as tmp, h5py.File(tmp, "w") as f:
        f.attrs["format_version"] = LABELS_FORMAT_VERSION
        f.attrs["view_names"] = json.dumps(list(labels.view_names))
        f.attrs["keypoint_names"] = json.dumps(list(labels.keypoint_names))
        f.attrs["n_frames"] = labels.n_frames
        for name, arr in coo.items():
            f.create_dataset(name, data=arr)
    if clear_dirty:
        labels.dirty = False

Ground-truth 2D labels for one project, and their labels.h5 file.

Labels is sparse (ADR-0026): a frame -> (V, K, 2) map in memory, COO arrays in HDF5. Beside the pixel an operator authors hidden, a per-cell flag that only excludes the cell from the loss, and absent, a per-frame declaration that a keypoint is not on this animal, which masks it at read time without destroying the pixels. Every cell is keyed by an instance slot, 0 being the one animal of a single-fly project. A file records the names and frame count it was written against, and load_labels remaps a reordering and refuses a file whose views or keypoints are gone.

Labels

Sparse per-(view, frame, keypoint) 2D ground truth for one project.

The V and K axes are positional, in the rig's and skeleton's order; an unlabeled cell is NaN, and dirty tracks unsaved edits. Every accessor takes a keyword-only instance, slot 0 by default.

Source code in src/deeperfly/labels/store.py

class Labels:
    """Sparse per-`(view, frame, keypoint)` 2D ground truth for one project.

    The `V` and `K` axes are positional, in the rig's and skeleton's order; an
    unlabeled cell is NaN, and `dirty` tracks unsaved edits. Every accessor takes a
    keyword-only `instance`, slot 0 by default.
    """

    def __init__(
        self,
        view_names: tuple[str, ...],
        keypoint_names: tuple[str, ...],
        n_frames: int,
    ) -> None:
        """An empty label set over the given views, keypoints and frame count."""
        self.view_names = tuple(view_names)
        self.keypoint_names = tuple(keypoint_names)
        self.n_frames = int(n_frames)
        # Keyed by (frame, instance): an instance slot is one animal of the
        # project, slot 0 being the only animal a single-fly project has.
        self._frames: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K, 2)
        # `hidden` only excludes a cell from the loss: it never moves, hides or
        # triangulates a point, and all four (gt, hidden) combinations are legal.
        self._hidden: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K) bool
        # `absent` is per (frame, keypoint), so it can vary over the frames (autotomy);
        # it masks a point out without destroying the pixels, so un-declaring restores.
        self._absent: dict[int, np.ndarray] = {}  # instance -> (T, K) bool, lazy
        self.dirty = False

    # -- sizes ---------------------------------------------------------------

    @property
    def n_views(self) -> int:
        """The number of views."""
        return len(self.view_names)

    @property
    def n_keypoints(self) -> int:
        """The number of keypoints."""
        return len(self.keypoint_names)

    # -- access --------------------------------------------------------------

    def frame(self, t: int, *, instance: int = 0) -> Float[np.ndarray, "V K 2"]:
        """The `(V, K, 2)` labels of one instance of frame `t`, all NaN if none."""
        self._check_frame(t)
        arr = self._frames.get((t, instance))
        if arr is not None:
            return arr.copy()
        return np.full((self.n_views, self.n_keypoints, 2), np.nan)

    def labeled_frames(self) -> list[int]:
        """Frame indices that carry a labeled point, sorted, each once."""
        return sorted(
            {t for (t, _), arr in self._frames.items() if np.isfinite(arr).any()}
        )

    def labeled_instances(self, t: int) -> list[int]:
        """The instance slots labeled in frame `t`, sorted (sparse track ids)."""
        self._check_frame(t)
        return sorted(
            i
            for (f, i), arr in self._frames.items()
            if f == t and np.isfinite(arr).any()
        )

    def frame_instances(self, t: int) -> list[int]:
        """Every instance slot that holds anything in frame `t`, sorted.

        A pixel, a hidden flag or an absent declaration counts, unlike
        `labeled_instances`, which takes pixels only.
        """
        self._check_frame(t)
        slots = {i for f, i in self._frames.keys() | self._hidden.keys() if f == t}
        slots.update(i for i, mask in self._absent.items() if mask[t].any())
        return sorted(slots)

    def label_counts(self) -> list[int]:
        """How many `(view, keypoint)` cells each of `labeled_frames` carries."""
        counts: dict[int, int] = {}
        for (t, _), arr in self._frames.items():
            counts[t] = counts.get(t, 0) + int(np.isfinite(arr).all(axis=-1).sum())
        return [counts[t] for t in self.labeled_frames()]

    def open_keypoints(self, min_views: int = MIN_CLICKED_VIEWS) -> list[int]:
        """How many keypoints each of `labeled_frames` still lacks to be complete.

        A frame is complete (0) when every non-absent keypoint of each labeled
        instance has at least `min_views` clicked views (ADR-0102). Counted per
        instance, so a second animal with nothing clicked is not owed; one with a
        single click owes its other keypoints.
        """
        owed: dict[int, int] = {}
        for (t, i), arr in self._frames.items():
            if not np.isfinite(arr).any():
                continue
            clicked = np.isfinite(arr).all(axis=-1).sum(axis=0)  # (K,)
            absent = self._absent.get(i)
            short = clicked < min_views
            if absent is not None:
                short &= ~absent[t]
            owed[t] = owed.get(t, 0) + int(short.sum())
        return [owed[t] for t in self.labeled_frames()]

    def frame_effective(
        self, t: int, *, instance: int = 0
    ) -> Float[np.ndarray, "V K 2"]:
        """The frame's GT with absent keypoints masked to NaN -- what the solve sees."""
        arr = self.frame(t, instance=instance)
        absent = self._absent.get(instance)
        if absent is not None:
            arr[:, absent[t]] = np.nan
        return arr

    def hidden_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
        """The `(V, K)` hidden mask for frame `t` (all False if none)."""
        self._check_frame(t)
        mask = self._hidden.get((t, instance))
        if mask is not None:
            return mask.copy()
        return np.zeros((self.n_views, self.n_keypoints), dtype=bool)

    def absent_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
        """The `(K,)` absent mask for frame `t` (all False if none)."""
        self._check_frame(t)
        absent = self._absent.get(instance)
        if absent is None:
            return np.zeros(self.n_keypoints, dtype=bool)
        return absent[t].copy()

    def is_absent(self, frame: int, keypoint: int, *, instance: int = 0) -> bool:
        """Whether a keypoint is declared not-on-this-animal in frame `t`."""
        self._check_frame(frame)
        absent = self._absent.get(instance)
        return bool(absent[frame, keypoint]) if absent is not None else False

    def absent_keypoint(
        self, keypoint: int, *, instance: int = 0
    ) -> Bool[np.ndarray, " T"]:
        """One keypoint's per-frame absent mask over every frame."""
        absent = self._absent.get(instance)
        if absent is None:
            return np.zeros(self.n_frames, dtype=bool)
        return absent[:, keypoint].copy()

    # -- mutation ------------------------------------------------------------

    def set_point(
        self, view: int, frame: int, keypoint: int, xy, *, instance: int = 0
    ) -> None:
        """Set the labeled pixel of one `(view, frame, keypoint)` cell."""
        self._check_frame(frame)
        arr = self._frames.get((frame, instance))
        if arr is None:
            arr = np.full((self.n_views, self.n_keypoints, 2), np.nan)
            self._frames[(frame, instance)] = arr
        arr[view, keypoint] = np.asarray(xy, dtype=float)
        self.dirty = True

    def clear_point(
        self, view: int, frame: int, keypoint: int, *, instance: int = 0
    ) -> None:
        """Remove the label of one `(view, frame, keypoint)` cell, if any."""
        self._check_frame(frame)
        arr = self._frames.get((frame, instance))
        if arr is None:
            return
        arr[view, keypoint] = np.nan
        if not np.isfinite(arr).any():
            del self._frames[(frame, instance)]
        self.dirty = True

    def set_hidden(
        self,
        view: int,
        frame: int,
        keypoint: int,
        value: bool = True,
        *,
        instance: int = 0,
    ) -> None:
        """Set or clear the hidden (exclude-from-loss) flag of one cell."""
        self._check_frame(frame)
        mask = self._hidden.get((frame, instance))
        if mask is None:
            if not value:
                return
            mask = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
            self._hidden[(frame, instance)] = mask
        mask[view, keypoint] = value
        if not mask.any():
            del self._hidden[(frame, instance)]
        self.dirty = True

    def set_absent(
        self, frame: int, keypoint: int, value: bool = True, *, instance: int = 0
    ) -> None:
        """Declare a keypoint absent (or present again) in one frame."""
        self._check_frame(frame)
        absent = self._absent.get(instance)
        if absent is None:
            if not value:
                return
            absent = self._absent_mask(instance)
        absent[frame, keypoint] = value
        self.dirty = True

    def set_absent_every_frame(
        self, keypoint: int, value: bool = True, *, instance: int = 0
    ) -> None:
        """Declare a keypoint absent (or present) in every frame."""
        self.set_absent_keypoint(
            keypoint, np.full(self.n_frames, value), instance=instance
        )

    def set_absent_keypoint(self, keypoint: int, values, *, instance: int = 0) -> None:
        """Replace one keypoint's per-frame absent flags with a whole `(T,)` mask.

        The general form of `set_absent_every_frame`, so an undo can put back any
        pattern.
        """
        values = np.asarray(values, dtype=bool)
        absent = self._absent.get(instance)
        if absent is None:
            if not values.any():
                return
            absent = self._absent_mask(instance)
        absent[:, keypoint] = values
        self.dirty = True

    def clear_instance(self, frame: int, instance: int) -> None:
        """Forget what one instance holds in one frame: points, flags, absences."""
        self._check_frame(frame)
        self._frames.pop((frame, instance), None)
        self._hidden.pop((frame, instance), None)
        absent = self._absent.get(instance)
        if absent is not None:
            absent[frame] = False
        self.dirty = True

    def changed_frames(self, other: "Labels") -> list[int]:
        """The frames whose authored state differs from `other`'s, sorted.

        A pixel, a hidden flag or an absent declaration of any instance counts; an
        empty cell and a cleared one are the same. The two sets share their shape.
        """
        changed: set[int] = set()
        nan = np.full((self.n_views, self.n_keypoints, 2), np.nan)
        for key in self._frames.keys() | other._frames.keys():
            mine, theirs = self._frames.get(key, nan), other._frames.get(key, nan)
            if not np.array_equal(mine, theirs, equal_nan=True):
                changed.add(key[0])
        none = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
        for key in self._hidden.keys() | other._hidden.keys():
            if not np.array_equal(
                self._hidden.get(key, none), other._hidden.get(key, none)
            ):
                changed.add(key[0])
        for instance in self._absent.keys() | other._absent.keys():
            mine, theirs = self._absent.get(instance), other._absent.get(instance)
            if mine is None:
                mine = np.zeros_like(theirs)
            if theirs is None:
                theirs = np.zeros_like(mine)
            changed.update(np.flatnonzero((mine != theirs).any(axis=1)).tolist())
        return sorted(changed)

    def _absent_mask(self, instance: int) -> np.ndarray:
        """One instance's `(T, K)` absent mask, allocated on first use."""
        mask = self._absent.get(instance)
        if mask is None:
            mask = np.zeros((self.n_frames, self.n_keypoints), dtype=bool)
            self._absent[instance] = mask
        return mask

    # -- COO (for persistence) -----------------------------------------------

    def to_coo(self) -> dict[str, np.ndarray]:
        """The authored state as flat COO arrays for persistence.

        Positions ride `view`/`frame`/`keypoint`/`xy`, hidden cells the `hidden_*`
        triples, and absence half-open spans per keypoint
        (`absent_keypoint`/`absent_start`/`absent_end`); each carries its instance
        slot, which `save_labels` leaves out when only slot 0 is named. Rows are in
        `(frame, instance)` order.
        """
        views, frames, kps, xys, insts = [], [], [], [], []
        for t, i in sorted(self._frames):
            arr = self._frames[(t, i)]
            vs, ks = np.nonzero(np.isfinite(arr).all(axis=-1))
            for v, k in zip(vs, ks, strict=True):
                views.append(v)
                frames.append(t)
                kps.append(k)
                xys.append(arr[v, k])
                insts.append(i)
        hv, hf, hk, hi = [], [], [], []
        for t, i in sorted(self._hidden):
            vs, ks = np.nonzero(self._hidden[(t, i)])
            for v, k in zip(vs, ks, strict=True):
                hv.append(v)
                hf.append(t)
                hk.append(k)
                hi.append(i)
        akp, astart, aend, ai = [], [], [], []
        for i in sorted(self._absent):
            absent = self._absent[i]
            for k in range(self.n_keypoints):
                diff = np.diff(np.concatenate(([0], absent[:, k].view(np.int8), [0])))
                for s, e in zip(
                    np.flatnonzero(diff == 1), np.flatnonzero(diff == -1), strict=True
                ):
                    akp.append(k)
                    astart.append(s)
                    aend.append(e)
                    ai.append(i)
        return {
            "view": np.asarray(views, dtype=np.int32),
            "frame": np.asarray(frames, dtype=np.int32),
            "keypoint": np.asarray(kps, dtype=np.int32),
            "xy": np.asarray(xys, dtype=np.float64).reshape(-1, 2),
            "hidden_view": np.asarray(hv, dtype=np.int32),
            "hidden_frame": np.asarray(hf, dtype=np.int32),
            "hidden_keypoint": np.asarray(hk, dtype=np.int32),
            "absent_keypoint": np.asarray(akp, dtype=np.int32),
            "absent_start": np.asarray(astart, dtype=np.int32),
            "absent_end": np.asarray(aend, dtype=np.int32),
            "instance": np.asarray(insts, dtype=np.int32),
            "hidden_instance": np.asarray(hi, dtype=np.int32),
            "absent_instance": np.asarray(ai, dtype=np.int32),
        }

    def _apply_coo(self, coo: Mapping[str, np.ndarray]) -> None:
        """Load the flat COO arrays of `to_coo` back into an empty store.

        Every column must be there; `load_labels` supplies zeros for one a file
        predates, so the strict zips stay a real check.
        """
        for v, t, k, i, p in zip(
            coo["view"],
            coo["frame"],
            coo["keypoint"],
            coo["instance"],
            coo["xy"],
            strict=True,
        ):
            self.set_point(int(v), int(t), int(k), p, instance=int(i))
        for v, t, k, i in zip(
            coo["hidden_view"],
            coo["hidden_frame"],
            coo["hidden_keypoint"],
            coo["hidden_instance"],
            strict=True,
        ):
            self.set_hidden(int(v), int(t), int(k), instance=int(i))
        for k, s, e, i in zip(
            coo["absent_keypoint"],
            coo["absent_start"],
            coo["absent_end"],
            coo["absent_instance"],
            strict=True,
        ):
            self._absent_mask(int(i))[int(s) : int(e), int(k)] = True
        self.dirty = False

    def _check_frame(self, t: int) -> None:
        if not 0 <= t < self.n_frames:
            raise IndexError(f"frame {t} out of range [0, {self.n_frames})")

n_keypoints property

n_keypoints: int

The number of keypoints.

n_views property

n_views: int

The number of views.

__init__

__init__(
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> None

An empty label set over the given views, keypoints and frame count.

Source code in src/deeperfly/labels/store.py

def __init__(
    self,
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> None:
    """An empty label set over the given views, keypoints and frame count."""
    self.view_names = tuple(view_names)
    self.keypoint_names = tuple(keypoint_names)
    self.n_frames = int(n_frames)
    # Keyed by (frame, instance): an instance slot is one animal of the
    # project, slot 0 being the only animal a single-fly project has.
    self._frames: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K, 2)
    # `hidden` only excludes a cell from the loss: it never moves, hides or
    # triangulates a point, and all four (gt, hidden) combinations are legal.
    self._hidden: dict[tuple[int, int], np.ndarray] = {}  # -> (V, K) bool
    # `absent` is per (frame, keypoint), so it can vary over the frames (autotomy);
    # it masks a point out without destroying the pixels, so un-declaring restores.
    self._absent: dict[int, np.ndarray] = {}  # instance -> (T, K) bool, lazy
    self.dirty = False

absent_frame

absent_frame(t: int, *, instance: int = 0) -> np.ndarray

The (K,) absent mask for frame t (all False if none).

Source code in src/deeperfly/labels/store.py

def absent_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
    """The `(K,)` absent mask for frame `t` (all False if none)."""
    self._check_frame(t)
    absent = self._absent.get(instance)
    if absent is None:
        return np.zeros(self.n_keypoints, dtype=bool)
    return absent[t].copy()

absent_keypoint

absent_keypoint(
    keypoint: int, *, instance: int = 0
) -> Bool[np.ndarray, " T"]

One keypoint's per-frame absent mask over every frame.

Source code in src/deeperfly/labels/store.py

def absent_keypoint(
    self, keypoint: int, *, instance: int = 0
) -> Bool[np.ndarray, " T"]:
    """One keypoint's per-frame absent mask over every frame."""
    absent = self._absent.get(instance)
    if absent is None:
        return np.zeros(self.n_frames, dtype=bool)
    return absent[:, keypoint].copy()

changed_frames

changed_frames(other: Labels) -> list[int]

The frames whose authored state differs from other's, sorted.

A pixel, a hidden flag or an absent declaration of any instance counts; an empty cell and a cleared one are the same. The two sets share their shape.

Source code in src/deeperfly/labels/store.py

def changed_frames(self, other: "Labels") -> list[int]:
    """The frames whose authored state differs from `other`'s, sorted.

    A pixel, a hidden flag or an absent declaration of any instance counts; an
    empty cell and a cleared one are the same. The two sets share their shape.
    """
    changed: set[int] = set()
    nan = np.full((self.n_views, self.n_keypoints, 2), np.nan)
    for key in self._frames.keys() | other._frames.keys():
        mine, theirs = self._frames.get(key, nan), other._frames.get(key, nan)
        if not np.array_equal(mine, theirs, equal_nan=True):
            changed.add(key[0])
    none = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
    for key in self._hidden.keys() | other._hidden.keys():
        if not np.array_equal(
            self._hidden.get(key, none), other._hidden.get(key, none)
        ):
            changed.add(key[0])
    for instance in self._absent.keys() | other._absent.keys():
        mine, theirs = self._absent.get(instance), other._absent.get(instance)
        if mine is None:
            mine = np.zeros_like(theirs)
        if theirs is None:
            theirs = np.zeros_like(mine)
        changed.update(np.flatnonzero((mine != theirs).any(axis=1)).tolist())
    return sorted(changed)

clear_instance

clear_instance(frame: int, instance: int) -> None

Forget what one instance holds in one frame: points, flags, absences.

Source code in src/deeperfly/labels/store.py

def clear_instance(self, frame: int, instance: int) -> None:
    """Forget what one instance holds in one frame: points, flags, absences."""
    self._check_frame(frame)
    self._frames.pop((frame, instance), None)
    self._hidden.pop((frame, instance), None)
    absent = self._absent.get(instance)
    if absent is not None:
        absent[frame] = False
    self.dirty = True

clear_point

clear_point(
    view: int,
    frame: int,
    keypoint: int,
    *,
    instance: int = 0,
) -> None

Remove the label of one (view, frame, keypoint) cell, if any.

Source code in src/deeperfly/labels/store.py

def clear_point(
    self, view: int, frame: int, keypoint: int, *, instance: int = 0
) -> None:
    """Remove the label of one `(view, frame, keypoint)` cell, if any."""
    self._check_frame(frame)
    arr = self._frames.get((frame, instance))
    if arr is None:
        return
    arr[view, keypoint] = np.nan
    if not np.isfinite(arr).any():
        del self._frames[(frame, instance)]
    self.dirty = True

frame

frame(
    t: int, *, instance: int = 0
) -> Float[np.ndarray, "V K 2"]

The (V, K, 2) labels of one instance of frame t, all NaN if none.

Source code in src/deeperfly/labels/store.py

def frame(self, t: int, *, instance: int = 0) -> Float[np.ndarray, "V K 2"]:
    """The `(V, K, 2)` labels of one instance of frame `t`, all NaN if none."""
    self._check_frame(t)
    arr = self._frames.get((t, instance))
    if arr is not None:
        return arr.copy()
    return np.full((self.n_views, self.n_keypoints, 2), np.nan)

frame_effective

frame_effective(
    t: int, *, instance: int = 0
) -> Float[np.ndarray, "V K 2"]

The frame's GT with absent keypoints masked to NaN -- what the solve sees.

Source code in src/deeperfly/labels/store.py

def frame_effective(
    self, t: int, *, instance: int = 0
) -> Float[np.ndarray, "V K 2"]:
    """The frame's GT with absent keypoints masked to NaN -- what the solve sees."""
    arr = self.frame(t, instance=instance)
    absent = self._absent.get(instance)
    if absent is not None:
        arr[:, absent[t]] = np.nan
    return arr

frame_instances

frame_instances(t: int) -> list[int]

Every instance slot that holds anything in frame t, sorted.

A pixel, a hidden flag or an absent declaration counts, unlike labeled_instances, which takes pixels only.

Source code in src/deeperfly/labels/store.py

def frame_instances(self, t: int) -> list[int]:
    """Every instance slot that holds anything in frame `t`, sorted.

    A pixel, a hidden flag or an absent declaration counts, unlike
    `labeled_instances`, which takes pixels only.
    """
    self._check_frame(t)
    slots = {i for f, i in self._frames.keys() | self._hidden.keys() if f == t}
    slots.update(i for i, mask in self._absent.items() if mask[t].any())
    return sorted(slots)

hidden_frame

hidden_frame(t: int, *, instance: int = 0) -> np.ndarray

The (V, K) hidden mask for frame t (all False if none).

Source code in src/deeperfly/labels/store.py

def hidden_frame(self, t: int, *, instance: int = 0) -> np.ndarray:
    """The `(V, K)` hidden mask for frame `t` (all False if none)."""
    self._check_frame(t)
    mask = self._hidden.get((t, instance))
    if mask is not None:
        return mask.copy()
    return np.zeros((self.n_views, self.n_keypoints), dtype=bool)

is_absent

is_absent(
    frame: int, keypoint: int, *, instance: int = 0
) -> bool

Whether a keypoint is declared not-on-this-animal in frame t.

Source code in src/deeperfly/labels/store.py

def is_absent(self, frame: int, keypoint: int, *, instance: int = 0) -> bool:
    """Whether a keypoint is declared not-on-this-animal in frame `t`."""
    self._check_frame(frame)
    absent = self._absent.get(instance)
    return bool(absent[frame, keypoint]) if absent is not None else False

label_counts

label_counts() -> list[int]

How many (view, keypoint) cells each of labeled_frames carries.

Source code in src/deeperfly/labels/store.py

def label_counts(self) -> list[int]:
    """How many `(view, keypoint)` cells each of `labeled_frames` carries."""
    counts: dict[int, int] = {}
    for (t, _), arr in self._frames.items():
        counts[t] = counts.get(t, 0) + int(np.isfinite(arr).all(axis=-1).sum())
    return [counts[t] for t in self.labeled_frames()]

labeled_frames

labeled_frames() -> list[int]

Frame indices that carry a labeled point, sorted, each once.

Source code in src/deeperfly/labels/store.py

def labeled_frames(self) -> list[int]:
    """Frame indices that carry a labeled point, sorted, each once."""
    return sorted(
        {t for (t, _), arr in self._frames.items() if np.isfinite(arr).any()}
    )

labeled_instances

labeled_instances(t: int) -> list[int]

The instance slots labeled in frame t, sorted (sparse track ids).

Source code in src/deeperfly/labels/store.py

def labeled_instances(self, t: int) -> list[int]:
    """The instance slots labeled in frame `t`, sorted (sparse track ids)."""
    self._check_frame(t)
    return sorted(
        i
        for (f, i), arr in self._frames.items()
        if f == t and np.isfinite(arr).any()
    )

open_keypoints

open_keypoints(
    min_views: int = MIN_CLICKED_VIEWS,
) -> list[int]

How many keypoints each of labeled_frames still lacks to be complete.

A frame is complete (0) when every non-absent keypoint of each labeled instance has at least min_views clicked views (ADR-0102). Counted per instance, so a second animal with nothing clicked is not owed; one with a single click owes its other keypoints.

Source code in src/deeperfly/labels/store.py

def open_keypoints(self, min_views: int = MIN_CLICKED_VIEWS) -> list[int]:
    """How many keypoints each of `labeled_frames` still lacks to be complete.

    A frame is complete (0) when every non-absent keypoint of each labeled
    instance has at least `min_views` clicked views (ADR-0102). Counted per
    instance, so a second animal with nothing clicked is not owed; one with a
    single click owes its other keypoints.
    """
    owed: dict[int, int] = {}
    for (t, i), arr in self._frames.items():
        if not np.isfinite(arr).any():
            continue
        clicked = np.isfinite(arr).all(axis=-1).sum(axis=0)  # (K,)
        absent = self._absent.get(i)
        short = clicked < min_views
        if absent is not None:
            short &= ~absent[t]
        owed[t] = owed.get(t, 0) + int(short.sum())
    return [owed[t] for t in self.labeled_frames()]

set_absent

set_absent(
    frame: int,
    keypoint: int,
    value: bool = True,
    *,
    instance: int = 0,
) -> None

Declare a keypoint absent (or present again) in one frame.

Source code in src/deeperfly/labels/store.py

def set_absent(
    self, frame: int, keypoint: int, value: bool = True, *, instance: int = 0
) -> None:
    """Declare a keypoint absent (or present again) in one frame."""
    self._check_frame(frame)
    absent = self._absent.get(instance)
    if absent is None:
        if not value:
            return
        absent = self._absent_mask(instance)
    absent[frame, keypoint] = value
    self.dirty = True

set_absent_every_frame

set_absent_every_frame(
    keypoint: int, value: bool = True, *, instance: int = 0
) -> None

Declare a keypoint absent (or present) in every frame.

Source code in src/deeperfly/labels/store.py

def set_absent_every_frame(
    self, keypoint: int, value: bool = True, *, instance: int = 0
) -> None:
    """Declare a keypoint absent (or present) in every frame."""
    self.set_absent_keypoint(
        keypoint, np.full(self.n_frames, value), instance=instance
    )

set_absent_keypoint

set_absent_keypoint(
    keypoint: int, values, *, instance: int = 0
) -> None

Replace one keypoint's per-frame absent flags with a whole (T,) mask.

The general form of set_absent_every_frame, so an undo can put back any pattern.

Source code in src/deeperfly/labels/store.py

def set_absent_keypoint(self, keypoint: int, values, *, instance: int = 0) -> None:
    """Replace one keypoint's per-frame absent flags with a whole `(T,)` mask.

    The general form of `set_absent_every_frame`, so an undo can put back any
    pattern.
    """
    values = np.asarray(values, dtype=bool)
    absent = self._absent.get(instance)
    if absent is None:
        if not values.any():
            return
        absent = self._absent_mask(instance)
    absent[:, keypoint] = values
    self.dirty = True

set_hidden

set_hidden(
    view: int,
    frame: int,
    keypoint: int,
    value: bool = True,
    *,
    instance: int = 0,
) -> None

Set or clear the hidden (exclude-from-loss) flag of one cell.

Source code in src/deeperfly/labels/store.py

def set_hidden(
    self,
    view: int,
    frame: int,
    keypoint: int,
    value: bool = True,
    *,
    instance: int = 0,
) -> None:
    """Set or clear the hidden (exclude-from-loss) flag of one cell."""
    self._check_frame(frame)
    mask = self._hidden.get((frame, instance))
    if mask is None:
        if not value:
            return
        mask = np.zeros((self.n_views, self.n_keypoints), dtype=bool)
        self._hidden[(frame, instance)] = mask
    mask[view, keypoint] = value
    if not mask.any():
        del self._hidden[(frame, instance)]
    self.dirty = True

set_point

set_point(
    view: int,
    frame: int,
    keypoint: int,
    xy,
    *,
    instance: int = 0,
) -> None

Set the labeled pixel of one (view, frame, keypoint) cell.

Source code in src/deeperfly/labels/store.py

def set_point(
    self, view: int, frame: int, keypoint: int, xy, *, instance: int = 0
) -> None:
    """Set the labeled pixel of one `(view, frame, keypoint)` cell."""
    self._check_frame(frame)
    arr = self._frames.get((frame, instance))
    if arr is None:
        arr = np.full((self.n_views, self.n_keypoints, 2), np.nan)
        self._frames[(frame, instance)] = arr
    arr[view, keypoint] = np.asarray(xy, dtype=float)
    self.dirty = True

to_coo

to_coo() -> dict[str, np.ndarray]

The authored state as flat COO arrays for persistence.

Positions ride view/frame/keypoint/xy, hidden cells the hidden_* triples, and absence half-open spans per keypoint (absent_keypoint/absent_start/absent_end); each carries its instance slot, which save_labels leaves out when only slot 0 is named. Rows are in (frame, instance) order.

Source code in src/deeperfly/labels/store.py

def to_coo(self) -> dict[str, np.ndarray]:
    """The authored state as flat COO arrays for persistence.

    Positions ride `view`/`frame`/`keypoint`/`xy`, hidden cells the `hidden_*`
    triples, and absence half-open spans per keypoint
    (`absent_keypoint`/`absent_start`/`absent_end`); each carries its instance
    slot, which `save_labels` leaves out when only slot 0 is named. Rows are in
    `(frame, instance)` order.
    """
    views, frames, kps, xys, insts = [], [], [], [], []
    for t, i in sorted(self._frames):
        arr = self._frames[(t, i)]
        vs, ks = np.nonzero(np.isfinite(arr).all(axis=-1))
        for v, k in zip(vs, ks, strict=True):
            views.append(v)
            frames.append(t)
            kps.append(k)
            xys.append(arr[v, k])
            insts.append(i)
    hv, hf, hk, hi = [], [], [], []
    for t, i in sorted(self._hidden):
        vs, ks = np.nonzero(self._hidden[(t, i)])
        for v, k in zip(vs, ks, strict=True):
            hv.append(v)
            hf.append(t)
            hk.append(k)
            hi.append(i)
    akp, astart, aend, ai = [], [], [], []
    for i in sorted(self._absent):
        absent = self._absent[i]
        for k in range(self.n_keypoints):
            diff = np.diff(np.concatenate(([0], absent[:, k].view(np.int8), [0])))
            for s, e in zip(
                np.flatnonzero(diff == 1), np.flatnonzero(diff == -1), strict=True
            ):
                akp.append(k)
                astart.append(s)
                aend.append(e)
                ai.append(i)
    return {
        "view": np.asarray(views, dtype=np.int32),
        "frame": np.asarray(frames, dtype=np.int32),
        "keypoint": np.asarray(kps, dtype=np.int32),
        "xy": np.asarray(xys, dtype=np.float64).reshape(-1, 2),
        "hidden_view": np.asarray(hv, dtype=np.int32),
        "hidden_frame": np.asarray(hf, dtype=np.int32),
        "hidden_keypoint": np.asarray(hk, dtype=np.int32),
        "absent_keypoint": np.asarray(akp, dtype=np.int32),
        "absent_start": np.asarray(astart, dtype=np.int32),
        "absent_end": np.asarray(aend, dtype=np.int32),
        "instance": np.asarray(insts, dtype=np.int32),
        "hidden_instance": np.asarray(hi, dtype=np.int32),
        "absent_instance": np.asarray(ai, dtype=np.int32),
    }

count_complete_frames

count_complete_frames(
    path: str | Path, min_views: int = MIN_CLICKED_VIEWS
) -> int

How many frames of a labels.h5 are complete (ADR-0102), from its columns.

Source code in src/deeperfly/labels/store.py

def count_complete_frames(path: str | Path, min_views: int = MIN_CLICKED_VIEWS) -> int:
    """How many frames of a `labels.h5` are complete (ADR-0102), from its columns."""
    return count_frames(path, min_views)[1]

count_frames

count_frames(
    path: str | Path, min_views: int = MIN_CLICKED_VIEWS
) -> tuple[int, int]

How many frames a labels.h5 labels, and how many of them are complete.

Complete by the same rule as Labels.open_keypoints returning 0 (ADR-0102), without a project, video or Labels object: only the sparse click and absent-range columns are read, in one open of the file.

Source code in src/deeperfly/labels/store.py

def count_frames(
    path: str | Path, min_views: int = MIN_CLICKED_VIEWS
) -> tuple[int, int]:
    """How many frames a `labels.h5` labels, and how many of them are complete.

    Complete by the same rule as `Labels.open_keypoints` returning 0 (ADR-0102),
    without a project, video or `Labels` object: only the sparse click and
    absent-range columns are read, in one open of the file.
    """
    with h5py.File(path, "r") as f:

        def column(name: str) -> np.ndarray:
            return np.asarray(cast(h5py.Dataset, f[name])[()])

        n_kp = len(json.loads(str(f.attrs["keypoint_names"])))
        frame, kp = column("frame"), column("keypoint")
        inst = column("instance") if "instance" in f else np.zeros_like(frame)
        a_kp, a_start, a_end = (column(k) for k in _ABSENT_COLUMNS)
        a_inst = column("absent_instance") if "absent_instance" in f else 0 * a_kp
    # One row per click, so a (frame, animal, keypoint)'s clicked views are its rows.
    cells, views = np.unique(
        np.stack([frame, inst, kp]).astype(np.int64), axis=1, return_counts=True
    )
    covered: dict[tuple[int, int], set[int]] = {}
    for (t, i, k), n in zip(cells.T.tolist(), views.tolist(), strict=True):
        covered.setdefault((t, i), set())
        if n >= min_views:
            covered[t, i].add(k)
    complete: dict[int, bool] = {}
    for (t, i), ok in covered.items():
        spans = (a_inst == i) & (a_start <= t) & (t < a_end)
        ok = ok | set(a_kp[spans].tolist())
        complete[t] = complete.get(t, True) and len(ok) >= n_kp
    return int(np.unique(frame).size), sum(complete.values())

count_labeled_frames

count_labeled_frames(path: str | Path) -> int

How many frames a labels.h5 labels, reading only its frame index.

Nothing is checked against a project, so every project of a collection can be summarized at once.

Source code in src/deeperfly/labels/store.py

def count_labeled_frames(path: str | Path) -> int:
    """How many frames a `labels.h5` labels, reading only its frame index.

    Nothing is checked against a project, so every project of a collection can be
    summarized at once.
    """
    with h5py.File(path, "r") as f:
        return int(np.unique(f["frame"][:]).size)

load_labels

load_labels(
    path: str | Path,
    *,
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> Labels

Load a labels.h5, checked and remapped against the current project.

A pure reordering of the views or keypoints is absorbed.

Raises:

Type	Description
ValueError	If a stored view or keypoint no longer exists, the frame count shrank below a labeled frame, or the format version is unknown.

Source code in src/deeperfly/labels/store.py

def load_labels(
    path: str | Path,
    *,
    view_names: tuple[str, ...],
    keypoint_names: tuple[str, ...],
    n_frames: int,
) -> Labels:
    """Load a `labels.h5`, checked and remapped against the current project.

    A pure reordering of the views or keypoints is absorbed.

    Raises:
        ValueError: If a stored view or keypoint no longer exists, the frame count
            shrank below a labeled frame, or the format version is unknown.
    """
    columns = (
        *("view", "frame", "keypoint", "xy"),
        *("hidden_view", "hidden_frame", "hidden_keypoint"),
        *("absent_keypoint", "absent_start", "absent_end"),
    )
    with h5py.File(path, "r") as f:
        version = int(f.attrs["format_version"])
        if version != LABELS_FORMAT_VERSION:
            why = CORNER_ERA if version < 3 else ""
            raise ValueError(
                f"{path} is labels format {version}, expected {LABELS_FORMAT_VERSION}"
                f"{why}"
            )
        stored_views = tuple(json.loads(f.attrs["view_names"]))
        stored_kps = tuple(json.loads(f.attrs["keypoint_names"]))
        coo = {k: f[k][()] for k in columns}
        # A file naming only slot 0 has no instance column: zeros as long as the column
        # beside it, so `_apply_coo` still zips strictly.
        for name, beside in _INSTANCE_COLUMNS_BESIDE:
            coo[name] = f[name][()] if name in f else np.zeros_like(coo[beside])

    view_map = name_positions(stored_views, view_names, "view")
    kp_map = name_positions(stored_kps, keypoint_names, "keypoint")
    if coo["frame"].size and int(coo["frame"].max()) >= n_frames:
        raise ValueError(
            f"{path} labels a frame beyond the recording's {n_frames} frames"
        )

    def remap(m: np.ndarray, a: np.ndarray) -> np.ndarray:
        return m[a] if a.size else a

    labels = Labels(view_names, keypoint_names, n_frames)
    labels._apply_coo(
        {
            **coo,
            "view": remap(view_map, coo["view"]),
            "keypoint": remap(kp_map, coo["keypoint"]),
            "xy": coo["xy"].reshape(-1, 2),
            "hidden_view": remap(view_map, coo["hidden_view"]),
            "hidden_keypoint": remap(kp_map, coo["hidden_keypoint"]),
            "absent_keypoint": remap(kp_map, coo["absent_keypoint"]),
        }
    )
    return labels

name_positions

name_positions(names, among, kind: str) -> np.ndarray

The index of each of names in among, matched by name.

Source code in src/deeperfly/labels/store.py

def name_positions(names, among, kind: str) -> np.ndarray:
    """The index of each of `names` in `among`, matched by name."""
    pos = {name: i for i, name in enumerate(among)}
    missing = [name for name in names if name not in pos]
    if missing:
        raise ValueError(f"{kind}s {missing} are not among {list(among)}")
    return np.asarray([pos[name] for name in names], dtype=np.int32)

reindex_keypoints

reindex_keypoints(
    labels: Labels, keypoint_names: tuple[str, ...]
) -> Labels

A copy of labels with its keypoint axis remapped by name to a new order.

Added keypoints get no labels. The instance slot, the hidden flags and the absent declarations are remapped as the points are; one on a dropped keypoint goes with it; the caller refuses any drop that would strand a label.

Source code in src/deeperfly/labels/store.py

def reindex_keypoints(labels: Labels, keypoint_names: tuple[str, ...]) -> Labels:
    """A copy of `labels` with its keypoint axis remapped by name to a new order.

    Added keypoints get no labels. The instance slot, the hidden flags and the absent
    declarations are remapped as the points are; one on a dropped keypoint goes with
    it; the caller refuses any drop that would strand a label.
    """
    out = Labels(labels.view_names, keypoint_names, labels.n_frames)
    pos = {name: i for i, name in enumerate(keypoint_names)}
    old_names = labels.keypoint_names
    coo = labels.to_coo()
    for v, t, k, i, xy in zip(
        coo["view"],
        coo["frame"],
        coo["keypoint"],
        coo["instance"],
        coo["xy"],
        strict=True,
    ):
        out.set_point(int(v), int(t), pos[old_names[int(k)]], xy, instance=int(i))
    for v, t, k, i in zip(
        coo["hidden_view"],
        coo["hidden_frame"],
        coo["hidden_keypoint"],
        coo["hidden_instance"],
        strict=True,
    ):
        new = pos.get(old_names[int(k)])
        if new is not None:
            out.set_hidden(int(v), int(t), new, True, instance=int(i))
    # Absence is stored as spans per keypoint; each keypoint's mask is rebuilt whole.
    absent: dict[tuple[int, int], np.ndarray] = {}
    for k, s, e, i in zip(
        coo["absent_keypoint"],
        coo["absent_start"],
        coo["absent_end"],
        coo["absent_instance"],
        strict=True,
    ):
        new = pos.get(old_names[int(k)])
        if new is None:
            continue
        mask = absent.setdefault((int(i), new), np.zeros(labels.n_frames, dtype=bool))
        mask[int(s) : int(e)] = True
    for (i, k), mask in absent.items():
        out.set_absent_keypoint(k, mask, instance=i)
    return out

save_labels

save_labels(
    labels: Labels,
    path: str | Path,
    *,
    clear_dirty: bool = True,
) -> None

Write a label set to a labels.h5, clearing its dirty flag unless told not to.

The file is replaced whole, so a reader never sees it half-written (ADR-0090). An autosave passes clear_dirty=False, since labels.h5 is still behind.

Source code in src/deeperfly/labels/store.py

def save_labels(labels: Labels, path: str | Path, *, clear_dirty: bool = True) -> None:
    """Write a label set to a `labels.h5`, clearing its dirty flag unless told not to.

    The file is replaced whole, so a reader never sees it half-written (ADR-0090). An
    autosave passes `clear_dirty=False`, since `labels.h5` is still behind.
    """
    coo = labels.to_coo()
    if not any(coo[name].any() for name in _INSTANCE_COLUMNS):
        # A project with one animal writes the file it always has.
        coo = {k: v for k, v in coo.items() if k not in _INSTANCE_COLUMNS}
    with atomic_write(path) as tmp, h5py.File(tmp, "w") as f:
        f.attrs["format_version"] = LABELS_FORMAT_VERSION
        f.attrs["view_names"] = json.dumps(list(labels.view_names))
        f.attrs["keypoint_names"] = json.dumps(list(labels.keypoint_names))
        f.attrs["n_frames"] = labels.n_frames
        for name, arr in coo.items():
            f.create_dataset(name, data=arr)
    if clear_dirty:
        labels.dirty = False

stored_n_frames

stored_n_frames(path: str | Path) -> int

The frame count a labels.h5 was written against, read from its attributes.

What a whole-file rewrite (a renamed view, a new skeleton) reloads it with, so no video is opened.

Source code in src/deeperfly/labels/store.py

def stored_n_frames(path: str | Path) -> int:
    """The frame count a `labels.h5` was written against, read from its attributes.

    What a whole-file rewrite (a renamed view, a new skeleton) reloads it with, so no
    video is opened.
    """
    with h5py.File(path, "r") as f:
        return int(np.asarray(f.attrs["n_frames"]))

A project's label files: labels.h5 and its recovery sidecar.

LabelFiles carries the directory with the key a set is checked against (the views, the skeleton's points, the frame count); the two lookups that need no key are functions over the directory. The sidecar, labels.autosave.h5, is written after each GUI edit and deleted on a save, and while it is at least as new as labels.h5 the labels load from it, marked dirty. A discard sets the unsaved labels aside as labels.discarded.h5, which a restore puts back until an edit or a save makes it stale (ADR-0120).

LabelFiles dataclass

One project's label files, keyed to the project they are read against.

Attributes:

Name	Type	Description
directory	Path	The project folder holding the files, whether or not it exists yet.
view_names	tuple[str, ...]	The project's views, in order.
keypoint_names	tuple[str, ...]	The skeleton's points, in order.
n_frames	int	The project's frame count.

Source code in src/deeperfly/labels/files.py

@dataclass(frozen=True)
class LabelFiles:
    """One project's label files, keyed to the project they are read against.

    Attributes:
        directory: The project folder holding the files, whether or not it exists yet.
        view_names: The project's views, in order.
        keypoint_names: The skeleton's points, in order.
        n_frames: The project's frame count.
    """

    directory: Path
    view_names: tuple[str, ...]
    keypoint_names: tuple[str, ...]
    n_frames: int

    @property
    def labels_path(self) -> Path:
        """Where the project's `labels.h5` lives."""
        return self.directory / LABELS_FILENAME

    @property
    def recovery_path(self) -> Path:
        """Where the crash-recovery autosave sidecar lives."""
        return self.directory / LABELS_RECOVERY_FILENAME

    def load(self) -> Labels:
        """The saved labels, or an empty set; a reordered skeleton is remapped."""
        return self._read(self.labels_path)

    def open(self) -> tuple[Labels, bool]:
        """The labels for editing and whether they came from the recovery sidecar.

        Recovered labels load dirty, so the operator saves or discards them.
        """
        path, recovered = labels_source(self.directory)
        if recovered and path is not None:
            labels = self._read(path)
            labels.dirty = True
            return labels, True
        return self.load(), False

    @property
    def discarded_path(self) -> Path:
        """Where the labels the last discard threw away are kept (ADR-0120)."""
        return self.directory / LABELS_DISCARDED_FILENAME

    def save(self, labels: Labels) -> None:
        """Persist the labels to `labels.h5`, clearing the sidecar and any discard."""
        self.directory.mkdir(parents=True, exist_ok=True)
        save_labels(labels, self.labels_path)
        self.recovery_path.unlink(missing_ok=True)
        self.discarded_path.unlink(missing_ok=True)

    def autosave(self, labels: Labels) -> None:
        """Write in-progress labels to the recovery sidecar, keeping them dirty.

        A discard kept from before is stale now: restoring it would drop this edit.
        """
        self.directory.mkdir(parents=True, exist_ok=True)
        save_labels(labels, self.recovery_path, clear_dirty=False)
        self.discarded_path.unlink(missing_ok=True)

    def discard_unsaved(self, labels: Labels | None = None) -> Labels:
        """Set the unsaved labels aside and reload the saved ones.

        `labels` are the unsaved ones, an open session's; without them the sidecar is
        what is set aside. Either way it lands at `discarded_path`, for
        `restore_discarded`.
        """
        if labels is not None and labels.dirty:
            save_labels(labels, self.discarded_path, clear_dirty=False)
            self.recovery_path.unlink(missing_ok=True)
        elif self.recovery_path.exists():
            os.replace(self.recovery_path, self.discarded_path)
        return self.load()

    def restore_discarded(self) -> bool:
        """Put what the last discard set aside back as the sidecar; whether there was.

        It is touched as it lands, so the labels load from it again, dirty.
        """
        if not self.discarded_path.exists():
            return False
        os.replace(self.discarded_path, self.recovery_path)
        os.utime(self.recovery_path)
        return True

    def _read(self, path: Path) -> Labels:
        if path.exists():
            return load_labels(
                path,
                view_names=self.view_names,
                keypoint_names=self.keypoint_names,
                n_frames=self.n_frames,
            )
        return Labels(self.view_names, self.keypoint_names, self.n_frames)

discarded_path property

discarded_path: Path

Where the labels the last discard threw away are kept (ADR-0120).

labels_path property

labels_path: Path

Where the project's labels.h5 lives.

recovery_path property

recovery_path: Path

Where the crash-recovery autosave sidecar lives.

autosave

autosave(labels: Labels) -> None

Write in-progress labels to the recovery sidecar, keeping them dirty.

A discard kept from before is stale now: restoring it would drop this edit.

Source code in src/deeperfly/labels/files.py

def autosave(self, labels: Labels) -> None:
    """Write in-progress labels to the recovery sidecar, keeping them dirty.

    A discard kept from before is stale now: restoring it would drop this edit.
    """
    self.directory.mkdir(parents=True, exist_ok=True)
    save_labels(labels, self.recovery_path, clear_dirty=False)
    self.discarded_path.unlink(missing_ok=True)

discard_unsaved

discard_unsaved(labels: Labels | None = None) -> Labels

Set the unsaved labels aside and reload the saved ones.

labels are the unsaved ones, an open session's; without them the sidecar is what is set aside. Either way it lands at discarded_path, for restore_discarded.

Source code in src/deeperfly/labels/files.py

def discard_unsaved(self, labels: Labels | None = None) -> Labels:
    """Set the unsaved labels aside and reload the saved ones.

    `labels` are the unsaved ones, an open session's; without them the sidecar is
    what is set aside. Either way it lands at `discarded_path`, for
    `restore_discarded`.
    """
    if labels is not None and labels.dirty:
        save_labels(labels, self.discarded_path, clear_dirty=False)
        self.recovery_path.unlink(missing_ok=True)
    elif self.recovery_path.exists():
        os.replace(self.recovery_path, self.discarded_path)
    return self.load()

load

load() -> Labels

The saved labels, or an empty set; a reordered skeleton is remapped.

Source code in src/deeperfly/labels/files.py

def load(self) -> Labels:
    """The saved labels, or an empty set; a reordered skeleton is remapped."""
    return self._read(self.labels_path)

open

open() -> tuple[Labels, bool]

The labels for editing and whether they came from the recovery sidecar.

Recovered labels load dirty, so the operator saves or discards them.

Source code in src/deeperfly/labels/files.py

def open(self) -> tuple[Labels, bool]:
    """The labels for editing and whether they came from the recovery sidecar.

    Recovered labels load dirty, so the operator saves or discards them.
    """
    path, recovered = labels_source(self.directory)
    if recovered and path is not None:
        labels = self._read(path)
        labels.dirty = True
        return labels, True
    return self.load(), False

restore_discarded

restore_discarded() -> bool

Put what the last discard set aside back as the sidecar; whether there was.

It is touched as it lands, so the labels load from it again, dirty.

Source code in src/deeperfly/labels/files.py

def restore_discarded(self) -> bool:
    """Put what the last discard set aside back as the sidecar; whether there was.

    It is touched as it lands, so the labels load from it again, dirty.
    """
    if not self.discarded_path.exists():
        return False
    os.replace(self.discarded_path, self.recovery_path)
    os.utime(self.recovery_path)
    return True

save

save(labels: Labels) -> None

Persist the labels to labels.h5, clearing the sidecar and any discard.

Source code in src/deeperfly/labels/files.py

def save(self, labels: Labels) -> None:
    """Persist the labels to `labels.h5`, clearing the sidecar and any discard."""
    self.directory.mkdir(parents=True, exist_ok=True)
    save_labels(labels, self.labels_path)
    self.recovery_path.unlink(missing_ok=True)
    self.discarded_path.unlink(missing_ok=True)

label_summary

label_summary(
    directory: str | Path,
) -> tuple[int, int, bool]

A project's labeled and complete (ADR-0102) frame counts, and its unsaved.

From one open of the file open would read; unsaved says it is the sidecar.

Source code in src/deeperfly/labels/files.py

def label_summary(directory: str | Path) -> tuple[int, int, bool]:
    """A project's labeled and complete (ADR-0102) frame counts, and its `unsaved`.

    From one open of the file `open` would read; `unsaved` says it is the sidecar.
    """
    path, unsaved = labels_source(directory)
    if path is None:
        return 0, 0, False
    return *count_frames(path), unsaved

labeled_frame_count

labeled_frame_count(directory: str | Path) -> int

How many frames a project has labels for, from the file open would read.

Source code in src/deeperfly/labels/files.py

def labeled_frame_count(directory: str | Path) -> int:
    """How many frames a project has labels for, from the file `open` would read."""
    path, _ = labels_source(directory)
    return count_labeled_frames(path) if path is not None else 0

labels_source

labels_source(
    directory: str | Path,
) -> tuple[Path | None, bool]

The file a project's labels come from, and whether it is the sidecar.

Source code in src/deeperfly/labels/files.py

def labels_source(directory: str | Path) -> tuple[Path | None, bool]:
    """The file a project's labels come from, and whether it is the sidecar."""
    saved = Path(directory) / LABELS_FILENAME
    recovery = Path(directory) / LABELS_RECOVERY_FILENAME
    if recovery.exists() and (
        not saved.exists() or recovery.stat().st_mtime >= saved.stat().st_mtime
    ):
        return recovery, True
    return (saved, False) if saved.exists() else (None, False)

A project's detections: its runs' pose2d, merged per frame by precedence.

Every run under the project's results/ is a source (ADR-0083). Per frame the winner is the run written last among those whose pose2d covers it, a tie in written going to the name that sorts last; where the manifest's [detections] runs lists runs, the winner is the first listed run covering the frame, and a run left off the list is not a source. One frame's points, conf, crops and track flags come from the winner, read on demand, in the caller's view and keypoint order.

detections = Detections.open(
    project, view_names=views, keypoint_names=names, n_frames=n_frames
)
points, conf = detections.frame(120)     # (V, K, 2), (V, K), or (None, None)

A file that is not a readable results file, is of another format, or was run on another recording, another skeleton or another set of views is listed in runs with usable false and why, and logged, never raised. A same-set other-order skeleton or view list is remapped by name.

Readers open without HDF5's file lock (Results.open in mode "r"), so a run that deeperfly run holds open in another process is read when it is consistent. Every request revalidates against signature(), the (path, mtime_ns, size) of each results file, and reinspects the runs that changed; a read that fails there yields (None, None) and a runs_error, never an exception. The files are listed again at most every RELIST_SECONDS (ADR-0141), so another process's write is seen within that. In one process HDF5 refuses to open a file twice with different locking flags, so a writer in the same process as a held reader proceeds by release(name), then Results.open(path, "r+"), the write and its close; the write changes the signature, and the release has the next request list the files at once, so it reopens the run. Such a writer holds lock over that whole sequence: every method that touches a reader takes it too, so a request in another thread waits for the write instead of reopening the run under it (which would refuse the writer's open).

Detections

A project's merged detections, served one frame at a time.

Holds one reader per usable run for the session; winner, runs, empty and n_animals are as of the last revalidation, which every request makes. lock, reentrant, serializes the readers across threads; a writer in this process holds it from release to its close (see the module).

Source code in src/deeperfly/labels/detections.py

class Detections:
    """A project's merged detections, served one frame at a time.

    Holds one reader per usable run for the session; `winner`, `runs`, `empty` and
    `n_animals` are as of the last revalidation, which every request makes. `lock`,
    reentrant, serializes the readers across threads; a writer in this process holds
    it from `release` to its close (see the module).
    """

    def __init__(
        self,
        project: Project,
        *,
        view_names: Sequence[str],
        keypoint_names: Sequence[str],
        n_frames: int,
    ) -> None:
        """An empty set over the given axes; `open` builds and fills one."""
        self.project = project
        self.view_names = tuple(view_names)
        self.keypoint_names = tuple(keypoint_names)
        self.n_frames = int(n_frames)
        self._runs: list[RunInfo] = []
        self._handles: dict[str, Results] = {}
        self._maps: dict[str, tuple[np.ndarray, np.ndarray]] = {}
        self._signature: Signature = ()
        self._open_error: str | None = None
        self._read_error: str | None = None
        # The runs another writer held open at the last build, retried every request.
        self._locked: set[str] = set()
        # When `revalidate` last listed the files, on `time.monotonic`.
        self._listed_at = -np.inf
        self.winner = np.full(self.n_frames, -1, dtype=np.int16)
        self.lock = threading.RLock()

    @classmethod
    def open(
        cls,
        project: Project,
        *,
        view_names: Sequence[str],
        keypoint_names: Sequence[str],
        n_frames: int,
    ) -> Detections:
        """The detections of `project`'s runs, in these views, keypoints and frames."""
        out = cls(
            project,
            view_names=view_names,
            keypoint_names=keypoint_names,
            n_frames=n_frames,
        )
        out.refresh()
        return out

    # -- state -----------------------------------------------------------------

    @property
    def runs(self) -> list[RunInfo]:
        """Every results file of the project, usable or not, in path order."""
        return list(self._runs)

    @property
    def runs_error(self) -> str | None:
        """Why the last request, or a run held open elsewhere, gave no detections."""
        parts = [e for e in (self._open_error, self._read_error) if e]
        return "; ".join(parts) or None

    @property
    def empty(self) -> bool:
        """Whether no frame has a detection."""
        return not bool((self.winner >= 0).any())

    @property
    def n_animals(self) -> int:
        """The most animals a winning run tracked; 0 without detections."""
        won = np.unique(self.winner[self.winner >= 0])
        return max((self._runs[int(i)].animals for i in won), default=0)

    def signature(self) -> Signature:
        """The `(path, mtime_ns, size)` of every results file, sorted by path."""
        out = []
        for path in find_results(self.project):
            try:
                st = path.stat()
            except OSError:
                continue  # removed since it was listed
            out.append((str(path), st.st_mtime_ns, st.st_size))
        return tuple(out)

    @property
    def built(self) -> Signature:
        """The `signature()` the runs were last built from, which lists no file.

        What a cache of something computed from the detections keys on, after a
        `revalidate`.
        """
        return self._signature

    def stale(self) -> bool:
        """Whether a results file appeared, went or changed since the last build."""
        return self.signature() != self._signature

    @_serialized
    def revalidate(self) -> None:
        """Rebuild what changed, and clear the last request's error.

        Only the runs whose `(path, mtime_ns, size)` moved, or that appeared, are
        inspected again; the rest keep their readers, so a run being written costs a
        request one reopen, not one per run. The files are listed at most every
        `RELIST_SECONDS`, but a run held open for writing at the last build is retried
        at every request until it opens.
        """
        self._read_error = None
        now = time.monotonic()
        if now - self._listed_at < RELIST_SECONDS and not self._open_error:
            return
        signature = self.signature()
        self._listed_at = now
        if signature != self._signature or self._open_error:
            self._rebuild(signature)

    @_serialized
    def refresh(self) -> None:
        """Close every reader, list and inspect the runs again, recompute the winner."""
        self.close()
        self._maps.clear()
        self._runs = []
        self._listed_at = time.monotonic()
        self._rebuild(self.signature())

    @_serialized
    def release(self, name: str) -> None:
        """Close one run's reader, reopened by the next request that needs it.

        What a writer in this process calls before it opens the run (see the module);
        the next `revalidate` lists the files whatever the time since the last.
        """
        self._listed_at = -np.inf
        res = self._handles.pop(name, None)
        if res is not None:
            res.close()

    @_serialized
    def close(self) -> None:
        """Close every reader."""
        for res in self._handles.values():
            res.close()
        self._handles.clear()

    # -- requests --------------------------------------------------------------

    def covered_frames(self) -> np.ndarray:
        """The frames with a winning run, ascending."""
        self.revalidate()
        return np.flatnonzero(self.winner >= 0)

    def is_covered(self, t: int) -> bool:
        """Whether frame `t` has a winning run."""
        self.revalidate()
        return 0 <= t < self.n_frames and bool(self.winner[t] >= 0)

    @_serialized
    def frame(
        self, t: int, animal: int | None = 0, *, min_conf: float = DEFAULT_MIN_CONF
    ) -> tuple[np.ndarray, np.ndarray] | tuple[None, None]:
        """One animal's `(V, K, 2)` pixels and `(V, K)` confidences at frame `t`.

        `animal=None` keeps every animal the winner tracked, `(N, V, K, 2)` and
        `(N, V, K)`. A pixel below `min_conf` is NaN. `(None, None)` where no run
        covers the frame, for an animal the winner did not track, or when the read
        fails.
        """
        self.revalidate()
        run = self._winning(t, 0 if animal is None else animal)
        if run is None:
            return None, None
        try:
            res = self._reader(run)
            xy = res.read_frame("pose2d/points", t, animal)
            conf = res.read_frame("pose2d/conf", t, animal)
        except _READ_ERRORS as exc:
            self._failed(run, exc)
            return None, None
        vm, km = self._maps[run.name]
        xy = np.asarray(xy, dtype=float)[..., vm, :, :][..., km, :]
        conf = np.asarray(conf, dtype=float)[..., vm, :][..., km]
        xy[~(conf >= min_conf)] = np.nan
        return xy, conf

    @_serialized
    def read(
        self, frames: Sequence[int] | np.ndarray, animal: int = 0
    ) -> tuple[Float[np.ndarray, "V F K 2"], Float[np.ndarray, "V F K"]]:
        """One animal's pixels `(V, F, K, 2)` and confidences `(V, F, K)` at `frames`.

        Unfiltered by confidence; NaN pixels and zero confidence where no run covers a
        frame, the winner did not track the animal, or its read failed.
        """
        self.revalidate()
        wanted = np.asarray(frames, dtype=np.int64).ravel()
        shape = (len(self.view_names), wanted.size, len(self.keypoint_names))
        points = np.full((*shape, 2), np.nan)
        conf = np.full(shape, float(fill_value("pose2d/conf", "f")))
        inside = (wanted >= 0) & (wanted < self.n_frames)
        winner = np.full(wanted.size, -1)
        winner[inside] = self.winner[wanted[inside]]
        for i in np.unique(winner[winner >= 0]):
            run = self._runs[int(i)]
            if animal >= run.animals:
                continue
            at = np.flatnonzero(winner == i)
            try:
                res = self._reader(run)
                xy = res.read_frames("pose2d/points", wanted[at])[animal]
                c = res.read_frames("pose2d/conf", wanted[at])[animal]
            except _READ_ERRORS as exc:
                self._failed(run, exc)
                continue
            vm, km = self._maps[run.name]
            points[:, at] = np.asarray(xy, dtype=float)[vm][:, :, km]
            conf[:, at] = np.asarray(c, dtype=float)[vm][:, :, km]
        return points, conf

    @_serialized
    def crops(self, t: int, animal: int = 0) -> np.ndarray | None:
        """The `(V, 4)` rectangles the winner's detector read frame `t` through.

        `None` where no run covers the frame, the run stores no crops, or the read
        fails.
        """
        self.revalidate()
        run = self._winning(t, animal)
        if run is None:
            return None
        try:
            res = self._reader(run)
            if not res.has("pose2d/crops"):
                return None
            crops = res.read_frame("pose2d/crops", t, animal)
        except _READ_ERRORS as exc:
            self._failed(run, exc)
            return None
        return np.asarray(crops, dtype=float)[self._maps[run.name][0]]

    @_serialized
    def track_flags(self) -> np.ndarray:
        """`(N, T)` int8 tracking flags of the winners, `LOST` where none covers.

        A covered frame of a run that did not track (one animal, no tracker) is
        `TRACKED`, and an animal the winner does not have is `LOST`.
        """
        self.revalidate()
        out = np.full((self.n_animals, self.n_frames), LOST, dtype=np.int8)
        for i in np.unique(self.winner[self.winner >= 0]):
            run = self._runs[int(i)]
            mask = self.winner == i
            try:
                res = self._reader(run)
                if res.has("pose2d/track_flags"):
                    flags = np.asarray(res.read("pose2d/track_flags"))
                    out[: run.animals, mask] = flags[:, mask]
                else:
                    out[: run.animals, mask] = TRACKED
            except _READ_ERRORS as exc:
                self._failed(run, exc)
        return out

    # -- internals -------------------------------------------------------------

    def _rebuild(self, signature: Signature) -> None:
        """Adopt `signature`: inspect each run whose entry is new, keep the others."""
        before = dict(
            zip((e[0] for e in self._signature), self._signature, strict=True)
        )
        kept = {
            str(run.path): run
            for run in self._runs
            if run.name not in self._locked and before.get(str(run.path)) in signature
        }
        for run in self._runs:
            if str(run.path) not in kept:
                self.release(run.name)
                self._maps.pop(run.name, None)
        self._signature = signature
        self._open_error = None
        self._locked = set()
        runs = []
        for entry in signature:
            info = kept.get(entry[0])
            if info is None:
                info = self._inspect(Path(entry[0]))
                if not info.usable:
                    log.warning(
                        "%s: skipped as a source of detections, %s", info.path, info.why
                    )
            runs.append(info)
        self._runs = runs
        self.winner = precedence(self.project, runs, self.n_frames)

    def _winning(self, t: int, animal: int) -> RunInfo | None:
        """The run that serves `animal` at frame `t`, if any."""
        if not 0 <= t < self.n_frames or self.winner[t] < 0:
            return None
        run = self._runs[int(self.winner[t])]
        return run if 0 <= animal < run.animals else None

    def _reader(self, run: RunInfo) -> Results:
        """The run's open reader, reopened after a `release` or a failed read."""
        res = self._handles.get(run.name)
        if res is None:
            res = Results.open(run.path)
            self._handles[run.name] = res
        return res

    def _failed(self, run: RunInfo, exc: BaseException) -> None:
        """Record a failed read and drop the reader, so the next request reopens."""
        self._read_error = f"{run.name}: {exc}"
        self.release(run.name)

    def _inspect(self, path: Path) -> RunInfo:
        """One results file as a source; its reader is kept when it is usable."""
        name = run_name(self.project, path) or path.stem
        try:
            res = Results.open(path)
        except ValueError as exc:
            return RunInfo(name, path, why=str(exc))
        except OSError as exc:
            if "lock" in str(exc):
                self._open_error = f"{name}: held open for writing ({exc})"
                self._locked.add(name)
                return RunInfo(name, path, why=f"held open for writing ({exc})")
            return RunInfo(name, path, why=f"not a readable results file ({exc})")
        except (KeyError, TypeError) as exc:
            return RunInfo(name, path, why=f"not a results file ({exc})")
        try:
            info, maps = self._describe(name, path, res)
        except _READ_ERRORS as exc:
            info, maps = RunInfo(name, path, why=f"not readable ({exc})"), None
        if maps is None:
            res.close()
        else:
            self._handles[name] = res
            self._maps[name] = maps
        return info

    def _describe(
        self, name: str, path: Path, res: Results
    ) -> tuple[RunInfo, tuple[np.ndarray, np.ndarray] | None]:
        """What an open run offers, and its view and keypoint maps if it is a source."""
        info = RunInfo(
            name,
            path,
            animals=res.animals,
            stages=tuple(res.stages()),
            written=res.written,
            pose3d_ref=next((r for r in _POSE3D_REFS if res.has(r)), None),
        )
        why = None
        if res.n_frames != self.n_frames:
            why = f"was run on {res.n_frames} frames, not {self.n_frames}"
        elif sorted(res.views) != sorted(self.view_names):
            why = "was run on views " + ", ".join(res.views)
        elif sorted(res.keypoints) != sorted(self.keypoint_names):
            why = "was run against another skeleton"
        elif not res.has("pose2d/points"):
            why = "has no pose2d"
        if why is not None:
            return replace(info, why=why), None
        covered = res.covered("pose2d")
        maps = (
            name_positions(self.view_names, res.views, "view"),
            name_positions(self.keypoint_names, res.keypoints, "keypoint"),
        )
        info = replace(
            info,
            covered_count=int(covered.sum()),
            stretches=tuple(stretches(np.flatnonzero(covered))),
            usable=True,
            covered=covered,
        )
        return info, maps

built property

built: Signature

The signature() the runs were last built from, which lists no file.

What a cache of something computed from the detections keys on, after a revalidate.

empty property

empty: bool

Whether no frame has a detection.

n_animals property

n_animals: int

The most animals a winning run tracked; 0 without detections.

runs property

runs: list[RunInfo]

Every results file of the project, usable or not, in path order.

runs_error property

runs_error: str | None

Why the last request, or a run held open elsewhere, gave no detections.

__init__

__init__(
    project: Project,
    *,
    view_names: Sequence[str],
    keypoint_names: Sequence[str],
    n_frames: int,
) -> None

An empty set over the given axes; open builds and fills one.

Source code in src/deeperfly/labels/detections.py

def __init__(
    self,
    project: Project,
    *,
    view_names: Sequence[str],
    keypoint_names: Sequence[str],
    n_frames: int,
) -> None:
    """An empty set over the given axes; `open` builds and fills one."""
    self.project = project
    self.view_names = tuple(view_names)
    self.keypoint_names = tuple(keypoint_names)
    self.n_frames = int(n_frames)
    self._runs: list[RunInfo] = []
    self._handles: dict[str, Results] = {}
    self._maps: dict[str, tuple[np.ndarray, np.ndarray]] = {}
    self._signature: Signature = ()
    self._open_error: str | None = None
    self._read_error: str | None = None
    # The runs another writer held open at the last build, retried every request.
    self._locked: set[str] = set()
    # When `revalidate` last listed the files, on `time.monotonic`.
    self._listed_at = -np.inf
    self.winner = np.full(self.n_frames, -1, dtype=np.int16)
    self.lock = threading.RLock()

close

close() -> None

Close every reader.

Source code in src/deeperfly/labels/detections.py

@_serialized
def close(self) -> None:
    """Close every reader."""
    for res in self._handles.values():
        res.close()
    self._handles.clear()

covered_frames

covered_frames() -> np.ndarray

The frames with a winning run, ascending.

Source code in src/deeperfly/labels/detections.py

def covered_frames(self) -> np.ndarray:
    """The frames with a winning run, ascending."""
    self.revalidate()
    return np.flatnonzero(self.winner >= 0)

crops

crops(t: int, animal: int = 0) -> np.ndarray | None

The (V, 4) rectangles the winner's detector read frame t through.

None where no run covers the frame, the run stores no crops, or the read fails.

Source code in src/deeperfly/labels/detections.py

@_serialized
def crops(self, t: int, animal: int = 0) -> np.ndarray | None:
    """The `(V, 4)` rectangles the winner's detector read frame `t` through.

    `None` where no run covers the frame, the run stores no crops, or the read
    fails.
    """
    self.revalidate()
    run = self._winning(t, animal)
    if run is None:
        return None
    try:
        res = self._reader(run)
        if not res.has("pose2d/crops"):
            return None
        crops = res.read_frame("pose2d/crops", t, animal)
    except _READ_ERRORS as exc:
        self._failed(run, exc)
        return None
    return np.asarray(crops, dtype=float)[self._maps[run.name][0]]

frame

frame(
    t: int,
    animal: int | None = 0,
    *,
    min_conf: float = DEFAULT_MIN_CONF,
) -> tuple[np.ndarray, np.ndarray] | tuple[None, None]

One animal's (V, K, 2) pixels and (V, K) confidences at frame t.

animal=None keeps every animal the winner tracked, (N, V, K, 2) and (N, V, K). A pixel below min_conf is NaN. (None, None) where no run covers the frame, for an animal the winner did not track, or when the read fails.

Source code in src/deeperfly/labels/detections.py

@_serialized
def frame(
    self, t: int, animal: int | None = 0, *, min_conf: float = DEFAULT_MIN_CONF
) -> tuple[np.ndarray, np.ndarray] | tuple[None, None]:
    """One animal's `(V, K, 2)` pixels and `(V, K)` confidences at frame `t`.

    `animal=None` keeps every animal the winner tracked, `(N, V, K, 2)` and
    `(N, V, K)`. A pixel below `min_conf` is NaN. `(None, None)` where no run
    covers the frame, for an animal the winner did not track, or when the read
    fails.
    """
    self.revalidate()
    run = self._winning(t, 0 if animal is None else animal)
    if run is None:
        return None, None
    try:
        res = self._reader(run)
        xy = res.read_frame("pose2d/points", t, animal)
        conf = res.read_frame("pose2d/conf", t, animal)
    except _READ_ERRORS as exc:
        self._failed(run, exc)
        return None, None
    vm, km = self._maps[run.name]
    xy = np.asarray(xy, dtype=float)[..., vm, :, :][..., km, :]
    conf = np.asarray(conf, dtype=float)[..., vm, :][..., km]
    xy[~(conf >= min_conf)] = np.nan
    return xy, conf

is_covered

is_covered(t: int) -> bool

Whether frame t has a winning run.

Source code in src/deeperfly/labels/detections.py

def is_covered(self, t: int) -> bool:
    """Whether frame `t` has a winning run."""
    self.revalidate()
    return 0 <= t < self.n_frames and bool(self.winner[t] >= 0)

open classmethod

open(
    project: Project,
    *,
    view_names: Sequence[str],
    keypoint_names: Sequence[str],
    n_frames: int,
) -> Detections

The detections of project's runs, in these views, keypoints and frames.

Source code in src/deeperfly/labels/detections.py

@classmethod
def open(
    cls,
    project: Project,
    *,
    view_names: Sequence[str],
    keypoint_names: Sequence[str],
    n_frames: int,
) -> Detections:
    """The detections of `project`'s runs, in these views, keypoints and frames."""
    out = cls(
        project,
        view_names=view_names,
        keypoint_names=keypoint_names,
        n_frames=n_frames,
    )
    out.refresh()
    return out

read

read(
    frames: Sequence[int] | ndarray, animal: int = 0
) -> tuple[
    Float[np.ndarray, "V F K 2"], Float[np.ndarray, "V F K"]
]

One animal's pixels (V, F, K, 2) and confidences (V, F, K) at frames.

Unfiltered by confidence; NaN pixels and zero confidence where no run covers a frame, the winner did not track the animal, or its read failed.

Source code in src/deeperfly/labels/detections.py

@_serialized
def read(
    self, frames: Sequence[int] | np.ndarray, animal: int = 0
) -> tuple[Float[np.ndarray, "V F K 2"], Float[np.ndarray, "V F K"]]:
    """One animal's pixels `(V, F, K, 2)` and confidences `(V, F, K)` at `frames`.

    Unfiltered by confidence; NaN pixels and zero confidence where no run covers a
    frame, the winner did not track the animal, or its read failed.
    """
    self.revalidate()
    wanted = np.asarray(frames, dtype=np.int64).ravel()
    shape = (len(self.view_names), wanted.size, len(self.keypoint_names))
    points = np.full((*shape, 2), np.nan)
    conf = np.full(shape, float(fill_value("pose2d/conf", "f")))
    inside = (wanted >= 0) & (wanted < self.n_frames)
    winner = np.full(wanted.size, -1)
    winner[inside] = self.winner[wanted[inside]]
    for i in np.unique(winner[winner >= 0]):
        run = self._runs[int(i)]
        if animal >= run.animals:
            continue
        at = np.flatnonzero(winner == i)
        try:
            res = self._reader(run)
            xy = res.read_frames("pose2d/points", wanted[at])[animal]
            c = res.read_frames("pose2d/conf", wanted[at])[animal]
        except _READ_ERRORS as exc:
            self._failed(run, exc)
            continue
        vm, km = self._maps[run.name]
        points[:, at] = np.asarray(xy, dtype=float)[vm][:, :, km]
        conf[:, at] = np.asarray(c, dtype=float)[vm][:, :, km]
    return points, conf

refresh

refresh() -> None

Close every reader, list and inspect the runs again, recompute the winner.

Source code in src/deeperfly/labels/detections.py

@_serialized
def refresh(self) -> None:
    """Close every reader, list and inspect the runs again, recompute the winner."""
    self.close()
    self._maps.clear()
    self._runs = []
    self._listed_at = time.monotonic()
    self._rebuild(self.signature())

release

release(name: str) -> None

Close one run's reader, reopened by the next request that needs it.

What a writer in this process calls before it opens the run (see the module); the next revalidate lists the files whatever the time since the last.

Source code in src/deeperfly/labels/detections.py

@_serialized
def release(self, name: str) -> None:
    """Close one run's reader, reopened by the next request that needs it.

    What a writer in this process calls before it opens the run (see the module);
    the next `revalidate` lists the files whatever the time since the last.
    """
    self._listed_at = -np.inf
    res = self._handles.pop(name, None)
    if res is not None:
        res.close()

revalidate

revalidate() -> None

Rebuild what changed, and clear the last request's error.

Only the runs whose (path, mtime_ns, size) moved, or that appeared, are inspected again; the rest keep their readers, so a run being written costs a request one reopen, not one per run. The files are listed at most every RELIST_SECONDS, but a run held open for writing at the last build is retried at every request until it opens.

Source code in src/deeperfly/labels/detections.py

@_serialized
def revalidate(self) -> None:
    """Rebuild what changed, and clear the last request's error.

    Only the runs whose `(path, mtime_ns, size)` moved, or that appeared, are
    inspected again; the rest keep their readers, so a run being written costs a
    request one reopen, not one per run. The files are listed at most every
    `RELIST_SECONDS`, but a run held open for writing at the last build is retried
    at every request until it opens.
    """
    self._read_error = None
    now = time.monotonic()
    if now - self._listed_at < RELIST_SECONDS and not self._open_error:
        return
    signature = self.signature()
    self._listed_at = now
    if signature != self._signature or self._open_error:
        self._rebuild(signature)

signature

signature() -> Signature

The (path, mtime_ns, size) of every results file, sorted by path.

Source code in src/deeperfly/labels/detections.py

def signature(self) -> Signature:
    """The `(path, mtime_ns, size)` of every results file, sorted by path."""
    out = []
    for path in find_results(self.project):
        try:
            st = path.stat()
        except OSError:
            continue  # removed since it was listed
        out.append((str(path), st.st_mtime_ns, st.st_size))
    return tuple(out)

stale

stale() -> bool

Whether a results file appeared, went or changed since the last build.

Source code in src/deeperfly/labels/detections.py

def stale(self) -> bool:
    """Whether a results file appeared, went or changed since the last build."""
    return self.signature() != self._signature

track_flags

track_flags() -> np.ndarray

(N, T) int8 tracking flags of the winners, LOST where none covers.

A covered frame of a run that did not track (one animal, no tracker) is TRACKED, and an animal the winner does not have is LOST.

Source code in src/deeperfly/labels/detections.py

@_serialized
def track_flags(self) -> np.ndarray:
    """`(N, T)` int8 tracking flags of the winners, `LOST` where none covers.

    A covered frame of a run that did not track (one animal, no tracker) is
    `TRACKED`, and an animal the winner does not have is `LOST`.
    """
    self.revalidate()
    out = np.full((self.n_animals, self.n_frames), LOST, dtype=np.int8)
    for i in np.unique(self.winner[self.winner >= 0]):
        run = self._runs[int(i)]
        mask = self.winner == i
        try:
            res = self._reader(run)
            if res.has("pose2d/track_flags"):
                flags = np.asarray(res.read("pose2d/track_flags"))
                out[: run.animals, mask] = flags[:, mask]
            else:
                out[: run.animals, mask] = TRACKED
        except _READ_ERRORS as exc:
            self._failed(run, exc)
    return out

RunInfo dataclass

One results file of a project, as a source of detections.

covered is the (T,) mask of frames its pose2d wrote, covered_count its sum and stretches its (start, stop) ranges; written is the run's write time in whole unix seconds. pose3d_ref names the most refined 3D the run holds (correction, then smoothing, then pose3d), None without one. A run that cannot be a source has usable false and says why.

Source code in src/deeperfly/labels/detections.py

@dataclass(frozen=True)
class RunInfo:
    """One results file of a project, as a source of detections.

    `covered` is the `(T,)` mask of frames its pose2d wrote, `covered_count` its sum
    and `stretches` its `(start, stop)` ranges; `written` is the run's write time in
    whole unix seconds. `pose3d_ref` names the most refined 3D the run holds
    (`correction`, then `smoothing`, then `pose3d`), `None` without one. A run that
    cannot be a source has `usable` false and says `why`.
    """

    name: str
    path: Path
    animals: int = 0
    stages: tuple[str, ...] = ()
    covered_count: int = 0
    written: int = 0
    stretches: tuple[tuple[int, int], ...] = ()
    pose3d_ref: str | None = None
    usable: bool = False
    why: str | None = None
    covered: np.ndarray = field(
        default_factory=lambda: np.zeros(0, dtype=bool), repr=False, compare=False
    )

find_results

find_results(project: Project) -> list[Path]

Every run's results file of project, results/**/*.h5, sorted by path.

Runs written elsewhere with --out are not found.

Source code in src/deeperfly/labels/detections.py

def find_results(project: Project) -> list[Path]:
    """Every run's results file of `project`, `results/**/*.h5`, sorted by path.

    Runs written elsewhere with `--out` are not found.
    """
    return project.results_paths()

precedence

precedence(
    project: Project, runs: Sequence[RunInfo], n_frames: int
) -> np.ndarray

The (T,) int16 index into runs of each frame's winner, -1 where none.

With the manifest's [detections] runs, the first listed run covering a frame wins and only listed runs are sources (a listed name with no file is skipped); otherwise the latest-written covering run wins, the name sorting last breaking a tie. A run that is not usable never wins.

Source code in src/deeperfly/labels/detections.py

def precedence(project: Project, runs: Sequence[RunInfo], n_frames: int) -> np.ndarray:
    """The `(T,)` int16 index into `runs` of each frame's winner, -1 where none.

    With the manifest's `[detections] runs`, the first listed run covering a frame
    wins and only listed runs are sources (a listed name with no file is skipped);
    otherwise the latest-written covering run wins, the name sorting last breaking a
    tie. A run that is not `usable` never wins.
    """
    winner = np.full(int(n_frames), -1, dtype=np.int16)
    listed = project.manifest.detections.runs
    if listed:
        index = {run.name: i for i, run in enumerate(runs)}
        order = [index[name] for name in listed if name in index]
    else:
        order = sorted(
            range(len(runs)),
            key=lambda i: (runs[i].written, runs[i].name),
            reverse=True,
        )
    for i in order:
        run = runs[i]
        if not run.usable or run.covered.shape != winner.shape:
            continue
        winner[(winner < 0) & run.covered] = i
    return winner

run_name

run_name(
    project: Project, results: str | Path
) -> str | None

A results file's run name (nightly/full), None if not under results/.

Source code in src/deeperfly/labels/detections.py

def run_name(project: Project, results: str | Path) -> str | None:
    """A results file's run name (`nightly/full`), `None` if not under `results/`."""
    return project.run_name(results)

Which detection slot is which labeled animal, frame by frame (ADR-0128).

A project that labels several animals numbers them twice: the operator numbers the label instances, and a run's tracker numbers its slots, arbitrarily, afresh in every run and possibly switching along a recording. Wherever the two meet (the labeling solve and its guides, the fill of a training target's unclicked views, the joints a suggestion skips), an instance reads the detections of the slot its clicks sit on.

match_slots is that pairing for one frame. A click within MATCH_GATE_PX of a slot's detection of the same (view, keypoint) cell sits on it and votes for the slot, the more the nearer, and the instances are assigned to the slots on their votes. A click farther off votes for nothing: the operator clicks first where the detector is wrong, so a half-labeled animal's clicks are mostly off its own detections (ADR-0138). An instance with no vote takes a slot no instance won: its own number when that one is left, else the lowest left. So in a frame nobody labeled slot i is instance i, and the skeleton the labeling aid draws for an animal before its first click is the one that click keeps it on, as does a joint dragged off it. Only a slot the clicks say is another animal's, over MATCH_SLACK more of them off its detections than on, is refused, and the labeling aid, which wants a skeleton to drag, refuses none.

frame_slots

frame_slots(
    labels: Labels,
    t: int,
    detected: Float[ndarray, "S V K 2"],
    *,
    gate_px: float = MATCH_GATE_PX,
    refuse: bool = True,
) -> Int[np.ndarray, " I"]

match_slots of frame t's label instances against the slots' detections.

Covers instances 0 to I - 1, I the larger of the slot count and one past the highest instance labeled; an instance beyond has no slot. Absent keypoints are no clicks.

Source code in src/deeperfly/labels/identity.py

def frame_slots(
    labels: Labels,
    t: int,
    detected: Float[np.ndarray, "S V K 2"],
    *,
    gate_px: float = MATCH_GATE_PX,
    refuse: bool = True,
) -> Int[np.ndarray, " I"]:
    """[match_slots][] of frame `t`'s label instances against the slots' detections.

    Covers instances `0` to `I - 1`, `I` the larger of the slot count and one past
    the highest instance labeled; an instance beyond has no slot. Absent keypoints
    are no clicks.
    """
    labeled = labels.labeled_instances(t)
    n_inst = max(detected.shape[0], labeled[-1] + 1 if labeled else 0)
    clicks = np.full((n_inst, labels.n_views, labels.n_keypoints, 2), np.nan)
    for i in labeled:
        clicks[i] = labels.frame_effective(t, instance=i)
    return match_slots(clicks, detected, gate_px=gate_px, refuse=refuse)

instance_frame

instance_frame(
    detections: Detections,
    labels: Labels,
    t: int,
    instance: int = 0,
    *,
    refuse: bool = True,
) -> tuple[np.ndarray, np.ndarray] | tuple[None, None]

The (V, K, 2) pixels and (V, K) confidences of the slot instance matches.

Detections.frame of that slot, a pixel below its min_conf NaN; (None, None) where no run covers frame t or no slot is this instance. refuse is as for match_slots.

Source code in src/deeperfly/labels/identity.py

def instance_frame(
    detections: Detections,
    labels: Labels,
    t: int,
    instance: int = 0,
    *,
    refuse: bool = True,
) -> tuple[np.ndarray, np.ndarray] | tuple[None, None]:
    """The `(V, K, 2)` pixels and `(V, K)` confidences of the slot `instance` matches.

    [Detections.frame][deeperfly.labels.detections.Detections.frame] of that slot, a
    pixel below its `min_conf` NaN; `(None, None)` where no run covers frame `t` or
    no slot is this instance. `refuse` is as for [match_slots][].
    """
    det, conf = detections.frame(t, None)
    if det is None or conf is None:
        return None, None
    slots = frame_slots(labels, t, det, refuse=refuse)
    if instance >= slots.size or slots[instance] < 0:
        return None, None
    return det[slots[instance]], conf[slots[instance]]

instance_read

instance_read(
    detections: Detections,
    labels: Labels | None,
    frames: Sequence[int] | ndarray,
    instance: int = 0,
) -> tuple[
    Float[np.ndarray, "V F K 2"], Float[np.ndarray, "V F K"]
]

Detections.read of one instance.

Slot instance wherever no label says otherwise, and at a labeled frame the slot instance matches there, NaN pixels and zero confidence where none does.

Source code in src/deeperfly/labels/identity.py

def instance_read(
    detections: Detections,
    labels: Labels | None,
    frames: Sequence[int] | np.ndarray,
    instance: int = 0,
) -> tuple[Float[np.ndarray, "V F K 2"], Float[np.ndarray, "V F K"]]:
    """[Detections.read][deeperfly.labels.detections.Detections.read] of one instance.

    Slot `instance` wherever no label says otherwise, and at a labeled frame the slot
    `instance` matches there, NaN pixels and zero confidence where none does.
    """
    points, conf = detections.read(frames, instance)
    if labels is None:
        return points, conf
    wanted = np.asarray(frames, dtype=np.int64).ravel()
    for j in np.flatnonzero(np.isin(wanted, labels.labeled_frames())):
        t = int(wanted[j])
        det, _ = detections.frame(t, None)
        if det is None:
            continue
        slots = frame_slots(labels, t, det)
        slot = int(slots[instance]) if instance < slots.size else -1
        if slot == instance:
            continue
        if slot < 0:
            points[:, j], conf[:, j] = np.nan, 0.0
        else:
            xy, c = detections.read([t], slot)
            points[:, j], conf[:, j] = xy[:, 0], c[:, 0]
    return points, conf

match_slots

match_slots(
    clicks: Float[ndarray, "I V K 2"],
    detected: Float[ndarray, "S V K 2"],
    *,
    gate_px: float = MATCH_GATE_PX,
    refuse: bool = True,
) -> Int[np.ndarray, " I"]

Each label instance's detection slot in one frame, -1 for none.

clicks are the instances' labels and detected the slots' detections, NaN where none. Of the cells both hold, a click within gate_px of the detection sits on it and votes for the pair, the more the nearer it sits; the instances with a vote are assigned by Hungarian assignment on their votes, a tie going to an instance's own number, and a pair without a vote being no match. The others take the slots no instance won: their own number where it is left, then the lowest left.

With refuse, a pair with over MATCH_SLACK more clicks off the detections than on them is another animal's: its votes do not count, and the instance never takes that slot, so an instance that every slot left refuses reads none. Without, an instance keeps a slot whatever its clicks say of it, as long as one is left.

Source code in src/deeperfly/labels/identity.py

def match_slots(
    clicks: Float[np.ndarray, "I V K 2"],
    detected: Float[np.ndarray, "S V K 2"],
    *,
    gate_px: float = MATCH_GATE_PX,
    refuse: bool = True,
) -> Int[np.ndarray, " I"]:
    """Each label instance's detection slot in one frame, -1 for none.

    `clicks` are the instances' labels and `detected` the slots' detections, NaN
    where none. Of the cells both hold, a click within `gate_px` of the detection
    sits on it and votes for the pair, the more the nearer it sits; the instances
    with a vote are assigned by Hungarian assignment on their votes, a tie going to
    an instance's own number, and a pair without a vote being no match. The others
    take the slots no instance won: their own number where it is left, then the
    lowest left.

    With `refuse`, a pair with over `MATCH_SLACK` more clicks off the detections than
    on them is another animal's: its votes do not count, and the instance never takes
    that slot, so an instance that every slot left refuses reads none. Without, an
    instance keeps a slot whatever its clicks say of it, as long as one is left.
    """
    from scipy.optimize import linear_sum_assignment

    clicks = np.asarray(clicks, dtype=float)
    detected = np.asarray(detected, dtype=float)
    n_inst, n_slots = clicks.shape[0], detected.shape[0]
    both = np.isfinite(clicks).all(-1)[:, None] & np.isfinite(detected).all(-1)[None]
    dist = np.linalg.norm(clicks[:, None] - detected[None], axis=-1)  # (I, S, V, K)
    on = both & (dist < gate_px)
    # The clicks off a slot's detections, less those on them.
    excess = both.sum(axis=(2, 3)) - 2 * on.sum(axis=(2, 3))  # (I, S)
    refused = (excess > MATCH_SLACK) & refuse
    votes = np.where(on, 1.0 - dist / gate_px, 0.0).sum(axis=(2, 3))
    votes[refused] = 0.0
    out = np.full(n_inst, -1, dtype=int)
    placed = np.flatnonzero(votes.any(axis=1))
    if placed.size:
        # A tie goes to the instance's own number.
        own = 1e-6 * (placed[:, None] == np.arange(n_slots))
        rows, cols = linear_sum_assignment(votes[placed] + own, maximize=True)
        keep = votes[placed[rows], cols] > 0
        out[placed[rows[keep]]] = cols[keep]
    free = set(range(n_slots)) - set(out.tolist())
    rest = []
    for i in np.flatnonzero(out < 0):
        if i in free and not refused[i, i]:
            out[i] = i
            free.discard(i)
        else:
            rest.append(i)
    for i in rest:
        left = [s for s in sorted(free) if not refused[i, s]]
        if left:
            out[i] = left[0]
            free.discard(left[0])
    return out

run_slots

run_slots(
    labels: Labels | None,
    frames: Sequence[int] | ndarray,
    points: Float[ndarray, "N V F K 2"],
    conf: Float[ndarray, "N V F K"],
    *,
    view_names: Sequence[str],
    keypoint_names: Sequence[str],
    min_conf: float = DEFAULT_MIN_CONF,
) -> Int[np.ndarray, "N F"]

The slot of a run each label instance reads at each of frames, -1 for none.

Slot i is instance i wherever no label says otherwise; at a labeled frame, frame_slots on the run's own points there, those under min_conf left out, as Detections.frame leaves them. The arrays' views and keypoints are view_names and keypoint_names, met with the labels' by name.

Source code in src/deeperfly/labels/identity.py

def run_slots(
    labels: Labels | None,
    frames: Sequence[int] | np.ndarray,
    points: Float[np.ndarray, "N V F K 2"],
    conf: Float[np.ndarray, "N V F K"],
    *,
    view_names: Sequence[str],
    keypoint_names: Sequence[str],
    min_conf: float = DEFAULT_MIN_CONF,
) -> Int[np.ndarray, "N F"]:
    """The slot of a run each label instance reads at each of `frames`, -1 for none.

    Slot `i` is instance `i` wherever no label says otherwise; at a labeled frame,
    [frame_slots][] on the run's own `points` there, those under `min_conf` left
    out, as [Detections.frame][deeperfly.labels.detections.Detections.frame] leaves
    them. The arrays' views and keypoints are `view_names` and `keypoint_names`, met
    with the labels' by name.
    """
    n = points.shape[0]
    wanted = np.asarray(frames, dtype=np.int64).ravel()
    out = np.tile(np.arange(n)[:, None], (1, wanted.size))
    if labels is None:
        return out
    vl, va = _common(labels.view_names, view_names)
    kl, ka = _common(labels.keypoint_names, keypoint_names)
    for j in np.flatnonzero(np.isin(wanted, labels.labeled_frames())):
        xy = np.where((conf[:, :, j] >= min_conf)[..., None], points[:, :, j], np.nan)
        det = np.full((n, labels.n_views, labels.n_keypoints, 2), np.nan)
        det[:, vl[:, None], kl] = xy[:, va[:, None], ka]
        out[:, j] = frame_slots(labels, int(wanted[j]), det)[:n]
    return out

take_slots

take_slots(
    arr: ndarray,
    slots: Int[ndarray, "N F"],
    axis: int,
    fill: float,
) -> None

Index arr's leading animal axis by label instance, in place.

axis is arr's frame axis; at frame j, index i takes what slot slots[i, j] held there (run_slots), fill where that is -1.

Source code in src/deeperfly/labels/identity.py

def take_slots(
    arr: np.ndarray, slots: Int[np.ndarray, "N F"], axis: int, fill: float
) -> None:
    """Index `arr`'s leading animal axis by label instance, in place.

    `axis` is `arr`'s frame axis; at frame `j`, index `i` takes what slot
    `slots[i, j]` held there ([run_slots][]), `fill` where that is -1.
    """
    n = slots.shape[0]
    for j in np.flatnonzero((slots != np.arange(n)[:, None]).any(axis=0)):
        at = (slice(None),) * axis + (int(j),)
        was = arr[at].copy()
        arr[at] = fill
        have = slots[:, j] >= 0
        arr[(np.flatnonzero(have), *at[1:])] = was[slots[have, j]]

The labeling aid's geometry: the solve behind a reprojected guide, and the checks.

One 3D per keypoint from the labels, the detections and a prior (ADR-0027, ADR-0074), and the checks that flag a label as off. Nothing here knows about a session or HTTP, so the routes of deeperfly.gui.labeling stay thin; deeperfly.labels.suggest is the offline counterpart.

bone_reference

bone_reference(
    rig: Rig | None,
    skeleton: Skeleton,
    labels: Labels,
    detections: Detections | None,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]

Each skeleton edge's usual 3D length in a recording: (median, mad, count).

From up to BONE_REF_FRAMES frames of the detections (the frames they cover, for a project's merged ones), triangulated robustly as the pictorial bone prior is, else from the labeled frames; NaN where an edge was never measured, empty without a rig.

Source code in src/deeperfly/labels/check.py

def bone_reference(
    rig: Rig | None,
    skeleton: Skeleton,
    labels: Labels,
    detections: Detections | None,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Each skeleton edge's usual 3D length in a recording: `(median, mad, count)`.

    From up to `BONE_REF_FRAMES` frames of the detections (the frames they cover, for
    a project's merged ones), triangulated robustly as the pictorial bone prior is,
    else from the labeled frames; NaN where an edge was never measured, empty without
    a rig.
    """
    edges = np.asarray(skeleton.edges, dtype=int).reshape(-1, 2)
    empty = np.full(edges.shape[0], np.nan)
    if rig is None or not edges.size:
        return empty, empty.copy(), np.zeros(edges.shape[0], dtype=int)
    xy = _bone_sample(detections)
    if xy is not None:
        pts3d, _ = triangulate_ransac(rig, xy, threshold=DETECTION_RANSAC_PX)
    else:
        frames = labels.labeled_frames()
        pts3d = np.full((len(frames), skeleton.n_points, 3), np.nan)
        for i, t in enumerate(frames):
            gt = labels.frame_effective(t)
            if (np.isfinite(gt).all(-1).sum(0) >= 2).any():
                pts3d[i] = triangulate(rig, gt)
    lengths = edge_lengths(pts3d, edges)  # (S, E)
    count = np.isfinite(lengths).sum(0)
    median = np.full(edges.shape[0], np.nan)
    mad = np.full(edges.shape[0], np.nan)
    have = count > 0  # only these have a finite entry to take a median over
    if have.any():
        median[have] = np.nanmedian(lengths[:, have], axis=0)
        mad[have] = np.nanmedian(np.abs(lengths[:, have] - median[have]), axis=0)
    return median, mad, count

detected_frame

detected_frame(
    detections: Detections | None,
    labels: Labels,
    frame: int,
    instance: int = 0,
) -> tuple[np.ndarray | None, np.ndarray | None]

One instance's detector pixels (V, K, 2) and confidences (V, K).

From a project's merged Detections (ADR-0083), the slot whose detections the instance's clicks sit on (instance_frame, ADR-0128). No slot is refused for the clicks that sit off it: the solve's guide is the skeleton the operator drags into place, and dragging its joints must not take it away (ADR-0138). NaN where the detector did not fire, absent keypoints masked, and (None, None) without detections at this frame, or for an instance no slot is.

Source code in src/deeperfly/labels/check.py

def detected_frame(
    detections: Detections | None,
    labels: Labels,
    frame: int,
    instance: int = 0,
) -> tuple[np.ndarray | None, np.ndarray | None]:
    """One instance's detector pixels `(V, K, 2)` and confidences `(V, K)`.

    From a project's merged [Detections][deeperfly.labels.detections.Detections]
    (ADR-0083), the slot whose detections the instance's clicks sit on
    ([instance_frame][deeperfly.labels.identity.instance_frame], ADR-0128). No slot
    is refused for the clicks that sit off it: the solve's guide is the skeleton the
    operator drags into place, and dragging its joints must not take it away
    (ADR-0138). NaN where the detector did not fire, absent keypoints masked, and
    `(None, None)` without detections at this frame, or for an instance no slot is.
    """
    if detections is None:
        return None, None
    det, conf = instance_frame(detections, labels, frame, instance, refuse=False)
    if det is None or conf is None:
        return None, None
    absent = labels.absent_frame(frame, instance=instance)  # (K,)
    det[:, absent] = np.nan
    conf[:, absent] = 0.0
    return det, conf

detection_precision

detection_precision(
    rig: Rig | None, detections: Detections | None
) -> np.ndarray | None

Each view's measured precision over a project's detections, (V,).

view_precision over up to BONE_REF_FRAMES covered frames of every animal, weighted by confidence (ADR-0129); None without a rig or a covered frame.

Source code in src/deeperfly/labels/check.py

def detection_precision(
    rig: Rig | None, detections: Detections | None
) -> np.ndarray | None:
    """Each view's measured precision over a project's detections, `(V,)`.

    [view_precision][deeperfly.pose3d.triangulation.view_precision] over up to
    `BONE_REF_FRAMES` covered frames of every animal, weighted by confidence
    (ADR-0129); `None` without a rig or a covered frame.
    """
    if rig is None or detections is None:
        return None
    sel = detection_sample(detections)
    if not sel.size:
        return None
    reads = [detections.read(sel, n) for n in range(max(detections.n_animals, 1))]
    xy = np.concatenate([r[0] for r in reads], axis=1)
    weights = precision_from_conf(np.concatenate([r[1] for r in reads], axis=1))
    return view_precision(rig, xy, weights, threshold=DETECTION_RANSAC_PX)

detection_sample

detection_sample(detections: Detections) -> np.ndarray

Up to BONE_REF_FRAMES covered frames, spread over the detections, sorted.

The covered frames are first cut to those of at most SAMPLE_CHUNKS chunks of T_CHUNK frames, taken where evenly spread frames fall, so a sample of a long recording reads a few dozen chunks rather than one per frame (ADR-0141). Empty where the detections cover no frame.

Source code in src/deeperfly/labels/check.py

def detection_sample(detections: Detections) -> np.ndarray:
    """Up to `BONE_REF_FRAMES` covered frames, spread over the detections, sorted.

    The covered frames are first cut to those of at most `SAMPLE_CHUNKS` chunks of
    `T_CHUNK` frames, taken where evenly spread frames fall, so a sample of a long
    recording reads a few dozen chunks rather than one per frame (ADR-0141). Empty
    where the detections cover no frame.
    """
    frames = detections.covered_frames()
    chunk = frames // T_CHUNK
    if np.unique(chunk).size > SAMPLE_CHUNKS:
        at = np.linspace(0, frames.size - 1, SAMPLE_CHUNKS).round().astype(int)
        frames = frames[np.isin(chunk, chunk[at])]
    pick = np.linspace(0, frames.size - 1, min(BONE_REF_FRAMES, frames.size))
    return frames[np.unique(pick.round().astype(int))]

detections_3d

detections_3d(
    rig: Rig,
    det: ndarray,
    conf: ndarray | None,
    precision: ndarray | None = None,
) -> tuple[np.ndarray, np.ndarray]

The detections' own 3D per keypoint, (K, 3) NaN where no two views agree.

A confidence-weighted RANSAC, not a least-squares fit (ADR-0074), whose consensus is voted by the confidences and whose fits weigh each view by its (V,) precision (detection_precision, ADR-0129); a detection below DEFAULT_MIN_CONF takes no part, and a keypoint no two views place in depth is NaN (ADR-0130). Also returns the (V, K) inlier mask.

Source code in src/deeperfly/labels/check.py

def detections_3d(
    rig: Rig,
    det: np.ndarray,
    conf: np.ndarray | None,
    precision: np.ndarray | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    """The detections' own 3D per keypoint, `(K, 3)` NaN where no two views agree.

    A confidence-weighted RANSAC, not a least-squares fit (ADR-0074), whose consensus
    is voted by the confidences and whose fits weigh each view by its `(V,)`
    `precision` ([detection_precision][], ADR-0129); a detection below
    `DEFAULT_MIN_CONF` takes no part, and a keypoint no two views place in depth is
    NaN (ADR-0130). Also returns the `(V, K)` inlier mask.
    """
    weights = None if conf is None else precision_from_conf(conf)
    if conf is not None:
        # Below the floor the detector saw nothing there; as a view of its own such a
        # pixel would still give the direct views' rays a depth (ADR-0130).
        det = np.where((np.asarray(conf) >= DEFAULT_MIN_CONF)[..., None], det, np.nan)
    return triangulate_ransac(
        rig,
        det,
        threshold=DETECTION_RANSAC_PX,
        min_views=2,
        weights=weights,
        weighted_vote=True,
        view_precision=precision,
    )

epipolar_lines

epipolar_lines(
    rig: Rig, names: Sequence[str], src: int, pixel: ndarray
) -> dict[str, list]

The source pixel's viewing ray, as a segment in every other view, by name.

Sampled across a band around its closest approach to the world origin, so the segment spans the animal rather than running off to infinity.

Source code in src/deeperfly/labels/check.py

def epipolar_lines(
    rig: Rig, names: Sequence[str], src: int, pixel: np.ndarray
) -> dict[str, list]:
    """The source pixel's viewing ray, as a segment in every other view, by name.

    Sampled across a band around its closest approach to the world origin, so the
    segment spans the animal rather than running off to infinity.
    """
    origin, direction = rig.views[names[src]].backproject_ray(pixel)
    t0 = -float(origin @ direction) / float(direction @ direction)
    span = max(abs(t0), 1.0)
    ends = np.stack(
        [origin + (t0 - span) * direction, origin + (t0 + span) * direction]
    )
    lines = {}
    for v, name in enumerate(names):
        if v == src:
            continue
        lines[name] = rig.views[name].project(ends).tolist()
    return lines

frame_checks

frame_checks(
    rig: Rig,
    skeleton: Skeleton,
    bone_ref: tuple[ndarray, ndarray, ndarray],
    gt: ndarray,
) -> dict[str, list]

The label checks of one frame, for the editor to flag (ADR-0074).

label_checks on this frame alone, so a frame's flags are the rows the project's list holds for it (ADR-0117). outliers are labeled views the consensus of four or more left out; inconsistent joints whose labeled views do not meet within CHECK_PX, none to blame; bones edges whose length between two anchors is off the recording's own distribution, which catches a swap coherent across every view. gt is the frame's effective (V, K, 2) labels.

Source code in src/deeperfly/labels/check.py

def frame_checks(
    rig: Rig,
    skeleton: Skeleton,
    bone_ref: tuple[np.ndarray, np.ndarray, np.ndarray],
    gt: np.ndarray,
) -> dict[str, list]:
    """The label checks of one frame, for the editor to flag (ADR-0074).

    [label_checks][] on this frame alone, so a frame's flags are the rows the
    project's list holds for it (ADR-0117). `outliers` are labeled views the consensus
    of four or more left out; `inconsistent` joints whose labeled views do not meet
    within `CHECK_PX`, none to blame; `bones` edges whose length between two anchors
    is off the recording's own distribution, which catches a swap coherent across
    every view. `gt` is the frame's effective `(V, K, 2)` labels.
    """
    checks: dict[str, list] = {key: [] for key in _KINDS.values()}
    for _, flag in _checks(rig, skeleton, bone_ref, gt[:, None])[0]:
        del flag["severity"]
        kind = flag.pop("kind")
        if kind == "bone":
            flag["a"], flag["b"] = flag.pop("keypoint"), flag.pop("other")
        checks[_KINDS[kind]].append(flag)
    return checks

gt_anchor

gt_anchor(
    rig: Rig, gt: ndarray
) -> tuple[np.ndarray | None, np.ndarray]

One keypoint's GT-only 3D anchor, and the (V,) labeled views it rests on.

Two or three labels are triangulated together; four or more take the RANSAC consensus within CHECK_PX of at least three views (ADR-0074). (None, seen) below two labels, when nothing triangulates, and when no two labeled views are far enough from parallel and from opposite to place the point in depth (ADR-0130).

Source code in src/deeperfly/labels/check.py

def gt_anchor(rig: Rig, gt: np.ndarray) -> tuple[np.ndarray | None, np.ndarray]:
    """One keypoint's GT-only 3D anchor, and the `(V,)` labeled views it rests on.

    Two or three labels are triangulated together; four or more take the RANSAC
    consensus within `CHECK_PX` of at least three views (ADR-0074). `(None, seen)`
    below two labels, when nothing triangulates, and when no two labeled views are far
    enough from parallel and from opposite to place the point in depth (ADR-0130).
    """
    seen = np.isfinite(gt).all(axis=-1)
    if int(seen.sum()) < 2:
        return None, seen
    if int(seen.sum()) >= 4:
        x, inliers = triangulate_ransac(
            rig, gt[:, None], threshold=CHECK_PX, min_views=3
        )
        if np.isfinite(x[0]).all():
            return x[0], inliers[:, 0] & seen
    x = triangulate(rig, gt)
    if not (np.isfinite(x).all() and placed_in_depth(rig, x, seen)):
        return None, seen
    return x, seen

label_checks

label_checks(
    rig: Rig,
    skeleton: Skeleton,
    bone_ref: tuple[ndarray, ndarray, ndarray],
    labels: Labels,
) -> dict

The label checks of every labeled frame at once, worst first (ADR-0102).

Each keypoint with two or more labeled views is anchored by gt_anchor's rule on the labels alone, with no detections or prior, and every frame is checked in one batch; frame_checks is the same check on one frame. Returns flags, {frame, instance, kind, keypoint, view?, views?, residual_px?, edge?, other?, length?, expected?, ratio?, severity} with kind "outlier", "inconsistent" or "bone" (severity is the multiple of the threshold the flag exceeds, so pixels and bone lengths sort together); views and keypoints, {name, n, median_px, p95_px, max_px} of the labeled cells' reprojection error to their anchor; and frames_checked.

Source code in src/deeperfly/labels/check.py

def label_checks(
    rig: Rig,
    skeleton: Skeleton,
    bone_ref: tuple[np.ndarray, np.ndarray, np.ndarray],
    labels: Labels,
) -> dict:
    """The label checks of every labeled frame at once, worst first (ADR-0102).

    Each keypoint with two or more labeled views is anchored by [gt_anchor][]'s rule
    on the labels alone, with no detections or prior, and every frame is checked in
    one batch; [frame_checks][] is the same check on one frame. Returns `flags`,
    `{frame, instance, kind, keypoint, view?, views?, residual_px?, edge?, other?,
    length?, expected?, ratio?, severity}` with `kind` `"outlier"`, `"inconsistent"`
    or `"bone"` (`severity` is the multiple of the threshold the flag exceeds, so
    pixels and bone lengths sort together); `views` and `keypoints`, `{name, n,
    median_px, p95_px, max_px}` of the labeled cells' reprojection error to their
    anchor; and `frames_checked`.
    """
    cells = [
        (t, i) for t in labels.labeled_frames() for i in labels.labeled_instances(t)
    ]
    out: dict = {"flags": [], "views": [], "keypoints": [], "frames_checked": 0}
    if not cells:
        return out
    gt = np.stack([labels.frame_effective(t, instance=i) for t, i in cells], axis=1)
    solvable = (np.isfinite(gt).all(axis=-1).sum(axis=0) >= 2).any(axis=-1)  # (S,)
    out["frames_checked"] = len(
        {t for (t, _), ok in zip(cells, solvable, strict=True) if ok}
    )
    if not solvable.any():
        return out
    flags, res = _checks(rig, skeleton, bone_ref, gt)
    out["flags"] = sorted(
        ({"frame": cells[s][0], "instance": cells[s][1], **f} for s, f in flags),
        key=lambda f: (-f["severity"], f["frame"], f["instance"]),
    )
    for key, names, axis in (
        ("views", rig.names, 1),
        ("keypoints", skeleton.point_names, 0),
    ):
        for j, name in enumerate(names):
            r = res[j] if axis else res[:, :, j]
            r = r[np.isfinite(r)]
            row: dict = {"name": name, "n": int(r.size)}
            if r.size:
                row |= {
                    "median_px": _round(np.median(r)),
                    "p95_px": _round(np.percentile(r, 95)),
                    "max_px": _round(r.max()),
                }
            out[key].append(row)
    return out

prior_3d

prior_3d(
    rig: Rig | None,
    skeleton: Skeleton,
    labels: Labels,
    detections: Detections | None,
    frame: int,
    instance: int = 0,
    precision: ndarray | None = None,
) -> tuple[np.ndarray, np.ndarray]

A (K, 3) prior 3D per keypoint (NaN where absent), and which are default pose.

Layered per keypoint (ADR-0074): the skeleton's default pose, overridden by the nearest earlier frame of this animal that triangulates the keypoint, overridden by this frame's detections (detections_3d with the views' precision). The (K,) mask names the keypoints left on the default pose alone, a starting instance rather than a guide. Absent keypoints are NaN.

Source code in src/deeperfly/labels/check.py

def prior_3d(
    rig: Rig | None,
    skeleton: Skeleton,
    labels: Labels,
    detections: Detections | None,
    frame: int,
    instance: int = 0,
    precision: np.ndarray | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    """A `(K, 3)` prior 3D per keypoint (NaN where absent), and which are default pose.

    Layered per keypoint (ADR-0074): the skeleton's default pose, overridden by the
    nearest earlier frame of this animal that triangulates the keypoint, overridden by
    this frame's detections ([detections_3d][] with the views' `precision`). The `(K,)`
    mask names the keypoints left on the default pose alone, a starting instance rather
    than a guide. Absent keypoints are NaN.
    """
    src = skeleton.resolved_pose()
    from_pose = np.ones(skeleton.n_points, dtype=bool)
    if rig is not None:
        # Newest earlier frame first, filling only the keypoints still missing; a frame
        # is triangulated only when it has two labeled views of one of them.
        need = np.ones(skeleton.n_points, dtype=bool)
        # Only the frames where *this* slot is labeled: the prior is what the same
        # animal did last, and animal 0's last pose is no evidence about animal 1.
        earlier = [
            t
            for t in labels.labeled_frames()
            if t < frame and instance in labels.labeled_instances(t)
        ]
        for t in reversed(earlier):
            if not need.any():
                break
            gt = labels.frame_effective(t, instance=instance)  # (V, K, 2)
            seen = np.isfinite(gt).all(axis=-1).sum(axis=0) >= 2  # (K,) triangulable
            if not (need & seen).any():
                continue
            tri = triangulate(rig, gt)  # (K, 3)
            # Only where the labels place it in depth, not on one line's two ends.
            labeled = np.isfinite(gt).all(axis=-1)  # (V, K)
            finite = need & placed_in_depth(rig, tri, labeled)
            src[finite] = tri[finite]
            from_pose &= ~finite
            need &= ~finite
        det, conf = detected_frame(detections, labels, frame, instance)
        if det is not None:
            tri, _ = detections_3d(rig, det, conf, precision)  # (K, 3) detections
            finite = np.isfinite(tri).all(axis=-1)
            src[finite] = tri[finite]
            from_pose &= ~finite
    src[labels.absent_frame(frame, instance=instance)] = np.nan
    return src, from_pose

project_all

project_all(
    rig: Rig, view_names: Sequence[str], pts3d: ndarray
) -> np.ndarray

A (K, 3) pose projected into every view as (V, K, 2) (NaN where none).

Source code in src/deeperfly/labels/check.py

def project_all(rig: Rig, view_names: Sequence[str], pts3d: np.ndarray) -> np.ndarray:
    """A `(K, 3)` pose projected into every view as `(V, K, 2)` (`NaN` where none)."""
    out = np.full((len(view_names), pts3d.shape[0], 2), np.nan)
    for v, name in enumerate(view_names):
        if name in rig.names:
            out[v] = rig.views[name].project(pts3d)
    return out

solve_keypoint

solve_keypoint(
    rig: Rig,
    gt: ndarray,
    det: ndarray,
    prior: ndarray,
    det_conf: ndarray | None = None,
    precision: ndarray | None = None,
) -> tuple[np.ndarray | None, str]

One keypoint's 3D from this frame's evidence, and what anchored it.

The cascade of ADR-0027, strongest first: GT views that place the keypoint in depth, or GT views with detections, give "gt", solved by triangulate_stabilized with the detections (else the prior reprojected) as the stabilizer (ADR-0068); GT views and a prior give "ray", the prior slid onto a label's ray; two or more detected views give "detected" (detections_3d); the prior alone gives "prior"; nothing gives (None, "none"). GT views that look along one line, such as a ventral and a dorsal label, fix the keypoint across that line only, so its depth comes from what follows them (ADR-0130). gt, det are (V, 2) and prior (3,), NaN where missing; det_conf drops the detections below DEFAULT_MIN_CONF, votes in the consensus solves and weighs their fits with the views' (V,) precision (ADR-0129).

Source code in src/deeperfly/labels/check.py

def solve_keypoint(
    rig: Rig,
    gt: np.ndarray,
    det: np.ndarray,
    prior: np.ndarray,
    det_conf: np.ndarray | None = None,
    precision: np.ndarray | None = None,
) -> tuple[np.ndarray | None, str]:
    """One keypoint's 3D from this frame's evidence, and what anchored it.

    The cascade of ADR-0027, strongest first: GT views that place the keypoint in depth,
    or GT views with detections, give `"gt"`, solved by
    [triangulate_stabilized][deeperfly.pose3d.triangulation.triangulate_stabilized]
    with the detections (else the prior reprojected) as the stabilizer (ADR-0068); GT
    views and a prior give `"ray"`, the prior slid onto a label's ray; two or more
    detected views give `"detected"` ([detections_3d][]); the prior alone gives
    `"prior"`; nothing gives `(None, "none")`. GT views that look along one line, such
    as a ventral and a dorsal label, fix the keypoint across that line only, so its
    depth comes from what follows them (ADR-0130). `gt`, `det` are `(V, 2)` and `prior`
    `(3,)`, NaN where missing; `det_conf` drops the detections below `DEFAULT_MIN_CONF`,
    votes in the consensus solves and weighs their fits with the views' `(V,)`
    `precision` (ADR-0129).
    """
    if det_conf is not None:
        det = np.where((np.asarray(det_conf) >= DEFAULT_MIN_CONF)[:, None], det, np.nan)
    seen = np.isfinite(gt).all(axis=-1)
    det_seen = np.isfinite(det).all(axis=-1)
    if int(seen.sum()) >= 2:
        x0, used = gt_anchor(rig, gt)
        if x0 is not None:
            # The stabilizer is the detector's pixels where it fired and the prior's
            # reprojection elsewhere, so nothing invents an observation.
            stab = det.copy()
            if np.isfinite(prior).all():
                fill = ~np.isfinite(stab).all(axis=-1)
                stab[fill] = np.asarray(rig.project(prior[None, :]))[:, 0][fill]
            gt_used = np.where(used[:, None], gt, np.nan)
            return triangulate_stabilized(rig, x0, gt_used, stab), "gt"
    if seen.any() and det_seen.any():
        # The start is the consensus of the labels and the detections, a label weighing
        # as a sure observation; the stabilized solve then anchors on it.
        combo = gt.copy()
        fill = ~seen & det_seen
        combo[fill] = det[fill]
        weights = (
            np.ones(len(combo)) if det_conf is None else precision_from_conf(det_conf)
        )
        weights = np.where(seen, 1.0, weights)
        x0, _ = triangulate_ransac(
            rig,
            combo,
            threshold=DETECTION_RANSAC_PX,
            min_views=2,
            weights=weights,
            weighted_vote=True,
            view_precision=precision,
        )
        if np.isfinite(x0).all():
            return triangulate_stabilized(rig, x0, gt, det), "gt"
    if seen.any() and np.isfinite(prior).all():
        v = int(np.flatnonzero(seen)[0])
        origin, direction = rig.views[rig.names[v]].backproject_ray(gt[v])
        x = np.asarray(closest_point_on_ray(origin, direction, prior), dtype=float)
        if np.isfinite(x).all():
            return x, "ray"
    if int(det_seen.sum()) >= 2:
        x0, _ = detections_3d(rig, det, det_conf, precision)
        if np.isfinite(x0).all():
            return x0, "detected"
    if np.isfinite(prior).all():
        return prior, "prior"
    return None, "none"

Label suggestions: the joints of a project a label would most likely fix.

Every (view, frame, keypoint) cell of the detector's 2D is scored by its residual against the other views' RANSAC consensus, in units of the project's typical residual; a joint takes its worst view, boosted by its drivers (the views disagree, a bone off its usual length, a detection off the image, a view's sides exchanged), and picks are whole frames ranked by the evidence their suspicious joints carry (ADR-0073). suggest_sample is the cold start, whole frames spread over the unlabeled part. The GUI's Suggested panel writes a project's suggestions.json and walks it (ADR-0084).

BONE_MIN_SAMPLES module-attribute

BONE_MIN_SAMPLES = 20

Frames a bone must be measured in before its usual length counts (ADR-0074).

BONE_RATIO module-attribute

BONE_RATIO = 1.5

A bone longer or shorter than its usual length by this factor is a driver.

GATE_PX module-attribute

GATE_PX = 12.0

RANSAC inlier gate, in view pixels: the consensus a view must fall within.

MIN_SCORE module-attribute

MIN_SCORE = 5.0

A joint at or above this many typical residuals is suspicious and gets listed.

PER_FRAME module-attribute

PER_FRAME = 10

Joints a picked frame lists at most: a labeling stop, not the whole budget.

SIDES_MARGIN module-attribute

SIDES_MARGIN = 5.0

Typical residuals a view's mirrored fit must win by to call its sides exchanged.

CellEvidence dataclass

What a run's 2D says about each (view, frame, keypoint) cell.

residual is in pixels against the consensus (or a run's 3D), NaN without a detection or a reference, and at least the gate where no two views agree. inlier marks the views the consensus kept and off_frame a detection outside its image; bone_ratio and bone_edge are, per joint, the bone that strays farthest from its usual length, and sides_exchanged marks a (view, frame) whose paired joints fit the others better mirrored.

Source code in src/deeperfly/labels/suggest.py

@dataclass(frozen=True)
class CellEvidence:
    """What a run's 2D says about each `(view, frame, keypoint)` cell.

    `residual` is in pixels against the consensus (or a run's 3D), NaN without a
    detection or a reference, and at least the gate where no two views agree.
    `inlier` marks the views the consensus kept and `off_frame` a detection outside
    its image; `bone_ratio` and `bone_edge` are, per joint, the bone that strays
    farthest from its usual length, and `sides_exchanged` marks a `(view, frame)`
    whose paired joints fit the others better mirrored.
    """

    residual: Float[np.ndarray, "V T K"]
    conf: Float[np.ndarray, "V T K"]
    detected: Bool[np.ndarray, "V T K"]
    inlier: Bool[np.ndarray, "V T K"]
    off_frame: Bool[np.ndarray, "V T K"]
    sides_exchanged: Bool[np.ndarray, "V T"]
    bone_ratio: Float[np.ndarray, "T K"]
    bone_edge: Int[np.ndarray, "T K"]
    typical_residual: float
    gate_px: float

    @property
    def n_detected(self) -> Int[np.ndarray, "T K"]:
        """Views with a detection, per joint."""
        return self.detected.sum(0)

    @property
    def n_inlier(self) -> Int[np.ndarray, "T K"]:
        """Views the consensus kept, per joint."""
        return self.inlier.sum(0)

    @property
    def disagree(self) -> Bool[np.ndarray, "T K"]:
        """Joints whose views disagree: two or more rejected, or no two agreeing."""
        n_d, n_i = self.n_detected, self.n_inlier
        return (n_d >= 2) & ((n_i < 2) | (n_d - n_i >= 2))

    @property
    def bone_off(self) -> Bool[np.ndarray, "T K"]:
        """Joints with a bone off its usual length by `BONE_RATIO` either way."""
        return np.abs(np.log(self.bone_ratio)) > np.log(BONE_RATIO)

bone_off property

bone_off: Bool[ndarray, 'T K']

Joints with a bone off its usual length by BONE_RATIO either way.

disagree property

disagree: Bool[ndarray, 'T K']

Joints whose views disagree: two or more rejected, or no two agreeing.

n_detected property

n_detected: Int[ndarray, 'T K']

Views with a detection, per joint.

n_inlier property

n_inlier: Int[ndarray, 'T K']

Views the consensus kept, per joint.

FlaggedView dataclass

One view a suggestion says to look at, with the numbers behind it.

score is the residual in typical residuals; rejected says the consensus left the view out, off_frame that its detection lies outside its image.

Source code in src/deeperfly/labels/suggest.py

@dataclass(frozen=True)
class FlaggedView:
    """One view a suggestion says to look at, with the numbers behind it.

    `score` is the residual in typical residuals; `rejected` says the consensus left
    the view out, `off_frame` that its detection lies outside its image.
    """

    view: str
    score: float
    residual_px: float | None
    conf: float | None
    rejected: bool = False
    off_frame: bool = False

Ranked dataclass

A project's picks, best first, with the ranking's stats for the document.

Source code in src/deeperfly/labels/suggest.py

@dataclass(frozen=True)
class Ranked:
    """A project's picks, best first, with the ranking's `stats` for the document."""

    picks: list[Suggestion]
    stats: dict[str, Any]

Suggestion dataclass

One joint to label: a frame, a keypoint, the views that make the case, and why.

score is the worst view's, boosted by the drivers; keypoint=None is a whole frame to label, naming no views. instance is the animal the pick is about.

Source code in src/deeperfly/labels/suggest.py

@dataclass(frozen=True)
class Suggestion:
    """One joint to label: a frame, a keypoint, the views that make the case, and why.

    `score` is the worst view's, boosted by the `drivers`; `keypoint=None` is a whole
    frame to label, naming no views. `instance` is the animal the pick is about.
    """

    frame: int
    keypoint: str | None
    score: float
    views: tuple[FlaggedView, ...]
    drivers: tuple[str, ...] = ()
    instance: int = 0

cell_evidence

cell_evidence(
    rig: Rig,
    points: Float[ndarray, "V T K 2"],
    conf: Float[ndarray, "V T K"],
    *,
    sizes: Sequence[tuple[int, int] | None] | None = None,
    pts3d: Float[ndarray, "T K 3"] | None = None,
    edges: Int[ndarray, "E 2"] | None = None,
    flip_perm: Int[ndarray, " K"] | None = None,
    gate_px: float = GATE_PX,
    sides_margin: float = SIDES_MARGIN,
) -> CellEvidence

Triangulate the 2D by consensus and measure every cell against it.

sizes (per view (height, width)), edges and flip_perm enable the off-frame, bone and exchanged-sides drivers, each skipped when None; pts3d replaces the consensus as the residuals' reference where it is finite, though the consensus still decides which views agree.

Source code in src/deeperfly/labels/suggest.py

def cell_evidence(
    rig: Rig,
    points: Float[np.ndarray, "V T K 2"],
    conf: Float[np.ndarray, "V T K"],
    *,
    sizes: Sequence[tuple[int, int] | None] | None = None,
    pts3d: Float[np.ndarray, "T K 3"] | None = None,
    edges: Int[np.ndarray, "E 2"] | None = None,
    flip_perm: Int[np.ndarray, " K"] | None = None,
    gate_px: float = GATE_PX,
    sides_margin: float = SIDES_MARGIN,
) -> CellEvidence:
    """Triangulate the 2D by consensus and measure every cell against it.

    `sizes` (per view `(height, width)`), `edges` and `flip_perm` enable the off-frame,
    bone and exchanged-sides drivers, each skipped when `None`; `pts3d` replaces the
    consensus as the residuals' reference where it is finite, though the consensus
    still decides which views agree.
    """
    points = np.asarray(points, dtype=float)
    conf = np.asarray(conf, dtype=float)
    n_v, n_t, n_k, _ = points.shape
    flat = points.reshape(n_v, n_t * n_k, 2)
    # The consensus is precision-weighted, a better reference to rank on (ADR-0067);
    # which views agree is still decided unweighted.
    consensus, inlier = triangulate_ransac(
        rig,
        flat,
        threshold=gate_px,
        min_views=2,
        weights=precision_from_conf(conf).reshape(n_v, -1),
    )
    consensus = consensus.reshape(n_t, n_k, 3)
    inlier = inlier.reshape(n_v, n_t, n_k)
    if pts3d is None:
        reference = consensus
    else:
        pts3d = np.asarray(pts3d, dtype=float)
        reference = np.where(np.isfinite(pts3d).all(-1)[..., None], pts3d, consensus)
    residual = np.asarray(
        reprojection_error(rig, reference.reshape(-1, 3), flat), dtype=float
    ).reshape(n_v, n_t, n_k)
    detected = np.isfinite(points).all(-1)
    residual = np.where(detected, residual, np.nan)
    # No two views agree: every detection is at least the gate from any agreement.
    none = ~np.isfinite(reference).all(-1) & (detected.sum(0) >= 2)
    residual = np.where(none[None] & detected, np.fmax(residual, gate_px), residual)
    finite = residual[np.isfinite(residual)]
    typical = float(np.median(finite)) if finite.size else np.nan
    typical = max(typical, 1.0) if np.isfinite(typical) else 1.0

    exchanged = _exchanged_sides(
        rig, points, reference, residual, flip_perm, sides_margin * typical
    )

    if sizes is not None and len(sizes) != n_v:
        raise ValueError(f"{len(sizes)} frame sizes for {n_v} views")
    off = np.zeros_like(detected)
    for v, size in enumerate(sizes or ()):
        if size is None:
            continue
        h, w = size
        x, y = points[v, ..., 0], points[v, ..., 1]
        off[v] = detected[v] & ((x < -0.5) | (y < -0.5) | (x > w - 0.5) | (y > h - 0.5))

    ratio = np.ones((n_t, n_k))
    which = np.full((n_t, n_k), -1, dtype=int)
    if edges is not None and len(edges):
        e = np.asarray(edges, dtype=int).reshape(-1, 2)
        length = edge_lengths(reference, e)  # (T, E)
        usual = np.full(e.shape[0], np.nan)
        enough = np.isfinite(length).sum(0) >= BONE_MIN_SAMPLES
        if enough.any():
            usual[enough] = np.nanmedian(length[:, enough], axis=0)
        rel = length / np.where(usual > 0, usual, np.nan)
        rel = np.where(np.isfinite(rel), np.clip(rel, 1e-6, None), 1.0)
        for j, (a, b) in enumerate(e):
            for k in (int(a), int(b)):
                farther = np.abs(np.log(rel[:, j])) > np.abs(np.log(ratio[:, k]))
                ratio[:, k] = np.where(farther, rel[:, j], ratio[:, k])
                which[:, k] = np.where(farther, j, which[:, k])
    return CellEvidence(
        residual=residual,
        conf=conf,
        detected=detected,
        inlier=inlier & detected,
        off_frame=off,
        sides_exchanged=exchanged,
        bone_ratio=ratio,
        bone_edge=which,
        typical_residual=typical,
        gate_px=gate_px,
    )

labeled_joints

labeled_joints(
    labels: Labels,
    n_frames: int,
    keypoints: Sequence[str],
    *,
    instance: int = 0,
) -> Bool[np.ndarray, "T K"]

(T, K): one animal's joints that carry a label in any view or an absent mark.

Matched by name, so a run made before a skeleton edit still lines up.

Source code in src/deeperfly/labels/suggest.py

def labeled_joints(
    labels: Labels, n_frames: int, keypoints: Sequence[str], *, instance: int = 0
) -> Bool[np.ndarray, "T K"]:
    """`(T, K)`: one animal's joints that carry a label in any view or an absent mark.

    Matched by name, so a run made before a skeleton edit still lines up.
    """
    out = np.zeros((n_frames, len(keypoints)), dtype=bool)
    index = []
    for name in keypoints:
        try:
            index.append(labels.keypoint_names.index(name))
        except KeyError, ValueError:
            index.append(-1)
    cols = np.asarray(index)
    known = cols >= 0
    for t in labels.labeled_frames():
        if t < n_frames:
            arr = labels.frame(t, instance=instance)
            has_label = np.isfinite(arr).all(-1).any(0)
            settled = has_label | labels.absent_frame(t, instance=instance)
            out[t, known] = settled[cols[known]]
    return out

read_suggestions

read_suggestions(path: str | Path) -> dict[str, Any]

Read a suggestions document of the current format, FORMAT_VERSION.

Raises:

Type	Description
ValueError	If the file is not a suggestions document, or one of another format version, which is refused rather than migrated (ADR-0061): suggesting again rewrites it.

Source code in src/deeperfly/labels/suggest.py

def read_suggestions(path: str | Path) -> dict[str, Any]:
    """Read a suggestions document of the current format, `FORMAT_VERSION`.

    Raises:
        ValueError: If the file is not a suggestions document, or one of another
            format version, which is refused rather than migrated (ADR-0061):
            suggesting again rewrites it.
    """
    document = json.loads(Path(path).read_text())
    if not isinstance(document, dict) or "suggestions" not in document:
        raise ValueError(f"{path}: not a suggestions document")
    version = document.get("format_version")
    if version != FORMAT_VERSION:
        raise ValueError(
            f"{path}: suggestions format_version {version!r}, expected "
            f"{FORMAT_VERSION}; suggest again to rewrite it (ADR-0061)"
        )
    return document

results_source

results_source(res: Results) -> dict[str, Any]

The provenance a document records for picks scored from a run's results.

Source code in src/deeperfly/labels/suggest.py

def results_source(res: Results) -> dict[str, Any]:
    """The provenance a document records for picks scored from a run's results."""
    return {
        "results": str(res.path),
        "pose2d": res.fingerprint("pose2d"),
        "pose3d": res.fingerprint("pose3d"),
    }

score_cells

score_cells(
    evidence: CellEvidence,
) -> Float[np.ndarray, "V T K"]

Per-cell scores: the residual in typical residuals, NaN without a residual.

Source code in src/deeperfly/labels/suggest.py

def score_cells(evidence: CellEvidence) -> Float[np.ndarray, "V T K"]:
    """Per-cell scores: the residual in typical residuals, NaN without a residual."""
    return evidence.residual / evidence.typical_residual

score_joints

score_joints(
    evidence: CellEvidence,
    cells: Float[ndarray, "V T K"] | None = None,
    *,
    disagree_boost: float = DISAGREE_BOOST,
    bone_boost: float = BONE_BOOST,
    off_frame_boost: float = OFF_FRAME_BOOST,
) -> Float[np.ndarray, "T K"]

Per-joint scores: the worst view's cell score times the drivers' boosts.

NaN below two detections, so such a joint is never suggested.

Source code in src/deeperfly/labels/suggest.py

def score_joints(
    evidence: CellEvidence,
    cells: Float[np.ndarray, "V T K"] | None = None,
    *,
    disagree_boost: float = DISAGREE_BOOST,
    bone_boost: float = BONE_BOOST,
    off_frame_boost: float = OFF_FRAME_BOOST,
) -> Float[np.ndarray, "T K"]:
    """Per-joint scores: the worst view's cell score times the drivers' boosts.

    NaN below two detections, so such a joint is never suggested.
    """
    if cells is None:
        cells = score_cells(evidence)
    worst = np.nanmax(np.where(np.isfinite(cells), cells, -np.inf), axis=0)
    worst = np.where(np.isfinite(worst), worst, np.nan)
    boost = np.where(evidence.disagree, disagree_boost, 1.0)
    boost = boost * np.where(evidence.bone_off, bone_boost, 1.0)
    boost = boost * np.where(evidence.off_frame.any(0), off_frame_boost, 1.0)
    return worst * boost

select_frames

select_frames(
    joint: Float[ndarray, "T K"],
    *,
    labeled: Bool[ndarray, "T K"] | None = None,
    frames: Sequence[int] | ndarray | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
) -> list[tuple[int, list[int]]]

Rows to label, best first, each with its suspicious joints, best first.

A joint is suspicious at or above min_score, labeled ones never. A row's evidence is sum(log2(score)) over the per_frame joints it would list, each weighted 1 / (1 + n) by how often its keypoint was picked; rows are taken greedily, no two within gap frames (frames maps rows to recording frames), until count joints are listed. Returns (row, keypoint indices) pairs.

Source code in src/deeperfly/labels/suggest.py

def select_frames(
    joint: Float[np.ndarray, "T K"],
    *,
    labeled: Bool[np.ndarray, "T K"] | None = None,
    frames: Sequence[int] | np.ndarray | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
) -> list[tuple[int, list[int]]]:
    """Rows to label, best first, each with its suspicious joints, best first.

    A joint is suspicious at or above `min_score`, labeled ones never. A row's evidence
    is `sum(log2(score))` over the `per_frame` joints it would list, each weighted
    `1 / (1 + n)` by how often its keypoint was picked; rows are taken greedily, no two
    within `gap` frames (`frames` maps rows to recording frames), until `count` joints
    are listed. Returns `(row, keypoint indices)` pairs.
    """
    n_t, n_k = joint.shape
    per_frame = max(1, min(int(per_frame), n_k))
    where = np.arange(n_t) if frames is None else np.asarray(frames, dtype=int)
    s = np.where(np.isfinite(joint), joint, 0.0)
    if labeled is not None:
        s = np.where(labeled, 0.0, s)
    bits = np.where(s >= min_score, np.log2(np.maximum(s, 1.0)), 0.0)
    frame_bits = _top_sum(bits, per_frame)
    pool = np.flatnonzero(frame_bits > 0)
    if not pool.size or count < 1:
        return []
    # The strongest frames are candidates enough: a pick lists at least one joint.
    pool = pool[np.argsort(-frame_bits[pool], kind="stable")[: 20 * count]]
    pool_bits = bits[pool]  # (P, K)
    available = np.ones(len(pool), dtype=bool)
    picked_kp = np.zeros(n_k)
    picks: list[tuple[int, list[int]]] = []
    remaining = count
    while remaining > 0 and available.any():
        weighted = _top_sum(pool_bits / (1.0 + picked_kp), per_frame)
        weighted[~available] = -np.inf
        i = int(np.argmax(weighted))
        if not weighted[i] > 0:
            break
        t = int(pool[i])
        available[i] = False
        if gap > 0:
            available &= np.abs(where[pool] - where[t]) >= gap
        ks = np.flatnonzero(bits[t] > 0)
        ks = ks[np.argsort(-s[t, ks], kind="stable")][: min(per_frame, remaining)]
        picks.append((t, [int(k) for k in ks]))
        picked_kp[ks] += 1
        remaining -= len(ks)
    return picks

suggest_detections

suggest_detections(
    rig: Rig,
    detections: Detections,
    labels: Labels | None = None,
    *,
    sizes: Sequence[tuple[int, int] | None] | None = None,
    edges: Int[ndarray, "E 2"] | None = None,
    flip_perm: Int[ndarray, " K"] | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
    gate_px: float = GATE_PX,
) -> Ranked

The suggestions of one project from its runs' merged detections (ADR-0083).

Only the frames the detections cover are scored, each from its winning run, and residuals are against the views' consensus, never a run's 3D; several animals are ranked one at a time and merged, each label instance from the slot its clicks sit on (instance_read, ADR-0128). The arguments are as for suggest_points.

Source code in src/deeperfly/labels/suggest.py

def suggest_detections(
    rig: Rig,
    detections: Detections,
    labels: Labels | None = None,
    *,
    sizes: Sequence[tuple[int, int] | None] | None = None,
    edges: Int[np.ndarray, "E 2"] | None = None,
    flip_perm: Int[np.ndarray, " K"] | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
    gate_px: float = GATE_PX,
) -> Ranked:
    """The suggestions of one project from its runs' merged detections (ADR-0083).

    Only the frames the detections cover are scored, each from its winning run, and
    residuals are against the views' consensus, never a run's 3D; several animals
    are ranked one at a time and merged, each label instance from the slot its
    clicks sit on ([instance_read][deeperfly.labels.identity.instance_read],
    ADR-0128). The arguments are as for `suggest_points`.
    """
    covered = detections.covered_frames()
    parts = []
    for n in range(max(detections.n_animals, 1)):
        points, conf = instance_read(detections, labels, covered, n)
        found = np.isfinite(points).all(-1).any((0, 2))
        parts.append(
            suggest_points(
                rig,
                points[:, found],
                conf[:, found],
                keypoints=detections.keypoint_names,
                labels=labels,
                sizes=sizes,
                frames=covered[found].tolist(),
                edges=edges,
                flip_perm=flip_perm,
                count=count,
                gap=gap,
                min_score=min_score,
                per_frame=per_frame,
                gate_px=gate_px,
                instance=n,
            )
        )
    return _merge_ranked(parts, count)

suggest_points

suggest_points(
    rig: Rig,
    points: Float[ndarray, "V T K 2"],
    conf: Float[ndarray, "V T K"],
    *,
    keypoints: Sequence[str],
    labels: Labels | None = None,
    sizes: Sequence[tuple[int, int] | None] | None = None,
    frames: Sequence[int] | ndarray | None = None,
    pts3d: Float[ndarray, "T K 3"] | None = None,
    edges: Int[ndarray, "E 2"] | None = None,
    flip_perm: Int[ndarray, " K"] | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
    gate_px: float = GATE_PX,
    instance: int = 0,
) -> Ranked

Rank the joints of one animal's 2D (ADR-0073).

points (V, T, K, 2) is the detector's 2D over the frames scored, keypoints the names along K, and frames the recording frame of each T index. labels are skipped, pts3d replaces the consensus as the reference, and sizes, edges and flip_perm enable the drivers as for cell_evidence; the rest are select_frames' knobs. Every pick is tagged with instance, and only that animal's labels are skipped.

Source code in src/deeperfly/labels/suggest.py

def suggest_points(
    rig: Rig,
    points: Float[np.ndarray, "V T K 2"],
    conf: Float[np.ndarray, "V T K"],
    *,
    keypoints: Sequence[str],
    labels: Labels | None = None,
    sizes: Sequence[tuple[int, int] | None] | None = None,
    frames: Sequence[int] | np.ndarray | None = None,
    pts3d: Float[np.ndarray, "T K 3"] | None = None,
    edges: Int[np.ndarray, "E 2"] | None = None,
    flip_perm: Int[np.ndarray, " K"] | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
    gate_px: float = GATE_PX,
    instance: int = 0,
) -> Ranked:
    """Rank the joints of one animal's 2D (ADR-0073).

    `points` `(V, T, K, 2)` is the detector's 2D over the frames scored, `keypoints`
    the names along `K`, and `frames` the recording frame of each `T` index. `labels`
    are skipped, `pts3d` replaces the consensus as the reference, and `sizes`,
    `edges` and `flip_perm` enable the drivers as for `cell_evidence`; the rest are
    `select_frames`' knobs. Every pick is tagged with `instance`, and only that
    animal's labels are skipped.
    """
    frame_index = (
        np.arange(np.asarray(points).shape[1])
        if frames is None
        else np.asarray(frames, dtype=int)
    )
    if frame_index.size == 0:
        # Nothing was detected in any frame (an animal the tracker never found), so
        # there is nothing to rank.
        return Ranked([], _empty_stats(min_score, per_frame, gate_px))
    evidence = cell_evidence(
        rig,
        points,
        conf,
        sizes=sizes,
        pts3d=pts3d,
        edges=edges,
        flip_perm=flip_perm,
        gate_px=gate_px,
    )
    cells = score_cells(evidence)
    joint = score_joints(evidence, cells)
    labeled = None
    if labels is not None:
        labeled = labeled_joints(labels, labels.n_frames, keypoints, instance=instance)[
            np.clip(frame_index, 0, labels.n_frames - 1)
        ]
        labeled &= frame_index[:, None] < labels.n_frames
    picks = select_frames(
        joint,
        labeled=labeled,
        frames=frame_index,
        count=count,
        gap=gap,
        min_score=min_score,
        per_frame=per_frame,
    )
    names = list(rig.names)
    edge_array = None if edges is None else np.asarray(edges, dtype=int).reshape(-1, 2)
    n_inlier = evidence.n_inlier
    n_detected = evidence.n_detected
    disagree = evidence.disagree
    bone_off = evidence.bone_off
    out: list[Suggestion] = []
    for t, ks in picks:
        for k in ks:
            c = cells[:, t, k]
            order = np.argsort(-np.where(np.isfinite(c), c, -np.inf), kind="stable")
            flagged = []
            for rank, v in enumerate(int(i) for i in order):
                if not evidence.detected[v, t, k]:
                    continue
                rejected = not evidence.inlier[v, t, k]
                off = bool(evidence.off_frame[v, t, k])
                cell = float(c[v]) if np.isfinite(c[v]) else 0.0
                if rank and not (cell >= min_score or rejected or off):
                    continue
                flagged.append(
                    FlaggedView(
                        view=names[v],
                        score=round(cell, 2),
                        residual_px=_finite(evidence.residual[v, t, k], 1),
                        conf=_finite(evidence.conf[v, t, k], 3),
                        rejected=rejected,
                        off_frame=off,
                    )
                )
            drivers = []
            if disagree[t, k]:
                n_i, n_d = int(n_inlier[t, k]), int(n_detected[t, k])
                drivers.append(
                    "no two views agree"
                    if n_i < 2
                    else f"only {n_i} of {n_d} views agree"
                )
            if bone_off[t, k] and edge_array is not None:
                a, b = edge_array[int(evidence.bone_edge[t, k])]
                drivers.append(
                    f"bone {keypoints[a]}-{keypoints[b]} at "
                    f"{evidence.bone_ratio[t, k]:.1f}x its usual length"
                )
            drivers.extend(
                f"left and right exchanged in {names[int(v)]}"
                for v in np.flatnonzero(evidence.sides_exchanged[:, t])
            )
            drivers.extend(f"off the image in {f.view}" for f in flagged if f.off_frame)
            out.append(
                Suggestion(
                    frame=int(frame_index[t]),
                    keypoint=str(keypoints[k]),
                    score=round(float(joint[t, k]), 2),
                    views=tuple(flagged),
                    drivers=tuple(drivers),
                    instance=instance,
                )
            )
    suspicious = np.isfinite(joint) & (joint >= min_score)
    if labeled is not None:
        suspicious &= ~labeled
    stats = {
        "typical_residual_px": round(evidence.typical_residual, 2),
        "gate_px": gate_px,
        "min_score": min_score,
        "per_frame": per_frame,
        "frames_scored": len(frame_index),
        "joints_suspicious": int(suspicious.sum()),
        "frames_suspicious": int(suspicious.any(1).sum()),
    }
    return Ranked(out, stats)

suggest_results

suggest_results(
    res: Results,
    labels: Labels | None = None,
    *,
    rig: Rig | None = None,
    edges: Int[ndarray, "E 2"] | None = None,
    flip_perm: Int[ndarray, " K"] | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
    gate_px: float = GATE_PX,
) -> Ranked

The suggestions of one project, from its run's pose2d (and pose3d).

Only the frames pose2d covers are scored (ADR-0082). The residuals are against pose3d/points on the frames pose3d covers, else against the views' consensus, and several tracked animals are ranked one at a time and merged, each label instance from the slot its clicks sit on (run_slots, ADR-0128). The views are the run's own when it stores them, else rig's, taken by name in the run's view order: a pose2d-only run of one animal stores none (ADR-0086). The file's frame sizes drive the off-frame driver; edges and flip_perm index the run's keypoints; the other arguments are as for suggest_points.

Raises:

Type	Description
ValueError	If the results have no 2D stage, or store no views and rig is missing or names other views.

Source code in src/deeperfly/labels/suggest.py

def suggest_results(
    res: Results,
    labels: Labels | None = None,
    *,
    rig: Rig | None = None,
    edges: Int[np.ndarray, "E 2"] | None = None,
    flip_perm: Int[np.ndarray, " K"] | None = None,
    count: int = 100,
    gap: int = 0,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
    gate_px: float = GATE_PX,
) -> Ranked:
    """The suggestions of one project, from its run's `pose2d` (and `pose3d`).

    Only the frames pose2d covers are scored (ADR-0082). The residuals are against
    `pose3d/points` on the frames pose3d covers, else against the views' consensus,
    and several tracked animals are ranked one at a time and merged, each label
    instance from the slot its clicks sit on
    ([run_slots][deeperfly.labels.identity.run_slots], ADR-0128). The views are
    the run's own when it stores them, else `rig`'s, taken by name in the run's view
    order: a pose2d-only run of one animal stores none (ADR-0086). The file's frame
    sizes drive the off-frame driver; `edges` and `flip_perm` index the run's
    keypoints; the other arguments are as for `suggest_points`.

    Raises:
        ValueError: If the results have no 2D stage, or store no views and `rig` is
            missing or names other views.
    """
    if not res.has("pose2d/points"):
        raise ValueError("suggestions need a [pose2d] stage in the results")
    if res.has("views"):
        rig = res.read_rig()
    elif rig is None:
        raise ValueError(f"{res.path} stores no calibration; pass the rig to score it")
    elif sorted(rig.names) != sorted(res.views):
        raise ValueError(
            f"{res.path} was run on views {', '.join(res.views)}, not the rig's "
            + ", ".join(rig.names)
        )
    else:
        rig = Rig({name: rig.views[name] for name in res.views})
    sizes = [res.frame_sizes.get(v) for v in res.views]
    # A frame a stage does not cover may hold stale values (`Results.uncover`).
    frames = np.flatnonzero(res.covered("pose2d"))
    points = np.asarray(res.read_frames("pose2d/points", frames), dtype=float)
    conf = np.asarray(res.read_frames("pose2d/conf", frames), dtype=float)
    pts3d = None
    if res.has("pose3d/points"):
        pts3d = np.asarray(res.read_frames("pose3d/points", frames), dtype=float)
        pts3d[:, ~res.covered("pose3d")[frames]] = np.nan
    # Animal `n` is label instance `n`, from the slot its clicks sit on (ADR-0128).
    slots = run_slots(
        labels, frames, points, conf, view_names=res.views, keypoint_names=res.keypoints
    )
    take_slots(points, slots, 2, np.nan)
    take_slots(conf, slots, 2, 0.0)
    if pts3d is not None:
        take_slots(pts3d, slots, 1, np.nan)
    parts = [
        suggest_points(
            rig,
            points[n],
            conf[n],
            keypoints=res.keypoints,
            labels=labels,
            sizes=sizes,
            frames=frames,
            pts3d=None if pts3d is None else pts3d[n],
            edges=edges,
            flip_perm=flip_perm,
            count=count,
            gap=gap,
            min_score=min_score,
            per_frame=per_frame,
            gate_px=gate_px,
            instance=n,
        )
        for n in range(res.animals)
    ]
    return _merge_ranked(parts, count)

suggest_sample

suggest_sample(
    n_frames: int,
    labels: Labels | None = None,
    *,
    count: int = 100,
) -> list[Suggestion]

count whole frames spread over the unlabeled part of the recording.

Farthest-first from every labeled frame, so however few get labeled they cover the recording; score is the distance to the nearest covered frame when picked, as a fraction of the recording.

Source code in src/deeperfly/labels/suggest.py

def suggest_sample(
    n_frames: int, labels: Labels | None = None, *, count: int = 100
) -> list[Suggestion]:
    """`count` whole frames spread over the unlabeled part of the recording.

    Farthest-first from every labeled frame, so however few get labeled they cover
    the recording; `score` is the distance to the nearest covered frame when picked, as
    a fraction of the recording.
    """
    covered = np.zeros(n_frames, dtype=bool)
    if labels is not None:
        for t in labels.labeled_frames():
            if t < n_frames:
                covered[t] = True
    frames = np.arange(n_frames)
    out = []
    if covered.any():
        anchors = frames[covered]
        dist = np.abs(frames[:, None] - anchors[None, :]).min(1).astype(float)
    elif n_frames:
        # Nothing labeled yet: the middle frame first, then farthest from it.
        t = n_frames // 2
        out.append(Suggestion(frame=t, keypoint=None, score=1.0, views=()))
        dist = np.abs(frames - t).astype(float)
        covered[t] = True
    else:
        return out
    dist[covered] = -np.inf
    while len(out) < count and np.isfinite(dist).any():
        t = int(np.argmax(dist))  # the first of equals: the earliest frame
        score = round(float(dist[t]) / n_frames, 4)
        out.append(Suggestion(frame=t, keypoint=None, score=score, views=()))
        dist = np.minimum(dist, np.abs(frames - t).astype(float))
        dist[t] = -np.inf
    return out

to_document

to_document(
    project: str,
    suggestions: list[Suggestion],
    *,
    source: dict[str, Any],
    params: dict[str, Any],
    stats: dict[str, Any] | None = None,
) -> dict[str, Any]

The suggestions.json document: provenance, parameters, stats and the picks.

Source code in src/deeperfly/labels/suggest.py

def to_document(
    project: str,
    suggestions: list[Suggestion],
    *,
    source: dict[str, Any],
    params: dict[str, Any],
    stats: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """The `suggestions.json` document: provenance, parameters, stats and the picks."""
    return {
        "format_version": FORMAT_VERSION,
        "project": project,
        "created": datetime.now(UTC).isoformat(timespec="seconds"),
        "source": source,
        "params": params,
        "stats": stats or {},
        "suggestions": [_pick_document(s) for s in suggestions],
    }

write_run_suggestions

write_run_suggestions(
    results_path: str | Path,
    labels: Labels,
    skeleton,
    *,
    project: str,
    target: str | Path,
    count: int = 100,
    gap: int = 20,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
) -> Ranked

Score a run's results against a project's labels; write its suggestions.json.

suggest_results on results_path, written to target with its source and parameters.

Raises:

Type	Description
ValueError	If the results hold no pose2d.

Source code in src/deeperfly/labels/suggest.py

def write_run_suggestions(
    results_path: str | Path,
    labels: Labels,
    skeleton,
    *,
    project: str,
    target: str | Path,
    count: int = 100,
    gap: int = 20,
    min_score: float = MIN_SCORE,
    per_frame: int = PER_FRAME,
) -> Ranked:
    """Score a run's results against a project's labels; write its `suggestions.json`.

    [suggest_results][deeperfly.labels.suggest.suggest_results] on `results_path`,
    written to `target` with its source and parameters.

    Raises:
        ValueError: If the results hold no `pose2d`.
    """
    with Results.open(results_path) as res:
        if not res.has("pose2d/points"):
            raise ValueError(f"no pose2d in {results_path}; run a config with [pose2d]")
        ranked = suggest_results(
            res,
            labels,
            edges=skeleton.edges,
            flip_perm=skeleton.flip_perm(),
            count=count,
            gap=gap,
            min_score=min_score,
            per_frame=per_frame,
        )
        source = results_source(res)
    document = to_document(
        project,
        ranked.picks,
        source=source,
        params={
            "count": count,
            "gap": gap,
            "min_score": min_score,
            "per_frame": per_frame,
        },
        stats=ranked.stats,
    )
    write_suggestions(target, document)
    return ranked

write_suggestions

write_suggestions(
    path: str | Path, document: dict[str, Any]
) -> Path

Write a suggestions document as JSON, creating the directory.

Source code in src/deeperfly/labels/suggest.py

def write_suggestions(path: str | Path, document: dict[str, Any]) -> Path:
    """Write a suggestions document as JSON, creating the directory."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(document, indent=1) + "\n")
    return path