Skip to content

pipeline

Orchestration: a run over projects, covering a frame set, reproducible.

It owns which projects a run covers, where its files live, which frames each stage already covers and what to do when one project fails, and does no math: run is the driver, stages one function per stage table, plan what a run means before it runs.

Context dataclass

What every stage is handed: the run config, the recording, and where to write.

Attributes:

Name Type Description
cfg RunConfig

The typed run config.
recording Recording

The project's recording: its videos and frame count.
skeleton Skeleton

The project's skeleton, the K axis.
results Results

Its open results file.
rig Rig | None

The geometry in force, the V axis: the run's initial calibration on the project's views, replaced by the [bundle_adjustment] stage for the stages after it; None for a run that reads no camera parameters (a pose2d-only run of one animal).
facts Facts

What the fingerprints hash besides the tables.
layout Layout

Which views each video carries (the project's).
labels Callable[[], Labels] | None

Loads the project's hand labels, raising a StageError that says why it cannot; None when none can be had.
detector DetectorCache | None

The shared detector handle, or None when [pose2d] is off.
log Callable[[str], None]

Where progress lines go.
todo Mapping[str, ndarray]

Per stage, the frames it computes in this invocation (the coverage plan, coverage_plan); a stage absent here computes every frame.
renders Path | None

The folder [render] writes into; None is beside the results.
force bool

Whether the invocation recomputes what it could reuse, such as the static crop an earlier invocation found.
observer RunObserver | None

Who watches the run, handed each detector pass; None for no one.
Source code in src/deeperfly/pipeline/stages.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
@dataclass
class Context:
    """What every stage is handed: the run config, the recording, and where to write.

    Attributes:
        cfg: The typed run config.
        recording: The project's recording: its videos and frame count.
        skeleton: The project's skeleton, the `K` axis.
        results: Its open results file.
        rig: The geometry in force, the `V` axis: the run's initial calibration on
            the project's views, replaced by the `[bundle_adjustment]` stage for the
            stages after it; `None` for a run that reads no camera parameters (a
            pose2d-only run of one animal).
        facts: What the fingerprints hash besides the tables.
        layout: Which views each video carries (the project's).
        labels: Loads the project's hand labels, raising a `StageError` that says
            why it cannot; `None` when none can be had.
        detector: The shared detector handle, or `None` when `[pose2d]` is off.
        log: Where progress lines go.
        todo: Per stage, the frames it computes in this invocation (the coverage
            plan, [coverage_plan][deeperfly.pipeline.plan.coverage_plan]); a stage
            absent here computes every frame.
        renders: The folder `[render]` writes into; `None` is beside the results.
        force: Whether the invocation recomputes what it could reuse, such as the
            static crop an earlier invocation found.
        observer: Who watches the run, handed each detector pass; `None` for no one.
    """

    cfg: RunConfig
    recording: Recording
    skeleton: Skeleton
    results: Results
    rig: Rig | None
    facts: Facts
    layout: Layout = field(default_factory=Layout)
    labels: Callable[[], Labels] | None = None
    detector: DetectorCache | None = None
    log: Callable[[str], None] = print
    todo: Mapping[str, np.ndarray] = field(default_factory=dict)
    renders: Path | None = None
    force: bool = False
    observer: RunObserver | None = None

    def frames(self, stage: str) -> np.ndarray:
        """The frames `stage` computes here, increasing absolute indices."""
        if stage in self.todo:
            return np.asarray(self.todo[stage], dtype=np.int64)
        return np.arange(self.results.n_frames)

    @property
    def animals(self) -> int:
        """How many animals the project holds, the `N` axis."""
        return self.results.animals

    def need_rig(self) -> Rig:
        """The geometry in force, for a stage that reads it.

        Raises:
            StageError: If the run reads no camera parameters, which
                [prepare][deeperfly.pipeline.plan.prepare] rules out for every
                stage that needs them.
        """
        if self.rig is None:
            raise StageError(
                f"{self.recording.name}: this stage reads camera parameters, and the "
                "run starts from none; add [calibration] to the config"
            )
        return self.rig

    @property
    def keypoints(self) -> tuple[str, ...]:
        """The skeleton's point names, the `K` axis order."""
        return tuple(self.skeleton.point_names)

    def frame_size(self, view: str) -> tuple[int, int]:
        """The `(height, width)` a view's pixels live in: its video's frame."""
        return self.recording.videos[self.layout.video_of(view)].size

    def read(self, ref: str, frames: np.ndarray | None = None) -> np.ndarray:
        """Resolve a data reference: a stored stage array, or the hand labels.

        With `frames`, only those frames, which the stage that wrote the array must
        cover; without, the whole array.

        Raises:
            StageError: If no stage has written the array (one left out with `--skip`),
                or it does not cover `frames`.
        """
        if ref.startswith("labels/"):
            arr = self._labels(ref)
            return arr if frames is None else np.take(arr, frames, axis=_LABEL_T[ref])
        stage = ref.split("/")[0]
        if not self.results.has(ref):
            raise StageError(
                f"{self.recording.name}: {ref!r} is not in {self.results.path.name}; "
                f"run the [{stage}] stage (it was skipped, or its results were "
                "removed)"
            )
        if frames is None:
            return self.results.read(ref)
        covered = self.results.covered(stage)[frames]
        if not covered.all():
            raise StageError(
                f"{self.recording.name}: [{stage}] covers {int(covered.sum())} of the "
                f"{covered.size} frames asked for; run [{stage}] on them too"
            )
        return self.results.read_frames(ref, frames)

    def read_animal(
        self, ref: str, n: int, frames: np.ndarray | None = None
    ) -> np.ndarray:
        """`read`, with the animal axis peeled off: one animal's copy of an array.

        An array about the recording rather than an animal, and the hand labels, come
        back whole.
        """
        arr = self.read(ref, frames)
        if ref.startswith("labels/"):
            return arr
        return arr[n] if self.results.axes(ref)[:1] == ("N",) else arr

    def _labels(self, ref: str) -> np.ndarray:
        """The hand labels as the array `ref` names, in the run's view order."""
        if self.labels is None:
            raise StageError(f"{self.recording.name}: no hand labels to read")
        labels = self.labels()
        frames = labels.labeled_frames()
        views = list(labels.view_names)
        missing = [v for v in self.results.views if v not in views]
        if missing:
            raise StageError(
                f"{self.recording.name}: the labels have no view named "
                + ", ".join(missing)
                + " (they know "
                + ", ".join(views)
                + ")"
            )
        order = [views.index(v) for v in self.results.views]
        n_v, n_k = len(order), labels.n_keypoints
        n_t = self.results.n_frames
        if ref == "labels/points":
            out = np.full((n_v, n_t, n_k, 2), np.nan)
            for t in frames:
                out[:, t] = labels.frame_effective(t)[order]
            return out
        if ref == "labels/hidden":
            out = np.zeros((n_v, n_t, n_k), dtype=bool)
            for t in frames:
                out[:, t] = labels.hidden_frame(t)[order]
            return out
        if ref == "labels/absent":
            out = np.zeros((n_t, n_k), dtype=bool)
            for t in frames:
                out[t] = labels.absent_frame(t)
            return out
        raise StageError(f"{ref!r} is not a label array")

    def points(self, selector: tuple[str, ...], where: str) -> list[int]:
        """The keypoint indices a point selector claims, in skeleton order."""
        names = self.keypoints
        chosen = match_names(selector, names, where)
        return [names.index(n) for n in chosen]

animals property 

animals: int

How many animals the project holds, the N axis.

keypoints property 

keypoints: tuple[str, ...]

The skeleton's point names, the K axis order.

frame_size 

frame_size(view: str) -> tuple[int, int]

The (height, width) a view's pixels live in: its video's frame.

Source code in src/deeperfly/pipeline/stages.py 
180
181
182
def frame_size(self, view: str) -> tuple[int, int]:
    """The `(height, width)` a view's pixels live in: its video's frame."""
    return self.recording.videos[self.layout.video_of(view)].size

frames 

frames(stage: str) -> np.ndarray

The frames stage computes here, increasing absolute indices.

Source code in src/deeperfly/pipeline/stages.py 
149
150
151
152
153
def frames(self, stage: str) -> np.ndarray:
    """The frames `stage` computes here, increasing absolute indices."""
    if stage in self.todo:
        return np.asarray(self.todo[stage], dtype=np.int64)
    return np.arange(self.results.n_frames)

need_rig 

need_rig() -> Rig

The geometry in force, for a stage that reads it.

Raises:

Type Description
StageError

If the run reads no camera parameters, which prepare rules out for every stage that needs them.
Source code in src/deeperfly/pipeline/stages.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def need_rig(self) -> Rig:
    """The geometry in force, for a stage that reads it.

    Raises:
        StageError: If the run reads no camera parameters, which
            [prepare][deeperfly.pipeline.plan.prepare] rules out for every
            stage that needs them.
    """
    if self.rig is None:
        raise StageError(
            f"{self.recording.name}: this stage reads camera parameters, and the "
            "run starts from none; add [calibration] to the config"
        )
    return self.rig

points 

points(selector: tuple[str, ...], where: str) -> list[int]

The keypoint indices a point selector claims, in skeleton order.

Source code in src/deeperfly/pipeline/stages.py 
263
264
265
266
267
def points(self, selector: tuple[str, ...], where: str) -> list[int]:
    """The keypoint indices a point selector claims, in skeleton order."""
    names = self.keypoints
    chosen = match_names(selector, names, where)
    return [names.index(n) for n in chosen]

read 

read(ref: str, frames: ndarray | None = None) -> np.ndarray

Resolve a data reference: a stored stage array, or the hand labels.

With frames, only those frames, which the stage that wrote the array must cover; without, the whole array.

Raises:

Type Description
StageError

If no stage has written the array (one left out with --skip), or it does not cover frames.
Source code in src/deeperfly/pipeline/stages.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def read(self, ref: str, frames: np.ndarray | None = None) -> np.ndarray:
    """Resolve a data reference: a stored stage array, or the hand labels.

    With `frames`, only those frames, which the stage that wrote the array must
    cover; without, the whole array.

    Raises:
        StageError: If no stage has written the array (one left out with `--skip`),
            or it does not cover `frames`.
    """
    if ref.startswith("labels/"):
        arr = self._labels(ref)
        return arr if frames is None else np.take(arr, frames, axis=_LABEL_T[ref])
    stage = ref.split("/")[0]
    if not self.results.has(ref):
        raise StageError(
            f"{self.recording.name}: {ref!r} is not in {self.results.path.name}; "
            f"run the [{stage}] stage (it was skipped, or its results were "
            "removed)"
        )
    if frames is None:
        return self.results.read(ref)
    covered = self.results.covered(stage)[frames]
    if not covered.all():
        raise StageError(
            f"{self.recording.name}: [{stage}] covers {int(covered.sum())} of the "
            f"{covered.size} frames asked for; run [{stage}] on them too"
        )
    return self.results.read_frames(ref, frames)

read_animal 

read_animal(
    ref: str, n: int, frames: ndarray | None = None
) -> np.ndarray

read, with the animal axis peeled off: one animal's copy of an array.

An array about the recording rather than an animal, and the hand labels, come back whole.

Source code in src/deeperfly/pipeline/stages.py 
214
215
216
217
218
219
220
221
222
223
224
225
def read_animal(
    self, ref: str, n: int, frames: np.ndarray | None = None
) -> np.ndarray:
    """`read`, with the animal axis peeled off: one animal's copy of an array.

    An array about the recording rather than an animal, and the hand labels, come
    back whole.
    """
    arr = self.read(ref, frames)
    if ref.startswith("labels/"):
        return arr
    return arr[n] if self.results.axes(ref)[:1] == ("N",) else arr

DetectorCache dataclass

The run's nets, loaded once, shared by every project; torch on first use.

Source code in src/deeperfly/pipeline/stages.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass
class DetectorCache:
    """The run's nets, loaded once, shared by every project; torch on first use."""

    weights: tuple[str, ...]
    locator: str | None = None
    device: str | None = None
    precision: str = "auto"
    _loaded: tuple | None = field(default=None, repr=False)
    _locator: tuple | None = field(default=None, repr=False)

    def get(self) -> tuple:
        """`(models, device)` for `predict_recording`, loading on first call."""
        if self._loaded is None:
            from ..pose2d.models import load_models

            self._loaded = load_models(
                list(self.weights), device=self.device, precision=self.precision
            )
        return self._loaded

    @property
    def loaded_device(self) -> str | None:
        """The device the nets were loaded on, or `None` before the first `get`."""
        return None if self._loaded is None else str(self._loaded[1])

    def get_locator(self):
        """The run's full-frame locator, loaded once; `None` when it names none."""
        if self.locator is None:
            return None
        if self._locator is None:
            from ..pose2d.models import load_locator

            self._locator = load_locator(
                self.locator, device=self.device, precision=self.precision
            )
        return self._locator[0]

loaded_device property 

loaded_device: str | None

The device the nets were loaded on, or None before the first get.

get 

get() -> tuple

(models, device) for predict_recording, loading on first call.

Source code in src/deeperfly/pipeline/stages.py 
78
79
80
81
82
83
84
85
86
def get(self) -> tuple:
    """`(models, device)` for `predict_recording`, loading on first call."""
    if self._loaded is None:
        from ..pose2d.models import load_models

        self._loaded = load_models(
            list(self.weights), device=self.device, precision=self.precision
        )
    return self._loaded

get_locator 

get_locator()

The run's full-frame locator, loaded once; None when it names none.

Source code in src/deeperfly/pipeline/stages.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_locator(self):
    """The run's full-frame locator, loaded once; `None` when it names none."""
    if self.locator is None:
        return None
    if self._locator is None:
        from ..pose2d.models import load_locator

        self._locator = load_locator(
            self.locator, device=self.device, precision=self.precision
        )
    return self._locator[0]

ProjectOutcome dataclass

What happened for one project.

Attributes:

Name Type Description
project str

Its name in the invocation (project_names).
results Path

Its run's results file.
ran tuple[str, ...]

The stages that computed frames.
cached tuple[str, ...]

The stages with nothing to compute: the frames asked for were covered, under a matching fingerprint.
error str | None

The message of the failure that stopped it, or None.
Source code in src/deeperfly/pipeline/run.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@dataclass(frozen=True)
class ProjectOutcome:
    """What happened for one project.

    Attributes:
        project: Its name in the invocation
            ([project_names][deeperfly.project.project_names]).
        results: Its run's results file.
        ran: The stages that computed frames.
        cached: The stages with nothing to compute: the frames asked for were
            covered, under a matching fingerprint.
        error: The message of the failure that stopped it, or `None`.
    """

    project: str
    results: Path
    ran: tuple[str, ...] = ()
    cached: tuple[str, ...] = ()
    error: str | None = None

RunObserver

Bases: Protocol

The calls a run makes, in order, to whoever watches it.

Source code in src/deeperfly/pipeline/observe.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class RunObserver(Protocol):
    """The calls a run makes, in order, to whoever watches it."""

    def run_started(
        self, run: str, labels: Sequence[str], stages: Sequence[str]
    ) -> None:
        """The run's name, its projects in the order they run, and its stages."""

    def project_started(self, label: str, ctx: "Context") -> None:
        """A project's recording and results are open, about to walk its stages."""

    def stage_cached(self, stage: str) -> None:
        """`stage` has nothing to compute here: an earlier invocation covered it."""

    def stage_started(self, stage: str, n_frames: int) -> None:
        """`stage` begins computing `n_frames` frames."""

    def pose2d_block(
        self,
        frames: np.ndarray,
        points: np.ndarray,
        conf: np.ndarray,
        boxes: np.ndarray,
        images: list[np.ndarray],
    ) -> None:
        """One detector pass, as `predict_recording`'s `on_block` hands it over."""

    def stage_finished(self, stage: str, elapsed: float, ctx: "Context") -> None:
        """`stage` wrote what it computed, in `elapsed` seconds."""

    def project_finished(self, label: str, error: str | None) -> None:
        """A project is done, or failed with `error`."""

pose2d_block 

pose2d_block(
    frames: ndarray,
    points: ndarray,
    conf: ndarray,
    boxes: ndarray,
    images: list[ndarray],
) -> None

One detector pass, as predict_recording's on_block hands it over.

Source code in src/deeperfly/pipeline/observe.py 
41
42
43
44
45
46
47
48
49
def pose2d_block(
    self,
    frames: np.ndarray,
    points: np.ndarray,
    conf: np.ndarray,
    boxes: np.ndarray,
    images: list[np.ndarray],
) -> None:
    """One detector pass, as `predict_recording`'s `on_block` hands it over."""

project_finished 

project_finished(label: str, error: str | None) -> None

A project is done, or failed with error.

Source code in src/deeperfly/pipeline/observe.py 
54
55
def project_finished(self, label: str, error: str | None) -> None:
    """A project is done, or failed with `error`."""

project_started 

project_started(label: str, ctx: Context) -> None

A project's recording and results are open, about to walk its stages.

Source code in src/deeperfly/pipeline/observe.py 
32
33
def project_started(self, label: str, ctx: "Context") -> None:
    """A project's recording and results are open, about to walk its stages."""

run_started 

run_started(
    run: str, labels: Sequence[str], stages: Sequence[str]
) -> None

The run's name, its projects in the order they run, and its stages.

Source code in src/deeperfly/pipeline/observe.py 
27
28
29
30
def run_started(
    self, run: str, labels: Sequence[str], stages: Sequence[str]
) -> None:
    """The run's name, its projects in the order they run, and its stages."""

stage_cached 

stage_cached(stage: str) -> None

stage has nothing to compute here: an earlier invocation covered it.

Source code in src/deeperfly/pipeline/observe.py 
35
36
def stage_cached(self, stage: str) -> None:
    """`stage` has nothing to compute here: an earlier invocation covered it."""

stage_finished 

stage_finished(
    stage: str, elapsed: float, ctx: Context
) -> None

stage wrote what it computed, in elapsed seconds.

Source code in src/deeperfly/pipeline/observe.py 
51
52
def stage_finished(self, stage: str, elapsed: float, ctx: "Context") -> None:
    """`stage` wrote what it computed, in `elapsed` seconds."""

stage_started 

stage_started(stage: str, n_frames: int) -> None

stage begins computing n_frames frames.

Source code in src/deeperfly/pipeline/observe.py 
38
39
def stage_started(self, stage: str, n_frames: int) -> None:
    """`stage` begins computing `n_frames` frames."""

RunStopped

Bases: KeyboardInterrupt

Someone watching asked the run to stop: it ends as Ctrl-C would end it.

Source code in src/deeperfly/pipeline/observe.py 
20
21
class RunStopped(KeyboardInterrupt):
    """Someone watching asked the run to stop: it ends as Ctrl-C would end it."""

RunSummary dataclass

The whole invocation's outcome.

Attributes:

Name Type Description
out Path | None

The --out root, or None when results went into the projects.
stages tuple[str, ...]

The stages selected for this run.
outcomes tuple[ProjectOutcome, ...]

One entry per project, in run order.
name str

The run's name, shared by every project.
Source code in src/deeperfly/pipeline/run.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass(frozen=True)
class RunSummary:
    """The whole invocation's outcome.

    Attributes:
        out: The `--out` root, or `None` when results went into the projects.
        stages: The stages selected for this run.
        outcomes: One entry per project, in run order.
        name: The run's name, shared by every project.
    """

    out: Path | None
    stages: tuple[str, ...]
    outcomes: tuple[ProjectOutcome, ...]
    name: str = ""

    @property
    def failed(self) -> tuple[ProjectOutcome, ...]:
        """The projects that stopped on an error."""
        return tuple(o for o in self.outcomes if o.error is not None)

failed property 

failed: tuple[ProjectOutcome, ...]

The projects that stopped on an error.

StageError

Bases: RuntimeError

A stage cannot run: an unbuilt path, or an input nothing has written.

Source code in src/deeperfly/pipeline/stages.py 
63
64
class StageError(RuntimeError):
    """A stage cannot run: an unbuilt path, or an input nothing has written."""

run_projects 

run_projects(
    cfg: RunConfig,
    paths: Sequence[str | Path],
    *,
    only: Sequence[str] = (),
    skip: Sequence[str] = (),
    out: Path | None = None,
    force: bool = False,
    name: str | None = None,
    frames: Sequence[int | str] | None = None,
    log: Callable[[str], None] = print,
    observer: RunObserver | None = None,
) -> RunSummary

Apply a run config to the projects some paths stand for, and say what happened.

Each path is resolved as every command resolves one, a directory of videos getting the default project (resolve_projects). The run's name is name, else the config's, else the time of this call, shared by every project (run_name) and claimed in each, with -2, -3 when another invocation holds it (claim_run_name). frames stands in for the config's frames; out puts each project's files in <out>/<project name>/ (run_paths). force recomputes every frame asked for of every selected stage and lets another config take over a run of this name, dropping the stages whose fingerprint moved. A project that fails carries its message in the summary instead of raising. observer is told each step as it happens (RunObserver).

Raises:

Type Description
ConfigError

If the stage selection or the name fails, which is the invocation being wrong.
(FileNotFoundError, ValueError)

If a path stands for no project, or two paths for one; nothing has run then.
Source code in src/deeperfly/pipeline/run.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def run_projects(
    cfg: RunConfig,
    paths: Sequence[str | Path],
    *,
    only: Sequence[str] = (),
    skip: Sequence[str] = (),
    out: Path | None = None,
    force: bool = False,
    name: str | None = None,
    frames: Sequence[int | str] | None = None,
    log: Callable[[str], None] = print,
    observer: RunObserver | None = None,
) -> RunSummary:
    """Apply a run config to the projects some paths stand for, and say what happened.

    Each path is resolved as every command resolves one, a directory of videos
    getting the default project
    ([resolve_projects][deeperfly.pipeline.plan.resolve_projects]). The run's name is
    `name`, else the config's, else the time of this call, shared by every project
    ([run_name][deeperfly.pipeline.plan.run_name]) and claimed in each, with `-2`,
    `-3` when another invocation holds it
    ([claim_run_name][deeperfly.pipeline.plan.claim_run_name]). `frames` stands
    in for the config's `frames`; `out` puts each project's files in
    `<out>/<project name>/` ([run_paths][deeperfly.pipeline.plan.run_paths]). `force`
    recomputes every frame asked for of every selected stage and lets another config
    take over a run of this name, dropping the stages whose fingerprint moved. A
    project that fails carries its message in the summary instead of raising.
    `observer` is told each step as it happens
    ([RunObserver][deeperfly.pipeline.observe.RunObserver]).

    Raises:
        ConfigError: If the stage selection or the name fails, which is the
            invocation being wrong.
        FileNotFoundError, ValueError: If a path stands for no project, or two paths
            for one; nothing has run then.
    """
    stages = select_stages(cfg, tuple(only), tuple(skip))
    run = run_name(cfg, name)
    default = name is None and cfg.name is None
    fresh = {
        project.folder
        for _, project in resolve_projects(paths, create=False)
        if not project.manifest_path.is_file()
    }
    projects = resolve_projects(paths, create=True, log=log)
    detector = (
        DetectorCache(
            tuple(str(w) for w in cfg.pose2d.weights),
            locator=(None if cfg.pose2d.locator is None else str(cfg.pose2d.locator)),
            device=cfg.compute.device,
            precision=cfg.compute.precision,
        )
        if cfg.pose2d is not None and "pose2d" in stages
        else None
    )
    config_text = cfg.recorded_text()
    claimed: list[Path | OSError | None] = [None] * len(projects)
    if default:
        run, made = claim_run_name(run, projects, out, config_text)
        claimed = list(made)
    log(f"run {run}: {len(projects)} project(s), stages " + ", ".join(stages))
    if observer is not None:
        observer.run_started(run, [label for label, _ in projects], stages)

    outcomes = []
    for (label, project), claim in zip(projects, claimed, strict=True):
        outcome = _run_project(
            cfg,
            project,
            label,
            stages,
            run=run,
            out=out,
            frames=cfg.frames if frames is None else frames,
            config_text=config_text,
            detector=detector,
            force=force,
            fresh=project.folder in fresh,
            claimed=claim,
            log=log,
            observer=observer,
        )
        if observer is not None:
            observer.project_finished(label, outcome.error)
        outcomes.append(outcome)
    failed = [o for o in outcomes if o.error is not None]
    if failed:
        log(f"{len(failed)} of {len(outcomes)} project(s) failed:")
        for o in failed:
            log(f"  {o.project}: {o.error}")
    return RunSummary(out, tuple(stages), tuple(outcomes), run)

What a run tells someone watching it, as it goes (ADR-0126).

run_projects hands each step of a run to an observer, such as deeperfly run's live view, which the pipeline does not import. An observer that wants the run to end raises RunStopped from any call; it unwinds as Ctrl-C does.

RunObserver

Bases: Protocol

The calls a run makes, in order, to whoever watches it.

Source code in src/deeperfly/pipeline/observe.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class RunObserver(Protocol):
    """The calls a run makes, in order, to whoever watches it."""

    def run_started(
        self, run: str, labels: Sequence[str], stages: Sequence[str]
    ) -> None:
        """The run's name, its projects in the order they run, and its stages."""

    def project_started(self, label: str, ctx: "Context") -> None:
        """A project's recording and results are open, about to walk its stages."""

    def stage_cached(self, stage: str) -> None:
        """`stage` has nothing to compute here: an earlier invocation covered it."""

    def stage_started(self, stage: str, n_frames: int) -> None:
        """`stage` begins computing `n_frames` frames."""

    def pose2d_block(
        self,
        frames: np.ndarray,
        points: np.ndarray,
        conf: np.ndarray,
        boxes: np.ndarray,
        images: list[np.ndarray],
    ) -> None:
        """One detector pass, as `predict_recording`'s `on_block` hands it over."""

    def stage_finished(self, stage: str, elapsed: float, ctx: "Context") -> None:
        """`stage` wrote what it computed, in `elapsed` seconds."""

    def project_finished(self, label: str, error: str | None) -> None:
        """A project is done, or failed with `error`."""

pose2d_block 

pose2d_block(
    frames: ndarray,
    points: ndarray,
    conf: ndarray,
    boxes: ndarray,
    images: list[ndarray],
) -> None

One detector pass, as predict_recording's on_block hands it over.

Source code in src/deeperfly/pipeline/observe.py 
41
42
43
44
45
46
47
48
49
def pose2d_block(
    self,
    frames: np.ndarray,
    points: np.ndarray,
    conf: np.ndarray,
    boxes: np.ndarray,
    images: list[np.ndarray],
) -> None:
    """One detector pass, as `predict_recording`'s `on_block` hands it over."""

project_finished 

project_finished(label: str, error: str | None) -> None

A project is done, or failed with error.

Source code in src/deeperfly/pipeline/observe.py 
54
55
def project_finished(self, label: str, error: str | None) -> None:
    """A project is done, or failed with `error`."""

project_started 

project_started(label: str, ctx: Context) -> None

A project's recording and results are open, about to walk its stages.

Source code in src/deeperfly/pipeline/observe.py 
32
33
def project_started(self, label: str, ctx: "Context") -> None:
    """A project's recording and results are open, about to walk its stages."""

run_started 

run_started(
    run: str, labels: Sequence[str], stages: Sequence[str]
) -> None

The run's name, its projects in the order they run, and its stages.

Source code in src/deeperfly/pipeline/observe.py 
27
28
29
30
def run_started(
    self, run: str, labels: Sequence[str], stages: Sequence[str]
) -> None:
    """The run's name, its projects in the order they run, and its stages."""

stage_cached 

stage_cached(stage: str) -> None

stage has nothing to compute here: an earlier invocation covered it.

Source code in src/deeperfly/pipeline/observe.py 
35
36
def stage_cached(self, stage: str) -> None:
    """`stage` has nothing to compute here: an earlier invocation covered it."""

stage_finished 

stage_finished(
    stage: str, elapsed: float, ctx: Context
) -> None

stage wrote what it computed, in elapsed seconds.

Source code in src/deeperfly/pipeline/observe.py 
51
52
def stage_finished(self, stage: str, elapsed: float, ctx: "Context") -> None:
    """`stage` wrote what it computed, in `elapsed` seconds."""

stage_started 

stage_started(stage: str, n_frames: int) -> None

stage begins computing n_frames frames.

Source code in src/deeperfly/pipeline/observe.py 
38
39
def stage_started(self, stage: str, n_frames: int) -> None:
    """`stage` begins computing `n_frames` frames."""

RunStopped

Bases: KeyboardInterrupt

Someone watching asked the run to stop: it ends as Ctrl-C would end it.

Source code in src/deeperfly/pipeline/observe.py 
20
21
class RunStopped(KeyboardInterrupt):
    """Someone watching asked the run to stop: it ends as Ctrl-C would end it."""

What a run means before it runs: its projects, stages, fingerprints and frames.

The paths an invocation names resolved to projects (ADR-0081), the stage selection (--only, --skip), the camera parameters a run starts from resolved against a project (ADR-0086), the run's name and files (results/<name>.h5 and its .toml, renders/<name>/), the frame set an invocation asks for, each stage's fingerprint (ADR-0035), and the coverage plan: which frames each stage computes, given what the run already covers (ADR-0082).

Facts dataclass

What a fingerprint hashes besides the config's stage tables.

Attributes:

Name Type Description
animals int

How many animals are tracked.
mirrored tuple[str, ...]

The mirrored views, in the rig's order.
views dict[str, Any]

The initial camera parameters, per view, as data; empty when the run reads none.
Source code in src/deeperfly/pipeline/plan.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@dataclass(frozen=True)
class Facts:
    """What a fingerprint hashes besides the config's stage tables.

    Attributes:
        animals: How many animals are tracked.
        mirrored: The mirrored views, in the rig's order.
        views: The initial camera parameters, per view, as data; empty when the run
            reads none.
    """

    animals: int
    mirrored: tuple[str, ...]
    views: dict[str, Any]

    @classmethod
    def of(cls, project: Project, calibration: Calibration | None) -> Facts:
        """A project's facts: its animals and mirrored views, and the initial values.

        From the manifest and the resolved `[calibration]`
        ([initial_calibration][deeperfly.pipeline.plan.initial_calibration]); no
        video is opened.
        """
        mirrored = set(project.manifest.mirrored)
        return cls(
            animals=project.animals,
            mirrored=tuple(v for v in project.views if v in mirrored),
            views=(
                {}
                if calibration is None
                else {n: _canonical(v) for n, v in calibration.views.items()}
            ),
        )

of classmethod 

of(
    project: Project, calibration: Calibration | None
) -> Facts

A project's facts: its animals and mirrored views, and the initial values.

From the manifest and the resolved [calibration] (initial_calibration); no video is opened.

Source code in src/deeperfly/pipeline/plan.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@classmethod
def of(cls, project: Project, calibration: Calibration | None) -> Facts:
    """A project's facts: its animals and mirrored views, and the initial values.

    From the manifest and the resolved `[calibration]`
    ([initial_calibration][deeperfly.pipeline.plan.initial_calibration]); no
    video is opened.
    """
    mirrored = set(project.manifest.mirrored)
    return cls(
        animals=project.animals,
        mirrored=tuple(v for v in project.views if v in mirrored),
        views=(
            {}
            if calibration is None
            else {n: _canonical(v) for n, v in calibration.views.items()}
        ),
    )

RunPaths dataclass

Where one project's run lives.

Attributes:

Name Type Description
results Path

The results file, <name>.h5.
config Path

The run config copied verbatim, <name>.toml beside it.
renders Path

The folder the [render] stage writes its videos into.
Source code in src/deeperfly/pipeline/plan.py 
441
442
443
444
445
446
447
448
449
450
451
452
453
@dataclass(frozen=True)
class RunPaths:
    """Where one project's run lives.

    Attributes:
        results: The results file, `<name>.h5`.
        config: The run config copied verbatim, `<name>.toml` beside it.
        renders: The folder the `[render]` stage writes its videos into.
    """

    results: Path
    config: Path
    renders: Path

claim_run_name 

claim_run_name(
    stamp: str,
    projects: Sequence[tuple[str, Project]],
    out: Path | None,
    config_text: str,
) -> tuple[str, list[Path | OSError]]

A default run name that no project of the invocation holds, claimed in each.

stamp, else stamp-2, stamp-3, ...: the first under which no project has the run's .h5 or .toml (run_paths). Each project's .toml is created exclusively, holding config_text, so another invocation in the same second, in this process or another, finds the name taken and moves on (ADR-0082). Returns the name and, per project in order, the file claimed or the error that kept the project from claiming it, which fails that project alone.

Source code in src/deeperfly/pipeline/plan.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def claim_run_name(
    stamp: str,
    projects: Sequence[tuple[str, Project]],
    out: Path | None,
    config_text: str,
) -> tuple[str, list[Path | OSError]]:
    """A default run name that no project of the invocation holds, claimed in each.

    `stamp`, else `stamp-2`, `stamp-3`, ...: the first under which no project has
    the run's `.h5` or `.toml` ([run_paths][deeperfly.pipeline.plan.run_paths]).
    Each project's `.toml` is created exclusively, holding `config_text`, so another
    invocation in the same second, in this process or another, finds the name taken
    and moves on (ADR-0082). Returns the name and, per project in order, the file
    claimed or the error that kept the project from claiming it, which fails that
    project alone.
    """
    n = 0
    while True:
        n += 1
        name = stamp if n == 1 else f"{stamp}-{n}"
        files = [run_paths(p, name, out, label=label).config for label, p in projects]
        if any(f.exists() or f.with_suffix(".h5").exists() for f in files):
            continue
        claimed: list[Path | OSError] = []
        try:
            for path in files:
                claimed.append(_claim(path, config_text))
        except FileExistsError:
            for path in claimed:
                if isinstance(path, Path):
                    path.unlink()
            continue
        return name, claimed

collision 

collision(name: str) -> str

What a run of this name holding another config says.

Source code in src/deeperfly/pipeline/plan.py 
484
485
486
487
488
489
def collision(name: str) -> str:
    """What a run of this name holding another config says."""
    return (
        f"run {name!r} exists with another config; pick another name, or --force to "
        "change it in place"
    )

coverage_plan 

coverage_plan(
    cfg: RunConfig,
    stages: Sequence[str],
    results: Results,
    requested: ndarray,
    facts: Facts,
    *,
    force: bool,
    log: Callable[[str], None],
) -> dict[str, np.ndarray]

The frames each selected stage computes in this invocation (ADR-0082).

Decided before any stage runs, in pipeline order. A stage whose stored fingerprint moved covers nothing (its group is dropped when it begins); it computes the frames asked for that it does not cover, plus the ones a stage it derives from, directly or not, computes in this invocation, and with force every frame asked for. bundle_adjustment solves one calibration, so it computes every frame or none: all of them when it never finished, its fingerprint moved, or force. render cannot patch a video, so it draws every frame asked for unless those are covered and nothing it reads is recomputed. A stage with nothing to compute is logged as cached, with how many frames it covers.

Source code in src/deeperfly/pipeline/plan.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
def coverage_plan(
    cfg: RunConfig,
    stages: Sequence[str],
    results: Results,
    requested: np.ndarray,
    facts: Facts,
    *,
    force: bool,
    log: Callable[[str], None],
) -> dict[str, np.ndarray]:
    """The frames each selected stage computes in this invocation (ADR-0082).

    Decided before any stage runs, in pipeline order. A stage whose stored
    fingerprint moved covers nothing (its group is dropped when it begins); it
    computes the frames asked for that it does not cover, plus the ones a stage it
    derives from, directly or not, computes in this invocation, and with `force`
    every frame asked for. `bundle_adjustment` solves one calibration, so it
    computes every frame or none: all of them when it never finished, its
    fingerprint moved, or `force`. `render` cannot patch a video, so it draws every
    frame asked for unless those are covered and nothing it reads is recomputed. A
    stage with nothing to compute is logged as cached, with how many frames it
    covers.
    """
    n_t = results.n_frames
    asked = np.zeros(n_t, dtype=bool)
    asked[np.asarray(requested, dtype=np.int64)] = True
    computed: dict[str, np.ndarray] = {}
    plan: dict[str, np.ndarray] = {}
    for stage in stages:
        wanted = fingerprint(cfg, stage, facts)
        stored = results.fingerprint(stage)
        moved = stored is not None and stored != wanted
        covered = np.zeros(n_t, dtype=bool) if moved else results.covered(stage)
        upstream = np.zeros(n_t, dtype=bool)
        for name, done in computed.items():
            if stage in downstream(name):
                upstream |= done
        if stage == "bundle_adjustment":
            whole = force or moved or not covered.all()
            todo = np.full(n_t, whole, dtype=bool)
        elif stage == "render":
            again = force or moved or upstream.any() or not covered[asked].all()
            todo = asked.copy() if again else np.zeros(n_t, dtype=bool)
        elif force:
            todo = asked.copy()
        else:
            todo = asked & (~covered | upstream)
        computed[stage] = todo
        plan[stage] = np.flatnonzero(todo)
        count = int(covered.sum())
        if plan[stage].size == 0:
            log(f"  {stage}: cached, {count} of {n_t} frames covered ({wanted})")
        else:
            log(
                f"  {stage}: {plan[stage].size} frame(s) to compute, {count} of "
                f"{n_t} covered ({wanted})"
            )
    return plan

describe 

describe(
    cfg: RunConfig,
    stages: Sequence[str],
    paths: Sequence[str | Path],
    out: Path | None = None,
    *,
    name: str,
    frames: Sequence[int | str] | None = None,
    force: bool = False,
) -> tuple[list[str], int]

The dry run: the selected stages, then what each project would do.

Each path is resolved as the run resolves it but nothing is created: a directory of videos is planned against the default project the run would create, built in memory. A project is checked, its calibration resolved and its recording opened exactly as a run would, so one the config does not fit shows here, with its frame set, its run's file and the coverage plan against what that file already covers (coverage_plan). frames stands in for the config's. Returns the report's lines and how many projects would fail.

Raises:

Type Description
(FileNotFoundError, ValueError)

As resolve_projects.
Source code in src/deeperfly/pipeline/plan.py 
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
def describe(
    cfg: RunConfig,
    stages: Sequence[str],
    paths: Sequence[str | Path],
    out: Path | None = None,
    *,
    name: str,
    frames: Sequence[int | str] | None = None,
    force: bool = False,
) -> tuple[list[str], int]:
    """The dry run: the selected stages, then what each project would do.

    Each path is resolved as the run resolves it but nothing is created: a directory
    of videos is planned against the default project the run would create, built in
    memory. A project is checked, its calibration resolved and its recording opened
    exactly as a run would, so one the config does not fit shows here, with its frame
    set, its run's file and the coverage plan against what that file already covers
    ([coverage_plan][deeperfly.pipeline.plan.coverage_plan]). `frames` stands in for
    the config's. Returns the report's lines and how many projects would fail.

    Raises:
        FileNotFoundError, ValueError: As
            [resolve_projects][deeperfly.pipeline.plan.resolve_projects].
    """
    report = [f"run {name}: " + ", ".join(stages)]
    off = [s for s in cfg.enabled if s not in stages]
    if off:
        report.append("not selected: " + ", ".join(off))
    config_text = cfg.recorded_text()
    bad = 0
    for label, project in resolve_projects(paths, create=False):
        if not project.manifest_path.is_file():
            report.append(
                f"{label}: no project yet; the run creates {project.folder}, one view "
                "per video file (" + ", ".join(project.views) + ")"
            )
        files = run_paths(project, name, out, label=label)
        try:
            initial, facts = prepare(cfg, project, project.skeleton().point_names)
            recording = open_recording(project)
            n_frames = recording.n_frames
            sizes = project.layout.frame_sizes(project.views, recording.videos)
            if initial is not None:
                project.rig(initial, frame_hw=sizes)
            spec = cfg.frames if frames is None else frames
            requested = frame_set(spec, n_frames)
        except (OSError, ValueError) as exc:
            report.append(f"{label}: {exc}")
            bad += 1
            continue
        report.append(
            f"{label}: {n_frames} frames, views "
            + ", ".join(project.views)
            + f", {requested.size} asked for -> {files.results}"
        )
        if initial is not None:
            source = (
                f", a copy of calibration {initial.copied_from}"
                if initial.copied_from is not None
                else ""
            )
            report.append(f"  starts from calibration {initial.label!r}{source}")
        if not files.results.exists():
            report.append("  a new run: every selected stage computes them")
            width = max(map(len, stages), default=0)
            report += [f"  {s:{width}s} {fingerprint(cfg, s, facts)}" for s in stages]
            continue
        try:
            with Results.open(files.results) as res:
                if res.n_frames != n_frames:
                    # What `Results.create` refuses, or replaces with --force.
                    if force:
                        report.append(
                            f"  replaced: it was written for {res.n_frames} frames"
                        )
                        continue
                    report.append(
                        f"  {files.results} was written for {res.n_frames} frames, "
                        f"not {n_frames}"
                    )
                    bad += 1
                    continue
                if not force and not res.same_config(config_text):
                    report.append(f"  {collision(name)}")
                    bad += 1
                    continue
                conflict = forced_conflict(cfg, stages, res, facts) if force else None
                if conflict is not None:
                    report.append(f"  {conflict}")
                    bad += 1
                    continue
                coverage_plan(
                    cfg, stages, res, requested, facts, force=force, log=report.append
                )
                if force:
                    drops = forced_drops(cfg, stages, res, facts)
                    report += [forced_drop_line(s, n) for s, n in drops.items()]
        except (OSError, ValueError) as exc:
            report.append(f"  {files.results}: {exc}")
            bad += 1
    return report, bad

downstream 

downstream(stage: str) -> list[str]

The stages whose output derives from stage's, in pipeline order.

Transitive over UPSTREAM: smoothing and ik read the 2D as well as the 3D their tables name, and whatever came from a stage's values is stale with them.

Source code in src/deeperfly/pipeline/plan.py 
82
83
84
85
86
87
88
89
90
91
92
def downstream(stage: str) -> list[str]:
    """The stages whose output derives from `stage`'s, in pipeline order.

    Transitive over `UPSTREAM`: `smoothing` and `ik` read the 2D as well as the 3D
    their tables name, and whatever came from a stage's values is stale with them.
    """
    found = {stage}
    for name in STAGES:
        if found.intersection(UPSTREAM[name]):
            found.add(name)
    return [s for s in STAGES if s in found and s != stage]

fingerprint 

fingerprint(
    cfg: RunConfig, stage: str, facts: Facts
) -> str

A stage's fingerprint: its table, everything upstream, and the rig (ADR-0035).

The payload is the stage's table and every table upstream, the animal count and the mirrored views (which decide what the detector sees), and the initial calibration's values past [pose2d] or when several animals are tracked. A key at its default and a key omitted hash the same; the [bundle_adjustment] stage stamps the fingerprint on the rig it writes.

Source code in src/deeperfly/pipeline/plan.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def fingerprint(cfg: RunConfig, stage: str, facts: Facts) -> str:
    """A stage's fingerprint: its table, everything upstream, and the rig (ADR-0035).

    The payload is the stage's table and every table upstream, the animal count and
    the mirrored views (which decide what the detector sees), and the initial
    calibration's values past `[pose2d]` or when several animals are tracked. A key
    at its default and a key omitted hash the same; the `[bundle_adjustment]` stage
    stamps the fingerprint on the rig it writes.
    """
    if stage not in UPSTREAM:
        raise ConfigError(f"{stage!r} is not a stage; " + ", ".join(STAGES))
    payload: dict[str, Any] = {
        name: _canonical(getattr(cfg, name))
        for name in (*UPSTREAM[stage], stage)
        if getattr(cfg, name, None) is not None
    }
    payload["animals"] = facts.animals
    payload["mirrored"] = list(facts.mirrored)
    if stage in GEOMETRY_STAGES or facts.animals > 1:
        payload["views"] = facts.views
    blob = json.dumps(payload, sort_keys=True, default=str)
    return hashlib.sha256(blob.encode()).hexdigest()[:16]

forced_conflict 

forced_conflict(
    cfg: RunConfig,
    stages: Sequence[str],
    results: Results,
    facts: Facts,
) -> str | None

Why --force refuses this selection over results; None when it does not.

A stage --force would drop unselected (forced_drops) that a selected stage reads would leave that stage reading nothing, or the initial rig in place of the solved one. The invocation is refused before anything is written instead, so the stored stage stays; selecting it too recomputes it under the new config. A stage whose table the config no longer has is read by none, so it is dropped.

Source code in src/deeperfly/pipeline/plan.py 
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
def forced_conflict(
    cfg: RunConfig, stages: Sequence[str], results: Results, facts: Facts
) -> str | None:
    """Why `--force` refuses this selection over `results`; `None` when it does not.

    A stage `--force` would drop unselected
    ([forced_drops][deeperfly.pipeline.plan.forced_drops]) that a selected stage
    reads would leave that stage reading nothing, or the initial rig in place of the
    solved one. The invocation is refused before anything is written instead, so the
    stored stage stays; selecting it too recomputes it under the new config. A stage
    whose table the config no longer has is read by none, so it is dropped.
    """
    dropped = forced_drops(cfg, stages, results, facts)
    for stage in stages:
        read = [
            name
            for name in UPSTREAM[stage]
            if name in dropped and getattr(cfg, name, None) is not None
        ]
        if read:
            return (
                f"[{stage}] reads [{read[0]}], which is not selected and whose "
                f"fingerprint moved, so --force would drop it; select {read[0]} too"
            )
    return None

forced_drop_line 

forced_drop_line(stage: str, lost: int) -> str

What a run and its dry run say of a stage --force drops unselected.

Source code in src/deeperfly/pipeline/plan.py 
599
600
601
602
603
604
def forced_drop_line(stage: str, lost: int) -> str:
    """What a run and its dry run say of a stage `--force` drops unselected."""
    return (
        f"  {stage}: not selected, and its fingerprint moved; --force drops it and "
        f"the {lost} frame(s) it covered"
    )

forced_drops 

forced_drops(
    cfg: RunConfig,
    stages: Sequence[str],
    results: Results,
    facts: Facts,
) -> dict[str, int]

The stored stages --force drops without running them, with what each covers.

--force lets a config take over a run and drops every stage whose fingerprint moved (ADR-0082). A selected stage is dropped when it begins; one left out with --only or --skip never begins, so it is one of these, lest the file and its .toml claim a config that did not compute it.

Source code in src/deeperfly/pipeline/plan.py 
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
def forced_drops(
    cfg: RunConfig, stages: Sequence[str], results: Results, facts: Facts
) -> dict[str, int]:
    """The stored stages `--force` drops without running them, with what each covers.

    `--force` lets a config take over a run and drops every stage whose fingerprint
    moved (ADR-0082). A selected stage is dropped when it begins; one left out with
    `--only` or `--skip` never begins, so it is one of these, lest the file and its
    `.toml` claim a config that did not compute it.
    """
    stored = results.stages()
    return {
        stage: int(results.covered(stage).sum())
        for stage in STAGES
        if stage in stored
        and stage not in stages
        and results.fingerprint(stage) != fingerprint(cfg, stage, facts)
    }

frame_set 

frame_set(
    spec: Sequence[int | str], n_frames: int
) -> np.ndarray

The frames an invocation covers: indices and slices, unioned and sorted.

Each entry is an index or a "start:stop:step" slice with Python's semantics over range(n_frames); empty means every frame. The frame set is outside every fingerprint.

Raises:

Type Description
ValueError

On a malformed or negative entry, an index at or past n_frames, or a set that selects no frame at all.
Source code in src/deeperfly/pipeline/plan.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
def frame_set(spec: Sequence[int | str], n_frames: int) -> np.ndarray:
    """The frames an invocation covers: indices and slices, unioned and sorted.

    Each entry is an index or a `"start:stop:step"` slice with Python's semantics
    over `range(n_frames)`; empty means every frame. The frame set is outside every
    fingerprint.

    Raises:
        ValueError: On a malformed or negative entry, an index at or past
            `n_frames`, or a set that selects no frame at all.
    """
    if not spec:
        return np.arange(n_frames, dtype=np.int64)
    picked: list[np.ndarray] = []
    for entry in parse_frames(spec):
        if isinstance(entry, slice):
            picked.append(np.arange(*entry.indices(n_frames), dtype=np.int64))
        elif entry >= n_frames:
            raise ValueError(
                f"frame {entry} is past the end of the recording's {n_frames} frames"
            )
        else:
            picked.append(np.array([entry], dtype=np.int64))
    frames = np.unique(np.concatenate(picked))
    if frames.size == 0:
        raise ValueError(
            "frames " + ", ".join(str(e) for e in spec) + f" select none of the "
            f"recording's {n_frames} frames"
        )
    return frames

initial_calibration 

initial_calibration(
    cfg: RunConfig, project: Project
) -> Calibration | None

[calibration] resolved into values for project; None without the table.

from = "active" and an id copy that calibration of the project, and a file is read; either way the values are a deep copy, so editing or deleting the file later does not reach the run, and a copy of a project calibration records its id as copied_from (ADR-0086). Inline placements are taken as written, each view mirrored as the manifest says. The views come out in the manifest's order, checked against it (placed_views).

Raises:

Type Description
ValueError

If the project has no active calibration or no such id, the file is not a calibration, or the views do not match the manifest.
OSError

If the file cannot be read.
Source code in src/deeperfly/pipeline/plan.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def initial_calibration(cfg: RunConfig, project: Project) -> Calibration | None:
    """`[calibration]` resolved into values for `project`; `None` without the table.

    `from = "active"` and an id copy that calibration of the project, and a file is
    read; either way the values are a deep copy, so editing or deleting the file
    later does not reach the run, and a copy of a project calibration records its id
    as `copied_from` (ADR-0086). Inline placements are taken as written, each view
    mirrored as the manifest says. The views come out in the manifest's order,
    checked against it ([placed_views][deeperfly.project.Project.placed_views]).

    Raises:
        ValueError: If the project has no active calibration or no such id, the file
            is not a calibration, or the views do not match the manifest.
        OSError: If the file cannot be read.
    """
    spec = cfg.calibration
    if spec is None:
        return None
    if spec.views:
        mirrored = set(project.manifest.mirrored)
        calib = Calibration(
            label="initial",
            views={
                name: view.model_copy(update={"mirrored": name in mirrored})
                for name, view in spec.views.items()
            },
        )
    elif spec.file is not None:
        calib = read_calibration(spec.file).model_copy(update={"copied_from": None})
    else:
        cid = (
            project.manifest.active_calibration
            if spec.source == ACTIVE
            else spec.source
        )
        if not isinstance(cid, int):
            raise ValueError(
                f'{project.folder}: [calibration] from = "active", but the project '
                "has no active calibration; activate one in the GUI, or name one "
                "by id"
            )
        try:
            found = project.calibration(cid)
        except KeyError as exc:
            said = f'"{ACTIVE}"' if spec.source == ACTIVE else cid
            raise ValueError(
                f"[calibration] from = {said}: {project.folder} has no calibration "
                f"{cid} in {project.calibrations_dir.name}/"
            ) from exc
        assert found is not None
        calib = found.model_copy(deep=True, update={"copied_from": cid})
    return calib.model_copy(update={"views": project.placed_views(calib)})

prepare 

prepare(
    cfg: RunConfig, project: Project, points: Sequence[str]
) -> tuple[Calibration | None, Facts]

Check a config against a project and resolve what the run starts from.

check_project against the project's views, animals and points, then the initial calibration when the run reads one (uses_rig); a pose2d-only run of one animal reads and stores none, whatever [calibration] says.

Raises:

Type Description
ConfigError

If the config does not fit the project.
ValueError

If the run needs camera parameters and [calibration] gives none, or they do not resolve.
OSError

If a calibration file cannot be read.
Source code in src/deeperfly/pipeline/plan.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def prepare(
    cfg: RunConfig, project: Project, points: Sequence[str]
) -> tuple[Calibration | None, Facts]:
    """Check a config against a project and resolve what the run starts from.

    [check_project][deeperfly.config.run.RunConfig.check_project] against the
    project's views, animals and `points`, then the initial calibration when the run
    reads one ([uses_rig][deeperfly.pipeline.plan.uses_rig]); a pose2d-only run of one
    animal reads and stores none, whatever `[calibration]` says.

    Raises:
        ConfigError: If the config does not fit the project.
        ValueError: If the run needs camera parameters and `[calibration]` gives
            none, or they do not resolve.
        OSError: If a calibration file cannot be read.
    """
    cfg.check_project(
        project.views, project.animals, points, videos=list(project.manifest.videos)
    )
    calibration = None
    if uses_rig(cfg, project.animals):
        calibration = initial_calibration(cfg, project)
        if calibration is None:
            why = (
                f"{project.animals} animals are tracked in 3D"
                if project.animals > 1
                else "[" + next(s for s in cfg.enabled if s in GEOMETRY_STAGES) + "] "
                "reads the camera parameters"
            )
            raise ValueError(
                f"{project.folder}: {why}, and the config has no [calibration] to "
                'start from; add one, from = "active", an id, a file, or '
                "[calibration.views.<name>] placements (ADR-0086)"
            )
    return calibration, Facts.of(project, calibration)

resolve_projects 

resolve_projects(
    paths: Sequence[str | Path],
    *,
    create: bool = True,
    log: Callable[[str], None] = print,
) -> list[tuple[str, Project]]

The project each path stands for, named as the invocation names them.

One rule for every path (ADR-0081): a project is used as is; else <path>/deeperfly/ if that is one; else a directory of video files gets the default project there (default), written and logged with create, and without it only built in memory. Every path is checked, its default project built and the names worked out (project_names), before anything is written.

Raises:

Type Description
FileNotFoundError

If a path does not exist.
ValueError

If a path is neither a project nor a directory of videos it can make the default project of, or two paths stand for one project.
Source code in src/deeperfly/pipeline/plan.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def resolve_projects(
    paths: Sequence[str | Path],
    *,
    create: bool = True,
    log: Callable[[str], None] = print,
) -> list[tuple[str, Project]]:
    """The project each path stands for, named as the invocation names them.

    One rule for every path (ADR-0081): a project is used as is; else
    `<path>/deeperfly/` if that is one; else a directory of video files gets the
    default project there ([default][deeperfly.project.Project.default]), written and
    logged with `create`, and without it only built in memory. Every path is
    checked, its default project built and the names worked out
    ([project_names][deeperfly.project.project_names]), before anything is written.

    Raises:
        FileNotFoundError: If a path does not exist.
        ValueError: If a path is neither a project nor a directory of videos it can
            make the default project of, or two paths stand for one project.
    """
    typed = [normalized(p) for p in paths]
    kinds = [classify(p) for p in typed]
    for path, kind in zip(typed, kinds, strict=True):
        if kind == "missing":
            raise FileNotFoundError(f"{path}: no such file or directory")
        if kind not in ("project", "videos"):
            raise ValueError(
                f"{path}: not a project, nor a directory of video files to make the "
                "default project of"
                + (
                    "; its children are projects, which a run takes one path each"
                    if kind == "collection"
                    else ""
                )
            )
    found = [
        Project.resolve(p) if kind == "project" else Project.default(p)
        for p, kind in zip(typed, kinds, strict=True)
    ]
    names = project_names([f.folder for f in found])
    if create:
        for i, (path, kind) in enumerate(zip(typed, kinds, strict=True)):
            if kind == "videos":
                found[i] = project = Project.resolve(path, create=True)
                log(
                    f"{path}: created the default project {project.folder}, one view "
                    f"per video file ({', '.join(project.views)})"
                )
    return list(zip(names, found, strict=True))

run_name 

run_name(
    cfg: RunConfig,
    override: str | None = None,
    *,
    now: float | None = None,
) -> str

What a run is called: --name, else the config's name, else the time.

The default is %y%m%d_%H%M%S at now (default: the call), taken once per invocation and shared by every project of it; the run claims it, or -2, -3 when another invocation holds it (claim_run_name; ADR-0082).

Raises:

Type Description
ConfigError

If override is not a run name.
Source code in src/deeperfly/pipeline/plan.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def run_name(
    cfg: RunConfig, override: str | None = None, *, now: float | None = None
) -> str:
    """What a run is called: `--name`, else the config's `name`, else the time.

    The default is `%y%m%d_%H%M%S` at `now` (default: the call), taken once per
    invocation and shared by every project of it; the run claims it, or `-2`, `-3`
    when another invocation holds it
    ([claim_run_name][deeperfly.pipeline.plan.claim_run_name]; ADR-0082).

    Raises:
        ConfigError: If `override` is not a run name.
    """
    if override is not None:
        check_run_name(override)
        return override
    if cfg.name is not None:
        return cfg.name
    return time.strftime("%y%m%d_%H%M%S", time.localtime(now))

run_paths 

run_paths(
    project: Project | Path,
    name: str,
    out: Path | None = None,
    *,
    label: str | None = None,
) -> RunPaths

A run's files: results/<name>.{h5,toml} and renders/<name>/ (ADR-0082).

They go in the project folder (a Project, or the folder one would be created in); with out, in <out>/<label>/ instead, as <name>.{h5,toml} and renders/<name>/, the label being the project's name on its own unless given (project_names).

Source code in src/deeperfly/pipeline/plan.py 
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
def run_paths(
    project: Project | Path,
    name: str,
    out: Path | None = None,
    *,
    label: str | None = None,
) -> RunPaths:
    """A run's files: `results/<name>.{h5,toml}` and `renders/<name>/` (ADR-0082).

    They go in the project folder (a `Project`, or the folder one would be created
    in); with `out`, in `<out>/<label>/` instead, as `<name>.{h5,toml}` and
    `renders/<name>/`, the label being the project's name on its own unless given
    ([project_names][deeperfly.project.project_names]).
    """
    folder = project.folder if isinstance(project, Project) else Path(project)
    if out is None:
        base = folder
        results = folder / RESULTS_DIRNAME
    else:
        label = label or project_names([folder])[0]
        base = results = Path(out).expanduser() / label
    return RunPaths(
        results=results / f"{name}.h5",
        config=results / f"{name}.toml",
        renders=base / RENDERS_DIRNAME / name,
    )

select_stages 

select_stages(
    cfg: RunConfig,
    only: tuple[str, ...] = (),
    skip: tuple[str, ...] = (),
) -> list[str]

The stages to execute: the enabled ones, narrowed by --only and --skip.

Narrowing does not re-validate: --only render renders from results a previous run wrote, so the references it checks are still the whole config's.

Raises:

Type Description
ConfigError

If a name is not a stage, or is not enabled here.
Source code in src/deeperfly/pipeline/plan.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def select_stages(
    cfg: RunConfig, only: tuple[str, ...] = (), skip: tuple[str, ...] = ()
) -> list[str]:
    """The stages to execute: the enabled ones, narrowed by `--only` and `--skip`.

    Narrowing does not re-validate: `--only render` renders from results a previous
    run wrote, so the references it checks are still the whole config's.

    Raises:
        ConfigError: If a name is not a stage, or is not enabled here.
    """
    chosen = list(cfg.enabled)
    for names, flag in ((only, "--only"), (skip, "--skip")):
        for name in names:
            if name not in STAGES:
                raise ConfigError(f"{flag} {name}: not a stage; " + ", ".join(STAGES))
            if name not in chosen:
                raise ConfigError(
                    f"{flag} {name}: this config has no [{name}] table, so the stage "
                    "is off"
                )
    if only:
        chosen = [s for s in chosen if s in only]
    return [s for s in chosen if s not in skip]

uses_rig 

uses_rig(cfg: RunConfig, animals: int) -> bool

Whether a run reads camera parameters: a geometry stage, or the tracker.

Source code in src/deeperfly/pipeline/plan.py 
261
262
263
def uses_rig(cfg: RunConfig, animals: int) -> bool:
    """Whether a run reads camera parameters: a geometry stage, or the tracker."""
    return animals > 1 or any(s in GEOMETRY_STAGES for s in cfg.enabled)

The driver: a run over projects, stage by stage, over a frame set.

A run is one results file per project, results/<name>.h5 with its config copied to results/<name>.toml in the project folder (ADR-0082). Each path is resolved to a project, a directory of videos getting the default one (ADR-0081). For each project the driver checks the config against it, resolves the camera parameters the run starts from (ADR-0086), opens its recording, works out the frames asked for, continues the run of that name when its config is the same apart from name and frames (another is refused unless force), decides from each stage's coverage and fingerprint which frames it computes (coverage_plan), and walks the selected stages in order. Recomputing frames clears the coverage of every stage that reads them, so a later invocation recomputes those too; a stage whose fingerprint moved is dropped when it begins (ADR-0035). A cached [bundle_adjustment] still hands its rig to the stages after it. A project is the unit of failure (ADR-0037): one that fails is reported and the run moves on.

ProjectOutcome dataclass

What happened for one project.

Attributes:

Name Type Description
project str

Its name in the invocation (project_names).
results Path

Its run's results file.
ran tuple[str, ...]

The stages that computed frames.
cached tuple[str, ...]

The stages with nothing to compute: the frames asked for were covered, under a matching fingerprint.
error str | None

The message of the failure that stopped it, or None.
Source code in src/deeperfly/pipeline/run.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@dataclass(frozen=True)
class ProjectOutcome:
    """What happened for one project.

    Attributes:
        project: Its name in the invocation
            ([project_names][deeperfly.project.project_names]).
        results: Its run's results file.
        ran: The stages that computed frames.
        cached: The stages with nothing to compute: the frames asked for were
            covered, under a matching fingerprint.
        error: The message of the failure that stopped it, or `None`.
    """

    project: str
    results: Path
    ran: tuple[str, ...] = ()
    cached: tuple[str, ...] = ()
    error: str | None = None

RunSummary dataclass

The whole invocation's outcome.

Attributes:

Name Type Description
out Path | None

The --out root, or None when results went into the projects.
stages tuple[str, ...]

The stages selected for this run.
outcomes tuple[ProjectOutcome, ...]

One entry per project, in run order.
name str

The run's name, shared by every project.
Source code in src/deeperfly/pipeline/run.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass(frozen=True)
class RunSummary:
    """The whole invocation's outcome.

    Attributes:
        out: The `--out` root, or `None` when results went into the projects.
        stages: The stages selected for this run.
        outcomes: One entry per project, in run order.
        name: The run's name, shared by every project.
    """

    out: Path | None
    stages: tuple[str, ...]
    outcomes: tuple[ProjectOutcome, ...]
    name: str = ""

    @property
    def failed(self) -> tuple[ProjectOutcome, ...]:
        """The projects that stopped on an error."""
        return tuple(o for o in self.outcomes if o.error is not None)

failed property 

failed: tuple[ProjectOutcome, ...]

The projects that stopped on an error.

run_projects 

run_projects(
    cfg: RunConfig,
    paths: Sequence[str | Path],
    *,
    only: Sequence[str] = (),
    skip: Sequence[str] = (),
    out: Path | None = None,
    force: bool = False,
    name: str | None = None,
    frames: Sequence[int | str] | None = None,
    log: Callable[[str], None] = print,
    observer: RunObserver | None = None,
) -> RunSummary

Apply a run config to the projects some paths stand for, and say what happened.

Each path is resolved as every command resolves one, a directory of videos getting the default project (resolve_projects). The run's name is name, else the config's, else the time of this call, shared by every project (run_name) and claimed in each, with -2, -3 when another invocation holds it (claim_run_name). frames stands in for the config's frames; out puts each project's files in <out>/<project name>/ (run_paths). force recomputes every frame asked for of every selected stage and lets another config take over a run of this name, dropping the stages whose fingerprint moved. A project that fails carries its message in the summary instead of raising. observer is told each step as it happens (RunObserver).

Raises:

Type Description
ConfigError

If the stage selection or the name fails, which is the invocation being wrong.
(FileNotFoundError, ValueError)

If a path stands for no project, or two paths for one; nothing has run then.
Source code in src/deeperfly/pipeline/run.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def run_projects(
    cfg: RunConfig,
    paths: Sequence[str | Path],
    *,
    only: Sequence[str] = (),
    skip: Sequence[str] = (),
    out: Path | None = None,
    force: bool = False,
    name: str | None = None,
    frames: Sequence[int | str] | None = None,
    log: Callable[[str], None] = print,
    observer: RunObserver | None = None,
) -> RunSummary:
    """Apply a run config to the projects some paths stand for, and say what happened.

    Each path is resolved as every command resolves one, a directory of videos
    getting the default project
    ([resolve_projects][deeperfly.pipeline.plan.resolve_projects]). The run's name is
    `name`, else the config's, else the time of this call, shared by every project
    ([run_name][deeperfly.pipeline.plan.run_name]) and claimed in each, with `-2`,
    `-3` when another invocation holds it
    ([claim_run_name][deeperfly.pipeline.plan.claim_run_name]). `frames` stands
    in for the config's `frames`; `out` puts each project's files in
    `<out>/<project name>/` ([run_paths][deeperfly.pipeline.plan.run_paths]). `force`
    recomputes every frame asked for of every selected stage and lets another config
    take over a run of this name, dropping the stages whose fingerprint moved. A
    project that fails carries its message in the summary instead of raising.
    `observer` is told each step as it happens
    ([RunObserver][deeperfly.pipeline.observe.RunObserver]).

    Raises:
        ConfigError: If the stage selection or the name fails, which is the
            invocation being wrong.
        FileNotFoundError, ValueError: If a path stands for no project, or two paths
            for one; nothing has run then.
    """
    stages = select_stages(cfg, tuple(only), tuple(skip))
    run = run_name(cfg, name)
    default = name is None and cfg.name is None
    fresh = {
        project.folder
        for _, project in resolve_projects(paths, create=False)
        if not project.manifest_path.is_file()
    }
    projects = resolve_projects(paths, create=True, log=log)
    detector = (
        DetectorCache(
            tuple(str(w) for w in cfg.pose2d.weights),
            locator=(None if cfg.pose2d.locator is None else str(cfg.pose2d.locator)),
            device=cfg.compute.device,
            precision=cfg.compute.precision,
        )
        if cfg.pose2d is not None and "pose2d" in stages
        else None
    )
    config_text = cfg.recorded_text()
    claimed: list[Path | OSError | None] = [None] * len(projects)
    if default:
        run, made = claim_run_name(run, projects, out, config_text)
        claimed = list(made)
    log(f"run {run}: {len(projects)} project(s), stages " + ", ".join(stages))
    if observer is not None:
        observer.run_started(run, [label for label, _ in projects], stages)

    outcomes = []
    for (label, project), claim in zip(projects, claimed, strict=True):
        outcome = _run_project(
            cfg,
            project,
            label,
            stages,
            run=run,
            out=out,
            frames=cfg.frames if frames is None else frames,
            config_text=config_text,
            detector=detector,
            force=force,
            fresh=project.folder in fresh,
            claimed=claim,
            log=log,
            observer=observer,
        )
        if observer is not None:
            observer.project_finished(label, outcome.error)
        outcomes.append(outcome)
    failed = [o for o in outcomes if o.error is not None]
    if failed:
        log(f"{len(failed)} of {len(outcomes)} project(s) failed:")
        for o in failed:
            log(f"  {o.project}: {o.error}")
    return RunSummary(out, tuple(stages), tuple(outcomes), run)

One function per run stage: read what is upstream, compute, write the arrays.

Each stage takes a Context (the run config, the project's recording and skeleton, its results file, the geometry in force and the frames the coverage plan hands it), reads its inputs at those frames by the references the schema declares and writes its group's arrays there; the driver owns the order, the coverage plan and the logging. A temporal stage (smoothing, correction, IK) runs once per stretch of its frames and writes each as it completes, so a stretch sees no frame outside it (ADR-0082); an array without T is rewritten by every stretch, the last one winning.

Context dataclass

What every stage is handed: the run config, the recording, and where to write.

Attributes:

Name Type Description
cfg RunConfig

The typed run config.
recording Recording

The project's recording: its videos and frame count.
skeleton Skeleton

The project's skeleton, the K axis.
results Results

Its open results file.
rig Rig | None

The geometry in force, the V axis: the run's initial calibration on the project's views, replaced by the [bundle_adjustment] stage for the stages after it; None for a run that reads no camera parameters (a pose2d-only run of one animal).
facts Facts

What the fingerprints hash besides the tables.
layout Layout

Which views each video carries (the project's).
labels Callable[[], Labels] | None

Loads the project's hand labels, raising a StageError that says why it cannot; None when none can be had.
detector DetectorCache | None

The shared detector handle, or None when [pose2d] is off.
log Callable[[str], None]

Where progress lines go.
todo Mapping[str, ndarray]

Per stage, the frames it computes in this invocation (the coverage plan, coverage_plan); a stage absent here computes every frame.
renders Path | None

The folder [render] writes into; None is beside the results.
force bool

Whether the invocation recomputes what it could reuse, such as the static crop an earlier invocation found.
observer RunObserver | None

Who watches the run, handed each detector pass; None for no one.
Source code in src/deeperfly/pipeline/stages.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
@dataclass
class Context:
    """What every stage is handed: the run config, the recording, and where to write.

    Attributes:
        cfg: The typed run config.
        recording: The project's recording: its videos and frame count.
        skeleton: The project's skeleton, the `K` axis.
        results: Its open results file.
        rig: The geometry in force, the `V` axis: the run's initial calibration on
            the project's views, replaced by the `[bundle_adjustment]` stage for the
            stages after it; `None` for a run that reads no camera parameters (a
            pose2d-only run of one animal).
        facts: What the fingerprints hash besides the tables.
        layout: Which views each video carries (the project's).
        labels: Loads the project's hand labels, raising a `StageError` that says
            why it cannot; `None` when none can be had.
        detector: The shared detector handle, or `None` when `[pose2d]` is off.
        log: Where progress lines go.
        todo: Per stage, the frames it computes in this invocation (the coverage
            plan, [coverage_plan][deeperfly.pipeline.plan.coverage_plan]); a stage
            absent here computes every frame.
        renders: The folder `[render]` writes into; `None` is beside the results.
        force: Whether the invocation recomputes what it could reuse, such as the
            static crop an earlier invocation found.
        observer: Who watches the run, handed each detector pass; `None` for no one.
    """

    cfg: RunConfig
    recording: Recording
    skeleton: Skeleton
    results: Results
    rig: Rig | None
    facts: Facts
    layout: Layout = field(default_factory=Layout)
    labels: Callable[[], Labels] | None = None
    detector: DetectorCache | None = None
    log: Callable[[str], None] = print
    todo: Mapping[str, np.ndarray] = field(default_factory=dict)
    renders: Path | None = None
    force: bool = False
    observer: RunObserver | None = None

    def frames(self, stage: str) -> np.ndarray:
        """The frames `stage` computes here, increasing absolute indices."""
        if stage in self.todo:
            return np.asarray(self.todo[stage], dtype=np.int64)
        return np.arange(self.results.n_frames)

    @property
    def animals(self) -> int:
        """How many animals the project holds, the `N` axis."""
        return self.results.animals

    def need_rig(self) -> Rig:
        """The geometry in force, for a stage that reads it.

        Raises:
            StageError: If the run reads no camera parameters, which
                [prepare][deeperfly.pipeline.plan.prepare] rules out for every
                stage that needs them.
        """
        if self.rig is None:
            raise StageError(
                f"{self.recording.name}: this stage reads camera parameters, and the "
                "run starts from none; add [calibration] to the config"
            )
        return self.rig

    @property
    def keypoints(self) -> tuple[str, ...]:
        """The skeleton's point names, the `K` axis order."""
        return tuple(self.skeleton.point_names)

    def frame_size(self, view: str) -> tuple[int, int]:
        """The `(height, width)` a view's pixels live in: its video's frame."""
        return self.recording.videos[self.layout.video_of(view)].size

    def read(self, ref: str, frames: np.ndarray | None = None) -> np.ndarray:
        """Resolve a data reference: a stored stage array, or the hand labels.

        With `frames`, only those frames, which the stage that wrote the array must
        cover; without, the whole array.

        Raises:
            StageError: If no stage has written the array (one left out with `--skip`),
                or it does not cover `frames`.
        """
        if ref.startswith("labels/"):
            arr = self._labels(ref)
            return arr if frames is None else np.take(arr, frames, axis=_LABEL_T[ref])
        stage = ref.split("/")[0]
        if not self.results.has(ref):
            raise StageError(
                f"{self.recording.name}: {ref!r} is not in {self.results.path.name}; "
                f"run the [{stage}] stage (it was skipped, or its results were "
                "removed)"
            )
        if frames is None:
            return self.results.read(ref)
        covered = self.results.covered(stage)[frames]
        if not covered.all():
            raise StageError(
                f"{self.recording.name}: [{stage}] covers {int(covered.sum())} of the "
                f"{covered.size} frames asked for; run [{stage}] on them too"
            )
        return self.results.read_frames(ref, frames)

    def read_animal(
        self, ref: str, n: int, frames: np.ndarray | None = None
    ) -> np.ndarray:
        """`read`, with the animal axis peeled off: one animal's copy of an array.

        An array about the recording rather than an animal, and the hand labels, come
        back whole.
        """
        arr = self.read(ref, frames)
        if ref.startswith("labels/"):
            return arr
        return arr[n] if self.results.axes(ref)[:1] == ("N",) else arr

    def _labels(self, ref: str) -> np.ndarray:
        """The hand labels as the array `ref` names, in the run's view order."""
        if self.labels is None:
            raise StageError(f"{self.recording.name}: no hand labels to read")
        labels = self.labels()
        frames = labels.labeled_frames()
        views = list(labels.view_names)
        missing = [v for v in self.results.views if v not in views]
        if missing:
            raise StageError(
                f"{self.recording.name}: the labels have no view named "
                + ", ".join(missing)
                + " (they know "
                + ", ".join(views)
                + ")"
            )
        order = [views.index(v) for v in self.results.views]
        n_v, n_k = len(order), labels.n_keypoints
        n_t = self.results.n_frames
        if ref == "labels/points":
            out = np.full((n_v, n_t, n_k, 2), np.nan)
            for t in frames:
                out[:, t] = labels.frame_effective(t)[order]
            return out
        if ref == "labels/hidden":
            out = np.zeros((n_v, n_t, n_k), dtype=bool)
            for t in frames:
                out[:, t] = labels.hidden_frame(t)[order]
            return out
        if ref == "labels/absent":
            out = np.zeros((n_t, n_k), dtype=bool)
            for t in frames:
                out[t] = labels.absent_frame(t)
            return out
        raise StageError(f"{ref!r} is not a label array")

    def points(self, selector: tuple[str, ...], where: str) -> list[int]:
        """The keypoint indices a point selector claims, in skeleton order."""
        names = self.keypoints
        chosen = match_names(selector, names, where)
        return [names.index(n) for n in chosen]

animals property 

animals: int

How many animals the project holds, the N axis.

keypoints property 

keypoints: tuple[str, ...]

The skeleton's point names, the K axis order.

frame_size 

frame_size(view: str) -> tuple[int, int]

The (height, width) a view's pixels live in: its video's frame.

Source code in src/deeperfly/pipeline/stages.py 
180
181
182
def frame_size(self, view: str) -> tuple[int, int]:
    """The `(height, width)` a view's pixels live in: its video's frame."""
    return self.recording.videos[self.layout.video_of(view)].size

frames 

frames(stage: str) -> np.ndarray

The frames stage computes here, increasing absolute indices.

Source code in src/deeperfly/pipeline/stages.py 
149
150
151
152
153
def frames(self, stage: str) -> np.ndarray:
    """The frames `stage` computes here, increasing absolute indices."""
    if stage in self.todo:
        return np.asarray(self.todo[stage], dtype=np.int64)
    return np.arange(self.results.n_frames)

need_rig 

need_rig() -> Rig

The geometry in force, for a stage that reads it.

Raises:

Type Description
StageError

If the run reads no camera parameters, which prepare rules out for every stage that needs them.
Source code in src/deeperfly/pipeline/stages.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def need_rig(self) -> Rig:
    """The geometry in force, for a stage that reads it.

    Raises:
        StageError: If the run reads no camera parameters, which
            [prepare][deeperfly.pipeline.plan.prepare] rules out for every
            stage that needs them.
    """
    if self.rig is None:
        raise StageError(
            f"{self.recording.name}: this stage reads camera parameters, and the "
            "run starts from none; add [calibration] to the config"
        )
    return self.rig

points 

points(selector: tuple[str, ...], where: str) -> list[int]

The keypoint indices a point selector claims, in skeleton order.

Source code in src/deeperfly/pipeline/stages.py 
263
264
265
266
267
def points(self, selector: tuple[str, ...], where: str) -> list[int]:
    """The keypoint indices a point selector claims, in skeleton order."""
    names = self.keypoints
    chosen = match_names(selector, names, where)
    return [names.index(n) for n in chosen]

read 

read(ref: str, frames: ndarray | None = None) -> np.ndarray

Resolve a data reference: a stored stage array, or the hand labels.

With frames, only those frames, which the stage that wrote the array must cover; without, the whole array.

Raises:

Type Description
StageError

If no stage has written the array (one left out with --skip), or it does not cover frames.
Source code in src/deeperfly/pipeline/stages.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def read(self, ref: str, frames: np.ndarray | None = None) -> np.ndarray:
    """Resolve a data reference: a stored stage array, or the hand labels.

    With `frames`, only those frames, which the stage that wrote the array must
    cover; without, the whole array.

    Raises:
        StageError: If no stage has written the array (one left out with `--skip`),
            or it does not cover `frames`.
    """
    if ref.startswith("labels/"):
        arr = self._labels(ref)
        return arr if frames is None else np.take(arr, frames, axis=_LABEL_T[ref])
    stage = ref.split("/")[0]
    if not self.results.has(ref):
        raise StageError(
            f"{self.recording.name}: {ref!r} is not in {self.results.path.name}; "
            f"run the [{stage}] stage (it was skipped, or its results were "
            "removed)"
        )
    if frames is None:
        return self.results.read(ref)
    covered = self.results.covered(stage)[frames]
    if not covered.all():
        raise StageError(
            f"{self.recording.name}: [{stage}] covers {int(covered.sum())} of the "
            f"{covered.size} frames asked for; run [{stage}] on them too"
        )
    return self.results.read_frames(ref, frames)

read_animal 

read_animal(
    ref: str, n: int, frames: ndarray | None = None
) -> np.ndarray

read, with the animal axis peeled off: one animal's copy of an array.

An array about the recording rather than an animal, and the hand labels, come back whole.

Source code in src/deeperfly/pipeline/stages.py 
214
215
216
217
218
219
220
221
222
223
224
225
def read_animal(
    self, ref: str, n: int, frames: np.ndarray | None = None
) -> np.ndarray:
    """`read`, with the animal axis peeled off: one animal's copy of an array.

    An array about the recording rather than an animal, and the hand labels, come
    back whole.
    """
    arr = self.read(ref, frames)
    if ref.startswith("labels/"):
        return arr
    return arr[n] if self.results.axes(ref)[:1] == ("N",) else arr

DetectorCache dataclass

The run's nets, loaded once, shared by every project; torch on first use.

Source code in src/deeperfly/pipeline/stages.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass
class DetectorCache:
    """The run's nets, loaded once, shared by every project; torch on first use."""

    weights: tuple[str, ...]
    locator: str | None = None
    device: str | None = None
    precision: str = "auto"
    _loaded: tuple | None = field(default=None, repr=False)
    _locator: tuple | None = field(default=None, repr=False)

    def get(self) -> tuple:
        """`(models, device)` for `predict_recording`, loading on first call."""
        if self._loaded is None:
            from ..pose2d.models import load_models

            self._loaded = load_models(
                list(self.weights), device=self.device, precision=self.precision
            )
        return self._loaded

    @property
    def loaded_device(self) -> str | None:
        """The device the nets were loaded on, or `None` before the first `get`."""
        return None if self._loaded is None else str(self._loaded[1])

    def get_locator(self):
        """The run's full-frame locator, loaded once; `None` when it names none."""
        if self.locator is None:
            return None
        if self._locator is None:
            from ..pose2d.models import load_locator

            self._locator = load_locator(
                self.locator, device=self.device, precision=self.precision
            )
        return self._locator[0]

loaded_device property 

loaded_device: str | None

The device the nets were loaded on, or None before the first get.

get 

get() -> tuple

(models, device) for predict_recording, loading on first call.

Source code in src/deeperfly/pipeline/stages.py 
78
79
80
81
82
83
84
85
86
def get(self) -> tuple:
    """`(models, device)` for `predict_recording`, loading on first call."""
    if self._loaded is None:
        from ..pose2d.models import load_models

        self._loaded = load_models(
            list(self.weights), device=self.device, precision=self.precision
        )
    return self._loaded

get_locator 

get_locator()

The run's full-frame locator, loaded once; None when it names none.

Source code in src/deeperfly/pipeline/stages.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_locator(self):
    """The run's full-frame locator, loaded once; `None` when it names none."""
    if self.locator is None:
        return None
    if self._locator is None:
        from ..pose2d.models import load_locator

        self._locator = load_locator(
            self.locator, device=self.device, precision=self.precision
        )
    return self._locator[0]

StageError

Bases: RuntimeError

A stage cannot run: an unbuilt path, or an input nothing has written.

Source code in src/deeperfly/pipeline/stages.py 
63
64
class StageError(RuntimeError):
    """A stage cannot run: an unbuilt path, or an input nothing has written."""

predict_options 

predict_options(
    cfg: RunConfig,
    skeleton: Skeleton,
    animals: int,
    *,
    batch: int | None = None,
) -> PredictOptions

The config's [pose2d] knobs as the one record pose2d.predict reads.

animals is the project's; batch stands in for [compute] batch, and each frame runs once per animal.

Source code in src/deeperfly/pipeline/stages.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def predict_options(
    cfg: RunConfig, skeleton: Skeleton, animals: int, *, batch: int | None = None
) -> PredictOptions:
    """The config's `[pose2d]` knobs as the one record `pose2d.predict` reads.

    `animals` is the project's; `batch` stands in for `[compute] batch`, and each
    frame runs once per animal.
    """
    from ..pose2d.options import PredictOptions, Tracking
    from ..pose2d.predict import flip_permutation

    stage = cfg.pose2d
    assert stage is not None
    return PredictOptions(
        window=stage.window,
        animals=animals,
        tracking=Tracking(**stage.tracking.model_dump()),
        crop_aspect={view: float(a) for view, a in stage.crop_aspect.items()},
        combine=stage.combine,
        candidates=stage.candidates,
        tightness=stage.tightness,
        batch_frames=max(1, (cfg.compute.batch if batch is None else batch) // animals),
        flip_perm=flip_permutation(skeleton, stage.tta),
        point_names=skeleton.point_names,
        compile=cfg.compute.compile,
    )

resolved_video 

resolved_video(render, name: str) -> dict[str, Any]

One video's settings with [render]'s defaults filled in, layers as records.

The per-video overrides let one video render 20 frames while the others keep every frame.

Source code in src/deeperfly/pipeline/stages.py 
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
def resolved_video(render, name: str) -> dict[str, Any]:
    """One video's settings with `[render]`'s defaults filled in, layers as records.

    The per-video overrides let one video render 20 frames while the others keep
    every frame.
    """
    from ..render import Layer

    video = render.outputs[name]
    out: dict[str, Any] = {
        "grid": video.grid,
        "layers": tuple(Layer(**layer.model_dump()) for layer in video.layers),
        "crop": video.crop,
    }
    for key in _INHERITED:
        own = getattr(video, key)
        out[key] = own if own not in (None, ()) else getattr(render, key)
    out["codec"] = render.codec
    out["crf"] = render.crf
    return out

run_ball 

run_ball(ctx: Context) -> None

[ball]: track the treadmill ball -> ball/*, in the world frame.

The ball's radius is fitted once to its outline in the [ball] outline views, on frames spread over the opening contact_frames the 3D covers, starting from a guess from the leg tips that rest on it, which also orient the animal on it. Each stretch then has its center filtered and smoothed over every frame from the outlines, and its rotation tracked by spintrack in [ball] view from where the outline puts the ball at its start; rotation restarts at zero with every stretch, and a frame standing alone is refused (ADR-0144).

Source code in src/deeperfly/pipeline/stages.py 
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
def run_ball(ctx: Context) -> None:
    """`[ball]`: track the treadmill ball -> `ball/*`, in the world frame.

    The ball's radius is fitted once to its outline in the `[ball] outline` views, on
    frames spread over the opening `contact_frames` the 3D covers, starting from a
    guess from the leg tips that rest on it, which also orient the animal on it. Each
    stretch then has its center filtered and smoothed over every frame from the
    outlines, and its rotation tracked by spintrack in `[ball] view` from where the
    outline puts the ball at its start; `rotation` restarts at zero with every
    stretch, and a frame standing alone is refused (ADR-0144).
    """
    from ..ball import (
        MissingSpintrack,
        require_spintrack,
        rim_points,
        track_ball,
        track_outline,
    )

    try:
        require_spintrack()
    except MissingSpintrack as exc:
        raise StageError(f"[ball]: {exc}") from exc
    from spintrack.detect import DetectionError

    stage = ctx.cfg.ball
    assert stage is not None
    rig = ctx.need_rig()
    view = rig.views[stage.view]
    if view.mirrored:
        raise StageError(
            f"[ball] view = {stage.view!r} is mirrored; spintrack needs a camera that "
            "sees the ball directly"
        )
    names = match_names(stage.outline, rig.names, "[ball] outline")
    outlines = {name: rig.views[name] for name in names if not rig.views[name].mirrored}
    video = ctx.recording.videos[ctx.layout.video_of(stage.view)]
    spans = _spans(ctx, "ball")
    alone = next((span for span in spans if span.size == 1), None)
    if alone is not None:
        raise StageError(
            f"{ctx.recording.name}: [ball] tracks consecutive frames, and frame "
            f"{int(alone[0])} stands alone; give it stretches (frames = "
            '["0:1000"], not a step)'
        )
    try:
        contact = _ball_contact(ctx, view, video, outlines)
    except DetectionError as exc:
        raise StageError(
            f"{ctx.recording.name}: [ball] cannot find the ball in view "
            f"{stage.view}: {exc}"
        ) from exc
    center, radius = contact.center, contact.radius
    for i, span in enumerate(spans):
        _announce(ctx, spans, i)
        start, stop = int(span[0]), int(span[-1]) + 1
        # Where the ball was, give or take: the stretch's first frame searches wide.
        cov = (BALL_START_RADII * radius) ** 2 * np.eye(3)
        with ExitStack() as stack:
            frames = _ball_frames(ctx, outlines, start, stop, stack)
            track = track_outline(outlines, frames, center, cov, radius)
        centers, covs = track.smoothed()
        arrays, columns, quality = track_ball(
            view,
            video,
            start,
            stop,
            contact.axes,
            rim_points(view, centers[0], radius),
            tracking=stage.tracking,
            two_pass=stage.two_pass,
        )
        sd = np.sqrt(np.diagonal(covs, axis1=1, axis2=2))
        arrays.update(
            center=centers,
            center_sd=sd,
            radius=np.asarray(radius, dtype=np.float64),
            axes=np.asarray(contact.axes, dtype=np.float64),
        )
        ctx.results.write_frames("ball", arrays, span, names={"tracks": columns})
        center = centers[-1]
        ctx.log(
            f"    tracked {quality.n_tracked}/{quality.n_frames} frames in view "
            f"{stage.view}, the ball turning {quality.turned_deg:.0f} deg in all; its "
            f"outline measured on {int(track.seen.sum())}, its center known to "
            f"{np.median(sd.max(axis=1)) / radius:.4f} radii (median of the worst "
            f"axis), {sd.max() / radius:.4f} at worst"
        )

run_bundle_adjustment 

run_bundle_adjustment(ctx: Context) -> None

[bundle_adjustment]: refine the run's calibration on its own 2D.

The sampled detector points or hand labels are fitted from the rig in force, which the solved rig replaces for the stages after it and is stored as bundle_adjustment/views. Animal 0 only: the rig is the project's, not an animal's.

Source code in src/deeperfly/pipeline/stages.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
def run_bundle_adjustment(ctx: Context) -> None:
    """`[bundle_adjustment]`: refine the run's calibration on its own 2D.

    The sampled detector points or hand labels are fitted from the rig in force, which
    the solved rig replaces for the stages after it and is stored as
    `bundle_adjustment/views`. Animal 0 only: the rig is the project's, not an
    animal's.
    """
    from ..calibration import refine_bundle

    stage = ctx.cfg.bundle_adjustment
    assert stage is not None
    stamp = fingerprint(ctx.cfg, "bundle_adjustment", ctx.facts)
    pts2d = np.asarray(ctx.read_animal(stage.source, 0), dtype=float)
    conf = None
    if stage.weighted:
        conf = np.asarray(ctx.read_animal("pose2d/conf", 0), dtype=float)
    cols = ctx.points(stage.points, "[bundle_adjustment].points")
    frames = _sample_frames(pts2d[:, :, cols], stage.frames)
    if frames.size == 0:
        raise StageError(
            f"{ctx.recording.name}: no frame has a selected point in two views, so "
            f"there is nothing to solve ([bundle_adjustment] source = {stage.source!r})"
        )
    obs = pts2d[:, frames][:, :, cols].reshape(len(ctx.results.views), -1, 2)
    weights = (
        None
        if conf is None
        else conf[:, frames][:, :, cols].reshape(len(ctx.results.views), -1)
    )
    rig = ctx.need_rig()
    before = reprojection_error(rig, triangulate(rig, obs), obs)
    refined = refine_bundle(
        rig,
        obs,
        weights,
        fixed=stage.fixed_refs(rig.names),
        shared=[list(g) for g in stage.shared],
        loss=stage.loss,
        threshold_px=stage.threshold_px,
        max_nfev=stage.max_nfev,
    )
    after = reprojection_error(refined, triangulate(refined, obs), obs)
    ctx.rig = refined

    ctx.results.write_stage(
        "bundle_adjustment",
        {
            "reproj_before": _scatter(before, frames, cols, ctx),
            "reproj_after": _scatter(after, frames, cols, ctx),
        },
        fingerprint=stamp,
        rig=refined,
    )
    ctx.log(
        f"    {frames.size} frames, {np.nanmedian(before):.2f} -> "
        f"{np.nanmedian(after):.2f} px median"
    )

run_correction 

run_correction(ctx: Context) -> None

[correction]: hold the static points still, then symmetrize -> points.

In that order: with one plane per stretch the symmetrization is affine, so a point already made static stays static through it. One stretch at a time, each with its own static positions and plane.

Source code in src/deeperfly/pipeline/stages.py 
747
748
749
750
751
752
753
754
755
756
757
758
def run_correction(ctx: Context) -> None:
    """`[correction]`: hold the static points still, then symmetrize -> `points`.

    In that order: with one plane per stretch the symmetrization is affine, so a
    point already made static stays static through it. One stretch at a time, each
    with its own static positions and plane.
    """
    spans = _spans(ctx, "correction")
    for i, span in enumerate(spans):
        _announce(ctx, spans, i)
        arrays = _per_animal(ctx, _correction_one, span)
        ctx.results.write_frames("correction", arrays, span)

run_ik 

run_ik(ctx: Context) -> None

[ik]: fit the body model to the 3D -> angles, points, the chain fit.

points is the fitted joints in the skeleton's point order, so it reprojects like any 3D track; chain_scale and chain_offset size and place the head and abdomen (ADR-0066), which lets a mesh layer draw the fit rather than the model. One stretch at a time, the last fitted one's chains kept; a stretch no animal has any 3D in is written as NaN once another has given the arrays their shape.

Source code in src/deeperfly/pipeline/stages.py 
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
def run_ik(ctx: Context) -> None:
    """`[ik]`: fit the body model to the 3D -> `angles`, `points`, the chain fit.

    `points` is the fitted joints in the skeleton's point order, so it reprojects like
    any 3D track; `chain_scale` and `chain_offset` size and place the head and abdomen
    (ADR-0066), which lets a `mesh` layer draw the fit rather than the model. One
    stretch at a time, the last fitted one's chains kept; a stretch no animal has any
    3D in is written as NaN once another has given the arrays their shape.
    """
    shape: dict[str, np.ndarray] | None = None
    unfitted: list[np.ndarray] = []
    spans = _spans(ctx, "ik")
    for i, span in enumerate(spans):
        _announce(ctx, spans, i)
        parts: list[dict[str, np.ndarray] | None] = []
        names: dict[str, list[str]] = {}
        for n in range(ctx.animals):
            if ctx.animals > 1:
                ctx.log(f"    animal {n}")
            arrays, got = _ik_one(ctx, n, span)
            parts.append(arrays)
            names = got or names
        fitted = next((p for p in parts if p is not None), None)
        if fitted is None:
            unfitted.append(span)
            continue
        # An animal that was never tracked takes the shape of one that was.
        blank = {name: np.full_like(arr, np.nan) for name, arr in fitted.items()}
        filled = [blank if p is None else p for p in parts]
        stacked = {name: stack_animals([p[name] for p in filled]) for name in fitted}
        ctx.results.write_frames("ik", stacked, span, names=names)
        shape = shape or stacked
    if shape is None:
        raise StageError(
            f"{ctx.recording.name}: no animal has a 3D pose for [ik] to fit"
        )
    for span in unfitted:
        blank = {
            name: np.full((arr.shape[0], span.size, *arr.shape[2:]), np.nan)
            for name, arr in shape.items()
            if "T" in ctx.results.axes(f"ik/{name}")
        }
        ctx.results.write_frames("ik", blank, span)

run_pose2d 

run_pose2d(ctx: Context) -> None

[pose2d]: the detector over the run's frames -> pose2d/points, conf, ...

One write per stretch of the frames the coverage plan hands it (ADR-0082), each covering what it wrote as it completes, so a pass that fails keeps the stretches it finished; a tracker reading a locator writes a long stretch in parts as it goes, the tracker carrying on across them (ADR-0139). The static crop comes from the config, else from the frames this run already covers, else is discovered on the frames asked for (always, with --force; pose2d.predict); an ensemble collapses inside the forward pass. Every array leads with the animal, and a per-frame window above one animal also writes the tracker's centers, track_flags and center_candidates. crops holds what each animal ran through, per view and frame. Only the tracker reads the rig.

Source code in src/deeperfly/pipeline/stages.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def run_pose2d(ctx: Context) -> None:
    """`[pose2d]`: the detector over the run's frames -> `pose2d/points`, `conf`, ...

    One write per stretch of the frames the coverage plan hands it (ADR-0082), each
    covering what it wrote as it completes, so a pass that fails keeps the stretches
    it finished; a tracker reading a locator writes a long stretch in parts as it
    goes, the tracker carrying on across them (ADR-0139). The static crop comes from the
    config, else from the frames this run already covers, else is discovered on the
    frames asked for (always, with `--force`;
    [pose2d.predict][deeperfly.pose2d.predict]); an ensemble collapses inside the
    forward pass. Every array leads with the animal, and a per-frame window above one
    animal also writes the tracker's `centers`, `track_flags` and `center_candidates`.
    `crops` holds what each animal ran through, per view and frame. Only the tracker
    reads the rig.
    """
    stage = ctx.cfg.pose2d
    assert stage is not None and ctx.detector is not None
    from ..pose2d.crops import crop_boxes
    from ..pose2d.models import first_model, model_point_names
    from ..pose2d.predict import predict_recording

    models, device = ctx.detector.get()
    trained = model_point_names(models)
    if trained and tuple(trained) != ctx.keypoints:
        raise StageError(
            f"{ctx.recording.name}: the detector was trained on {len(trained)} "
            f"points, the project's skeleton names {len(ctx.keypoints)}"
            + (
                ""
                if len(trained) != len(ctx.keypoints)
                else " in another order or by other names"
            )
            + "; the project's skeleton must be the one it was trained with"
        )
    try:
        locator = ctx.detector.get_locator()
    except ValueError as exc:
        raise StageError(f"{ctx.recording.name}: [pose2d] locator: {exc}") from exc
    if ctx.animals > 1 and locator is None and not first_model(models).center_channel:
        raise StageError(
            f"{ctx.recording.name}: animals = {ctx.animals} needs something that "
            "locates animals: a detector trained with `[data] center_channel = true`, "
            "or a `[pose2d] locator` trained on several animals"
        )
    options = predict_options(ctx.cfg, ctx.skeleton, ctx.animals)
    frames = ctx.frames("pose2d")
    _log_seams(ctx, "pose2d", frames)
    seen = cells = done = 0
    tracked = np.zeros(options.animals)
    from ..pose2d.runtime import accumulation

    try:
        fixed = None
        if stage.window != "per_frame" and stage.crops:
            fixed = crop_boxes(dict(stage.crops), ctx.results.views, options.animals)
        elif stage.window != "per_frame" and not ctx.force:
            fixed = _stored_static_crop(ctx)
        with accumulation(ctx.cfg.compute.accumulate):
            for part in predict_recording(
                models,
                ctx.recording,
                views=ctx.results.views,
                layout=ctx.layout,
                frames=frames,
                crops=fixed,
                rig=ctx.rig,
                device=device,
                options=options,
                locator=locator,
                log=ctx.log,
                on_block=None if ctx.observer is None else ctx.observer.pose2d_block,
            ):
                ctx.results.write_frames("pose2d", _pose2d_arrays(part), part.frames)
                ok = np.isfinite(part.points).all(axis=-1)
                seen, cells = seen + int(ok.sum()), cells + ok.size
                done += part.stop - part.start
                if part.track_flags is not None:
                    tracked += (part.track_flags != LOST).sum(axis=1)
    except ValueError as exc:
        raise StageError(f"{ctx.recording.name}: {exc}") from exc
    ctx.log(f"    {done} frames, {seen / max(cells, 1):.1%} cells predicted")
    if options.window == "per_frame" and options.animals > 1:
        shown = ", ".join(
            f"animal {n} {f / max(done, 1):.1%}" for n, f in enumerate(tracked)
        )
        ctx.log(f"    frames with a track: {shown}")

run_pose3d 

run_pose3d(ctx: Context) -> None

[pose3d]: lift the 2D to one 3D per keypoint -> points, reproj_error.

inliers are the views each point's consensus kept and support their summed vote (consensus_support), what tells a point to trust from one to doubt (ADR-0132). A point whose kept views all look along one line, such as a ventral and a dorsal view alone, is NaN with its inliers kept: those views fix it across the line and nowhere along it (placed_in_depth, ADR-0135). "pictorial" and "election" read the candidate sets and also write selected/points and selected/conf, the candidate each view settled on, which their reproj_error is against (pictorial). A veto keeps its views out of every triangulation. One animal at a time.

Source code in src/deeperfly/pipeline/stages.py 
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
def run_pose3d(ctx: Context) -> None:
    """`[pose3d]`: lift the 2D to one 3D per keypoint -> `points`, `reproj_error`.

    `inliers` are the views each point's consensus kept and `support` their summed
    vote ([consensus_support][deeperfly.pose3d.triangulation.consensus_support]), what
    tells a point to trust from one to doubt (ADR-0132). A point whose kept views all
    look along one line, such as a ventral and a dorsal view alone, is NaN with its
    `inliers` kept: those views fix it across the line and nowhere along it
    ([placed_in_depth][deeperfly.pose3d.triangulation.placed_in_depth], ADR-0135).
    `"pictorial"` and `"election"` read the candidate sets and also write
    `selected/points` and `selected/conf`, the candidate each view settled on, which
    their `reproj_error` is against ([pictorial][deeperfly.pose3d.pictorial]). A `veto`
    keeps its views out of every triangulation. One animal at a time.
    """
    frames = ctx.frames("pose3d")
    arrays = _per_animal(ctx, _pose3d_one, frames)
    ctx.results.write_frames("pose3d", arrays, frames, rig=ctx.need_rig())

run_render 

run_render(ctx: Context) -> None

[render]: each [render.outputs.<name>] to renders/<run>/<name>.mp4.

A video cannot be patched, so every output is drawn anew over the frames asked for that its range keeps (render_video), and the stage's group, which holds no array, then covers exactly the frames asked for: a rerun asking for no more, with nothing upstream recomputed, skips the encode.

Source code in src/deeperfly/pipeline/stages.py 
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
def run_render(ctx: Context) -> None:
    """`[render]`: each `[render.outputs.<name>]` to `renders/<run>/<name>.mp4`.

    A video cannot be patched, so every output is drawn anew over the frames asked
    for that its range keeps ([render_video][deeperfly.render.render_video]), and the
    stage's group, which holds no array, then covers exactly the frames asked for:
    a rerun asking for no more, with nothing upstream recomputed, skips the encode.
    """
    from ..render import frames_in_range, render_video

    stage = ctx.cfg.render
    assert stage is not None
    out_dir = ctx.results.path.parent if ctx.renders is None else ctx.renders
    out_dir.mkdir(parents=True, exist_ok=True)
    todo = ctx.frames("render")
    rig = ctx.need_rig()
    here = rig.names
    mesh = _overlay_mesh(ctx, stage)
    # One reader per video, shared by its views' panels and its own.
    readers: dict = {}
    try:
        for name, output in stage.outputs.items():
            for panel in output.names:
                video = ctx.layout.video_of(panel) if panel in here else panel
                if video in ctx.recording.videos and video not in readers:
                    readers[video] = ctx.recording.videos[video].open()
            fps = next((r.fps() for r in readers.values() if r.fps()), None)
            resolved = resolved_video(stage, name)
            frames = frames_in_range(resolved["frames"], todo)
            if frames.size == 0:
                ctx.log(f"    {name}: no frame asked for is in its range; not written")
                continue
            try:
                render_video(
                    resolved,
                    out=out_dir / f"{name}.mp4",
                    project=ctx.recording.name,
                    videos=readers,
                    frames=frames,
                    fps=fps,
                    rig=rig,
                    skeleton=ctx.skeleton,
                    views=synthetic_views(stage),
                    read=ctx.read,
                    names=ctx.results.names,
                    axes=ctx.results.axes,
                    mesh=mesh,
                    layout=ctx.layout,
                    log=ctx.log,
                )
            except ValueError as exc:
                raise StageError(f"[render.outputs.{name}]: {exc}") from exc
    finally:
        for reader in readers.values():
            reader.close()
    ctx.results.uncover("render", np.arange(ctx.results.n_frames))
    ctx.results.write_frames("render", {}, todo)

run_smoothing 

run_smoothing(ctx: Context) -> None

[smoothing]: the geometric EKS over the 2D -> points, var, smooth_param.

The smoother observes the 2D through the rig (ADR-0062): the decode, or what the pictorial method re-selected; [pose3d] only initializes it. Where several animals share the frame it observes the views [pose3d]'s consensus kept and no others: a view that left the consensus there mostly saw the other animal (ADR-0133). One animal at a time, which keeps the animal axis off the smoother's own ensemble axis. One stretch at a time, smooth_param fitted on each and the last one's kept.

Source code in src/deeperfly/pipeline/stages.py 
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
def run_smoothing(ctx: Context) -> None:
    """`[smoothing]`: the geometric EKS over the 2D -> `points`, `var`, `smooth_param`.

    The smoother observes the 2D through the rig (ADR-0062): the decode, or what the
    pictorial method re-selected; `[pose3d]` only initializes it. Where several
    animals share the frame it observes the views `[pose3d]`'s consensus kept and no
    others: a view that left the consensus there mostly saw the other animal
    (ADR-0133). One animal at a time, which keeps the animal axis off the smoother's
    own ensemble axis. One stretch at a time, `smooth_param` fitted on each and the
    last one's kept.
    """
    spans = _spans(ctx, "smoothing")
    for i, span in enumerate(spans):
        _announce(ctx, spans, i)
        arrays = _per_animal(ctx, _smoothing_one, span)
        ctx.results.write_frames("smoothing", arrays, span)

stack_animals 

stack_animals(parts: Sequence[ndarray]) -> np.ndarray

One array per animal, stacked on the leading N axis the schema declares.

Source code in src/deeperfly/pipeline/stages.py 
270
271
272
def stack_animals(parts: Sequence[np.ndarray]) -> np.ndarray:
    """One array per animal, stacked on the leading `N` axis the schema declares."""
    return np.stack(parts)

synthetic_views 

synthetic_views(render) -> dict[str, SyntheticView]

[render.views] as the renderer's own records.

Source code in src/deeperfly/pipeline/stages.py 
1184
1185
1186
1187
1188
def synthetic_views(render) -> dict[str, SyntheticView]:
    """`[render.views]` as the renderer's own records."""
    return {
        name: SyntheticView(**view.model_dump()) for name, view in render.views.items()
    }