live

Watching poses as they are computed (ADR-0126).

feed holds what a producer publishes, with no transport; server serves a feed to a browser page; run is deeperfly run's producer and status. Nothing is imported here, so a producer outside the package can take the feed alone.

Poses as they are computed, held for a reader that may or may not be watching.

A producer, such as deeperfly run's detector passes or a tracker on a rig that is recording, publishes moments: a frame index, and per animal and view the keypoints in video pixels, their confidences and the window they were read through. A reader takes JSON-ready payloads off the feed, so any channel can carry them. A feed nobody watches counts what it is given and keeps nothing (ADR-0126).

Nothing here knows about a transport, a pipeline or a model, and nothing of the package is imported: plain arrays in, plain dicts out.

PoseFeed

Moments a producer publishes, kept only while a reader is watching.

A reader calls watched on every look; the feed is active for IDLE_S after the last one, or always when forced. While it is not, publish counts the moments and returns. While it is, the last history moments go into a ring, the per-view confidence is bucketed into a trend, and every so often the last frame's images are copied for a reader to take (capture). A new capture waits until the previous one has been taken, at most MAX_CAPTURE_HZ, and comes regardless after 1 / MIN_CAPTURE_HZ seconds.

Every method is safe to call from any thread.

Source code in src/deeperfly/live/feed.py

class PoseFeed:
    """Moments a producer publishes, kept only while a reader is watching.

    A reader calls `watched` on every look; the feed is `active` for `IDLE_S` after
    the last one, or always when `forced`. While it is not, `publish` counts the
    moments and returns. While it is, the last `history` moments go into a ring, the
    per-view confidence is bucketed into a trend, and every so often the last frame's
    images are copied for a reader to take (`capture`). A new capture waits until the
    previous one has been taken, at most `MAX_CAPTURE_HZ`, and comes regardless after
    `1 / MIN_CAPTURE_HZ` seconds.

    Every method is safe to call from any thread.
    """

    def __init__(
        self,
        *,
        forced: bool = False,
        history: int = HISTORY,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """An empty feed, `forced` to keep what it is given whether watched or not.

        `history` is how many moments it holds; `clock` the seconds it measures by.
        """
        self.forced = forced
        self._size = history
        self._clock = clock
        self._lock = threading.Lock()
        self._looked = -np.inf
        # Bumped whenever the parts a reader fetches only on change move: the meta,
        # the 3D, the errors.
        self.version = 0
        self.meta: dict | None = None
        self._reset()

    def _reset(self) -> None:
        self.published = 0
        self.last_frame: int | None = None
        self._samples: deque[tuple[float, int]] = deque()
        self._ring: dict[str, np.ndarray] | None = None
        self._head = 0
        self._bucket: list | None = None
        self._buckets: deque[tuple[int, int, list, list]] = deque(maxlen=BUCKETS)
        self._bucket_id = 0
        self._capture: tuple[int, list[np.ndarray]] | None = None
        self._taken: set[int] = set()
        self._captured_at = -np.inf
        self._three_d: dict | None = None
        self._errors: dict[str, dict[str, float]] = {}

    # -- the producer --------------------------------------------------------------

    def set_meta(
        self,
        *,
        views: Sequence[str],
        sources: Sequence[dict],
        point_names: Sequence[str],
        edges: Sequence[Sequence[int]],
        point_colors: Sequence[str] = (),
        edge_colors: Sequence[str] = (),
        animals: int = 1,
        n_frames: int | None = None,
        fps: float | None = None,
        name: str = "",
    ) -> None:
        """Describe what the moments are of, which forgets every moment before.

        `sources` is one `{"name", "views", "size"}` per image a moment carries:
        the views drawn on it by index into `views`, and its `[width, height]`.
        """
        with self._lock:
            self._reset()
            self.meta = {
                "name": name,
                "views": list(views),
                "sources": [dict(s) for s in sources],
                "point_names": list(point_names),
                "edges": [list(map(int, e)) for e in edges],
                "point_colors": list(point_colors),
                "edge_colors": list(edge_colors),
                "animals": int(animals),
                "n_frames": n_frames,
                "fps": fps,
            }
            self.version += 1

    @property
    def active(self) -> bool:
        """Whether a reader is watching, or the feed keeps everything anyway."""
        return self.forced or self.watching

    @property
    def watching(self) -> bool:
        """Whether a reader has looked in the last `IDLE_S` seconds."""
        return self._clock() - self._looked < IDLE_S

    def publish(
        self,
        frames,
        points,
        conf,
        boxes=None,
        images: Sequence[np.ndarray] | None = None,
        times=None,
    ) -> None:
        """Take `T` moments: their frames `(T,)` and what was found at them.

        `points (T, N, V, K, 2)` in video pixels and `conf (T, N, V, K)`; `boxes (T, N,
        V, 4)` the windows they were read through; `images` per source `(T, H, W)`
        uint8; `times (T,)` integer nanoseconds. Nothing is kept by reference.
        """
        frames = np.asarray(frames)
        n_t = len(frames)
        if n_t == 0:
            return
        now = self._clock()
        self.published += n_t
        self.last_frame = int(frames[-1])
        if not self._samples or now - self._samples[-1][0] >= BUCKET_S:
            # Under the lock, since `rate` copies the samples from another thread.
            with self._lock:
                self._samples.append((now, self.published))
                while now - self._samples[0][0] > RATE_S:
                    self._samples.popleft()
        if not self.active:
            return
        points = np.asarray(points, dtype=np.float32)
        conf = np.asarray(conf, dtype=np.float32)
        with self._lock:
            if self._write(frames, points, conf, boxes, times) is None:
                return
            self._add_to_trend(now, int(frames[-1]), points, conf)
            if images is not None and self._capture_due(now):
                frame = int(frames[-1])
                self._capture = (frame, [np.array(im[-1]) for im in images])
                self._taken = set()
                self._captured_at = now

    def fill(self, frames, points, conf, boxes=None) -> None:
        """Put moments read back from storage into the history, uncounted.

        For a reader that opened after the moments went by: the history takes them
        as `publish` would, but they are no part of the rate, the trend or a capture.
        """
        frames = np.asarray(frames)
        with self._lock:
            self._write(frames, points, conf, boxes, None)

    def _write(self, frames, points, conf, boxes, times) -> dict | None:
        """Moments into the history's ring, its arrays made on the first."""
        points = np.asarray(points, dtype=np.float32)
        ring = self._ring_for(points.shape[1:4])
        if ring is None:
            return None
        slots = (self._head + np.arange(len(frames))) % self._size
        ring["frames"][slots] = frames
        ring["points"][slots] = points
        ring["conf"][slots] = conf
        ring["boxes"][slots] = np.nan if boxes is None else boxes
        ring["times"][slots] = -1 if times is None else times
        self._head += len(frames)
        return ring

    def _ring_for(self, shape) -> dict[str, np.ndarray] | None:
        """The history's arrays for `(N, V, K)` moments, made on the first."""
        if self._ring is not None:
            if self._ring["conf"].shape[1:] != tuple(shape):
                return None  # another shape without new meta: drop it, keep the old
            return self._ring
        n_a, n_v, n_k = shape
        size = self._size
        self._ring = {
            "frames": np.full(size, -1, dtype=np.int64),
            "points": np.full((size, n_a, n_v, n_k, 2), np.nan, dtype=np.float32),
            "conf": np.full((size, n_a, n_v, n_k), np.nan, dtype=np.float32),
            "boxes": np.full((size, n_a, n_v, 4), np.nan, dtype=np.float32),
            "times": np.full(size, -1, dtype=np.int64),
        }
        return self._ring

    def _add_to_trend(self, now: float, frame: int, points, conf) -> None:
        """Fold a publish into the open bucket; close it once `BUCKET_S` old."""
        # Per view: summed confidence, the cells it is over, and the cells located.
        axes = (0, 1, 3)
        found = np.isfinite(points).all(axis=-1)
        part = np.stack(
            [
                np.nansum(conf, axis=axes),
                np.isfinite(conf).sum(axis=axes),
                found.sum(axis=axes),
                np.full(conf.shape[2], found[:, :, 0].size, dtype=float),
            ]
        )
        if self._bucket is None:
            self._bucket = [now, frame, part]
            return
        self._bucket[1] = frame
        self._bucket[2] = self._bucket[2] + part
        if now - self._bucket[0] >= BUCKET_S:
            self._close_bucket()

    def _close_bucket(self) -> None:
        if self._bucket is None:
            return
        _, frame, (conf_sum, conf_n, found, cells_n) = self._bucket
        with np.errstate(invalid="ignore", divide="ignore"):
            mean = conf_sum / conf_n
            share = found / cells_n
        self._buckets.append((self._bucket_id, frame, cells(mean, 3), cells(share, 3)))
        self._bucket_id += 1
        self._bucket = None

    def _capture_due(self, now: float) -> bool:
        since = now - self._captured_at
        if since < 1 / MAX_CAPTURE_HZ:
            return False
        if self._capture is None or since >= 1 / MIN_CAPTURE_HZ:
            return True
        return len(self._taken) >= len(self._capture[1])

    def attach_3d(self, stage: str, frames, points3d, reprojected) -> None:
        """What a 3D stage made of some frames, replacing what an earlier one made.

        `points3d (T, N, K, 3)` and their reprojection `(T, N, V, K, 2)` in video
        pixels, at `frames (T,)`.
        """
        with self._lock:
            self._three_d = {
                "stage": stage,
                "frames": np.asarray(frames, dtype=np.int64),
                "points3d": np.asarray(points3d, dtype=float),
                "reprojected": np.asarray(reprojected, dtype=float),
            }
            self.version += 1

    def set_errors(self, label: str, errors: dict[str, float]) -> None:
        """A reprojection error per view name, in pixels, kept beside the others."""
        with self._lock:
            self._errors[label] = {k: float(v) for k, v in errors.items()}
            self.version += 1

    # -- the reader ----------------------------------------------------------------

    def watched(self) -> None:
        """A reader is looking: keep the feed active for `IDLE_S` more seconds."""
        self._looked = self._clock()

    @property
    def rate(self) -> float | None:
        """Moments published per second over the last `RATE_S` seconds."""
        with self._lock:
            samples = list(self._samples)
        if len(samples) < 2:
            return None
        (t0, n0), (t1, n1) = samples[0], samples[-1]
        return (n1 - n0) / (t1 - t0) if t1 > t0 else None

    def capture(self, source: int) -> tuple[int, np.ndarray] | None:
        """The captured frame's image of `source`, `(frame, (H, W) uint8)`, if any."""
        with self._lock:
            if self._capture is None or not 0 <= source < len(self._capture[1]):
                return None
            self._taken.add(source)
            frame, images = self._capture
            return frame, images[source]

    def held_frames(self) -> np.ndarray:
        """The frames the history holds, increasing."""
        with self._lock:
            return self._held()[1]

    def _held(self) -> tuple[np.ndarray, np.ndarray]:
        """The history's slots and their frames, ordered by frame."""
        if self._ring is None:
            return np.zeros(0, dtype=np.int64), np.zeros(0, dtype=np.int64)
        count = min(self._head, self._size)
        slots = (self._head - count + np.arange(count)) % self._size
        frames = self._ring["frames"][slots]
        order = np.argsort(frames, kind="stable")
        return slots[order], frames[order]

    def moment_payload(self, frame: int | None = None, step: int = 0) -> dict | None:
        """One held moment as JSON-ready data, or `None` when there is none.

        `frame` alone is that frame, or the nearest held one; with `step` 1 or -1 it
        is the next held frame after it or the last before it. `None` is the newest.
        Points and boxes are rounded to 0.1 px, `None` where missing; `three_d` is
        the 3D attached at the frame, if any.
        """
        with self._lock:
            slots, frames = self._held()
            if frames.size == 0:
                return None
            if frame is None:
                i = frames.size - 1
            elif step > 0:
                i = int(np.searchsorted(frames, frame, side="right"))
            elif step < 0:
                i = int(np.searchsorted(frames, frame, side="left")) - 1
            else:
                i = int(np.argmin(np.abs(frames - frame)))
            ring = self._ring
            if ring is None or not 0 <= i < frames.size:
                return None
            slot = slots[i]
            return {
                "frame": int(frames[i]),
                "time": int(ring["times"][slot]) if ring["times"][slot] >= 0 else None,
                "points": points(ring["points"][slot], 1),
                "conf": cells(ring["conf"][slot], 3),
                "boxes": points(ring["boxes"][slot], 1),
                "three_d": self._three_d_at(int(frames[i])),
            }

    def _three_d_at(self, frame: int) -> dict | None:
        held = self._three_d
        if held is None:
            return None
        (hit,) = np.nonzero(held["frames"] == frame)
        if hit.size == 0:
            return None
        return {
            "stage": held["stage"],
            "points3d": points(held["points3d"][hit[0]], 4),
            "reprojected": points(held["reprojected"][hit[0]], 1),
        }

    def state_payload(self, since: int = -1, trend_from: int = 0) -> dict:
        """What a reader polls: counts, the live moment, the history, the trend.

        `meta`, `three_d` (the frames 3D is attached at) and `errors` come only when
        the feed's `version` is past `since`; `trend` holds the buckets from id
        `trend_from` on.
        """
        rate = self.rate
        with self._lock:
            _, frames = self._held()
            live = self._capture[0] if self._capture is not None else None
            out: dict[str, Any] = {
                "version": self.version,
                "active": self.active,
                "published": self.published,
                "last_frame": self.last_frame,
                "rate": None if rate is None else round(rate, 2),
                "history": {
                    "count": int(frames.size),
                    "first": int(frames[0]) if frames.size else None,
                    "last": int(frames[-1]) if frames.size else None,
                },
                "captured": live,
                "trend": [
                    {"id": i, "frame": f, "conf": c, "found": s}
                    for i, f, c, s in self._buckets
                    if i >= trend_from
                ],
            }
            if self.version > since:
                three_d = self._three_d
                out["meta"] = self.meta
                out["three_d"] = (
                    None
                    if three_d is None
                    else {
                        "stage": three_d["stage"],
                        "frames": three_d["frames"].tolist(),
                    }
                )
                out["errors"] = {k: dict(v) for k, v in self._errors.items()}
        out["live"] = None if live is None else self.moment_payload(live)
        return out

active property

active: bool

Whether a reader is watching, or the feed keeps everything anyway.

rate property

rate: float | None

Moments published per second over the last RATE_S seconds.

watching property

watching: bool

Whether a reader has looked in the last IDLE_S seconds.

__init__

__init__(
    *,
    forced: bool = False,
    history: int = HISTORY,
    clock: Callable[[], float] = time.monotonic,
) -> None

An empty feed, forced to keep what it is given whether watched or not.

history is how many moments it holds; clock the seconds it measures by.

Source code in src/deeperfly/live/feed.py

def __init__(
    self,
    *,
    forced: bool = False,
    history: int = HISTORY,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    """An empty feed, `forced` to keep what it is given whether watched or not.

    `history` is how many moments it holds; `clock` the seconds it measures by.
    """
    self.forced = forced
    self._size = history
    self._clock = clock
    self._lock = threading.Lock()
    self._looked = -np.inf
    # Bumped whenever the parts a reader fetches only on change move: the meta,
    # the 3D, the errors.
    self.version = 0
    self.meta: dict | None = None
    self._reset()

attach_3d

attach_3d(
    stage: str, frames, points3d, reprojected
) -> None

What a 3D stage made of some frames, replacing what an earlier one made.

points3d (T, N, K, 3) and their reprojection (T, N, V, K, 2) in video pixels, at frames (T,).

Source code in src/deeperfly/live/feed.py

def attach_3d(self, stage: str, frames, points3d, reprojected) -> None:
    """What a 3D stage made of some frames, replacing what an earlier one made.

    `points3d (T, N, K, 3)` and their reprojection `(T, N, V, K, 2)` in video
    pixels, at `frames (T,)`.
    """
    with self._lock:
        self._three_d = {
            "stage": stage,
            "frames": np.asarray(frames, dtype=np.int64),
            "points3d": np.asarray(points3d, dtype=float),
            "reprojected": np.asarray(reprojected, dtype=float),
        }
        self.version += 1

capture

capture(source: int) -> tuple[int, np.ndarray] | None

The captured frame's image of source, (frame, (H, W) uint8), if any.

Source code in src/deeperfly/live/feed.py

def capture(self, source: int) -> tuple[int, np.ndarray] | None:
    """The captured frame's image of `source`, `(frame, (H, W) uint8)`, if any."""
    with self._lock:
        if self._capture is None or not 0 <= source < len(self._capture[1]):
            return None
        self._taken.add(source)
        frame, images = self._capture
        return frame, images[source]

fill

fill(frames, points, conf, boxes=None) -> None

Put moments read back from storage into the history, uncounted.

For a reader that opened after the moments went by: the history takes them as publish would, but they are no part of the rate, the trend or a capture.

Source code in src/deeperfly/live/feed.py

def fill(self, frames, points, conf, boxes=None) -> None:
    """Put moments read back from storage into the history, uncounted.

    For a reader that opened after the moments went by: the history takes them
    as `publish` would, but they are no part of the rate, the trend or a capture.
    """
    frames = np.asarray(frames)
    with self._lock:
        self._write(frames, points, conf, boxes, None)

held_frames

held_frames() -> np.ndarray

The frames the history holds, increasing.

Source code in src/deeperfly/live/feed.py

def held_frames(self) -> np.ndarray:
    """The frames the history holds, increasing."""
    with self._lock:
        return self._held()[1]

moment_payload

moment_payload(
    frame: int | None = None, step: int = 0
) -> dict | None

One held moment as JSON-ready data, or None when there is none.

frame alone is that frame, or the nearest held one; with step 1 or -1 it is the next held frame after it or the last before it. None is the newest. Points and boxes are rounded to 0.1 px, None where missing; three_d is the 3D attached at the frame, if any.

Source code in src/deeperfly/live/feed.py

def moment_payload(self, frame: int | None = None, step: int = 0) -> dict | None:
    """One held moment as JSON-ready data, or `None` when there is none.

    `frame` alone is that frame, or the nearest held one; with `step` 1 or -1 it
    is the next held frame after it or the last before it. `None` is the newest.
    Points and boxes are rounded to 0.1 px, `None` where missing; `three_d` is
    the 3D attached at the frame, if any.
    """
    with self._lock:
        slots, frames = self._held()
        if frames.size == 0:
            return None
        if frame is None:
            i = frames.size - 1
        elif step > 0:
            i = int(np.searchsorted(frames, frame, side="right"))
        elif step < 0:
            i = int(np.searchsorted(frames, frame, side="left")) - 1
        else:
            i = int(np.argmin(np.abs(frames - frame)))
        ring = self._ring
        if ring is None or not 0 <= i < frames.size:
            return None
        slot = slots[i]
        return {
            "frame": int(frames[i]),
            "time": int(ring["times"][slot]) if ring["times"][slot] >= 0 else None,
            "points": points(ring["points"][slot], 1),
            "conf": cells(ring["conf"][slot], 3),
            "boxes": points(ring["boxes"][slot], 1),
            "three_d": self._three_d_at(int(frames[i])),
        }

publish

publish(
    frames,
    points,
    conf,
    boxes=None,
    images: Sequence[ndarray] | None = None,
    times=None,
) -> None

Take T moments: their frames (T,) and what was found at them.

points (T, N, V, K, 2) in video pixels and conf (T, N, V, K); boxes (T, N, V, 4) the windows they were read through; images per source (T, H, W) uint8; times (T,) integer nanoseconds. Nothing is kept by reference.

Source code in src/deeperfly/live/feed.py

def publish(
    self,
    frames,
    points,
    conf,
    boxes=None,
    images: Sequence[np.ndarray] | None = None,
    times=None,
) -> None:
    """Take `T` moments: their frames `(T,)` and what was found at them.

    `points (T, N, V, K, 2)` in video pixels and `conf (T, N, V, K)`; `boxes (T, N,
    V, 4)` the windows they were read through; `images` per source `(T, H, W)`
    uint8; `times (T,)` integer nanoseconds. Nothing is kept by reference.
    """
    frames = np.asarray(frames)
    n_t = len(frames)
    if n_t == 0:
        return
    now = self._clock()
    self.published += n_t
    self.last_frame = int(frames[-1])
    if not self._samples or now - self._samples[-1][0] >= BUCKET_S:
        # Under the lock, since `rate` copies the samples from another thread.
        with self._lock:
            self._samples.append((now, self.published))
            while now - self._samples[0][0] > RATE_S:
                self._samples.popleft()
    if not self.active:
        return
    points = np.asarray(points, dtype=np.float32)
    conf = np.asarray(conf, dtype=np.float32)
    with self._lock:
        if self._write(frames, points, conf, boxes, times) is None:
            return
        self._add_to_trend(now, int(frames[-1]), points, conf)
        if images is not None and self._capture_due(now):
            frame = int(frames[-1])
            self._capture = (frame, [np.array(im[-1]) for im in images])
            self._taken = set()
            self._captured_at = now

set_errors

set_errors(label: str, errors: dict[str, float]) -> None

A reprojection error per view name, in pixels, kept beside the others.

Source code in src/deeperfly/live/feed.py

def set_errors(self, label: str, errors: dict[str, float]) -> None:
    """A reprojection error per view name, in pixels, kept beside the others."""
    with self._lock:
        self._errors[label] = {k: float(v) for k, v in errors.items()}
        self.version += 1

set_meta

set_meta(
    *,
    views: Sequence[str],
    sources: Sequence[dict],
    point_names: Sequence[str],
    edges: Sequence[Sequence[int]],
    point_colors: Sequence[str] = (),
    edge_colors: Sequence[str] = (),
    animals: int = 1,
    n_frames: int | None = None,
    fps: float | None = None,
    name: str = "",
) -> None

Describe what the moments are of, which forgets every moment before.

sources is one {"name", "views", "size"} per image a moment carries: the views drawn on it by index into views, and its [width, height].

Source code in src/deeperfly/live/feed.py

def set_meta(
    self,
    *,
    views: Sequence[str],
    sources: Sequence[dict],
    point_names: Sequence[str],
    edges: Sequence[Sequence[int]],
    point_colors: Sequence[str] = (),
    edge_colors: Sequence[str] = (),
    animals: int = 1,
    n_frames: int | None = None,
    fps: float | None = None,
    name: str = "",
) -> None:
    """Describe what the moments are of, which forgets every moment before.

    `sources` is one `{"name", "views", "size"}` per image a moment carries:
    the views drawn on it by index into `views`, and its `[width, height]`.
    """
    with self._lock:
        self._reset()
        self.meta = {
            "name": name,
            "views": list(views),
            "sources": [dict(s) for s in sources],
            "point_names": list(point_names),
            "edges": [list(map(int, e)) for e in edges],
            "point_colors": list(point_colors),
            "edge_colors": list(edge_colors),
            "animals": int(animals),
            "n_frames": n_frames,
            "fps": fps,
        }
        self.version += 1

state_payload

state_payload(since: int = -1, trend_from: int = 0) -> dict

What a reader polls: counts, the live moment, the history, the trend.

meta, three_d (the frames 3D is attached at) and errors come only when the feed's version is past since; trend holds the buckets from id trend_from on.

Source code in src/deeperfly/live/feed.py

def state_payload(self, since: int = -1, trend_from: int = 0) -> dict:
    """What a reader polls: counts, the live moment, the history, the trend.

    `meta`, `three_d` (the frames 3D is attached at) and `errors` come only when
    the feed's `version` is past `since`; `trend` holds the buckets from id
    `trend_from` on.
    """
    rate = self.rate
    with self._lock:
        _, frames = self._held()
        live = self._capture[0] if self._capture is not None else None
        out: dict[str, Any] = {
            "version": self.version,
            "active": self.active,
            "published": self.published,
            "last_frame": self.last_frame,
            "rate": None if rate is None else round(rate, 2),
            "history": {
                "count": int(frames.size),
                "first": int(frames[0]) if frames.size else None,
                "last": int(frames[-1]) if frames.size else None,
            },
            "captured": live,
            "trend": [
                {"id": i, "frame": f, "conf": c, "found": s}
                for i, f, c, s in self._buckets
                if i >= trend_from
            ],
        }
        if self.version > since:
            three_d = self._three_d
            out["meta"] = self.meta
            out["three_d"] = (
                None
                if three_d is None
                else {
                    "stage": three_d["stage"],
                    "frames": three_d["frames"].tolist(),
                }
            )
            out["errors"] = {k: dict(v) for k, v in self._errors.items()}
    out["live"] = None if live is None else self.moment_payload(live)
    return out

watched

watched() -> None

A reader is looking: keep the feed active for IDLE_S more seconds.

Source code in src/deeperfly/live/feed.py

def watched(self) -> None:
    """A reader is looking: keep the feed active for `IDLE_S` more seconds."""
    self._looked = self._clock()

cells

cells(a, digits: int | None = None) -> list

An array as nested lists, NaN as None, rounded to digits if given.

Source code in src/deeperfly/live/feed.py

def cells(a, digits: int | None = None) -> list:
    """An array as nested lists, NaN as `None`, rounded to `digits` if given."""
    a = np.asarray(a, dtype=float)
    if digits is not None:
        a = np.round(a, digits)
    out = a.astype(object)
    out[~np.isfinite(a)] = None
    return out.tolist()

points

points(a, digits: int | None = None) -> list

Points (or boxes) along the last axis as nested lists, None for any with NaN.

The labeling GUI's shape for a missing point ([x, y] | null), so a page tests one value, not each coordinate.

Source code in src/deeperfly/live/feed.py

def points(a, digits: int | None = None) -> list:
    """Points (or boxes) along the last axis as nested lists, `None` for any with NaN.

    The labeling GUI's shape for a missing point (`[x, y] | null`), so a page tests
    one value, not each coordinate.
    """
    a = np.asarray(a, dtype=float)
    if digits is not None:
        a = np.round(a, digits)
    rows = a.reshape(-1, a.shape[-1])
    out = np.empty(len(rows), dtype=object)
    for i, (row, ok) in enumerate(
        zip(rows.tolist(), np.isfinite(rows).all(axis=1), strict=True)
    ):
        out[i] = row if ok else None
    return out.reshape(a.shape[:-1]).tolist()

deeperfly run as its live view sees it (ADR-0126).

RunWatch is the run's observer (RunObserver): - it feeds each detector pass to a PoseFeed; - it keeps the stages' progress and the log's tail as the page's status; - it reads back what the 3D stages made of the frames the feed holds; - it turns a Stop into RunStopped.

The results file is read only from the observer calls, which run on the run's own thread. Nothing here may fail the run: an error of its own turns the watch off, says so once in the log, and the run goes on.

RunWatch

What a live page reads about a run: its feed, its status and its Stop.

The observer calls come from the run's thread; status, read_frame and request_stop from the server's.

Source code in src/deeperfly/live/run.py

class RunWatch:
    """What a live page reads about a run: its feed, its status and its Stop.

    The observer calls come from the run's thread; `status`, `read_frame` and
    `request_stop` from the server's.
    """

    def __init__(
        self,
        feed: PoseFeed,
        *,
        facts: Sequence[tuple[str, str]] = (),
        config_text: str = "",
        log: Callable[[str], None] = print,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """A watch over a run that has not started, feeding `feed`.

        `facts` are `(label, value)` pairs the page lists about the run, such as
        its config file and weights; `log` is where the watch reports its own
        trouble.
        """
        self.feed = feed
        self.facts = [(str(k), str(v)) for k, v in facts]
        self.config_text = config_text
        self._say = log
        self._clock = clock
        self._lock = threading.Lock()
        self._lines: deque[str] = deque(maxlen=LOG_LINES)
        self.run = ""
        self.labels: list[str] = []
        self.stages: list[str] = []
        self.outcomes: dict[str, str | None] = {}
        self.project: str | None = None
        self.state = "starting"
        self.broken: str | None = None
        self.stop_requested = False
        self.stopped = False
        self._steps: dict[str, dict] = {}
        self._ctx = None
        self._attached: str | None = None
        self._summarized: set[str] = set()
        self._videos: list = []
        self._readers: dict[int, tuple[threading.Lock, FrameReader]] = {}
        self._readers_lock = threading.Lock()

    # -- what the CLI tells it -----------------------------------------------------

    def log(self, line: str) -> None:
        """Keep a line of the run's log for the page."""
        with self._lock:
            self._lines.append(line)

    def finish(self, state: str) -> None:
        """The run ended: `done`, `failed` or `stopped`, the stage it was in with it."""
        with self._lock:
            self.state = state
            for step in self._steps.values():
                if step["state"] == "running" and state != "done":
                    step["state"] = state

    def close(self) -> None:
        """Let go of the decoders the page's frames were read with."""
        with self._readers_lock:
            readers, self._readers = self._readers, {}
        for lock, reader in readers.values():
            with lock:
                reader.close()

    # -- the observer --------------------------------------------------------------

    @_guarded
    def run_started(
        self, run: str, labels: Sequence[str], stages: Sequence[str]
    ) -> None:
        """Note the run's name, its projects and its stages."""
        with self._lock:
            self.run, self.labels, self.stages = run, list(labels), list(stages)
            if self.state == "starting":  # not a stop asked for before the run began
                self.state = "running"

    @_guarded
    def project_started(self, label: str, ctx) -> None:
        """Describe the project's views and videos to the feed, which starts afresh."""
        views = list(ctx.results.views)
        sizes = ctx.results.frame_sizes
        names = ctx.layout.videos_of(views)
        sources = []
        for name in names:
            mine = [i for i, v in enumerate(views) if ctx.layout.video_of(v) == name]
            height, width = sizes[views[mine[0]]]
            sources.append({"name": name, "views": mine, "size": [width, height]})
        skel = ctx.skeleton
        self.feed.set_meta(
            views=views,
            sources=sources,
            point_names=skel.point_names,
            edges=skel.edges.tolist(),
            point_colors=skel.point_colors,
            edge_colors=skel.edge_colors,
            animals=ctx.animals,
            n_frames=ctx.results.n_frames,
            fps=ctx.recording.fps,
            name=label,
        )
        self.close()
        with self._lock:
            self._videos = [ctx.recording.videos[n] for n in names]
            self.project = label
            self._steps = {s: {"state": "pending"} for s in self.stages}
            self._ctx = ctx
            self._attached = None
            self._summarized = set()
        self._check_stop()

    @_guarded
    def stage_cached(self, stage: str) -> None:
        """Mark a stage an earlier invocation already covered."""
        with self._lock:
            self._steps[stage] = {"state": "cached"}

    @_guarded
    def stage_started(self, stage: str, n_frames: int) -> None:
        """Mark a stage running, after reading back 3D the page has not seen yet."""
        self._check_stop()
        self._read_back()
        with self._lock:
            self._steps[stage] = {
                "state": "running",
                "total": int(n_frames),
                "started": self._clock(),
                "published": self.feed.published,
            }

    @_guarded
    def pose2d_block(self, frames, points, conf, boxes, images) -> None:
        """Hand a detector pass to the feed."""
        self._check_stop()
        self.feed.publish(frames, points, conf, boxes, images)

    @_guarded
    def stage_finished(self, stage: str, elapsed: float, ctx) -> None:
        """Mark a stage done and read back the 3D it made."""
        with self._lock:
            step = self._steps.get(stage, {})
            self._steps[stage] = {
                "state": "done",
                "total": step.get("total"),
                "elapsed": float(elapsed),
            }
        self._read_back(ctx)

    @_guarded
    def project_finished(self, label: str, error: str | None) -> None:
        """Note how a project ended; a failure marks the stage it failed in."""
        with self._lock:
            self.outcomes[label] = error
            if error is None:
                return
            for step in self._steps.values():
                if step["state"] == "running":
                    step.update(state="failed", error=error)

    # -- the 3D checks ---------------------------------------------------------------

    def _read_back(self, ctx=None) -> None:
        """Give the feed the newest 3D and the reprojection errors it lacks.

        Only while someone watches, so a page opened late gets them at the next
        stage boundary. The 3D is read at the frames the feed holds, or spread over
        what the stage covers when it holds none, and reprojected through the rig
        the stages after bundle adjustment use.
        """
        ctx = ctx or self._ctx
        if ctx is None or not self.feed.active:
            return
        res = ctx.results
        for stage, refs in ERRORS.items():
            if stage in self._summarized:
                continue
            if not all(res.has(f"{stage}/{ref}") for ref, _ in refs):
                continue
            self._summarized.add(stage)
            covered = np.flatnonzero(res.covered(stage))
            if covered.size == 0:
                continue
            frames = _spread(covered, ERROR_FRAMES)
            for ref, label in refs:
                err = res.read_frames(f"{stage}/{ref}", frames)
                v_axis = res.axes(f"{stage}/{ref}").index("V")
                err = np.moveaxis(err, v_axis, 0).reshape(err.shape[v_axis], -1)
                with np.errstate(all="ignore"):
                    med = np.nanmedian(np.where(np.isfinite(err), err, np.nan), axis=1)
                self.feed.set_errors(
                    label,
                    {
                        v: m
                        for v, m in zip(res.views, med, strict=True)
                        if np.isfinite(m)
                    },
                )
        held = [s for s in POSE3D_STAGES if res.has(f"{s}/points")]
        if not held or held[0] == self._attached or ctx.rig is None:
            return
        stage = held[0]
        frames = self.feed.held_frames()
        if frames.size == 0:
            frames = _spread(np.flatnonzero(res.covered(stage)), SPREAD)
            if frames.size == 0:
                return
            self._fill_2d(res, frames)
        self._attached = stage
        # `(N, T, K, 3)` as stored, to `(T, N, K, 3)`; NaN where the stage did not
        # cover the frame, so nothing an earlier invocation left is shown.
        points = np.moveaxis(res.read_frames(f"{stage}/points", frames), 1, 0)
        points = points.astype(float)
        points[~res.covered(stage)[frames]] = np.nan
        reprojected = np.stack(
            [ctx.rig.views[v].project(points) for v in res.views], axis=2
        )
        self.feed.attach_3d(stage, frames, points, reprojected)

    def _fill_2d(self, res, frames: np.ndarray) -> None:
        """The detections at `frames` into the feed's history, for 3D to ride on."""
        if not res.has("pose2d/points"):
            return
        covered = res.covered("pose2d")[frames]
        # Stored `N`-leading with `T` behind the view: `(N, V, T, ...)` to `(T, N, V,
        # ...)`, the order a detector pass comes in.
        points, conf, crops = (
            np.moveaxis(res.read_frames(f"pose2d/{name}", frames), 2, 0).astype(float)
            for name in ("points", "conf", "crops")
        )
        points[~covered] = np.nan
        self.feed.fill(frames, points, conf, crops)

    # -- stopping ----------------------------------------------------------------------

    def request_stop(self) -> None:
        """Ask the run to stop at its next pass or stage, or in `STOP_GRACE_S` anyway.

        A run deep in one long call (a fit, a solve) reaches no boundary for a
        while; it is then interrupted as Ctrl-C would interrupt it.
        """
        with self._lock:
            if self.stop_requested or self.state not in ("starting", "running"):
                return
            self.stop_requested = True
            self.state = "stopping"
        timer = threading.Timer(STOP_GRACE_S, self._interrupt)
        timer.daemon = True
        timer.start()

    def _interrupt(self) -> None:
        if not self.stopped and self.state == "stopping":
            self.stopped = True
            _thread.interrupt_main()

    def _check_stop(self) -> None:
        if self.stop_requested and not self.stopped:
            self.stopped = True
            raise RunStopped("stopped from the live view")

    # -- what the page reads ---------------------------------------------------------

    def read_frame(self, source: int, frame: int) -> np.ndarray | None:
        """One video's frame, decoded on the reader's thread, or `None` if out of reach.

        Each video keeps one decoder open while the project runs, as the labeling
        GUI's do, so stepping through the history decodes forward instead of seeking.
        """
        with self._readers_lock:
            if not 0 <= source < len(self._videos):
                return None
            if source not in self._readers:
                reader = self._videos[source].open().hold_open(gray_ok=True)
                self._readers[source] = (threading.Lock(), reader)
            lock, reader = self._readers[source]
        with lock:
            try:
                return np.asarray(reader[frame])
            except IndexError, ValueError, OSError:
                return None

    def status(self) -> dict:
        """The page's status block: facts, projects, steps, the log, the actions."""
        now = self._clock()
        rate = self.feed.rate
        ctx = self._ctx
        with self._lock:
            facts = [("run", self.run)]
            if self.project is not None:
                at = self.labels.index(self.project) + 1 if self.labels else 1
                facts.append(("project", f"{self.project} ({at}/{len(self.labels)})"))
            detector = None if ctx is None else ctx.detector
            if detector is not None and detector.loaded_device is not None:
                facts.append(("loaded on", detector.loaded_device))
            steps = [self._step(name, now, rate) for name in self.stages]
            projects = [
                {
                    "name": label,
                    "state": self._project_state(label),
                    "error": self.outcomes.get(label),
                }
                for label in self.labels
            ]
            stoppable = self.state in ("starting", "running")
            return {
                "title": f"deeperfly run {self.run}".strip(),
                "state": self.state,
                "facts": [list(f) for f in [*facts, *self.facts]],
                "projects": projects,
                "steps": steps,
                "log": list(self._lines),
                "config": self.config_text,
                "error": self.broken,
                "actions": [
                    {
                        "name": "stop",
                        "label": "Stop run",
                        "confirm": (
                            "Stop the run? The stage running now loses what it has "
                            "computed; the stages already finished are kept."
                        ),
                        "danger": True,
                        "enabled": stoppable,
                    }
                ],
            }

    def _project_state(self, label: str) -> str:
        if label in self.outcomes:
            return "done" if self.outcomes[label] is None else "failed"
        return "running" if label == self.project else "pending"

    def _step(self, name: str, now: float, rate: float | None) -> dict:
        step = self._steps.get(name, {"state": "pending"})
        out = {"name": name, "state": step["state"], "total": step.get("total")}
        if step["state"] == "running":
            out["elapsed"] = now - step["started"]
            if name == "pose2d":
                done = self.feed.published - step["published"]
                out["done"] = done
                out["rate"] = rate
                if rate and out["total"]:
                    out["eta"] = max(out["total"] - done, 0) / rate
        elif step["state"] == "done":
            out["elapsed"] = step.get("elapsed")
            out["done"] = step.get("total")
        if "error" in step:
            out["error"] = step["error"]
        return out

__init__

__init__(
    feed: PoseFeed,
    *,
    facts: Sequence[tuple[str, str]] = (),
    config_text: str = "",
    log: Callable[[str], None] = print,
    clock: Callable[[], float] = time.monotonic,
) -> None

A watch over a run that has not started, feeding feed.

facts are (label, value) pairs the page lists about the run, such as its config file and weights; log is where the watch reports its own trouble.

Source code in src/deeperfly/live/run.py

def __init__(
    self,
    feed: PoseFeed,
    *,
    facts: Sequence[tuple[str, str]] = (),
    config_text: str = "",
    log: Callable[[str], None] = print,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    """A watch over a run that has not started, feeding `feed`.

    `facts` are `(label, value)` pairs the page lists about the run, such as
    its config file and weights; `log` is where the watch reports its own
    trouble.
    """
    self.feed = feed
    self.facts = [(str(k), str(v)) for k, v in facts]
    self.config_text = config_text
    self._say = log
    self._clock = clock
    self._lock = threading.Lock()
    self._lines: deque[str] = deque(maxlen=LOG_LINES)
    self.run = ""
    self.labels: list[str] = []
    self.stages: list[str] = []
    self.outcomes: dict[str, str | None] = {}
    self.project: str | None = None
    self.state = "starting"
    self.broken: str | None = None
    self.stop_requested = False
    self.stopped = False
    self._steps: dict[str, dict] = {}
    self._ctx = None
    self._attached: str | None = None
    self._summarized: set[str] = set()
    self._videos: list = []
    self._readers: dict[int, tuple[threading.Lock, FrameReader]] = {}
    self._readers_lock = threading.Lock()

close

close() -> None

Let go of the decoders the page's frames were read with.

Source code in src/deeperfly/live/run.py

def close(self) -> None:
    """Let go of the decoders the page's frames were read with."""
    with self._readers_lock:
        readers, self._readers = self._readers, {}
    for lock, reader in readers.values():
        with lock:
            reader.close()

finish

finish(state: str) -> None

The run ended: done, failed or stopped, the stage it was in with it.

Source code in src/deeperfly/live/run.py

def finish(self, state: str) -> None:
    """The run ended: `done`, `failed` or `stopped`, the stage it was in with it."""
    with self._lock:
        self.state = state
        for step in self._steps.values():
            if step["state"] == "running" and state != "done":
                step["state"] = state

log

log(line: str) -> None

Keep a line of the run's log for the page.

Source code in src/deeperfly/live/run.py

def log(self, line: str) -> None:
    """Keep a line of the run's log for the page."""
    with self._lock:
        self._lines.append(line)

pose2d_block

pose2d_block(frames, points, conf, boxes, images) -> None

Hand a detector pass to the feed.

Source code in src/deeperfly/live/run.py

@_guarded
def pose2d_block(self, frames, points, conf, boxes, images) -> None:
    """Hand a detector pass to the feed."""
    self._check_stop()
    self.feed.publish(frames, points, conf, boxes, images)

project_finished

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

Note how a project ended; a failure marks the stage it failed in.

Source code in src/deeperfly/live/run.py

@_guarded
def project_finished(self, label: str, error: str | None) -> None:
    """Note how a project ended; a failure marks the stage it failed in."""
    with self._lock:
        self.outcomes[label] = error
        if error is None:
            return
        for step in self._steps.values():
            if step["state"] == "running":
                step.update(state="failed", error=error)

project_started

project_started(label: str, ctx) -> None

Describe the project's views and videos to the feed, which starts afresh.

Source code in src/deeperfly/live/run.py

@_guarded
def project_started(self, label: str, ctx) -> None:
    """Describe the project's views and videos to the feed, which starts afresh."""
    views = list(ctx.results.views)
    sizes = ctx.results.frame_sizes
    names = ctx.layout.videos_of(views)
    sources = []
    for name in names:
        mine = [i for i, v in enumerate(views) if ctx.layout.video_of(v) == name]
        height, width = sizes[views[mine[0]]]
        sources.append({"name": name, "views": mine, "size": [width, height]})
    skel = ctx.skeleton
    self.feed.set_meta(
        views=views,
        sources=sources,
        point_names=skel.point_names,
        edges=skel.edges.tolist(),
        point_colors=skel.point_colors,
        edge_colors=skel.edge_colors,
        animals=ctx.animals,
        n_frames=ctx.results.n_frames,
        fps=ctx.recording.fps,
        name=label,
    )
    self.close()
    with self._lock:
        self._videos = [ctx.recording.videos[n] for n in names]
        self.project = label
        self._steps = {s: {"state": "pending"} for s in self.stages}
        self._ctx = ctx
        self._attached = None
        self._summarized = set()
    self._check_stop()

read_frame

read_frame(source: int, frame: int) -> np.ndarray | None

One video's frame, decoded on the reader's thread, or None if out of reach.

Each video keeps one decoder open while the project runs, as the labeling GUI's do, so stepping through the history decodes forward instead of seeking.

Source code in src/deeperfly/live/run.py

def read_frame(self, source: int, frame: int) -> np.ndarray | None:
    """One video's frame, decoded on the reader's thread, or `None` if out of reach.

    Each video keeps one decoder open while the project runs, as the labeling
    GUI's do, so stepping through the history decodes forward instead of seeking.
    """
    with self._readers_lock:
        if not 0 <= source < len(self._videos):
            return None
        if source not in self._readers:
            reader = self._videos[source].open().hold_open(gray_ok=True)
            self._readers[source] = (threading.Lock(), reader)
        lock, reader = self._readers[source]
    with lock:
        try:
            return np.asarray(reader[frame])
        except IndexError, ValueError, OSError:
            return None

request_stop

request_stop() -> None

Ask the run to stop at its next pass or stage, or in STOP_GRACE_S anyway.

A run deep in one long call (a fit, a solve) reaches no boundary for a while; it is then interrupted as Ctrl-C would interrupt it.

Source code in src/deeperfly/live/run.py

def request_stop(self) -> None:
    """Ask the run to stop at its next pass or stage, or in `STOP_GRACE_S` anyway.

    A run deep in one long call (a fit, a solve) reaches no boundary for a
    while; it is then interrupted as Ctrl-C would interrupt it.
    """
    with self._lock:
        if self.stop_requested or self.state not in ("starting", "running"):
            return
        self.stop_requested = True
        self.state = "stopping"
    timer = threading.Timer(STOP_GRACE_S, self._interrupt)
    timer.daemon = True
    timer.start()

run_started

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

Note the run's name, its projects and its stages.

Source code in src/deeperfly/live/run.py

@_guarded
def run_started(
    self, run: str, labels: Sequence[str], stages: Sequence[str]
) -> None:
    """Note the run's name, its projects and its stages."""
    with self._lock:
        self.run, self.labels, self.stages = run, list(labels), list(stages)
        if self.state == "starting":  # not a stop asked for before the run began
            self.state = "running"

stage_cached

stage_cached(stage: str) -> None

Mark a stage an earlier invocation already covered.

Source code in src/deeperfly/live/run.py

@_guarded
def stage_cached(self, stage: str) -> None:
    """Mark a stage an earlier invocation already covered."""
    with self._lock:
        self._steps[stage] = {"state": "cached"}

stage_finished

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

Mark a stage done and read back the 3D it made.

Source code in src/deeperfly/live/run.py

@_guarded
def stage_finished(self, stage: str, elapsed: float, ctx) -> None:
    """Mark a stage done and read back the 3D it made."""
    with self._lock:
        step = self._steps.get(stage, {})
        self._steps[stage] = {
            "state": "done",
            "total": step.get("total"),
            "elapsed": float(elapsed),
        }
    self._read_back(ctx)

stage_started

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

Mark a stage running, after reading back 3D the page has not seen yet.

Source code in src/deeperfly/live/run.py

@_guarded
def stage_started(self, stage: str, n_frames: int) -> None:
    """Mark a stage running, after reading back 3D the page has not seen yet."""
    self._check_stop()
    self._read_back()
    with self._lock:
        self._steps[stage] = {
            "state": "running",
            "total": int(n_frames),
            "started": self._clock(),
            "published": self.feed.published,
        }

status

status() -> dict

The page's status block: facts, projects, steps, the log, the actions.

Source code in src/deeperfly/live/run.py

def status(self) -> dict:
    """The page's status block: facts, projects, steps, the log, the actions."""
    now = self._clock()
    rate = self.feed.rate
    ctx = self._ctx
    with self._lock:
        facts = [("run", self.run)]
        if self.project is not None:
            at = self.labels.index(self.project) + 1 if self.labels else 1
            facts.append(("project", f"{self.project} ({at}/{len(self.labels)})"))
        detector = None if ctx is None else ctx.detector
        if detector is not None and detector.loaded_device is not None:
            facts.append(("loaded on", detector.loaded_device))
        steps = [self._step(name, now, rate) for name in self.stages]
        projects = [
            {
                "name": label,
                "state": self._project_state(label),
                "error": self.outcomes.get(label),
            }
            for label in self.labels
        ]
        stoppable = self.state in ("starting", "running")
        return {
            "title": f"deeperfly run {self.run}".strip(),
            "state": self.state,
            "facts": [list(f) for f in [*facts, *self.facts]],
            "projects": projects,
            "steps": steps,
            "log": list(self._lines),
            "config": self.config_text,
            "error": self.broken,
            "actions": [
                {
                    "name": "stop",
                    "label": "Stop run",
                    "confirm": (
                        "Stop the run? The stage running now loses what it has "
                        "computed; the stages already finished are kept."
                    ),
                    "danger": True,
                    "enabled": stoppable,
                }
            ],
        }

A browser page over a pose feed, served from the process that feeds it (ADR-0126).

build_live_app makes the FastAPI app: the page, the GUI's static modules, and an API that a token guards. serve_live runs it on a thread of its own beside the producer. The feed, the status and the actions are the caller's; deeperfly run passes its RunWatch's.

Every API request marks the feed watched, so a page that is open is what keeps a run capturing, and closing it lets the run go back to counting.

build_live_app

build_live_app(
    feed: PoseFeed,
    *,
    status: Callable[[], dict] = dict,
    actions: Mapping[str, Callable[[], None]] | None = None,
    frame_reader: Callable[[int, int], ndarray | None]
    | None = None,
    token: str | None = None,
) -> FastAPI

The live page's app over feed, no server attached.

status() is the producer's own block of the state (see RunWatch.status); actions are what the page's buttons call by name; frame_reader(source, frame) decodes a frame the feed did not capture, or answers None. With a token, every API request must carry it in an X-Deeperfly-Token header.

Source code in src/deeperfly/live/server.py

def build_live_app(
    feed: PoseFeed,
    *,
    status: Callable[[], dict] = dict,
    actions: Mapping[str, Callable[[], None]] | None = None,
    frame_reader: Callable[[int, int], np.ndarray | None] | None = None,
    token: str | None = None,
) -> FastAPI:
    """The live page's app over `feed`, no server attached.

    `status()` is the producer's own block of the state (see
    [RunWatch.status][deeperfly.live.run.RunWatch.status]); `actions` are what the
    page's buttons call by name; `frame_reader(source, frame)` decodes a frame the
    feed did not capture, or answers `None`. With a `token`, every API request must
    carry it in an `X-Deeperfly-Token` header.
    """
    actions = dict(actions or {})
    app = FastAPI(title="deeperfly live")

    @app.middleware("http")
    async def revalidate(request: Request, call_next):
        """Revalidate every response before reuse, as the GUI does (`gui/app.py`)."""
        response = await call_next(request)
        response.headers.setdefault("Cache-Control", "no-cache")
        return response

    def looking(x_deeperfly_token: str | None = Header(default=None)) -> None:
        """Refuse a request without the token; mark the feed watched by one with it."""
        if token is not None and not secrets.compare_digest(
            x_deeperfly_token or "", token
        ):
            raise HTTPException(status_code=403, detail="the live view's token")
        feed.watched()

    api = APIRouter(prefix="/api", dependencies=[Depends(looking)])

    @api.get("/state")
    def state(since: int = -1, trend_from: int = 0) -> dict:
        """The feed's state (`PoseFeed.state_payload`) and the producer's `status`."""
        return {**feed.state_payload(since, trend_from), "status": status()}

    @api.get("/moment")
    def moment(frame: int | None = None, step: int = 0) -> dict:
        """A held moment: `frame` or the nearest, or with `step` the next or last."""
        out = feed.moment_payload(frame, step)
        if out is None:
            raise HTTPException(status_code=404, detail="no such moment held")
        return out

    @api.get("/frame")
    def frame(source: int, frame: int) -> Response:
        """One source's image at `frame` as a JPEG: the capture's, else decoded."""
        held = feed.capture(source)
        if held is not None and held[0] == frame:
            image = held[1]
        else:
            image = None if frame_reader is None else frame_reader(source, frame)
        if image is None:
            raise HTTPException(status_code=404, detail=f"no image of frame {frame}")
        return Response(
            content=jpeg(image),
            media_type="image/jpeg",
            headers={"Cache-Control": "no-store"},
        )

    @api.post("/action/{name}")
    def action(name: str) -> dict:
        """Run the action the page's button names."""
        if name not in actions:
            raise HTTPException(status_code=404, detail=f"no action {name!r}")
        actions[name]()
        return {"ok": True}

    app.include_router(api)

    @app.get("/")
    def page() -> FileResponse:
        """The live page."""
        return FileResponse(WEB / "live.html")

    app.mount("/static", StaticFiles(directory=WEB / "static"), name="static")
    return app

jpeg

jpeg(image: ndarray) -> bytes

A frame as a JPEG, halved until it is at most MAX_WIDTH wide.

The page draws it over the video's own size, so a smaller image moves no point.

Source code in src/deeperfly/live/server.py

def jpeg(image: np.ndarray) -> bytes:
    """A frame as a JPEG, halved until it is at most `MAX_WIDTH` wide.

    The page draws it over the video's own size, so a smaller image moves no point.
    """
    img = Image.fromarray(np.asarray(image))
    factor = 1
    while img.width / factor > MAX_WIDTH:
        factor *= 2
    if factor > 1:
        img = img.reduce(factor)
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=JPEG_QUALITY)
    return buf.getvalue()

serve_live

serve_live(
    app: FastAPI,
    *,
    host: str = "127.0.0.1",
    port: int | None = None,
) -> tuple[str, Callable[[], None]]

Serve app on a daemon thread: its base URL, and a function that stops it.

port None is LIVE_PORT or the next free one.

Raises:

Type	Description
OSError	If no port could be bound.

Source code in src/deeperfly/live/server.py

def serve_live(
    app: FastAPI, *, host: str = "127.0.0.1", port: int | None = None
) -> tuple[str, Callable[[], None]]:
    """Serve `app` on a daemon thread: its base URL, and a function that stops it.

    `port` `None` is `LIVE_PORT` or the next free one.

    Raises:
        OSError: If no port could be bound.
    """
    import uvicorn

    from ..gui.instance import bind

    sock = bind(host, port, first=LIVE_PORT)
    server = uvicorn.Server(uvicorn.Config(app, log_level="warning"))
    # Off the main thread uvicorn leaves the signals alone, so Ctrl-C stays the run's.
    thread = threading.Thread(
        target=server.run, kwargs={"sockets": [sock]}, name="live-view", daemon=True
    )
    thread.start()

    def stop() -> None:
        server.should_exit = True
        thread.join(timeout=5.0)
        sock.close()

    return f"http://{host}:{sock.getsockname()[1]}/", stop