results

One run's outputs: the results file deeperfly run writes and a reader reads.

A run's file holds one group per stage, the arrays inventory.STAGE_ARRAYS declares under the names every reference uses, and nothing derived. The file describes the work it came from: the run config, the skeleton and the initial calibration verbatim, the geometry, the videos, the frame rate, the deeperfly that wrote it, per stage the fingerprint that says its arrays are valid (ADR-0035), and when it was last written.

A run covers a frame set (ADR-0082): every array with a T axis is dense over the absolute frame index, and each stage group's covered (T,) mask says which frames hold values; a frame never written holds the array's fill value (fill_value). An array without T is rewritten whole by every write, so the last one wins. The file is only appended to or refused, never overwritten for other video, and shapes are checked on write against the inventory. Every per-animal array leads with N (ADR-0052).

Results

Read/write access to one run's results file, as a context manager.

Attributes:

Name	Type	Description
path		The file on disk.
views	tuple[str, ...]	View names, in the V order every array uses.
keypoints	tuple[str, ...]	Keypoint names, in the K order.
n_frames		The recording's frame count, the T axis.
animals		How many animals the run tracked, the N axis.

Source code in src/deeperfly/results.py

class Results:
    """Read/write access to one run's results file, as a context manager.

    Attributes:
        path: The file on disk.
        views: View names, in the `V` order every array uses.
        keypoints: Keypoint names, in the `K` order.
        n_frames: The recording's frame count, the `T` axis.
        animals: How many animals the run tracked, the `N` axis.
    """

    #: The attributes that describe the run, refreshed when a rerun reopens the file.
    _DESCRIPTION = ("config", "skeleton", "videos", "fps", "frame_sizes", "calibration")

    def __init__(self, path: str | Path, file: h5py.File) -> None:
        """Wrap an already-open file; use `create` or `open` instead."""
        self.path = Path(path)
        self._f = file
        self.views: tuple[str, ...] = tuple(_attr_json(file.attrs, "views"))
        self.keypoints: tuple[str, ...] = tuple(_attr_json(file.attrs, "keypoints"))
        self.n_frames = _attr_int(file.attrs, "n_frames")
        self.animals = _attr_int(file.attrs, "animals")

    @classmethod
    def create(
        cls,
        path: str | Path,
        *,
        project: str,
        views: Sequence[str],
        keypoints: Sequence[str],
        n_frames: int,
        animals: int = 1,
        config_text: str = "",
        calibration_text: str | None = None,
        skeleton_text: str = "",
        videos: Mapping[str, Sequence[str | Path]] | None = None,
        fps: float | None = None,
        frame_sizes: Mapping[str, tuple[int, int]] | None = None,
        force: bool = False,
    ) -> Results:
        """Start a results file, or reopen one that already describes this recording.

        A file for the same views, keypoints, frame count and format is reopened for
        append, which is what lets a run grow and keep its matching stages; one
        describing other work is refused unless `force`. `config_text`,
        `skeleton_text` and `calibration_text` are stored verbatim, the skeleton
        being the only place a reader finds the edges and the calibration a
        [Calibration][deeperfly.calibration.Calibration] document of the parameters
        the run started from (ADR-0086; `None` stores none, as a pose2d-only run
        does). `videos` are each video's files and `frame_sizes` each view's frame.

        Raises:
            ValueError: If a file is there, describes other work, and not `force`;
                or if `calibration_text` is not a calibration document.
        """
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        if calibration_text is not None:
            parse_calibration(calibration_text)
        described = cls._description(
            config_text, skeleton_text, videos, fps, frame_sizes, calibration_text
        )
        if path.exists():
            mismatch = cls._mismatch(path, views, keypoints, n_frames, animals)
            if mismatch is None:
                file = _h5(path, "a")
                cls._describe(file, described)
                return cls(path, file)
            if not force:
                raise ValueError(
                    f"{path} {mismatch}; rerun with --force to replace it, or "
                    "name the run otherwise"
                )
        file = _h5(path, "w")
        file.attrs["format_version"] = RESULTS_FORMAT_VERSION
        file.attrs["deeperfly_version"] = __version__
        file.attrs["project"] = project
        file.attrs["created"] = _now()
        file.attrs["views"] = json.dumps(list(views))
        file.attrs["keypoints"] = json.dumps(list(keypoints))
        file.attrs["n_frames"] = int(n_frames)
        file.attrs["animals"] = int(animals)
        cls._describe(file, described)
        return cls(path, file)

    @staticmethod
    def _description(
        config_text: str,
        skeleton_text: str,
        videos: Mapping[str, Sequence[str | Path]] | None,
        fps: float | None,
        frame_sizes: Mapping[str, tuple[int, int]] | None = None,
        calibration_text: str | None = None,
    ) -> dict[str, Any]:
        """The run-describing attributes, as HDF5 takes them."""
        described: dict[str, Any] = {
            "config": config_text,
            "skeleton": skeleton_text,
            "videos": json.dumps(
                {v: [str(p) for p in paths] for v, paths in (videos or {}).items()}
            ),
        }
        if calibration_text is not None:
            described["calibration"] = calibration_text
        if fps is not None:
            described["fps"] = float(fps)
        if frame_sizes:
            described["frame_sizes"] = json.dumps(
                {v: [int(h), int(w)] for v, (h, w) in frame_sizes.items()}
            )
        return described

    @classmethod
    def _describe(cls, file: h5py.File, described: Mapping[str, Any]) -> None:
        """Stamp the run-describing attributes, dropping any this run cannot say."""
        for key in cls._DESCRIPTION:
            if key in described:
                file.attrs[key] = described[key]
            elif key in file.attrs:
                del file.attrs[key]

    @classmethod
    def open(
        cls,
        path: str | Path,
        mode: str = "r",
        *,
        locking: bool | None = None,
        rdcc_nbytes: int | None = None,
    ) -> Results:
        """Open an existing results file (read-only by default).

        `locking=None` opens a reader without HDF5's file lock and a writer with it;
        `rdcc_nbytes` is the chunk cache, `READ_CACHE_BYTES` by default.

        Raises:
            ValueError: If the file is not a results file, or another format
                version, so its arrays would be read under the wrong layout.
        """
        file = _h5(path, mode, locking=locking, rdcc_nbytes=rdcc_nbytes)
        if "format_version" not in file.attrs:
            file.close()
            raise ValueError(f"{path} {_FOREIGN}")
        version = _attr_int(file.attrs, "format_version")
        if version != RESULTS_FORMAT_VERSION:
            file.close()
            raise ValueError(
                f"{path} is results format {version}, expected "
                f"{RESULTS_FORMAT_VERSION}{_stale(version)}; run it again"
            )
        return cls(path, file)

    @staticmethod
    def _mismatch(
        path: Path,
        views: Sequence[str],
        keypoints: Sequence[str],
        n_frames: int,
        animals: int,
    ) -> str | None:
        """How the file at `path` differs from this recording, or `None`.

        Phrased as what the stored file is, since which side is wrong is the user's
        call.
        """
        try:
            with _h5(path, "r") as f:
                if "format_version" not in f.attrs:
                    return _FOREIGN
                version = _attr_int(f.attrs, "format_version")
                if version != RESULTS_FORMAT_VERSION:
                    return (
                        f"is format version {version}, and this deeperfly writes "
                        f"version {RESULTS_FORMAT_VERSION}{_stale(version)}"
                    )
                stored_views = tuple(_attr_json(f.attrs, "views"))
                if stored_views != tuple(views):
                    return "was written for views " + ", ".join(stored_views)
                if tuple(_attr_json(f.attrs, "keypoints")) != tuple(keypoints):
                    return "was written for another skeleton"
                stored_frames = _attr_int(f.attrs, "n_frames")
                if stored_frames != int(n_frames):
                    return f"was written for {stored_frames} frames, not {n_frames}"
                stored_animals = _attr_int(f.attrs, "animals")
                if stored_animals != int(animals):
                    return f"was written for {stored_animals} animals, not {animals}"
        except (OSError, KeyError, TypeError, ValueError) as exc:
            return f"is not a readable results file ({exc})"
        return None

    def __enter__(self) -> Self:
        """Enter the context; the file is already open."""
        return self

    def __exit__(self, *exc: object) -> None:
        """Close the file."""
        self.close()

    def close(self) -> None:
        """Close the underlying file."""
        self._f.close()

    @property
    def project(self) -> str:
        """The project this run is of."""
        return str(self._f.attrs["project"])

    @property
    def config_text(self) -> str:
        """The run config that produced it, verbatim."""
        return str(self._f.attrs["config"])

    def same_config(self, text: str) -> bool:
        """Whether `text` is the stored run config, apart from `name` and `frames`.

        Compared as parsed TOML, so layout and comments do not count; the name and
        the frame set are what may differ between invocations of one run (ADR-0082).
        Text that does not parse is compared verbatim.
        """
        try:
            stored, given = tomllib.loads(self.config_text), tomllib.loads(text)
        except tomllib.TOMLDecodeError:
            return self.config_text == text
        for doc in (stored, given):
            doc.pop("name", None)
            doc.pop("frames", None)
        return stored == given

    @property
    def written(self) -> int:
        """The unix time of the last write to any stage, or 0 if none was written."""
        return int(self._f.attrs.get("written", 0))

    def stamp_written(self, when: float | None = None) -> None:
        """Record that the run was written at `when` (unix time, default now)."""
        self._f.attrs["written"] = int(time.time() if when is None else when)
        self._f.flush()

    def initial_calibration(self) -> Calibration | None:
        """The camera parameters the run started from, or `None` if it stores none."""
        text = self._f.attrs.get("calibration")
        return None if not text else parse_calibration(str(text))

    def calibration(self) -> Calibration | None:
        """The run's calibration: the one bundle adjustment solved, else the initial.

        A solved one records, as `refined_from`, the project calibration its starting
        parameters were copied from, when they name one (ADR-0086): the copy the
        solve keeps ([write_frames][deeperfly.results.Results.write_frames]), not the
        file's, which a later invocation may have copied from another.
        """
        solved = "bundle_adjustment/views"
        if solved not in self._f:
            return self.initial_calibration()
        text = self._f["bundle_adjustment"].attrs.get("calibration")
        start = None if not text else parse_calibration(str(text))
        return Calibration.of_rig(
            "bundle-adjustment",
            self.read_rig(solved),
            solved=True,
            refined_from=None if start is None else start.copied_from,
            fingerprint=self.fingerprint("bundle_adjustment"),
            frame_hw=self.frame_sizes,
        )

    @property
    def skeleton_text(self) -> str:
        """The skeleton document the run used, verbatim."""
        return str(self._f.attrs["skeleton"])

    def skeleton(self) -> Skeleton:
        """The stored skeleton, parsed: the topology behind `keypoints`.

        Raises:
            ValueError: If the file stores no skeleton, or the text no longer parses.
        """
        text = self.skeleton_text
        if not text:
            raise ValueError(f"{self.path}: stores no skeleton")
        return Skeleton.from_toml_text(text)

    @property
    def videos(self) -> dict[str, list[str]]:
        """Per video, the files the run read, relative to the project folder.

        Relative to that folder also for a `--out` run, whose results lie outside it.
        """
        return dict(_attr_json(self._f.attrs, "videos"))

    @property
    def frame_sizes(self) -> dict[str, tuple[int, int]]:
        """Per view, the `(height, width)` of the video frames the run read."""
        attrs = self._f.attrs
        stored = _attr_json(attrs, "frame_sizes") if "frame_sizes" in attrs else {}
        return {v: (int(h), int(w)) for v, (h, w) in stored.items()}

    @property
    def fps(self) -> float | None:
        """The recording's frame rate, or `None` when the video did not say."""
        value = self._f.attrs.get("fps")
        return None if value is None else float(value)

    @property
    def deeperfly_version(self) -> str:
        """The deeperfly that wrote the file."""
        return str(self._f.attrs["deeperfly_version"])

    def stages(self) -> list[str]:
        """The stage groups present, in the file's order."""
        return [k for k in self._f if k in STAGES]

    def fingerprint(self, stage: str) -> str | None:
        """The fingerprint recorded for `stage`, or `None` if it never ran."""
        group = _member(self._f, stage, h5py.Group)
        if group is None:
            return None
        value = group.attrs.get("fingerprint")
        return None if value is None else str(value)

    def has(self, ref: str) -> bool:
        """Whether the array named by a `"stage/array"` reference is stored."""
        return ref in self._f

    def read(self, ref: str) -> np.ndarray:
        """The array a `"stage/array"` reference names (`KeyError` if none)."""
        return np.asarray(self._dataset(ref)[()])

    def _dataset(self, ref: str) -> h5py.Dataset:
        """The stored dataset of a reference, or a `KeyError` naming what is stored."""
        dset = _member(self._f, ref, h5py.Dataset)
        if dset is None:
            raise KeyError(f"{self.path}: no array {ref!r} (have {self._refs()})")
        return dset

    def read_frames(self, ref: str, frames: Sequence[int] | np.ndarray) -> np.ndarray:
        """An array at the given frames only, in the order given, `T` kept.

        Read one stretch at a time, so only the chunks holding those frames are
        decompressed; an array without `T` comes back whole.

        Raises:
            KeyError: If the file holds no such array.
            IndexError: If a frame is outside `[0, n_frames)`.
        """
        dset = self._dataset(ref)
        axes = self.axes(ref)
        if "T" not in axes:
            return np.asarray(dset[()])
        t = axes.index("T")
        wanted = np.asarray(frames, dtype=np.int64).ravel()
        self._check_frames(wanted)
        unique, inverse = np.unique(wanted, return_inverse=True)
        lead = (slice(None),) * t
        parts = [np.asarray(dset[(*lead, slice(a, b))]) for a, b in stretches(unique)]
        if not parts:
            return np.asarray(dset[(*lead, slice(0, 0))])
        return np.take(np.concatenate(parts, axis=t), inverse, axis=t)

    def read_frame(self, ref: str, t: int, instance: int | None = None) -> np.ndarray:
        """An array at one frame, its `T` axis dropped; `instance` also drops `N`.

        An array without `N` ignores `instance`, and one without `T` comes back
        whole.

        Raises:
            KeyError: If the file holds no such array.
            IndexError: If the frame, or the animal, is not in the file.
        """
        dset = self._dataset(ref)
        axes = self.axes(ref)
        index: list[Any] = [slice(None)] * len(axes)
        if "T" in axes:
            self._check_frames(np.asarray([t]))
            index[axes.index("T")] = int(t)
        if instance is not None and axes[:1] == ("N",):
            if not 0 <= instance < self.animals:
                raise IndexError(
                    f"{ref}: animal {instance}, and this file has {self.animals}"
                )
            index[0] = int(instance)
        return np.asarray(dset[tuple(index)])

    def _check_frames(self, frames: np.ndarray) -> None:
        """Refuse a frame index outside the file's `T` axis."""
        if frames.size and (frames.min() < 0 or frames.max() >= self.n_frames):
            bad = frames[(frames < 0) | (frames >= self.n_frames)][0]
            raise IndexError(
                f"{self.path}: frame {int(bad)}, and the file has {self.n_frames}"
            )

    def covered(self, stage: str) -> np.ndarray:
        """`(T,)` bool: the frames `stage` has written; all false if it never ran."""
        group = _member(self._f, stage, h5py.Group)
        covered = None if group is None else _member(group, COVERED, h5py.Dataset)
        if covered is None:
            return np.zeros(self.n_frames, dtype=bool)
        return np.asarray(covered[()], dtype=bool)

    def uncover(self, stage: str, frames: Sequence[int] | np.ndarray) -> None:
        """Mark frames of `stage` as not written, so a later run recomputes them.

        The arrays keep their values; a stage that never ran is left alone.
        """
        group = _member(self._f, stage, h5py.Group)
        covered = None if group is None else _member(group, COVERED, h5py.Dataset)
        if covered is None:
            return
        wanted = np.asarray(frames, dtype=np.int64).ravel()
        self._check_frames(wanted)
        for start, stop in stretches(wanted):
            covered[start:stop] = False
        self._f.flush()

    def names(self, ref: str) -> tuple[str, ...] | None:
        """The `attrs.names` labelling an array's named axis, if it has one."""
        attrs = self._f[ref].attrs
        return tuple(_attr_json(attrs, "names")) if "names" in attrs else None

    def axes(self, ref: str) -> tuple:
        """The declared axes of a `"stage/array"` reference, e.g. `("T", "K", 3)`."""
        stage, _, name = ref.partition("/")
        declared = STAGE_ARRAYS.get(stage, {})
        if name not in declared:
            raise KeyError(
                f"{ref!r} is not a declared array (see inventory.STAGE_ARRAYS)"
            )
        return declared[name]

    def read_animal(self, ref: str, n: int) -> np.ndarray:
        """One animal's copy of an array, with the leading `N` axis peeled off.

        `read_animal("pose3d/points", 0)` is `(T, K, 3)` however many animals were
        tracked; an array about the recording has no `N` and comes back whole.

        Raises:
            IndexError: If the file holds no animal `n`.
            KeyError: If the file holds no such array.
        """
        arr = self.read(ref)
        if self.axes(ref)[:1] != ("N",):
            return arr
        if not 0 <= n < self.animals:
            raise IndexError(f"{ref}: animal {n}, and this file has {self.animals}")
        return arr[n]

    def select(
        self,
        ref: str,
        *,
        points: Sequence[str] = (),
        views: Sequence[str] = (),
    ) -> np.ndarray:
        """Read an array and keep only the named keypoints and views, in stored order.

        `points` and `views` are the run config's selectors, applied along the
        reference's `K` and `V` axes; an array without the axis ignores its selector.

        Raises:
            KeyError: If the file holds no such array.
            ConfigError: If a selector matches nothing.
        """
        arr = self.read(ref)
        axes = self.axes(ref)
        for axis, patterns, pool in (
            ("K", points, self.keypoints),
            ("V", views, self.views),
        ):
            if axis in axes and patterns:
                chosen = match_names(patterns, list(pool), f"select {axis}")
                idx = [list(pool).index(n) for n in chosen]
                arr = np.take(arr, idx, axis=axes.index(axis))
        return arr

    def _refs(self) -> str:
        """Every stored reference, for an error message."""
        found: list[str] = []
        self._f.visititems(
            lambda name, obj: (
                found.append(name) if isinstance(obj, h5py.Dataset) else None
            )
        )
        return ", ".join(sorted(found))

    def write_stage(
        self,
        stage: str,
        arrays: Mapping[str, np.ndarray],
        *,
        fingerprint: str,
        names: Mapping[str, Sequence[str]] | None = None,
        rig: Rig | None = None,
    ) -> None:
        """Replace `stage`'s group with `arrays` over every frame, fingerprinted.

        Replaced, not updated, so an array a stage stopped producing does not linger:
        a fresh [begin_stage][deeperfly.results.Results.begin_stage] and one
        [write_frames][deeperfly.results.Results.write_frames] over the whole `T`
        axis, which marks every frame covered.

        Raises:
            KeyError: If `stage` or an array name is not declared.
            ValueError: If an array's shape disagrees with the declaration.
        """
        self._check_arrays(stage, arrays, self.n_frames)
        if stage in self._f:
            del self._f[stage]
        self.begin_stage(stage, fingerprint=fingerprint)
        self.write_frames(stage, arrays, np.arange(self.n_frames), names=names, rig=rig)

    def begin_stage(self, stage: str, *, fingerprint: str) -> bool:
        """Make `stage`'s group ready for frames written under `fingerprint`.

        A group of the same fingerprint is kept with its coverage; one of another is
        dropped, since none of its frames are valid now, and a fresh one covering
        nothing takes its place. Returns whether a stored group was dropped.

        Raises:
            KeyError: If `stage` is not a stage.
        """
        if stage not in STAGE_ARRAYS:
            raise KeyError(f"{stage!r} is not a stage; " + ", ".join(STAGES))
        group = _member(self._f, stage, h5py.Group)
        dropped = False
        if group is not None:
            if group.attrs.get("fingerprint") == fingerprint:
                return False
            del self._f[stage]
            dropped = True
        group = self._f.create_group(stage)
        group.attrs["fingerprint"] = fingerprint
        group.attrs["created"] = _now()
        group.attrs["deeperfly_version"] = __version__
        group.create_dataset(COVERED, data=np.zeros(self.n_frames, dtype=bool))
        self._f.flush()
        return dropped

    def write_frames(
        self,
        stage: str,
        arrays: Mapping[str, np.ndarray],
        frames: Sequence[int] | np.ndarray,
        *,
        names: Mapping[str, Sequence[str]] | None = None,
        rig: Rig | None = None,
    ) -> None:
        """Write a stage's arrays at `frames`, mark them covered, stamp `written`.

        `frames` are increasing absolute indices, and every array with a `T` axis
        holds exactly those frames along it; the file's copy is dense over `T`, and
        a frame never written keeps the array's fill value. An array without `T` is
        rewritten whole. `names` labels an array's named axis (`ik/angles`) and
        `rig` stores the geometry beside the arrays (`bundle_adjustment/views`), the
        group keeping a copy of the file's `calibration`, the one it derives from.

        Raises:
            KeyError: If the stage was not begun, or an array name is not declared.
            ValueError: If an array's shape disagrees with the declaration or with
                what the file already holds, or `frames` are not increasing.
            IndexError: If a frame is outside `[0, n_frames)`.
        """
        group = _member(self._f, stage, h5py.Group)
        if group is None:
            raise KeyError(f"{self.path}: begin stage {stage!r} before writing to it")
        wanted = np.asarray(frames, dtype=np.int64).ravel()
        if wanted.size > 1 and np.any(np.diff(wanted) <= 0):
            raise ValueError(f"{stage}: frames must be increasing")
        self._check_frames(wanted)
        self._check_arrays(stage, arrays, wanted.size)
        declared = STAGE_ARRAYS[stage]
        spans = stretches(wanted)
        for name, arr in arrays.items():
            data = np.asarray(arr)
            axes = declared[name]
            if "T" in axes:
                dset = self._frames_dataset(group, f"{stage}/{name}", data, axes)
                t = axes.index("T")
                lead = (slice(None),) * t
                at = 0
                for start, stop in spans:
                    span = stop - start
                    part = data[(*lead, slice(at, at + span))]
                    dset[(*lead, slice(start, stop))] = part
                    at += span
            else:
                if name in group:
                    del group[name]
                dset = group.create_dataset(
                    name, data=data, compression="gzip" if data.size > 4096 else None
                )
            labels = (names or {}).get(name)
            if labels is not None:
                dset.attrs["names"] = json.dumps(list(labels))
        if rig is not None:
            self._write_rig(group, rig, self.frame_sizes)
            # The file's `calibration` is rewritten by every invocation, so the rig
            # keeps its own copy of the one it derives from (ADR-0086).
            started = self._f.attrs.get("calibration")
            if started:
                group.attrs["calibration"] = str(started)
            elif "calibration" in group.attrs:
                del group.attrs["calibration"]
        covered = _member(group, COVERED, h5py.Dataset)
        if covered is None:
            raise KeyError(f"{self.path}: stage {stage!r} has no {COVERED!r} mask")
        for start, stop in spans:
            covered[start:stop] = True
        self.stamp_written()

    def _frames_dataset(
        self, group: h5py.Group, ref: str, data: np.ndarray, axes: tuple
    ) -> h5py.Dataset:
        """The stored `T`-bearing dataset `data` is written into, made if absent.

        Made dense over the whole `T` axis at `data`'s dtype, chunked along `T` only
        (`_t_chunks`), gzip 4 with shuffle, and filled with the array's fill value.
        """
        name = ref.partition("/")[2]
        t = axes.index("T")
        shape = tuple(self.n_frames if i == t else n for i, n in enumerate(data.shape))
        dset = _member(group, name, h5py.Dataset)
        if dset is not None:
            if dset.shape != shape:
                raise ValueError(
                    f"{ref}: the file holds {dset.shape}, and this write is "
                    f"{data.shape} over {data.shape[t]} frames"
                )
            return dset
        chunks = _t_chunks(shape, t, data.dtype.itemsize)
        packing: dict[str, Any] = {}
        if chunks is not None:
            packing = {
                "chunks": chunks,
                "compression": "gzip",
                "compression_opts": 4,
                "shuffle": True,
            }
        return group.create_dataset(
            name,
            shape=shape,
            dtype=data.dtype,
            fillvalue=fill_value(ref, data.dtype.kind),
            **packing,
        )

    def _check_arrays(
        self, stage: str, arrays: Mapping[str, np.ndarray], n_t: int
    ) -> None:
        """Check a write's array names and shapes, its `T` axis `n_t` frames long."""
        declared = STAGE_ARRAYS[stage]
        for name, arr in arrays.items():
            if name not in declared:
                raise KeyError(
                    f"{stage}: {name!r} is not one of " + ", ".join(sorted(declared))
                )
            self._check_shape(f"{stage}/{name}", np.shape(arr), declared[name], n_t)

    def stamp_duration(self, stage: str, seconds: float) -> None:
        """Record how long `stage` took, beside its fingerprint."""
        self._f[stage].attrs["duration_s"] = round(float(seconds), 3)
        self._f.flush()

    def provenance(self, stage: str) -> dict[str, Any]:
        """What a stage's group records: its fingerprint, when, how long, by whom.

        Raises:
            KeyError: If the stage has written nothing.
        """
        group = self._f[stage]
        keys = ("fingerprint", "created", "duration_s", "deeperfly_version")
        return {k: group.attrs[k] for k in keys if k in group.attrs}

    def replace(self, ref: str, array: np.ndarray) -> None:
        """Overwrite a stored array in place with one of the same shape.

        A repair ([swap_animals][deeperfly.results.swap_animals]), not a stage's write:
        the fingerprint and provenance stand, so a rerun still skips the stage.

        Raises:
            KeyError: If the file holds no such array.
            ValueError: If `array` has another shape.
        """
        dset = self._dataset(ref)
        if tuple(array.shape) != tuple(dset.shape):
            raise ValueError(
                f"{ref}: replacing {tuple(dset.shape)} with {tuple(array.shape)}"
            )
        dset[...] = array
        self._f.flush()

    def drop_stage(self, stage: str) -> None:
        """Delete a stage's group, so the next run recomputes it."""
        if stage not in self._f:
            raise KeyError(f"{self.path}: no stage {stage!r}")
        del self._f[stage]
        self._f.flush()

    def write_rig(self, rig: Rig) -> None:
        """Store the geometry the run used, at the file's top level."""
        if "views" in self._f:
            del self._f["views"]
        self._write_rig(self._f, rig, self.frame_sizes)
        self._f.flush()

    def read_rig(self, where: str = "views") -> Rig:
        """Read back a stored rig (`"views"`, or `"bundle_adjustment/views"`)."""
        group = self._f[where]
        tables = _attr_json(group.attrs, "views")["views"]
        return Rig.from_specs({n: RawView.model_validate(t) for n, t in tables.items()})

    @staticmethod
    def _write_rig(
        parent: Any, rig: Rig, frame_hw: Mapping[str, tuple[int, int]]
    ) -> None:
        """Write a rig as one JSON attribute on a `views` subgroup of `parent`.

        Each view records `frame_hw`, the frame the run read it against (ADR-0079).
        """
        group = (
            parent.create_group("views") if "views" not in parent else (parent["views"])
        )
        specs = {
            n: s.model_dump(exclude_defaults=True)
            for n, s in rig.specs(frame_hw).items()
        }
        group.attrs["views"] = json.dumps({"views": specs})
        group.attrs["names"] = json.dumps(rig.names)

    def _check_shape(
        self, ref: str, shape: tuple[int, ...], axes: tuple, n_t: int | None = None
    ) -> None:
        """Check one array against its declared axes, binding `V`, `T`, `K` and `N`.

        `T` is `n_t` frames, the file's whole axis by default.
        """
        n_t = self.n_frames if n_t is None else n_t
        sizes = (len(self.views), n_t, len(self.keypoints), self.animals)
        bound = dict(zip(_BOUND, sizes, strict=True))
        if len(shape) != len(axes):
            raise ValueError(
                f"{ref}: expected {len(axes)} axes {axes}, got shape {shape}"
            )
        for size, axis in zip(shape, axes, strict=True):
            want = axis if isinstance(axis, int) else bound.get(axis)
            if want is not None and size != want:
                raise ValueError(
                    f"{ref}: axis {axis} is {want} here, but the array is {shape}"
                )

config_text property

config_text: str

The run config that produced it, verbatim.

deeperfly_version property

deeperfly_version: str

The deeperfly that wrote the file.

fps property

fps: float | None

The recording's frame rate, or None when the video did not say.

frame_sizes property

frame_sizes: dict[str, tuple[int, int]]

Per view, the (height, width) of the video frames the run read.

project property

project: str

The project this run is of.

skeleton_text property

skeleton_text: str

The skeleton document the run used, verbatim.

videos property

videos: dict[str, list[str]]

Per video, the files the run read, relative to the project folder.

Relative to that folder also for a --out run, whose results lie outside it.

written property

written: int

The unix time of the last write to any stage, or 0 if none was written.

__enter__

__enter__() -> Self

Enter the context; the file is already open.

Source code in src/deeperfly/results.py

def __enter__(self) -> Self:
    """Enter the context; the file is already open."""
    return self

__exit__

__exit__(*exc: object) -> None

Close the file.

Source code in src/deeperfly/results.py

def __exit__(self, *exc: object) -> None:
    """Close the file."""
    self.close()

__init__

__init__(path: str | Path, file: File) -> None

Wrap an already-open file; use create or open instead.

Source code in src/deeperfly/results.py

def __init__(self, path: str | Path, file: h5py.File) -> None:
    """Wrap an already-open file; use `create` or `open` instead."""
    self.path = Path(path)
    self._f = file
    self.views: tuple[str, ...] = tuple(_attr_json(file.attrs, "views"))
    self.keypoints: tuple[str, ...] = tuple(_attr_json(file.attrs, "keypoints"))
    self.n_frames = _attr_int(file.attrs, "n_frames")
    self.animals = _attr_int(file.attrs, "animals")

axes

axes(ref: str) -> tuple

The declared axes of a "stage/array" reference, e.g. ("T", "K", 3).

Source code in src/deeperfly/results.py

def axes(self, ref: str) -> tuple:
    """The declared axes of a `"stage/array"` reference, e.g. `("T", "K", 3)`."""
    stage, _, name = ref.partition("/")
    declared = STAGE_ARRAYS.get(stage, {})
    if name not in declared:
        raise KeyError(
            f"{ref!r} is not a declared array (see inventory.STAGE_ARRAYS)"
        )
    return declared[name]

begin_stage

begin_stage(stage: str, *, fingerprint: str) -> bool

Make stage's group ready for frames written under fingerprint.

A group of the same fingerprint is kept with its coverage; one of another is dropped, since none of its frames are valid now, and a fresh one covering nothing takes its place. Returns whether a stored group was dropped.

Raises:

Type	Description
KeyError	If stage is not a stage.

Source code in src/deeperfly/results.py

def begin_stage(self, stage: str, *, fingerprint: str) -> bool:
    """Make `stage`'s group ready for frames written under `fingerprint`.

    A group of the same fingerprint is kept with its coverage; one of another is
    dropped, since none of its frames are valid now, and a fresh one covering
    nothing takes its place. Returns whether a stored group was dropped.

    Raises:
        KeyError: If `stage` is not a stage.
    """
    if stage not in STAGE_ARRAYS:
        raise KeyError(f"{stage!r} is not a stage; " + ", ".join(STAGES))
    group = _member(self._f, stage, h5py.Group)
    dropped = False
    if group is not None:
        if group.attrs.get("fingerprint") == fingerprint:
            return False
        del self._f[stage]
        dropped = True
    group = self._f.create_group(stage)
    group.attrs["fingerprint"] = fingerprint
    group.attrs["created"] = _now()
    group.attrs["deeperfly_version"] = __version__
    group.create_dataset(COVERED, data=np.zeros(self.n_frames, dtype=bool))
    self._f.flush()
    return dropped

calibration

calibration() -> Calibration | None

The run's calibration: the one bundle adjustment solved, else the initial.

A solved one records, as refined_from, the project calibration its starting parameters were copied from, when they name one (ADR-0086): the copy the solve keeps (write_frames), not the file's, which a later invocation may have copied from another.

Source code in src/deeperfly/results.py

def calibration(self) -> Calibration | None:
    """The run's calibration: the one bundle adjustment solved, else the initial.

    A solved one records, as `refined_from`, the project calibration its starting
    parameters were copied from, when they name one (ADR-0086): the copy the
    solve keeps ([write_frames][deeperfly.results.Results.write_frames]), not the
    file's, which a later invocation may have copied from another.
    """
    solved = "bundle_adjustment/views"
    if solved not in self._f:
        return self.initial_calibration()
    text = self._f["bundle_adjustment"].attrs.get("calibration")
    start = None if not text else parse_calibration(str(text))
    return Calibration.of_rig(
        "bundle-adjustment",
        self.read_rig(solved),
        solved=True,
        refined_from=None if start is None else start.copied_from,
        fingerprint=self.fingerprint("bundle_adjustment"),
        frame_hw=self.frame_sizes,
    )

close

close() -> None

Close the underlying file.

Source code in src/deeperfly/results.py

def close(self) -> None:
    """Close the underlying file."""
    self._f.close()

covered

covered(stage: str) -> np.ndarray

(T,) bool: the frames stage has written; all false if it never ran.

Source code in src/deeperfly/results.py

def covered(self, stage: str) -> np.ndarray:
    """`(T,)` bool: the frames `stage` has written; all false if it never ran."""
    group = _member(self._f, stage, h5py.Group)
    covered = None if group is None else _member(group, COVERED, h5py.Dataset)
    if covered is None:
        return np.zeros(self.n_frames, dtype=bool)
    return np.asarray(covered[()], dtype=bool)

create classmethod

create(
    path: str | Path,
    *,
    project: str,
    views: Sequence[str],
    keypoints: Sequence[str],
    n_frames: int,
    animals: int = 1,
    config_text: str = "",
    calibration_text: str | None = None,
    skeleton_text: str = "",
    videos: Mapping[str, Sequence[str | Path]]
    | None = None,
    fps: float | None = None,
    frame_sizes: Mapping[str, tuple[int, int]]
    | None = None,
    force: bool = False,
) -> Results

Start a results file, or reopen one that already describes this recording.

A file for the same views, keypoints, frame count and format is reopened for append, which is what lets a run grow and keep its matching stages; one describing other work is refused unless force. config_text, skeleton_text and calibration_text are stored verbatim, the skeleton being the only place a reader finds the edges and the calibration a Calibration document of the parameters the run started from (ADR-0086; None stores none, as a pose2d-only run does). videos are each video's files and frame_sizes each view's frame.

Raises:

Type	Description
ValueError	If a file is there, describes other work, and not force; or if calibration_text is not a calibration document.

Source code in src/deeperfly/results.py

@classmethod
def create(
    cls,
    path: str | Path,
    *,
    project: str,
    views: Sequence[str],
    keypoints: Sequence[str],
    n_frames: int,
    animals: int = 1,
    config_text: str = "",
    calibration_text: str | None = None,
    skeleton_text: str = "",
    videos: Mapping[str, Sequence[str | Path]] | None = None,
    fps: float | None = None,
    frame_sizes: Mapping[str, tuple[int, int]] | None = None,
    force: bool = False,
) -> Results:
    """Start a results file, or reopen one that already describes this recording.

    A file for the same views, keypoints, frame count and format is reopened for
    append, which is what lets a run grow and keep its matching stages; one
    describing other work is refused unless `force`. `config_text`,
    `skeleton_text` and `calibration_text` are stored verbatim, the skeleton
    being the only place a reader finds the edges and the calibration a
    [Calibration][deeperfly.calibration.Calibration] document of the parameters
    the run started from (ADR-0086; `None` stores none, as a pose2d-only run
    does). `videos` are each video's files and `frame_sizes` each view's frame.

    Raises:
        ValueError: If a file is there, describes other work, and not `force`;
            or if `calibration_text` is not a calibration document.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    if calibration_text is not None:
        parse_calibration(calibration_text)
    described = cls._description(
        config_text, skeleton_text, videos, fps, frame_sizes, calibration_text
    )
    if path.exists():
        mismatch = cls._mismatch(path, views, keypoints, n_frames, animals)
        if mismatch is None:
            file = _h5(path, "a")
            cls._describe(file, described)
            return cls(path, file)
        if not force:
            raise ValueError(
                f"{path} {mismatch}; rerun with --force to replace it, or "
                "name the run otherwise"
            )
    file = _h5(path, "w")
    file.attrs["format_version"] = RESULTS_FORMAT_VERSION
    file.attrs["deeperfly_version"] = __version__
    file.attrs["project"] = project
    file.attrs["created"] = _now()
    file.attrs["views"] = json.dumps(list(views))
    file.attrs["keypoints"] = json.dumps(list(keypoints))
    file.attrs["n_frames"] = int(n_frames)
    file.attrs["animals"] = int(animals)
    cls._describe(file, described)
    return cls(path, file)

drop_stage

drop_stage(stage: str) -> None

Delete a stage's group, so the next run recomputes it.

Source code in src/deeperfly/results.py

def drop_stage(self, stage: str) -> None:
    """Delete a stage's group, so the next run recomputes it."""
    if stage not in self._f:
        raise KeyError(f"{self.path}: no stage {stage!r}")
    del self._f[stage]
    self._f.flush()

fingerprint

fingerprint(stage: str) -> str | None

The fingerprint recorded for stage, or None if it never ran.

Source code in src/deeperfly/results.py

def fingerprint(self, stage: str) -> str | None:
    """The fingerprint recorded for `stage`, or `None` if it never ran."""
    group = _member(self._f, stage, h5py.Group)
    if group is None:
        return None
    value = group.attrs.get("fingerprint")
    return None if value is None else str(value)

has

has(ref: str) -> bool

Whether the array named by a "stage/array" reference is stored.

Source code in src/deeperfly/results.py

def has(self, ref: str) -> bool:
    """Whether the array named by a `"stage/array"` reference is stored."""
    return ref in self._f

initial_calibration

initial_calibration() -> Calibration | None

The camera parameters the run started from, or None if it stores none.

Source code in src/deeperfly/results.py

def initial_calibration(self) -> Calibration | None:
    """The camera parameters the run started from, or `None` if it stores none."""
    text = self._f.attrs.get("calibration")
    return None if not text else parse_calibration(str(text))

names

names(ref: str) -> tuple[str, ...] | None

The attrs.names labelling an array's named axis, if it has one.

Source code in src/deeperfly/results.py

def names(self, ref: str) -> tuple[str, ...] | None:
    """The `attrs.names` labelling an array's named axis, if it has one."""
    attrs = self._f[ref].attrs
    return tuple(_attr_json(attrs, "names")) if "names" in attrs else None

open classmethod

open(
    path: str | Path,
    mode: str = "r",
    *,
    locking: bool | None = None,
    rdcc_nbytes: int | None = None,
) -> Results

Open an existing results file (read-only by default).

locking=None opens a reader without HDF5's file lock and a writer with it; rdcc_nbytes is the chunk cache, READ_CACHE_BYTES by default.

Raises:

Type	Description
ValueError	If the file is not a results file, or another format version, so its arrays would be read under the wrong layout.

Source code in src/deeperfly/results.py

@classmethod
def open(
    cls,
    path: str | Path,
    mode: str = "r",
    *,
    locking: bool | None = None,
    rdcc_nbytes: int | None = None,
) -> Results:
    """Open an existing results file (read-only by default).

    `locking=None` opens a reader without HDF5's file lock and a writer with it;
    `rdcc_nbytes` is the chunk cache, `READ_CACHE_BYTES` by default.

    Raises:
        ValueError: If the file is not a results file, or another format
            version, so its arrays would be read under the wrong layout.
    """
    file = _h5(path, mode, locking=locking, rdcc_nbytes=rdcc_nbytes)
    if "format_version" not in file.attrs:
        file.close()
        raise ValueError(f"{path} {_FOREIGN}")
    version = _attr_int(file.attrs, "format_version")
    if version != RESULTS_FORMAT_VERSION:
        file.close()
        raise ValueError(
            f"{path} is results format {version}, expected "
            f"{RESULTS_FORMAT_VERSION}{_stale(version)}; run it again"
        )
    return cls(path, file)

provenance

provenance(stage: str) -> dict[str, Any]

What a stage's group records: its fingerprint, when, how long, by whom.

Raises:

Type	Description
KeyError	If the stage has written nothing.

Source code in src/deeperfly/results.py

def provenance(self, stage: str) -> dict[str, Any]:
    """What a stage's group records: its fingerprint, when, how long, by whom.

    Raises:
        KeyError: If the stage has written nothing.
    """
    group = self._f[stage]
    keys = ("fingerprint", "created", "duration_s", "deeperfly_version")
    return {k: group.attrs[k] for k in keys if k in group.attrs}

read

read(ref: str) -> np.ndarray

The array a "stage/array" reference names (KeyError if none).

Source code in src/deeperfly/results.py

def read(self, ref: str) -> np.ndarray:
    """The array a `"stage/array"` reference names (`KeyError` if none)."""
    return np.asarray(self._dataset(ref)[()])

read_animal

read_animal(ref: str, n: int) -> np.ndarray

One animal's copy of an array, with the leading N axis peeled off.

read_animal("pose3d/points", 0) is (T, K, 3) however many animals were tracked; an array about the recording has no N and comes back whole.

Raises:

Type	Description
IndexError	If the file holds no animal n.
KeyError	If the file holds no such array.

Source code in src/deeperfly/results.py

def read_animal(self, ref: str, n: int) -> np.ndarray:
    """One animal's copy of an array, with the leading `N` axis peeled off.

    `read_animal("pose3d/points", 0)` is `(T, K, 3)` however many animals were
    tracked; an array about the recording has no `N` and comes back whole.

    Raises:
        IndexError: If the file holds no animal `n`.
        KeyError: If the file holds no such array.
    """
    arr = self.read(ref)
    if self.axes(ref)[:1] != ("N",):
        return arr
    if not 0 <= n < self.animals:
        raise IndexError(f"{ref}: animal {n}, and this file has {self.animals}")
    return arr[n]

read_frame

read_frame(
    ref: str, t: int, instance: int | None = None
) -> np.ndarray

An array at one frame, its T axis dropped; instance also drops N.

An array without N ignores instance, and one without T comes back whole.

Raises:

Type	Description
KeyError	If the file holds no such array.
IndexError	If the frame, or the animal, is not in the file.

Source code in src/deeperfly/results.py

def read_frame(self, ref: str, t: int, instance: int | None = None) -> np.ndarray:
    """An array at one frame, its `T` axis dropped; `instance` also drops `N`.

    An array without `N` ignores `instance`, and one without `T` comes back
    whole.

    Raises:
        KeyError: If the file holds no such array.
        IndexError: If the frame, or the animal, is not in the file.
    """
    dset = self._dataset(ref)
    axes = self.axes(ref)
    index: list[Any] = [slice(None)] * len(axes)
    if "T" in axes:
        self._check_frames(np.asarray([t]))
        index[axes.index("T")] = int(t)
    if instance is not None and axes[:1] == ("N",):
        if not 0 <= instance < self.animals:
            raise IndexError(
                f"{ref}: animal {instance}, and this file has {self.animals}"
            )
        index[0] = int(instance)
    return np.asarray(dset[tuple(index)])

read_frames

read_frames(
    ref: str, frames: Sequence[int] | ndarray
) -> np.ndarray

An array at the given frames only, in the order given, T kept.

Read one stretch at a time, so only the chunks holding those frames are decompressed; an array without T comes back whole.

Raises:

Type	Description
KeyError	If the file holds no such array.
IndexError	If a frame is outside [0, n_frames).

Source code in src/deeperfly/results.py

def read_frames(self, ref: str, frames: Sequence[int] | np.ndarray) -> np.ndarray:
    """An array at the given frames only, in the order given, `T` kept.

    Read one stretch at a time, so only the chunks holding those frames are
    decompressed; an array without `T` comes back whole.

    Raises:
        KeyError: If the file holds no such array.
        IndexError: If a frame is outside `[0, n_frames)`.
    """
    dset = self._dataset(ref)
    axes = self.axes(ref)
    if "T" not in axes:
        return np.asarray(dset[()])
    t = axes.index("T")
    wanted = np.asarray(frames, dtype=np.int64).ravel()
    self._check_frames(wanted)
    unique, inverse = np.unique(wanted, return_inverse=True)
    lead = (slice(None),) * t
    parts = [np.asarray(dset[(*lead, slice(a, b))]) for a, b in stretches(unique)]
    if not parts:
        return np.asarray(dset[(*lead, slice(0, 0))])
    return np.take(np.concatenate(parts, axis=t), inverse, axis=t)

read_rig

read_rig(where: str = 'views') -> Rig

Read back a stored rig ("views", or "bundle_adjustment/views").

Source code in src/deeperfly/results.py

def read_rig(self, where: str = "views") -> Rig:
    """Read back a stored rig (`"views"`, or `"bundle_adjustment/views"`)."""
    group = self._f[where]
    tables = _attr_json(group.attrs, "views")["views"]
    return Rig.from_specs({n: RawView.model_validate(t) for n, t in tables.items()})

replace

replace(ref: str, array: ndarray) -> None

Overwrite a stored array in place with one of the same shape.

A repair (swap_animals), not a stage's write: the fingerprint and provenance stand, so a rerun still skips the stage.

Raises:

Type	Description
KeyError	If the file holds no such array.
ValueError	If array has another shape.

Source code in src/deeperfly/results.py

def replace(self, ref: str, array: np.ndarray) -> None:
    """Overwrite a stored array in place with one of the same shape.

    A repair ([swap_animals][deeperfly.results.swap_animals]), not a stage's write:
    the fingerprint and provenance stand, so a rerun still skips the stage.

    Raises:
        KeyError: If the file holds no such array.
        ValueError: If `array` has another shape.
    """
    dset = self._dataset(ref)
    if tuple(array.shape) != tuple(dset.shape):
        raise ValueError(
            f"{ref}: replacing {tuple(dset.shape)} with {tuple(array.shape)}"
        )
    dset[...] = array
    self._f.flush()

same_config

same_config(text: str) -> bool

Whether text is the stored run config, apart from name and frames.

Compared as parsed TOML, so layout and comments do not count; the name and the frame set are what may differ between invocations of one run (ADR-0082). Text that does not parse is compared verbatim.

Source code in src/deeperfly/results.py

def same_config(self, text: str) -> bool:
    """Whether `text` is the stored run config, apart from `name` and `frames`.

    Compared as parsed TOML, so layout and comments do not count; the name and
    the frame set are what may differ between invocations of one run (ADR-0082).
    Text that does not parse is compared verbatim.
    """
    try:
        stored, given = tomllib.loads(self.config_text), tomllib.loads(text)
    except tomllib.TOMLDecodeError:
        return self.config_text == text
    for doc in (stored, given):
        doc.pop("name", None)
        doc.pop("frames", None)
    return stored == given

select

select(
    ref: str,
    *,
    points: Sequence[str] = (),
    views: Sequence[str] = (),
) -> np.ndarray

Read an array and keep only the named keypoints and views, in stored order.

points and views are the run config's selectors, applied along the reference's K and V axes; an array without the axis ignores its selector.

Raises:

Type	Description
KeyError	If the file holds no such array.
ConfigError	If a selector matches nothing.

Source code in src/deeperfly/results.py

def select(
    self,
    ref: str,
    *,
    points: Sequence[str] = (),
    views: Sequence[str] = (),
) -> np.ndarray:
    """Read an array and keep only the named keypoints and views, in stored order.

    `points` and `views` are the run config's selectors, applied along the
    reference's `K` and `V` axes; an array without the axis ignores its selector.

    Raises:
        KeyError: If the file holds no such array.
        ConfigError: If a selector matches nothing.
    """
    arr = self.read(ref)
    axes = self.axes(ref)
    for axis, patterns, pool in (
        ("K", points, self.keypoints),
        ("V", views, self.views),
    ):
        if axis in axes and patterns:
            chosen = match_names(patterns, list(pool), f"select {axis}")
            idx = [list(pool).index(n) for n in chosen]
            arr = np.take(arr, idx, axis=axes.index(axis))
    return arr

skeleton

skeleton() -> Skeleton

The stored skeleton, parsed: the topology behind keypoints.

Raises:

Type	Description
ValueError	If the file stores no skeleton, or the text no longer parses.

Source code in src/deeperfly/results.py

def skeleton(self) -> Skeleton:
    """The stored skeleton, parsed: the topology behind `keypoints`.

    Raises:
        ValueError: If the file stores no skeleton, or the text no longer parses.
    """
    text = self.skeleton_text
    if not text:
        raise ValueError(f"{self.path}: stores no skeleton")
    return Skeleton.from_toml_text(text)

stages

stages() -> list[str]

The stage groups present, in the file's order.

Source code in src/deeperfly/results.py

def stages(self) -> list[str]:
    """The stage groups present, in the file's order."""
    return [k for k in self._f if k in STAGES]

stamp_duration

stamp_duration(stage: str, seconds: float) -> None

Record how long stage took, beside its fingerprint.

Source code in src/deeperfly/results.py

def stamp_duration(self, stage: str, seconds: float) -> None:
    """Record how long `stage` took, beside its fingerprint."""
    self._f[stage].attrs["duration_s"] = round(float(seconds), 3)
    self._f.flush()

stamp_written

stamp_written(when: float | None = None) -> None

Record that the run was written at when (unix time, default now).

Source code in src/deeperfly/results.py

def stamp_written(self, when: float | None = None) -> None:
    """Record that the run was written at `when` (unix time, default now)."""
    self._f.attrs["written"] = int(time.time() if when is None else when)
    self._f.flush()

uncover

uncover(
    stage: str, frames: Sequence[int] | ndarray
) -> None

Mark frames of stage as not written, so a later run recomputes them.

The arrays keep their values; a stage that never ran is left alone.

Source code in src/deeperfly/results.py

def uncover(self, stage: str, frames: Sequence[int] | np.ndarray) -> None:
    """Mark frames of `stage` as not written, so a later run recomputes them.

    The arrays keep their values; a stage that never ran is left alone.
    """
    group = _member(self._f, stage, h5py.Group)
    covered = None if group is None else _member(group, COVERED, h5py.Dataset)
    if covered is None:
        return
    wanted = np.asarray(frames, dtype=np.int64).ravel()
    self._check_frames(wanted)
    for start, stop in stretches(wanted):
        covered[start:stop] = False
    self._f.flush()

write_frames

write_frames(
    stage: str,
    arrays: Mapping[str, ndarray],
    frames: Sequence[int] | ndarray,
    *,
    names: Mapping[str, Sequence[str]] | None = None,
    rig: Rig | None = None,
) -> None

Write a stage's arrays at frames, mark them covered, stamp written.

frames are increasing absolute indices, and every array with a T axis holds exactly those frames along it; the file's copy is dense over T, and a frame never written keeps the array's fill value. An array without T is rewritten whole. names labels an array's named axis (ik/angles) and rig stores the geometry beside the arrays (bundle_adjustment/views), the group keeping a copy of the file's calibration, the one it derives from.

Raises:

Type	Description
KeyError	If the stage was not begun, or an array name is not declared.
ValueError	If an array's shape disagrees with the declaration or with what the file already holds, or frames are not increasing.
IndexError	If a frame is outside [0, n_frames).

Source code in src/deeperfly/results.py

def write_frames(
    self,
    stage: str,
    arrays: Mapping[str, np.ndarray],
    frames: Sequence[int] | np.ndarray,
    *,
    names: Mapping[str, Sequence[str]] | None = None,
    rig: Rig | None = None,
) -> None:
    """Write a stage's arrays at `frames`, mark them covered, stamp `written`.

    `frames` are increasing absolute indices, and every array with a `T` axis
    holds exactly those frames along it; the file's copy is dense over `T`, and
    a frame never written keeps the array's fill value. An array without `T` is
    rewritten whole. `names` labels an array's named axis (`ik/angles`) and
    `rig` stores the geometry beside the arrays (`bundle_adjustment/views`), the
    group keeping a copy of the file's `calibration`, the one it derives from.

    Raises:
        KeyError: If the stage was not begun, or an array name is not declared.
        ValueError: If an array's shape disagrees with the declaration or with
            what the file already holds, or `frames` are not increasing.
        IndexError: If a frame is outside `[0, n_frames)`.
    """
    group = _member(self._f, stage, h5py.Group)
    if group is None:
        raise KeyError(f"{self.path}: begin stage {stage!r} before writing to it")
    wanted = np.asarray(frames, dtype=np.int64).ravel()
    if wanted.size > 1 and np.any(np.diff(wanted) <= 0):
        raise ValueError(f"{stage}: frames must be increasing")
    self._check_frames(wanted)
    self._check_arrays(stage, arrays, wanted.size)
    declared = STAGE_ARRAYS[stage]
    spans = stretches(wanted)
    for name, arr in arrays.items():
        data = np.asarray(arr)
        axes = declared[name]
        if "T" in axes:
            dset = self._frames_dataset(group, f"{stage}/{name}", data, axes)
            t = axes.index("T")
            lead = (slice(None),) * t
            at = 0
            for start, stop in spans:
                span = stop - start
                part = data[(*lead, slice(at, at + span))]
                dset[(*lead, slice(start, stop))] = part
                at += span
        else:
            if name in group:
                del group[name]
            dset = group.create_dataset(
                name, data=data, compression="gzip" if data.size > 4096 else None
            )
        labels = (names or {}).get(name)
        if labels is not None:
            dset.attrs["names"] = json.dumps(list(labels))
    if rig is not None:
        self._write_rig(group, rig, self.frame_sizes)
        # The file's `calibration` is rewritten by every invocation, so the rig
        # keeps its own copy of the one it derives from (ADR-0086).
        started = self._f.attrs.get("calibration")
        if started:
            group.attrs["calibration"] = str(started)
        elif "calibration" in group.attrs:
            del group.attrs["calibration"]
    covered = _member(group, COVERED, h5py.Dataset)
    if covered is None:
        raise KeyError(f"{self.path}: stage {stage!r} has no {COVERED!r} mask")
    for start, stop in spans:
        covered[start:stop] = True
    self.stamp_written()

write_rig

write_rig(rig: Rig) -> None

Store the geometry the run used, at the file's top level.

Source code in src/deeperfly/results.py

def write_rig(self, rig: Rig) -> None:
    """Store the geometry the run used, at the file's top level."""
    if "views" in self._f:
        del self._f["views"]
    self._write_rig(self._f, rig, self.frame_sizes)
    self._f.flush()

write_stage

write_stage(
    stage: str,
    arrays: Mapping[str, ndarray],
    *,
    fingerprint: str,
    names: Mapping[str, Sequence[str]] | None = None,
    rig: Rig | None = None,
) -> None

Replace stage's group with arrays over every frame, fingerprinted.

Replaced, not updated, so an array a stage stopped producing does not linger: a fresh begin_stage and one write_frames over the whole T axis, which marks every frame covered.

Raises:

Type	Description
KeyError	If stage or an array name is not declared.
ValueError	If an array's shape disagrees with the declaration.

Source code in src/deeperfly/results.py

def write_stage(
    self,
    stage: str,
    arrays: Mapping[str, np.ndarray],
    *,
    fingerprint: str,
    names: Mapping[str, Sequence[str]] | None = None,
    rig: Rig | None = None,
) -> None:
    """Replace `stage`'s group with `arrays` over every frame, fingerprinted.

    Replaced, not updated, so an array a stage stopped producing does not linger:
    a fresh [begin_stage][deeperfly.results.Results.begin_stage] and one
    [write_frames][deeperfly.results.Results.write_frames] over the whole `T`
    axis, which marks every frame covered.

    Raises:
        KeyError: If `stage` or an array name is not declared.
        ValueError: If an array's shape disagrees with the declaration.
    """
    self._check_arrays(stage, arrays, self.n_frames)
    if stage in self._f:
        del self._f[stage]
    self.begin_stage(stage, fingerprint=fingerprint)
    self.write_frames(stage, arrays, np.arange(self.n_frames), names=names, rig=rig)

stretches

stretches(
    frames: Sequence[int] | ndarray,
) -> list[tuple[int, int]]

The maximal contiguous ranges of a frame set, as (start, stop) slices.

frames is unioned and sorted first, so [5, 0, 1, 2, 7, 6] is [(0, 3), (5, 8)], and a stepped range is one stretch per frame.

Source code in src/deeperfly/results.py

def stretches(frames: Sequence[int] | np.ndarray) -> list[tuple[int, int]]:
    """The maximal contiguous ranges of a frame set, as `(start, stop)` slices.

    `frames` is unioned and sorted first, so `[5, 0, 1, 2, 7, 6]` is
    `[(0, 3), (5, 8)]`, and a stepped range is one stretch per frame.
    """
    f = np.unique(np.asarray(frames, dtype=np.int64).ravel())
    if f.size == 0:
        return []
    breaks = np.flatnonzero(np.diff(f) != 1) + 1
    starts = np.concatenate([[0], breaks])
    stops = np.concatenate([breaks, [f.size]])
    return [(int(f[a]), int(f[b - 1]) + 1) for a, b in zip(starts, stops, strict=True)]

swap_animals

swap_animals(
    res: Results,
    a: int,
    b: int,
    start: int = 0,
    stop: int | None = None,
) -> list[str]

Exchange two animals of a run's results over [start, stop), and drop the rest.

Every pose2d array with an animal axis is exchanged, and every downstream group deleted, since its fingerprint still matches the config. The run's written is stamped, so the swapped run wins precedence where it covers a frame (ADR-0083). stop=None runs to the end. Returns the stage groups deleted, in pipeline order.

Raises:

Type	Description
ValueError	If the file holds one animal, a slot is not one of its animals, the two slots are the same, or the range is empty.

Source code in src/deeperfly/results.py

def swap_animals(
    res: Results, a: int, b: int, start: int = 0, stop: int | None = None
) -> list[str]:
    """Exchange two animals of a run's results over `[start, stop)`, and drop the rest.

    Every `pose2d` array with an animal axis is exchanged, and every downstream group
    deleted, since its fingerprint still matches the config. The run's `written` is
    stamped, so the swapped run wins precedence where it covers a frame (ADR-0083).
    `stop=None` runs to the end. Returns the stage groups deleted, in pipeline order.

    Raises:
        ValueError: If the file holds one animal, a slot is not one of its animals,
            the two slots are the same, or the range is empty.
    """
    if res.animals == 1:
        raise ValueError(f"{res.path}: holds one animal, so there is no swap")
    for slot in (a, b):
        if not 0 <= slot < res.animals:
            raise ValueError(
                f"animal {slot}: this file has {res.animals}, so 0 to {res.animals - 1}"
            )
    if a == b:
        raise ValueError(f"animal {a} is already itself")
    frames = slice(max(0, start), res.n_frames if stop is None else stop)
    if frames.start >= frames.stop:
        raise ValueError(f"frames [{frames.start}, {frames.stop}) is empty")
    for name, axes in STAGE_ARRAYS["pose2d"].items():
        ref = f"pose2d/{name}"
        if axes[0] != "N" or not res.has(ref):
            continue
        arr = res.read(ref)
        here, there = _animal_index(axes, a, frames), _animal_index(axes, b, frames)
        arr[here], arr[there] = arr[there].copy(), arr[here].copy()
        res.replace(ref, arr)
    dropped = [s for s in STAGES if s != "pose2d" and s in res.stages()]
    for stage in dropped:
        res.drop_stage(stage)
    res.stamp_written()
    return dropped