Skip to content

media

Read videos as frames -- video files (PyAV), image sequences (Pillow), and the parts of one split video read as one stream -- behind one backend-agnostic reader interface, and write frames back out as an MP4.

Read recordings as frames: video files (PyAV) and image sequences (Pillow).

open_reader resolves a source to its FrameReader once (ADR-0005); a consumer then indexes it or streams it without knowing the kind of source. A new decode backend is a FrameReader subclass and one entry in _VIDEO_BACKENDS.

reader = media.open_reader("clip.mp4")
block = reader[0:64]                                    # (64, H, W, 3) uint8
for b in media.prefetch(reader.stream_blocks(gray_ok=True)):
    model(b)

ConcatReader

Bases: FrameReader

The parts of one video, read as a single stream.

Parameters:

Name Type Description Default
paths

The parts, in decode order, two or more.

 required
factory Callable[[Path], FrameReader]

Builds a part's reader from its path: the video backend.

 required
counts list[int] | None

Each part's exact frame count if already known, which skips the measuring decode.

 None

Attributes:

Name Type Description
paths

The parts, in decode order.
Source code in src/deeperfly/io/media/concat.py 
 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
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
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
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
class ConcatReader(FrameReader):
    """The parts of one video, read as a single stream.

    Args:
        paths: The parts, in decode order, two or more.
        factory: Builds a part's reader from its path: the video backend.
        counts: Each part's exact frame count if already known, which skips the
            measuring decode.

    Attributes:
        paths: The parts, in decode order.
    """

    def __init__(
        self,
        paths,
        *,
        factory: Callable[[Path], FrameReader],
        counts: list[int] | None = None,
    ) -> None:
        """Open the parts and build the global index (measuring counts if needed)."""
        self.paths = [Path(p) for p in paths]
        if len(self.paths) < 2:
            raise ValueError(
                "ConcatReader is for two or more parts; open_reader returns a "
                "plain reader for one"
            )
        self._readers = [factory(p) for p in self.paths]
        self._counts = (
            [int(c) for c in counts] if counts is not None else self._measure()
        )
        # Exclusive prefix sums: `_starts[i]` is the global index of part i's
        # frame 0.
        self._starts: list[int] = []
        total = 0
        for n in self._counts:
            self._starts.append(total)
            total += n
        self._total = total
        # The part a single-frame read landed in last: the one a held-open stream
        # keeps its decoder for (see `__getitem__`).
        self._last_part: int | None = None
        self._check_agreement()

    def _measure(self) -> list[int]:
        """Each part's exact frame count, decoding a part whose header will not say.

        The last part too: `count` reports the total, and a wrong total truncates a
        run.
        """
        out: list[int] = []
        for reader in self._readers:
            n = reader.count()
            if n is None:
                started = time.perf_counter()
                n = sum(1 for _ in reader.stream_frames(gray_ok=True))
                log.warning(
                    "%s carries no frame count in its container header, so it "
                    "was counted by a full decode (%d frames in %.1f s). A "
                    "concatenated stream needs exact lengths -- an estimate "
                    "would shift every frame index after this part.",
                    self.paths[len(out)].name,
                    n,
                    time.perf_counter() - started,
                )
            out.append(int(n))
        return out

    def _check_agreement(self) -> None:
        """Refuse parts that disagree on frame size or frame rate."""
        sizes: dict[tuple[int, int], Path] = {}
        rates: dict[float, Path] = {}
        for reader, path in zip(self._readers, self.paths, strict=True):
            frame = reader[0]
            size = (int(frame.shape[0]), int(frame.shape[1]))
            if size not in sizes and sizes:
                (other_size, other_path) = next(iter(sizes.items()))
                raise ValueError(
                    f"{path.name} is {size[1]}x{size[0]} but {other_path.name} "
                    f"is {other_size[1]}x{other_size[0]}; the parts of one "
                    "video's parts have to agree on frame size -- a "
                    "concatenated stream with two resolutions has no single "
                    "intrinsics"
                )
            sizes[size] = path
            rate = reader.fps()
            if rate is None:
                continue
            rate = round(float(rate), 3)
            if rate not in rates and rates:
                other_rate, other_path = next(iter(rates.items()))
                raise ValueError(
                    f"{path.name} is {rate} fps but {other_path.name} is "
                    f"{other_rate} fps; the parts of one video have to "
                    "agree on frame rate"
                )
            rates[rate] = path

    def _locate(self, idx: int) -> tuple[int, int]:
        """A global frame index -> `(part, local index)`."""
        if idx < 0:
            idx += self._total
        if not 0 <= idx < self._total:
            raise IndexError(
                f"frame {idx} is outside this {self._total}-frame stream "
                f"({len(self.paths)} parts)"
            )
        part = bisect.bisect_right(self._starts, idx) - 1
        return part, idx - self._starts[part]

    def _indices(self, key) -> list[int]:
        if isinstance(key, slice):
            return list(range(*key.indices(self._total)))
        return [int(i) for i in key]

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames (int / list / slice) into a uint8 RGB array."""
        if isinstance(key, (int, np.integer)):
            part, local = self._locate(int(key))
            # A viewer holds one decoder, the part it is reading, not every part it
            # read on the way (a no-op on a stateless part).
            if self._last_part is not None and self._last_part != part:
                self._readers[self._last_part].release()
            self._last_part = part
            return self._readers[part][local]
        wanted = self._indices(key)
        if not wanted:
            raise ValueError("no frames selected (check index list / slice)")
        # One decode per part, then back into the caller's order, so a slice across a
        # boundary never re-opens per frame.
        per_part: dict[int, list[int]] = {}
        for i in wanted:
            part, local = self._locate(i)
            per_part.setdefault(part, []).append(local)
        decoded: dict[int, np.ndarray] = {}
        for part, locals_ in per_part.items():
            block = self._readers[part][locals_]
            for local, frame in zip(locals_, block, strict=True):
                decoded[self._starts[part] + local] = frame
        return np.stack([decoded[i] for i in wanted])

    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W C"]]:
        """Walk the parts in order, forward, never seeking across a boundary.

        The inherited `stream_blocks` then never breaks a block at a part boundary.
        """
        stop = self._total if stop is None else min(stop, self._total)
        wanted = range(start, stop, step)
        for part, reader in enumerate(self._readers):
            base, n = self._starts[part], self._counts[part]
            first = next((i for i in wanted if base <= i < base + n), None)
            if first is None:
                continue
            local_stop = min(stop, base + n) - base
            yield from reader.stream_frames(
                start=first - base,
                stop=local_stop,
                step=step,
                gray_ok=gray_ok,
                thread_count=thread_count,
            )

    def count(self) -> int:
        """The exact total, summed over parts (never `None` -- see `_measure`)."""
        return self._total

    def fps(self) -> float | None:
        """The parts' shared frame rate (they are checked to agree)."""
        return self._readers[0].fps()

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """This stream with each part held open, and returns itself.

        A part opens its decoder on its first read, so a viewer opens only the parts
        it reads.
        """
        held = []
        for reader in self._readers:
            one = reader.hold_open(gray_ok=gray_ok)
            if one is not reader:
                reader.close()
            held.append(one)
        self._readers = held
        return self

    def release(self) -> None:
        """Let go of every part's held decoder; each reopens on its next read."""
        for reader in self._readers:
            reader.release()

    def close(self) -> None:
        """Close every part's reader."""
        for reader in self._readers:
            reader.close()

    def __repr__(self) -> str:
        """The parts and the total frame count."""
        parts = ", ".join(p.name for p in self.paths)
        return f"ConcatReader({parts}; {self._total} frames)"

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames (int / list / slice) into a uint8 RGB array.

Source code in src/deeperfly/io/media/concat.py 
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
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames (int / list / slice) into a uint8 RGB array."""
    if isinstance(key, (int, np.integer)):
        part, local = self._locate(int(key))
        # A viewer holds one decoder, the part it is reading, not every part it
        # read on the way (a no-op on a stateless part).
        if self._last_part is not None and self._last_part != part:
            self._readers[self._last_part].release()
        self._last_part = part
        return self._readers[part][local]
    wanted = self._indices(key)
    if not wanted:
        raise ValueError("no frames selected (check index list / slice)")
    # One decode per part, then back into the caller's order, so a slice across a
    # boundary never re-opens per frame.
    per_part: dict[int, list[int]] = {}
    for i in wanted:
        part, local = self._locate(i)
        per_part.setdefault(part, []).append(local)
    decoded: dict[int, np.ndarray] = {}
    for part, locals_ in per_part.items():
        block = self._readers[part][locals_]
        for local, frame in zip(locals_, block, strict=True):
            decoded[self._starts[part] + local] = frame
    return np.stack([decoded[i] for i in wanted])

__init__ 

__init__(
    paths,
    *,
    factory: Callable[[Path], FrameReader],
    counts: list[int] | None = None,
) -> None

Open the parts and build the global index (measuring counts if needed).

Source code in src/deeperfly/io/media/concat.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    paths,
    *,
    factory: Callable[[Path], FrameReader],
    counts: list[int] | None = None,
) -> None:
    """Open the parts and build the global index (measuring counts if needed)."""
    self.paths = [Path(p) for p in paths]
    if len(self.paths) < 2:
        raise ValueError(
            "ConcatReader is for two or more parts; open_reader returns a "
            "plain reader for one"
        )
    self._readers = [factory(p) for p in self.paths]
    self._counts = (
        [int(c) for c in counts] if counts is not None else self._measure()
    )
    # Exclusive prefix sums: `_starts[i]` is the global index of part i's
    # frame 0.
    self._starts: list[int] = []
    total = 0
    for n in self._counts:
        self._starts.append(total)
        total += n
    self._total = total
    # The part a single-frame read landed in last: the one a held-open stream
    # keeps its decoder for (see `__getitem__`).
    self._last_part: int | None = None
    self._check_agreement()

__repr__ 

__repr__() -> str

The parts and the total frame count.

Source code in src/deeperfly/io/media/concat.py 
228
229
230
231
def __repr__(self) -> str:
    """The parts and the total frame count."""
    parts = ", ".join(p.name for p in self.paths)
    return f"ConcatReader({parts}; {self._total} frames)"

close 

close() -> None

Close every part's reader.

Source code in src/deeperfly/io/media/concat.py 
223
224
225
226
def close(self) -> None:
    """Close every part's reader."""
    for reader in self._readers:
        reader.close()

count 

count() -> int

The exact total, summed over parts (never None -- see _measure).

Source code in src/deeperfly/io/media/concat.py 
195
196
197
def count(self) -> int:
    """The exact total, summed over parts (never `None` -- see `_measure`)."""
    return self._total

fps 

fps() -> float | None

The parts' shared frame rate (they are checked to agree).

Source code in src/deeperfly/io/media/concat.py 
199
200
201
def fps(self) -> float | None:
    """The parts' shared frame rate (they are checked to agree)."""
    return self._readers[0].fps()

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

This stream with each part held open, and returns itself.

A part opens its decoder on its first read, so a viewer opens only the parts it reads.

Source code in src/deeperfly/io/media/concat.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """This stream with each part held open, and returns itself.

    A part opens its decoder on its first read, so a viewer opens only the parts
    it reads.
    """
    held = []
    for reader in self._readers:
        one = reader.hold_open(gray_ok=gray_ok)
        if one is not reader:
            reader.close()
        held.append(one)
    self._readers = held
    return self

release 

release() -> None

Let go of every part's held decoder; each reopens on its next read.

Source code in src/deeperfly/io/media/concat.py 
218
219
220
221
def release(self) -> None:
    """Let go of every part's held decoder; each reopens on its next read."""
    for reader in self._readers:
        reader.release()

stream_frames 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]

Walk the parts in order, forward, never seeking across a boundary.

The inherited stream_blocks then never breaks a block at a part boundary.

Source code in src/deeperfly/io/media/concat.py 
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
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]:
    """Walk the parts in order, forward, never seeking across a boundary.

    The inherited `stream_blocks` then never breaks a block at a part boundary.
    """
    stop = self._total if stop is None else min(stop, self._total)
    wanted = range(start, stop, step)
    for part, reader in enumerate(self._readers):
        base, n = self._starts[part], self._counts[part]
        first = next((i for i in wanted if base <= i < base + n), None)
        if first is None:
            continue
        local_stop = min(stop, base + n) - base
        yield from reader.stream_frames(
            start=first - base,
            stop=local_stop,
            step=step,
            gray_ok=gray_ok,
            thread_count=thread_count,
        )

FrameReader

Bases: ABC

Reads one source as (H, W, 3) uint8 RGB frames, on the CPU.

reader[5] is one frame; reader[[0, 3, 5]] and reader[2:8:2] are (T, H, W, 3) in the order asked; stream_frames and stream_blocks iterate lazily and take gray_ok. A reader is a context manager, and close is a no-op for the stateless ones.

Source code in src/deeperfly/io/media/reader.py 
 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
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
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
class FrameReader(ABC):
    """Reads one source as `(H, W, 3)` uint8 RGB frames, on the CPU.

    `reader[5]` is one frame; `reader[[0, 3, 5]]` and `reader[2:8:2]` are
    `(T, H, W, 3)` in the order asked; `stream_frames` and `stream_blocks` iterate
    lazily and take `gray_ok`. A reader is a context manager, and `close` is a no-op
    for the stateless ones.
    """

    def __enter__(self) -> Self:
        """Enter the context, returning the reader."""
        return self

    def __exit__(self, *exc) -> None:
        """Release resources on context exit."""
        self.close()

    def close(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
        """Release any resources held by the reader (a no-op by default)."""

    @abstractmethod
    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode one frame (`int`), several in order (`list[int]`), or a `slice`."""

    @abstractmethod
    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W C"]]:
        """Yield `(H, W, C)` uint8 frames from one forward decode.

        `start`, `stop` and `step` are as for `range`. `gray_ok` permits a single
        channel, kept as `(H, W, 1)`, for a frame that carries no color, which skips the
        color conversion; a source that cannot tell cheaply returns RGB.
        `thread_count` caps the decode threads, worth doing when several sources decode
        at once.
        """

    def stream_blocks(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        block_size: int = 64,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "T H W C"]]:
        """Yield `(T, H, W, C)` uint8 blocks (`T <= block_size`) from one decode.

        The default buffers `stream_frames` into blocks, with its arguments; a backend
        with a cheaper block decode overrides it.
        """
        if block_size < 1:
            raise ValueError(f"block_size must be >= 1, got {block_size}")
        buf: list[np.ndarray] = []
        for frame in self.stream_frames(
            start=start,
            stop=stop,
            step=step,
            gray_ok=gray_ok,
            thread_count=thread_count,
        ):
            buf.append(frame)
            if len(buf) >= block_size:
                yield _stack_frames(buf)
                buf = []
        if buf:
            yield _stack_frames(buf)

    @abstractmethod
    def count(self) -> int | None:
        """Best-effort frame count, or `None` when unknown: a progress hint only."""

    def fps(self) -> float | None:
        """Frame rate in frames per second, or `None` (an image sequence has none)."""
        return None

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """This source again, with its decoder held open between reads.

        For a viewer, which would otherwise pay a re-open and a seek per frame. The
        base returns `self` and a backend overrides it. The caller owns the result,
        which may be this same object, and closes it. `gray_ok` lets indexing one frame
        return `(H, W)` for a source that carries no color.
        """
        return self

    def release(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
        """Let go of what a held-open reader holds between reads; it reopens on demand.

        For a viewer that keeps readers of footage no longer on screen. The base holds
        nothing, so there is nothing to let go of.
        """

__enter__ 

__enter__() -> Self

Enter the context, returning the reader.

Source code in src/deeperfly/io/media/reader.py 
40
41
42
def __enter__(self) -> Self:
    """Enter the context, returning the reader."""
    return self

__exit__ 

__exit__(*exc) -> None

Release resources on context exit.

Source code in src/deeperfly/io/media/reader.py 
44
45
46
def __exit__(self, *exc) -> None:
    """Release resources on context exit."""
    self.close()

__getitem__ abstractmethod 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode one frame (int), several in order (list[int]), or a slice.

Source code in src/deeperfly/io/media/reader.py 
51
52
53
@abstractmethod
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode one frame (`int`), several in order (`list[int]`), or a `slice`."""

close 

close() -> None

Release any resources held by the reader (a no-op by default).

Source code in src/deeperfly/io/media/reader.py 
48
49
def close(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
    """Release any resources held by the reader (a no-op by default)."""

count abstractmethod 

count() -> int | None

Best-effort frame count, or None when unknown: a progress hint only.

Source code in src/deeperfly/io/media/reader.py 
106
107
108
@abstractmethod
def count(self) -> int | None:
    """Best-effort frame count, or `None` when unknown: a progress hint only."""

fps 

fps() -> float | None

Frame rate in frames per second, or None (an image sequence has none).

Source code in src/deeperfly/io/media/reader.py 
110
111
112
def fps(self) -> float | None:
    """Frame rate in frames per second, or `None` (an image sequence has none)."""
    return None

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

This source again, with its decoder held open between reads.

For a viewer, which would otherwise pay a re-open and a seek per frame. The base returns self and a backend overrides it. The caller owns the result, which may be this same object, and closes it. gray_ok lets indexing one frame return (H, W) for a source that carries no color.

Source code in src/deeperfly/io/media/reader.py 
114
115
116
117
118
119
120
121
122
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """This source again, with its decoder held open between reads.

    For a viewer, which would otherwise pay a re-open and a seek per frame. The
    base returns `self` and a backend overrides it. The caller owns the result,
    which may be this same object, and closes it. `gray_ok` lets indexing one frame
    return `(H, W)` for a source that carries no color.
    """
    return self

release 

release() -> None

Let go of what a held-open reader holds between reads; it reopens on demand.

For a viewer that keeps readers of footage no longer on screen. The base holds nothing, so there is nothing to let go of.

Source code in src/deeperfly/io/media/reader.py 
124
125
126
127
128
129
def release(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
    """Let go of what a held-open reader holds between reads; it reopens on demand.

    For a viewer that keeps readers of footage no longer on screen. The base holds
    nothing, so there is nothing to let go of.
    """

stream_blocks 

stream_blocks(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W C"]]

Yield (T, H, W, C) uint8 blocks (T <= block_size) from one decode.

The default buffers stream_frames into blocks, with its arguments; a backend with a cheaper block decode overrides it.

Source code in src/deeperfly/io/media/reader.py 
 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
104
def stream_blocks(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W C"]]:
    """Yield `(T, H, W, C)` uint8 blocks (`T <= block_size`) from one decode.

    The default buffers `stream_frames` into blocks, with its arguments; a backend
    with a cheaper block decode overrides it.
    """
    if block_size < 1:
        raise ValueError(f"block_size must be >= 1, got {block_size}")
    buf: list[np.ndarray] = []
    for frame in self.stream_frames(
        start=start,
        stop=stop,
        step=step,
        gray_ok=gray_ok,
        thread_count=thread_count,
    ):
        buf.append(frame)
        if len(buf) >= block_size:
            yield _stack_frames(buf)
            buf = []
    if buf:
        yield _stack_frames(buf)

stream_frames abstractmethod 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]

Yield (H, W, C) uint8 frames from one forward decode.

start, stop and step are as for range. gray_ok permits a single channel, kept as (H, W, 1), for a frame that carries no color, which skips the color conversion; a source that cannot tell cheaply returns RGB. thread_count caps the decode threads, worth doing when several sources decode at once.

Source code in src/deeperfly/io/media/reader.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@abstractmethod
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]:
    """Yield `(H, W, C)` uint8 frames from one forward decode.

    `start`, `stop` and `step` are as for `range`. `gray_ok` permits a single
    channel, kept as `(H, W, 1)`, for a frame that carries no color, which skips the
    color conversion; a source that cannot tell cheaply returns RGB.
    `thread_count` caps the decode threads, worth doing when several sources decode
    at once.
    """

ImageSequenceReader

Bases: FrameReader

Reads an ordered image sequence (a directory, glob, or explicit file list).

Files decode in parallel on up to workers threads; gray_ok and thread_count are accepted and ignored.

Source code in src/deeperfly/io/media/images.py 
 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
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
class ImageSequenceReader(FrameReader):
    """Reads an ordered image sequence (a directory, glob, or explicit file list).

    Files decode in parallel on up to `workers` threads; `gray_ok` and `thread_count`
    are accepted and ignored.
    """

    def __init__(self, files, *, workers: int | None = None) -> None:
        """Wrap an ordered list of image `files`, decoded on up to `workers` threads."""
        self.files = [Path(f) for f in files]
        self.workers = workers

    @classmethod
    def from_pattern(
        cls, pattern: str | Path, *, workers: int | None = None
    ) -> ImageSequenceReader:
        """A reader for a directory or glob, its files sorted by name."""
        return cls(list_image_files(pattern), workers=workers)

    def _n_workers(self, n: int) -> int:
        return max(1, min(n, self.workers or (os.process_cpu_count() or 4)))

    def _decode(self, files: list[Path]) -> np.ndarray:
        """Parallel CPU decode of `files` to a `(T, H, W, 3)` uint8 array."""
        from PIL import Image

        def decode(f: Path) -> np.ndarray:
            with Image.open(f) as img:
                return _to_rgb_uint8(np.asarray(img))

        with ThreadPoolExecutor(max_workers=self._n_workers(len(files))) as pool:
            frames = list(pool.map(decode, files))
        return np.stack(frames)

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames (int / list / slice) into a uint8 RGB array."""
        if isinstance(key, (int, np.integer)):
            return self._decode([self.files[int(key)]])[0]
        if isinstance(key, list):
            if not key:
                raise ValueError("index list must be non-empty")
            return self._decode([self.files[int(i)] for i in key])
        if isinstance(key, slice):
            files = self.files[key]
            if not files:
                raise ValueError("no frames selected (check slice)")
            return self._decode(files)
        raise TypeError(f"invalid index type {type(key).__name__!r}")

    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W 3"]]:
        """Yield `(H, W, 3)` uint8 frames, one per file, decoded a block at a time."""
        for block in self.stream_blocks(start=start, stop=stop, step=step):
            yield from block

    def stream_blocks(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        block_size: int = 64,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "T H W 3"]]:
        """Yield `(T, H, W, 3)` uint8 blocks, each decoded in parallel across files."""
        if block_size < 1:
            raise ValueError(f"block_size must be >= 1, got {block_size}")
        files = self.files[start:stop:step]
        for pos in range(0, len(files), block_size):
            yield self._decode(files[pos : pos + block_size])

    def count(self) -> int | None:
        """The number of files -- one frame each."""
        return len(self.files)

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames (int / list / slice) into a uint8 RGB array.

Source code in src/deeperfly/io/media/images.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames (int / list / slice) into a uint8 RGB array."""
    if isinstance(key, (int, np.integer)):
        return self._decode([self.files[int(key)]])[0]
    if isinstance(key, list):
        if not key:
            raise ValueError("index list must be non-empty")
        return self._decode([self.files[int(i)] for i in key])
    if isinstance(key, slice):
        files = self.files[key]
        if not files:
            raise ValueError("no frames selected (check slice)")
        return self._decode(files)
    raise TypeError(f"invalid index type {type(key).__name__!r}")

__init__ 

__init__(files, *, workers: int | None = None) -> None

Wrap an ordered list of image files, decoded on up to workers threads.

Source code in src/deeperfly/io/media/images.py 
83
84
85
86
def __init__(self, files, *, workers: int | None = None) -> None:
    """Wrap an ordered list of image `files`, decoded on up to `workers` threads."""
    self.files = [Path(f) for f in files]
    self.workers = workers

count 

count() -> int | None

The number of files -- one frame each.

Source code in src/deeperfly/io/media/images.py 
155
156
157
def count(self) -> int | None:
    """The number of files -- one frame each."""
    return len(self.files)

from_pattern classmethod 

from_pattern(
    pattern: str | Path, *, workers: int | None = None
) -> ImageSequenceReader

A reader for a directory or glob, its files sorted by name.

Source code in src/deeperfly/io/media/images.py 
88
89
90
91
92
93
@classmethod
def from_pattern(
    cls, pattern: str | Path, *, workers: int | None = None
) -> ImageSequenceReader:
    """A reader for a directory or glob, its files sorted by name."""
    return cls(list_image_files(pattern), workers=workers)

stream_blocks 

stream_blocks(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W 3"]]

Yield (T, H, W, 3) uint8 blocks, each decoded in parallel across files.

Source code in src/deeperfly/io/media/images.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def stream_blocks(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W 3"]]:
    """Yield `(T, H, W, 3)` uint8 blocks, each decoded in parallel across files."""
    if block_size < 1:
        raise ValueError(f"block_size must be >= 1, got {block_size}")
    files = self.files[start:stop:step]
    for pos in range(0, len(files), block_size):
        yield self._decode(files[pos : pos + block_size])

stream_frames 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W 3"]]

Yield (H, W, 3) uint8 frames, one per file, decoded a block at a time.

Source code in src/deeperfly/io/media/images.py 
125
126
127
128
129
130
131
132
133
134
135
136
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W 3"]]:
    """Yield `(H, W, 3)` uint8 frames, one per file, decoded a block at a time."""
    for block in self.stream_blocks(start=start, stop=stop, step=step):
        yield from block

PyavHeldReader

Bases: PyavVideoReader

One video file with the decoder held open between single-frame reads.

A frame a short way ahead is decoded on from the live decode, a frame among the last _KEPT_FRAMES decoded -- a seek's approach included -- is looked up, and any other jump seeks the open container; a list, a slice and the streams take the stateless paths. release lets go of the decoder until the next read, and a reader dropped without close closes its container as it goes. A container that cannot be opened or has no usable timestamps is logged once and read statelessly. Every read takes a lock, since an av container cannot decode from two threads.

Parameters:

Name Type Description Default
path str | Path

The video file.

 required
gray_ok bool

Whether a frame with no color may come back single-channel.

 False
thread_count int | None

Decode threads; None takes _held_thread_count(), 0 FFmpeg's own choice.

 None
Source code in src/deeperfly/io/media/pyav.py 
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
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
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
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
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
class PyavHeldReader(PyavVideoReader):
    """One video file with the decoder held open between single-frame reads.

    A frame a short way ahead is decoded on from the live decode, a frame among the
    last `_KEPT_FRAMES` decoded -- a seek's approach included -- is looked up, and any
    other jump seeks the open container; a list, a slice and the streams take the
    stateless paths. `release` lets go of the decoder until the next read, and a
    reader dropped without `close` closes its container as it goes. A container that
    cannot be opened or has no usable timestamps is logged once and read statelessly.
    Every read takes a lock, since an `av` container cannot decode from two threads.

    Args:
        path: The video file.
        gray_ok: Whether a frame with no color may come back single-channel.
        thread_count: Decode threads; `None` takes `_held_thread_count()`, `0`
            FFmpeg's own choice.
    """

    def __init__(
        self,
        path: str | Path,
        *,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> None:
        """Note what to open on the first read; open nothing yet."""
        super().__init__(path)
        self._gray_ok = gray_ok
        self._thread_count = thread_count
        # Held for the whole of every read, so close cannot free the container out
        # from under a decode in flight -- it waits for the reader to leave.
        self._lock = threading.Lock()
        self._container: Any = None
        # Closes the container once, at `close`, `release` or when the reader is
        # dropped, whichever comes first.
        self._finalizer: Any = None
        self._stream: Any = None
        self._rate: Any = None
        self._time_base: Any = None
        self._closed = False
        # Set once the container turns out not to be indexable by timestamp; every
        # later read then goes straight down the stateless path.
        self._stateless = False
        # The live decode generator and the index it will yield next -- the pair
        # that makes a step forward free. `None` / `-1` means nothing to continue.
        self._gen: Iterator[Any] | None = None
        self._next = -1
        # The last frames decoded, by index, oldest first: what a step back reads.
        self._kept: OrderedDict[int, Any] = OrderedDict()

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """This reader: its decoder is already held open."""
        return self

    def _open(self) -> bool:
        """Open the container on first use; `False` if it cannot be indexed."""
        import av

        if self._closed:
            raise ValueError(f"reader over {self.path.name!r} is closed")
        if self._stateless:
            return False
        if self._container is not None:
            return True
        thread_count = self._thread_count
        if thread_count is None:
            thread_count = _held_thread_count()
        container = None
        try:
            container = av.open(str(self.path))
            stream = container.streams.video[0]
            stream.thread_type = "AUTO"
            if thread_count:
                try:
                    stream.codec_context.thread_count = thread_count
                except Exception as exc:  # noqa: BLE001 -- some codecs refuse
                    log.debug("could not cap decode threads for %s: %s", self.path, exc)
            rate = stream.average_rate or stream.guessed_rate
            if not rate or stream.time_base is None:
                raise ValueError(f"{self.path.name!r} has no frame rate to index by")
        except Exception as exc:  # noqa: BLE001 -- unreadable / no PTS mapping
            if container is not None:
                container.close()
            log.debug(
                "cannot hold %s open (%s); using stateless reads", self.path.name, exc
            )
            self._stateless = True
            return False
        self._container = container
        # The container and its stream reference each other, so a reader dropped
        # unclosed would hold the file and the decode threads until the cycle
        # collector ran; this closes them as the reader goes.
        self._finalizer = weakref.finalize(self, _close_container, container, self.path)
        # Not at exit, when a daemon thread may still be decoding: the OS frees it.
        # `atexit` is a documented writable attribute that typeshed's slots omit.
        self._finalizer.atexit = False  # pyrefly: ignore[missing-attribute]
        self._stream = stream
        self._rate = rate
        self._time_base = stream.time_base
        return True

    def _index_of(self, frame) -> int:
        """The frame's own index, recovered from its presentation timestamp."""
        if frame.pts is None:
            raise ValueError(f"{self.path.name!r} has frames with no timestamp")
        return round(float(frame.pts * self._time_base * self._rate))

    def _keep(self, idx: int, frame) -> None:
        """Hold one decoded frame, dropping the oldest past `_KEPT_FRAMES`.

        The first frame decoded at an index stays, which is the one a seek returns
        when irregular timestamps round two frames to one index.
        """
        self._kept.setdefault(idx, frame)
        self._kept.move_to_end(idx)
        while len(self._kept) > _KEPT_FRAMES:
            self._kept.popitem(last=False)

    def _seek_to(self, idx: int):
        """Seek to `idx`'s keyframe and decode forward to it, leaving the gen live.

        The last `_KEPT_FRAMES` frames of the approach, the target's included, are
        kept.
        """
        if idx < 0:
            raise IndexError(f"frame index {idx} is negative")
        self._gen = None
        ts = int(idx / self._rate / self._time_base)
        self._container.seek(ts, stream=self._stream, backward=True, any_frame=False)
        gen = self._container.decode(self._stream)
        for frame in gen:
            at = self._index_of(frame)
            if at > idx - _KEPT_FRAMES:
                self._keep(at, frame)
            if at >= idx:
                # Keep the generator: the next frame is very often asked for next.
                self._gen, self._next = gen, at + 1
                return frame
        raise ValueError(f"could not seek to frame {idx} of {str(self.path)!r}")

    def _walk(self, idx: int):
        """Decode on from the live generator to `idx`, keeping every frame it passes.

        `None` (the stream ended, or skipped past `idx`) means seek; each index is
        re-derived from the timestamp, so a skip never returns the wrong moment.
        """
        gen = self._gen
        if gen is None:
            return None
        while True:
            try:
                frame = next(gen)
            except StopIteration:
                self._gen = None
                return None
            at = self._index_of(frame)
            self._next = at + 1
            self._keep(at, frame)
            if at == idx:
                return frame
            if at > idx:
                self._gen = None
                return None

    def _held_frame(self, idx: int) -> UInt8[np.ndarray, "..."]:
        """Frame `idx` as `(H, W, 3)`, or `(H, W)` when held `gray_ok` and no color."""
        with self._lock:
            if not self._open():
                return super().__getitem__(idx)
            frame = self._kept.get(idx)
            if frame is not None:
                self._kept.move_to_end(idx)
            elif self._gen is not None and 0 <= idx - self._next < _WALK_FRAMES:
                frame = self._walk(idx)
            if frame is None:
                frame = self._seek_to(idx)
            return _to_array(frame, gray_ok=self._gray_ok)

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames; one frame comes off the held decoder, the rest inherited."""
        if isinstance(key, (int, np.integer)):
            return self._held_frame(int(key))
        return super().__getitem__(key)

    def release(self) -> None:
        """Let go of the decoder and the kept frames; the next read opens them again.

        Waits for a read in flight to finish, as `close` does.
        """
        with self._lock:
            self._drop()

    def close(self) -> None:
        """Close the container (idempotent), waiting for a read in flight to finish."""
        with self._lock:
            self._closed = True
            self._drop()

    def _drop(self) -> None:
        """Close the container and forget the decode; the caller holds `_lock`."""
        self._gen = None
        self._next = -1
        self._kept.clear()
        if self._finalizer is not None:
            self._finalizer()
            self._finalizer = None
        self._container = None

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames; one frame comes off the held decoder, the rest inherited.

Source code in src/deeperfly/io/media/pyav.py 
529
530
531
532
533
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames; one frame comes off the held decoder, the rest inherited."""
    if isinstance(key, (int, np.integer)):
        return self._held_frame(int(key))
    return super().__getitem__(key)

__init__ 

__init__(
    path: str | Path,
    *,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> None

Note what to open on the first read; open nothing yet.

Source code in src/deeperfly/io/media/pyav.py 
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
def __init__(
    self,
    path: str | Path,
    *,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> None:
    """Note what to open on the first read; open nothing yet."""
    super().__init__(path)
    self._gray_ok = gray_ok
    self._thread_count = thread_count
    # Held for the whole of every read, so close cannot free the container out
    # from under a decode in flight -- it waits for the reader to leave.
    self._lock = threading.Lock()
    self._container: Any = None
    # Closes the container once, at `close`, `release` or when the reader is
    # dropped, whichever comes first.
    self._finalizer: Any = None
    self._stream: Any = None
    self._rate: Any = None
    self._time_base: Any = None
    self._closed = False
    # Set once the container turns out not to be indexable by timestamp; every
    # later read then goes straight down the stateless path.
    self._stateless = False
    # The live decode generator and the index it will yield next -- the pair
    # that makes a step forward free. `None` / `-1` means nothing to continue.
    self._gen: Iterator[Any] | None = None
    self._next = -1
    # The last frames decoded, by index, oldest first: what a step back reads.
    self._kept: OrderedDict[int, Any] = OrderedDict()

close 

close() -> None

Close the container (idempotent), waiting for a read in flight to finish.

Source code in src/deeperfly/io/media/pyav.py 
543
544
545
546
547
def close(self) -> None:
    """Close the container (idempotent), waiting for a read in flight to finish."""
    with self._lock:
        self._closed = True
        self._drop()

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

This reader: its decoder is already held open.

Source code in src/deeperfly/io/media/pyav.py 
401
402
403
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """This reader: its decoder is already held open."""
    return self

release 

release() -> None

Let go of the decoder and the kept frames; the next read opens them again.

Waits for a read in flight to finish, as close does.

Source code in src/deeperfly/io/media/pyav.py 
535
536
537
538
539
540
541
def release(self) -> None:
    """Let go of the decoder and the kept frames; the next read opens them again.

    Waits for a read in flight to finish, as `close` does.
    """
    with self._lock:
        self._drop()

PyavVideoReader

Bases: FrameReader

Frame-accurate decode of one video file via PyAV.

A forward read seeks to its start's keyframe and decodes on from there, and a single index or a list seeks per target; count and fps read the container header. Every method opens and closes the file.

Source code in src/deeperfly/io/media/pyav.py 
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
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
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
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
347
348
class PyavVideoReader(FrameReader):
    """Frame-accurate decode of one video file via PyAV.

    A forward read seeks to its start's keyframe and decodes on from there, and a
    single index or a list seeks per target; `count` and `fps` read the container
    header. Every method opens and closes the
    file.
    """

    def __init__(self, path: str | Path) -> None:
        """Open a reader over the video file at `path` (no decode yet)."""
        self.path = Path(path)

    def _decode_stream(
        self, *, start=0, step=1, stop=None, gray_ok=False, thread_count=None
    ) -> Iterator[np.ndarray]:
        """Yield uint8 frames from one forward decode, FFmpeg-multithreaded.

        A `start` past 0 seeks to its keyframe first, as random access does; a file
        that cannot seek there is walked from frame 0 instead. `thread_count` caps the
        pool for a rig decoding many streams at once.
        """
        import av

        for seek in (True, False) if start > 0 else (False,):
            with av.open(str(self.path)) as container:
                stream = container.streams.video[0]
                stream.thread_type = "AUTO"
                if thread_count:
                    # Before the first decode: read when the codec opens.
                    stream.codec_context.thread_count = int(thread_count)
                if seek:
                    frames = _from_keyframe(container, stream, start)
                    if frames is None:
                        log.debug(
                            "could not seek to frame %d of %s; walking instead",
                            start,
                            self.path.name,
                        )
                        continue
                else:
                    frames = enumerate(container.decode(stream))
                for i, frame in frames:
                    if i < start:
                        continue
                    if stop is not None and i >= stop:
                        break
                    if (i - start) % step == 0:
                        yield _to_array(frame, gray_ok=gray_ok, keep_channel_axis=True)
                return

    def _decode_range(self, start, stop, step) -> np.ndarray:
        """Decode `range(start, stop, step)` to a stacked `(T, H, W, 3)` array."""
        out = list(self._decode_stream(start=start, step=step, stop=stop))
        if not out:
            raise ValueError(f"pyav decoded no frames from {str(self.path)!r}")
        return np.stack(out)

    def _decode_indices(self, indices) -> np.ndarray:
        """Random access: seek to each target's keyframe and decode forward to it.

        A frame's index comes from its PTS, and the result is in `indices`' order.
        """
        import av

        picked: dict[int, np.ndarray] = {}
        with av.open(str(self.path)) as container:
            stream = container.streams.video[0]
            stream.thread_type = "AUTO"
            rate = stream.average_rate or stream.guessed_rate
            time_base = stream.time_base
            assert rate is not None and time_base is not None
            for target in sorted(set(indices)):
                ts = int(target / rate / time_base)  # PTS of the target frame
                container.seek(ts, stream=stream, backward=True, any_frame=False)
                for frame in container.decode(stream):
                    assert frame.pts is not None
                    idx = round(float(frame.pts * time_base * rate))
                    if idx >= target:
                        picked[target] = frame.to_ndarray(format="rgb24")
                        break
        try:
            return np.stack([picked[int(i)] for i in indices])
        except KeyError as exc:  # a seek overshot / frame missing
            raise ValueError(
                f"pyav could not seek to frame {exc} of {str(self.path)!r}"
            ) from None

    def _decode_one(self, idx: int) -> np.ndarray:
        """One frame, by seeking to its keyframe, falling back to a forward walk.

        The walk serves a container that cannot seek or has no timestamps.
        """
        if idx >= 0:
            try:
                return self._decode_indices([idx])[0]
            except Exception as exc:  # noqa: BLE001 -- unseekable / no PTS
                log.debug(
                    "could not seek to frame %d of %s (%s); walking instead",
                    idx,
                    self.path.name,
                    exc,
                )
        return self._decode_range(idx, idx + 1, 1)[0]

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames (int / list / slice) into a uint8 RGB array."""
        if isinstance(key, (int, np.integer)):
            return self._decode_one(int(key))
        if isinstance(key, list):
            idx = [int(i) for i in key]
            if not idx:
                raise ValueError("index list must be non-empty")
            return self._decode_indices(idx)
        if isinstance(key, slice):
            return self._decode_slice(key)
        raise TypeError(f"invalid index type {type(key).__name__!r}")

    def _decode_slice(self, key: slice) -> np.ndarray:
        """Decode a slice, resolving negative bounds and steps against the length.

        A plain forward slice is one walk and needs no length; a reverse one is served
        by per-target seeks.
        """
        negative = any(v is not None and v < 0 for v in (key.start, key.stop, key.step))
        if not negative:
            start, stop, step = key.start or 0, key.stop, key.step or 1
            return self._decode_range(int(start), stop, int(step))
        n = self.count()
        if n is None:
            raise ValueError(
                f"cannot resolve slice {key} of {str(self.path)!r}: its container "
                "header carries no frame count to resolve negative indices against"
            )
        start, stop, step = key.indices(n)
        if step < 0:
            indices = list(range(start, stop, step))
            if not indices:
                raise ValueError(f"slice {key} selects no frames of {str(self.path)!r}")
            return self._decode_indices(indices)
        return self._decode_range(start, stop, step)

    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W C"]]:
        """Yield `(H, W, C)` uint8 frames from one forward decode."""
        yield from self._decode_stream(
            start=start,
            stop=stop,
            step=step,
            gray_ok=gray_ok,
            thread_count=thread_count,
        )

    def count(self) -> int | None:
        """The container header's frame count, or `None`; never `duration * fps`."""
        import av

        try:
            with av.open(str(self.path)) as container:
                n = container.streams.video[0].frames
        except Exception:  # noqa: BLE001 -- unreadable / unsupported container
            return None
        return int(n) if n and n > 0 else None

    def fps(self) -> float | None:
        """Average frame rate from the container header, or `None` if absent."""
        import av

        try:
            with av.open(str(self.path)) as container:
                stream = container.streams.video[0]
                rate = stream.average_rate or stream.guessed_rate
        except Exception:  # noqa: BLE001 -- unreadable / unsupported container
            return None
        return float(rate) if rate else None

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """The same file as a [PyavHeldReader][]."""
        return PyavHeldReader(self.path, gray_ok=gray_ok)

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames (int / list / slice) into a uint8 RGB array.

Source code in src/deeperfly/io/media/pyav.py 
268
269
270
271
272
273
274
275
276
277
278
279
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames (int / list / slice) into a uint8 RGB array."""
    if isinstance(key, (int, np.integer)):
        return self._decode_one(int(key))
    if isinstance(key, list):
        idx = [int(i) for i in key]
        if not idx:
            raise ValueError("index list must be non-empty")
        return self._decode_indices(idx)
    if isinstance(key, slice):
        return self._decode_slice(key)
    raise TypeError(f"invalid index type {type(key).__name__!r}")

__init__ 

__init__(path: str | Path) -> None

Open a reader over the video file at path (no decode yet).

Source code in src/deeperfly/io/media/pyav.py 
172
173
174
def __init__(self, path: str | Path) -> None:
    """Open a reader over the video file at `path` (no decode yet)."""
    self.path = Path(path)

count 

count() -> int | None

The container header's frame count, or None; never duration * fps.

Source code in src/deeperfly/io/media/pyav.py 
323
324
325
326
327
328
329
330
331
332
def count(self) -> int | None:
    """The container header's frame count, or `None`; never `duration * fps`."""
    import av

    try:
        with av.open(str(self.path)) as container:
            n = container.streams.video[0].frames
    except Exception:  # noqa: BLE001 -- unreadable / unsupported container
        return None
    return int(n) if n and n > 0 else None

fps 

fps() -> float | None

Average frame rate from the container header, or None if absent.

Source code in src/deeperfly/io/media/pyav.py 
334
335
336
337
338
339
340
341
342
343
344
def fps(self) -> float | None:
    """Average frame rate from the container header, or `None` if absent."""
    import av

    try:
        with av.open(str(self.path)) as container:
            stream = container.streams.video[0]
            rate = stream.average_rate or stream.guessed_rate
    except Exception:  # noqa: BLE001 -- unreadable / unsupported container
        return None
    return float(rate) if rate else None

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

The same file as a PyavHeldReader.

Source code in src/deeperfly/io/media/pyav.py 
346
347
348
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """The same file as a [PyavHeldReader][]."""
    return PyavHeldReader(self.path, gray_ok=gray_ok)

stream_frames 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]

Yield (H, W, C) uint8 frames from one forward decode.

Source code in src/deeperfly/io/media/pyav.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]:
    """Yield `(H, W, C)` uint8 frames from one forward decode."""
    yield from self._decode_stream(
        start=start,
        stop=stop,
        step=step,
        gray_ok=gray_ok,
        thread_count=thread_count,
    )

VideoWriter

Incremental H.264 (libx264) MP4 encoder, backed by PyAV.

The file opens on the first frame, whose size (rounded down to even for yuv420p) sets the encode dimensions; later frames are cropped to it, and non-uint8 input is clipped to [0, 255].

Parameters:

Name Type Description Default
path str | Path

Output file.

 required
fps float

Frame rate written into the container.

 30.0
codec str | None

Encoder, or None for libx264.

 None
pix_fmt str

Pixel format; yuv420p is the widely playable default.

 'yuv420p'
options Mapping[str, str] | None

Private encoder options as FFmpeg spells them ({"crf": "15", "preset": "veryslow"}); None keeps the encoder's defaults (libx264's crf 23). Frames a detector trains on want a lower crf than frames a person watches.

 None
Source code in src/deeperfly/io/media/writer.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
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
class VideoWriter:
    """Incremental H.264 (libx264) MP4 encoder, backed by PyAV.

    The file opens on the first frame, whose size (rounded down to even for `yuv420p`)
    sets the encode dimensions; later frames are cropped to it, and non-uint8 input is
    clipped to `[0, 255]`.

    Args:
        path: Output file.
        fps: Frame rate written into the container.
        codec: Encoder, or `None` for `libx264`.
        pix_fmt: Pixel format; `yuv420p` is the widely playable default.
        options: Private encoder options as FFmpeg spells them
            (`{"crf": "15", "preset": "veryslow"}`); `None` keeps the encoder's
            defaults (libx264's crf 23). Frames a detector trains on want a lower crf
            than frames a person watches.
    """

    def __init__(
        self,
        path: str | Path,
        fps: float = 30.0,
        *,
        codec: str | None = None,
        pix_fmt: str = "yuv420p",
        options: Mapping[str, str] | None = None,
    ) -> None:
        """Prepare a writer for `path` (no file is opened until the first frame)."""
        self.path = Path(path)
        self.fps = fps
        self.codec = codec
        self.pix_fmt = pix_fmt
        self.options = dict(options or {})
        self._container: Any = None  # av OutputContainer (lazy, on the 1st frame)
        self._stream: Any = None  # av VideoStream
        self._size: tuple[int, int] | None = None  # (w, h), even

    def __enter__(self) -> Self:
        """Enter the context, returning the writer."""
        return self

    def __exit__(self, *exc) -> None:
        """Flush and close on context exit."""
        self.close()

    def _open(self, width: int, height: int) -> None:
        import av

        w, h = width & ~1, height & ~1  # yuv420p subsampling needs even dimensions
        rate = Fraction(self.fps).limit_denominator(1_000_000)
        self._container = av.open(str(self.path), mode="w")
        stream = self._container.add_stream(self.codec or "libx264", rate=rate)
        if not isinstance(stream, av.VideoStream):
            raise TypeError(f"{self.codec!r} is not a video codec")
        stream.width = w
        stream.height = h
        stream.pix_fmt = self.pix_fmt
        if self.options:
            stream.options = dict(self.options)
        # Multithreaded encode (frame/slice), several times faster than the
        # single-thread default on a multi-core host; the file is unaffected.
        try:
            stream.thread_type = "AUTO"
        except Exception as exc:  # noqa: BLE001 -- some codecs reject it
            log.debug("multithreaded encode unavailable for %s: %s", self.path, exc)
        self._stream = stream
        self._size = (w, h)
        log.info("writing %s via pyav: %dx%d @ %g fps", self.path.name, w, h, self.fps)

    def write_frame(self, frame) -> None:
        """Append one `(H, W, 3)` RGB frame (NumPy, or a torch or DLPack array)."""
        import av

        frame = _to_numpy(frame)
        if frame.dtype != np.uint8:
            frame = np.clip(frame, 0, 255).astype(np.uint8)
        if self._stream is None:
            self._open(frame.shape[1], frame.shape[0])
        assert self._stream is not None and self._size is not None
        w, h = self._size
        frame = np.ascontiguousarray(frame[:h, :w])
        if self.pix_fmt == "yuv420p":
            # Converted here in the encoder's BT.601 limited range: faster than
            # handing PyAV RGB to convert, and half the bytes to copy.
            vframe = av.VideoFrame.from_ndarray(_i420(frame), format="yuv420p")
        else:
            vframe = av.VideoFrame.from_ndarray(frame, format="rgb24")
        for packet in self._stream.encode(vframe):
            self._container.mux(packet)

    def write_frames(self, frames) -> None:
        """Append a `(T, H, W, 3)` batch, or any iterable of frames or blocks."""
        if isinstance(frames, np.ndarray):
            if frames.ndim == 4:
                for frame in frames:
                    self.write_frame(frame)
            else:
                self.write_frame(frames)
            return
        if hasattr(frames, "detach") or hasattr(frames, "__dlpack__"):  # torch/array
            self.write_frames(_to_numpy(frames))
            return
        for item in frames:  # a list / tuple / generator of frames (or batches)
            self.write_frames(item)

    def close(self) -> None:
        """Flush the encoder and close the file (idempotent)."""
        if self._container is None:
            return
        try:
            for packet in self._stream.encode():  # flush
                self._container.mux(packet)
        finally:
            self._container.close()
            self._container = None
            self._stream = None

__enter__ 

__enter__() -> Self

Enter the context, returning the writer.

Source code in src/deeperfly/io/media/writer.py 
92
93
94
def __enter__(self) -> Self:
    """Enter the context, returning the writer."""
    return self

__exit__ 

__exit__(*exc) -> None

Flush and close on context exit.

Source code in src/deeperfly/io/media/writer.py 
96
97
98
def __exit__(self, *exc) -> None:
    """Flush and close on context exit."""
    self.close()

__init__ 

__init__(
    path: str | Path,
    fps: float = 30.0,
    *,
    codec: str | None = None,
    pix_fmt: str = "yuv420p",
    options: Mapping[str, str] | None = None,
) -> None

Prepare a writer for path (no file is opened until the first frame).

Source code in src/deeperfly/io/media/writer.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    path: str | Path,
    fps: float = 30.0,
    *,
    codec: str | None = None,
    pix_fmt: str = "yuv420p",
    options: Mapping[str, str] | None = None,
) -> None:
    """Prepare a writer for `path` (no file is opened until the first frame)."""
    self.path = Path(path)
    self.fps = fps
    self.codec = codec
    self.pix_fmt = pix_fmt
    self.options = dict(options or {})
    self._container: Any = None  # av OutputContainer (lazy, on the 1st frame)
    self._stream: Any = None  # av VideoStream
    self._size: tuple[int, int] | None = None  # (w, h), even

close 

close() -> None

Flush the encoder and close the file (idempotent).

Source code in src/deeperfly/io/media/writer.py 
160
161
162
163
164
165
166
167
168
169
170
def close(self) -> None:
    """Flush the encoder and close the file (idempotent)."""
    if self._container is None:
        return
    try:
        for packet in self._stream.encode():  # flush
            self._container.mux(packet)
    finally:
        self._container.close()
        self._container = None
        self._stream = None

write_frame 

write_frame(frame) -> None

Append one (H, W, 3) RGB frame (NumPy, or a torch or DLPack array).

Source code in src/deeperfly/io/media/writer.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def write_frame(self, frame) -> None:
    """Append one `(H, W, 3)` RGB frame (NumPy, or a torch or DLPack array)."""
    import av

    frame = _to_numpy(frame)
    if frame.dtype != np.uint8:
        frame = np.clip(frame, 0, 255).astype(np.uint8)
    if self._stream is None:
        self._open(frame.shape[1], frame.shape[0])
    assert self._stream is not None and self._size is not None
    w, h = self._size
    frame = np.ascontiguousarray(frame[:h, :w])
    if self.pix_fmt == "yuv420p":
        # Converted here in the encoder's BT.601 limited range: faster than
        # handing PyAV RGB to convert, and half the bytes to copy.
        vframe = av.VideoFrame.from_ndarray(_i420(frame), format="yuv420p")
    else:
        vframe = av.VideoFrame.from_ndarray(frame, format="rgb24")
    for packet in self._stream.encode(vframe):
        self._container.mux(packet)

write_frames 

write_frames(frames) -> None

Append a (T, H, W, 3) batch, or any iterable of frames or blocks.

Source code in src/deeperfly/io/media/writer.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def write_frames(self, frames) -> None:
    """Append a `(T, H, W, 3)` batch, or any iterable of frames or blocks."""
    if isinstance(frames, np.ndarray):
        if frames.ndim == 4:
            for frame in frames:
                self.write_frame(frame)
        else:
            self.write_frame(frames)
        return
    if hasattr(frames, "detach") or hasattr(frames, "__dlpack__"):  # torch/array
        self.write_frames(_to_numpy(frames))
        return
    for item in frames:  # a list / tuple / generator of frames (or batches)
        self.write_frames(item)

is_video_file 

is_video_file(path: str | Path) -> bool

Whether path is an existing video file (decoded as video, not an image).

Source code in src/deeperfly/io/media/__init__.py 
51
52
53
54
def is_video_file(path: str | Path) -> bool:
    """Whether `path` is an existing video file (decoded as video, not an image)."""
    p = Path(path)
    return p.is_file() and p.suffix.lower() in VIDEO_EXTS

list_image_files 

list_image_files(pattern: str | Path) -> list[Path]

Sorted image files for a directory or glob pattern, ordered by name.

Raises:

Type Description
FileNotFoundError

If nothing matches pattern.
Source code in src/deeperfly/io/media/images.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def list_image_files(pattern: str | Path) -> list[Path]:
    """Sorted image files for a directory or glob pattern, ordered by name.

    Raises:
        FileNotFoundError: If nothing matches `pattern`.
    """
    p = Path(pattern)
    if p.is_dir():
        files = sorted(f for f in p.iterdir() if f.suffix.lower() in IMAGE_EXTS)
    else:
        files = sorted(Path(f) for f in glob.glob(str(pattern)))
    if not files:
        raise FileNotFoundError(f"no images matched {pattern!r}")
    return files

open_reader 

open_reader(
    source: str | Path | list[Path],
    *,
    backend: str = DEFAULT_BACKEND,
    workers: int | None = None,
) -> FrameReader

Open the right FrameReader for a source.

A video file opens with the video backend, several video files (the parts of one split video) as a ConcatReader, and a directory, glob or list of images as an ImageSequenceReader decoding with workers threads.

Source code in src/deeperfly/io/media/__init__.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def open_reader(
    source: str | Path | list[Path],
    *,
    backend: str = DEFAULT_BACKEND,
    workers: int | None = None,
) -> FrameReader:
    """Open the right [FrameReader][deeperfly.io.media.reader.FrameReader] for a source.

    A video file opens with the video `backend`, several video files (the parts of one
    split video) as a [ConcatReader][deeperfly.io.media.concat.ConcatReader], and a
    directory, glob or list of images as an
    [ImageSequenceReader][deeperfly.io.media.images.ImageSequenceReader] decoding with
    `workers` threads.
    """
    return _open_source(source, backend=backend, workers=workers)

The backend-agnostic frame-reading interface, FrameReader (ADR-0005).

A reader is stateless by default: it opens the source, decodes and lets go, so any number of callers may index one. A viewer, reading single frames in its own order, asks for hold_open instead and gets a reader that keeps its decoder between reads. A backend implements __getitem__, stream_frames and count; the base class builds the rest on a forward decode.

FrameReader

Bases: ABC

Reads one source as (H, W, 3) uint8 RGB frames, on the CPU.

reader[5] is one frame; reader[[0, 3, 5]] and reader[2:8:2] are (T, H, W, 3) in the order asked; stream_frames and stream_blocks iterate lazily and take gray_ok. A reader is a context manager, and close is a no-op for the stateless ones.

Source code in src/deeperfly/io/media/reader.py 
 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
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
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
class FrameReader(ABC):
    """Reads one source as `(H, W, 3)` uint8 RGB frames, on the CPU.

    `reader[5]` is one frame; `reader[[0, 3, 5]]` and `reader[2:8:2]` are
    `(T, H, W, 3)` in the order asked; `stream_frames` and `stream_blocks` iterate
    lazily and take `gray_ok`. A reader is a context manager, and `close` is a no-op
    for the stateless ones.
    """

    def __enter__(self) -> Self:
        """Enter the context, returning the reader."""
        return self

    def __exit__(self, *exc) -> None:
        """Release resources on context exit."""
        self.close()

    def close(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
        """Release any resources held by the reader (a no-op by default)."""

    @abstractmethod
    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode one frame (`int`), several in order (`list[int]`), or a `slice`."""

    @abstractmethod
    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W C"]]:
        """Yield `(H, W, C)` uint8 frames from one forward decode.

        `start`, `stop` and `step` are as for `range`. `gray_ok` permits a single
        channel, kept as `(H, W, 1)`, for a frame that carries no color, which skips the
        color conversion; a source that cannot tell cheaply returns RGB.
        `thread_count` caps the decode threads, worth doing when several sources decode
        at once.
        """

    def stream_blocks(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        block_size: int = 64,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "T H W C"]]:
        """Yield `(T, H, W, C)` uint8 blocks (`T <= block_size`) from one decode.

        The default buffers `stream_frames` into blocks, with its arguments; a backend
        with a cheaper block decode overrides it.
        """
        if block_size < 1:
            raise ValueError(f"block_size must be >= 1, got {block_size}")
        buf: list[np.ndarray] = []
        for frame in self.stream_frames(
            start=start,
            stop=stop,
            step=step,
            gray_ok=gray_ok,
            thread_count=thread_count,
        ):
            buf.append(frame)
            if len(buf) >= block_size:
                yield _stack_frames(buf)
                buf = []
        if buf:
            yield _stack_frames(buf)

    @abstractmethod
    def count(self) -> int | None:
        """Best-effort frame count, or `None` when unknown: a progress hint only."""

    def fps(self) -> float | None:
        """Frame rate in frames per second, or `None` (an image sequence has none)."""
        return None

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """This source again, with its decoder held open between reads.

        For a viewer, which would otherwise pay a re-open and a seek per frame. The
        base returns `self` and a backend overrides it. The caller owns the result,
        which may be this same object, and closes it. `gray_ok` lets indexing one frame
        return `(H, W)` for a source that carries no color.
        """
        return self

    def release(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
        """Let go of what a held-open reader holds between reads; it reopens on demand.

        For a viewer that keeps readers of footage no longer on screen. The base holds
        nothing, so there is nothing to let go of.
        """

__enter__ 

__enter__() -> Self

Enter the context, returning the reader.

Source code in src/deeperfly/io/media/reader.py 
40
41
42
def __enter__(self) -> Self:
    """Enter the context, returning the reader."""
    return self

__exit__ 

__exit__(*exc) -> None

Release resources on context exit.

Source code in src/deeperfly/io/media/reader.py 
44
45
46
def __exit__(self, *exc) -> None:
    """Release resources on context exit."""
    self.close()

__getitem__ abstractmethod 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode one frame (int), several in order (list[int]), or a slice.

Source code in src/deeperfly/io/media/reader.py 
51
52
53
@abstractmethod
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode one frame (`int`), several in order (`list[int]`), or a `slice`."""

close 

close() -> None

Release any resources held by the reader (a no-op by default).

Source code in src/deeperfly/io/media/reader.py 
48
49
def close(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
    """Release any resources held by the reader (a no-op by default)."""

count abstractmethod 

count() -> int | None

Best-effort frame count, or None when unknown: a progress hint only.

Source code in src/deeperfly/io/media/reader.py 
106
107
108
@abstractmethod
def count(self) -> int | None:
    """Best-effort frame count, or `None` when unknown: a progress hint only."""

fps 

fps() -> float | None

Frame rate in frames per second, or None (an image sequence has none).

Source code in src/deeperfly/io/media/reader.py 
110
111
112
def fps(self) -> float | None:
    """Frame rate in frames per second, or `None` (an image sequence has none)."""
    return None

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

This source again, with its decoder held open between reads.

For a viewer, which would otherwise pay a re-open and a seek per frame. The base returns self and a backend overrides it. The caller owns the result, which may be this same object, and closes it. gray_ok lets indexing one frame return (H, W) for a source that carries no color.

Source code in src/deeperfly/io/media/reader.py 
114
115
116
117
118
119
120
121
122
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """This source again, with its decoder held open between reads.

    For a viewer, which would otherwise pay a re-open and a seek per frame. The
    base returns `self` and a backend overrides it. The caller owns the result,
    which may be this same object, and closes it. `gray_ok` lets indexing one frame
    return `(H, W)` for a source that carries no color.
    """
    return self

release 

release() -> None

Let go of what a held-open reader holds between reads; it reopens on demand.

For a viewer that keeps readers of footage no longer on screen. The base holds nothing, so there is nothing to let go of.

Source code in src/deeperfly/io/media/reader.py 
124
125
126
127
128
129
def release(self) -> None:  # noqa: B027 -- a hook: a no-op unless overridden
    """Let go of what a held-open reader holds between reads; it reopens on demand.

    For a viewer that keeps readers of footage no longer on screen. The base holds
    nothing, so there is nothing to let go of.
    """

stream_blocks 

stream_blocks(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W C"]]

Yield (T, H, W, C) uint8 blocks (T <= block_size) from one decode.

The default buffers stream_frames into blocks, with its arguments; a backend with a cheaper block decode overrides it.

Source code in src/deeperfly/io/media/reader.py 
 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
104
def stream_blocks(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W C"]]:
    """Yield `(T, H, W, C)` uint8 blocks (`T <= block_size`) from one decode.

    The default buffers `stream_frames` into blocks, with its arguments; a backend
    with a cheaper block decode overrides it.
    """
    if block_size < 1:
        raise ValueError(f"block_size must be >= 1, got {block_size}")
    buf: list[np.ndarray] = []
    for frame in self.stream_frames(
        start=start,
        stop=stop,
        step=step,
        gray_ok=gray_ok,
        thread_count=thread_count,
    ):
        buf.append(frame)
        if len(buf) >= block_size:
            yield _stack_frames(buf)
            buf = []
    if buf:
        yield _stack_frames(buf)

stream_frames abstractmethod 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]

Yield (H, W, C) uint8 frames from one forward decode.

start, stop and step are as for range. gray_ok permits a single channel, kept as (H, W, 1), for a frame that carries no color, which skips the color conversion; a source that cannot tell cheaply returns RGB. thread_count caps the decode threads, worth doing when several sources decode at once.

Source code in src/deeperfly/io/media/reader.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@abstractmethod
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]:
    """Yield `(H, W, C)` uint8 frames from one forward decode.

    `start`, `stop` and `step` are as for `range`. `gray_ok` permits a single
    channel, kept as `(H, W, 1)`, for a frame that carries no color, which skips the
    color conversion; a source that cannot tell cheaply returns RGB.
    `thread_count` caps the decode threads, worth doing when several sources decode
    at once.
    """

The PyAV video backend: frame-accurate decode of one video file (ADR-0006).

PyavVideoReader decodes to uint8 RGB with per-target seeks; PyavHeldReader keeps the decoder open for a viewer, and the frames it last decoded. With gray_ok, a full-range frame with flat chroma (every fly-rig camera records grayscale as YUV) comes back as its luma plane, which skips the YUV->RGB conversion, 100-300x the plane copy's cost. TV range and a frame with color decode as RGB; the test is per frame, since a color video can open on black.

PyavHeldReader

Bases: PyavVideoReader

One video file with the decoder held open between single-frame reads.

A frame a short way ahead is decoded on from the live decode, a frame among the last _KEPT_FRAMES decoded -- a seek's approach included -- is looked up, and any other jump seeks the open container; a list, a slice and the streams take the stateless paths. release lets go of the decoder until the next read, and a reader dropped without close closes its container as it goes. A container that cannot be opened or has no usable timestamps is logged once and read statelessly. Every read takes a lock, since an av container cannot decode from two threads.

Parameters:

Name Type Description Default
path str | Path

The video file.

 required
gray_ok bool

Whether a frame with no color may come back single-channel.

 False
thread_count int | None

Decode threads; None takes _held_thread_count(), 0 FFmpeg's own choice.

 None
Source code in src/deeperfly/io/media/pyav.py 
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
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
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
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
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
class PyavHeldReader(PyavVideoReader):
    """One video file with the decoder held open between single-frame reads.

    A frame a short way ahead is decoded on from the live decode, a frame among the
    last `_KEPT_FRAMES` decoded -- a seek's approach included -- is looked up, and any
    other jump seeks the open container; a list, a slice and the streams take the
    stateless paths. `release` lets go of the decoder until the next read, and a
    reader dropped without `close` closes its container as it goes. A container that
    cannot be opened or has no usable timestamps is logged once and read statelessly.
    Every read takes a lock, since an `av` container cannot decode from two threads.

    Args:
        path: The video file.
        gray_ok: Whether a frame with no color may come back single-channel.
        thread_count: Decode threads; `None` takes `_held_thread_count()`, `0`
            FFmpeg's own choice.
    """

    def __init__(
        self,
        path: str | Path,
        *,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> None:
        """Note what to open on the first read; open nothing yet."""
        super().__init__(path)
        self._gray_ok = gray_ok
        self._thread_count = thread_count
        # Held for the whole of every read, so close cannot free the container out
        # from under a decode in flight -- it waits for the reader to leave.
        self._lock = threading.Lock()
        self._container: Any = None
        # Closes the container once, at `close`, `release` or when the reader is
        # dropped, whichever comes first.
        self._finalizer: Any = None
        self._stream: Any = None
        self._rate: Any = None
        self._time_base: Any = None
        self._closed = False
        # Set once the container turns out not to be indexable by timestamp; every
        # later read then goes straight down the stateless path.
        self._stateless = False
        # The live decode generator and the index it will yield next -- the pair
        # that makes a step forward free. `None` / `-1` means nothing to continue.
        self._gen: Iterator[Any] | None = None
        self._next = -1
        # The last frames decoded, by index, oldest first: what a step back reads.
        self._kept: OrderedDict[int, Any] = OrderedDict()

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """This reader: its decoder is already held open."""
        return self

    def _open(self) -> bool:
        """Open the container on first use; `False` if it cannot be indexed."""
        import av

        if self._closed:
            raise ValueError(f"reader over {self.path.name!r} is closed")
        if self._stateless:
            return False
        if self._container is not None:
            return True
        thread_count = self._thread_count
        if thread_count is None:
            thread_count = _held_thread_count()
        container = None
        try:
            container = av.open(str(self.path))
            stream = container.streams.video[0]
            stream.thread_type = "AUTO"
            if thread_count:
                try:
                    stream.codec_context.thread_count = thread_count
                except Exception as exc:  # noqa: BLE001 -- some codecs refuse
                    log.debug("could not cap decode threads for %s: %s", self.path, exc)
            rate = stream.average_rate or stream.guessed_rate
            if not rate or stream.time_base is None:
                raise ValueError(f"{self.path.name!r} has no frame rate to index by")
        except Exception as exc:  # noqa: BLE001 -- unreadable / no PTS mapping
            if container is not None:
                container.close()
            log.debug(
                "cannot hold %s open (%s); using stateless reads", self.path.name, exc
            )
            self._stateless = True
            return False
        self._container = container
        # The container and its stream reference each other, so a reader dropped
        # unclosed would hold the file and the decode threads until the cycle
        # collector ran; this closes them as the reader goes.
        self._finalizer = weakref.finalize(self, _close_container, container, self.path)
        # Not at exit, when a daemon thread may still be decoding: the OS frees it.
        # `atexit` is a documented writable attribute that typeshed's slots omit.
        self._finalizer.atexit = False  # pyrefly: ignore[missing-attribute]
        self._stream = stream
        self._rate = rate
        self._time_base = stream.time_base
        return True

    def _index_of(self, frame) -> int:
        """The frame's own index, recovered from its presentation timestamp."""
        if frame.pts is None:
            raise ValueError(f"{self.path.name!r} has frames with no timestamp")
        return round(float(frame.pts * self._time_base * self._rate))

    def _keep(self, idx: int, frame) -> None:
        """Hold one decoded frame, dropping the oldest past `_KEPT_FRAMES`.

        The first frame decoded at an index stays, which is the one a seek returns
        when irregular timestamps round two frames to one index.
        """
        self._kept.setdefault(idx, frame)
        self._kept.move_to_end(idx)
        while len(self._kept) > _KEPT_FRAMES:
            self._kept.popitem(last=False)

    def _seek_to(self, idx: int):
        """Seek to `idx`'s keyframe and decode forward to it, leaving the gen live.

        The last `_KEPT_FRAMES` frames of the approach, the target's included, are
        kept.
        """
        if idx < 0:
            raise IndexError(f"frame index {idx} is negative")
        self._gen = None
        ts = int(idx / self._rate / self._time_base)
        self._container.seek(ts, stream=self._stream, backward=True, any_frame=False)
        gen = self._container.decode(self._stream)
        for frame in gen:
            at = self._index_of(frame)
            if at > idx - _KEPT_FRAMES:
                self._keep(at, frame)
            if at >= idx:
                # Keep the generator: the next frame is very often asked for next.
                self._gen, self._next = gen, at + 1
                return frame
        raise ValueError(f"could not seek to frame {idx} of {str(self.path)!r}")

    def _walk(self, idx: int):
        """Decode on from the live generator to `idx`, keeping every frame it passes.

        `None` (the stream ended, or skipped past `idx`) means seek; each index is
        re-derived from the timestamp, so a skip never returns the wrong moment.
        """
        gen = self._gen
        if gen is None:
            return None
        while True:
            try:
                frame = next(gen)
            except StopIteration:
                self._gen = None
                return None
            at = self._index_of(frame)
            self._next = at + 1
            self._keep(at, frame)
            if at == idx:
                return frame
            if at > idx:
                self._gen = None
                return None

    def _held_frame(self, idx: int) -> UInt8[np.ndarray, "..."]:
        """Frame `idx` as `(H, W, 3)`, or `(H, W)` when held `gray_ok` and no color."""
        with self._lock:
            if not self._open():
                return super().__getitem__(idx)
            frame = self._kept.get(idx)
            if frame is not None:
                self._kept.move_to_end(idx)
            elif self._gen is not None and 0 <= idx - self._next < _WALK_FRAMES:
                frame = self._walk(idx)
            if frame is None:
                frame = self._seek_to(idx)
            return _to_array(frame, gray_ok=self._gray_ok)

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames; one frame comes off the held decoder, the rest inherited."""
        if isinstance(key, (int, np.integer)):
            return self._held_frame(int(key))
        return super().__getitem__(key)

    def release(self) -> None:
        """Let go of the decoder and the kept frames; the next read opens them again.

        Waits for a read in flight to finish, as `close` does.
        """
        with self._lock:
            self._drop()

    def close(self) -> None:
        """Close the container (idempotent), waiting for a read in flight to finish."""
        with self._lock:
            self._closed = True
            self._drop()

    def _drop(self) -> None:
        """Close the container and forget the decode; the caller holds `_lock`."""
        self._gen = None
        self._next = -1
        self._kept.clear()
        if self._finalizer is not None:
            self._finalizer()
            self._finalizer = None
        self._container = None

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames; one frame comes off the held decoder, the rest inherited.

Source code in src/deeperfly/io/media/pyav.py 
529
530
531
532
533
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames; one frame comes off the held decoder, the rest inherited."""
    if isinstance(key, (int, np.integer)):
        return self._held_frame(int(key))
    return super().__getitem__(key)

__init__ 

__init__(
    path: str | Path,
    *,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> None

Note what to open on the first read; open nothing yet.

Source code in src/deeperfly/io/media/pyav.py 
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
def __init__(
    self,
    path: str | Path,
    *,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> None:
    """Note what to open on the first read; open nothing yet."""
    super().__init__(path)
    self._gray_ok = gray_ok
    self._thread_count = thread_count
    # Held for the whole of every read, so close cannot free the container out
    # from under a decode in flight -- it waits for the reader to leave.
    self._lock = threading.Lock()
    self._container: Any = None
    # Closes the container once, at `close`, `release` or when the reader is
    # dropped, whichever comes first.
    self._finalizer: Any = None
    self._stream: Any = None
    self._rate: Any = None
    self._time_base: Any = None
    self._closed = False
    # Set once the container turns out not to be indexable by timestamp; every
    # later read then goes straight down the stateless path.
    self._stateless = False
    # The live decode generator and the index it will yield next -- the pair
    # that makes a step forward free. `None` / `-1` means nothing to continue.
    self._gen: Iterator[Any] | None = None
    self._next = -1
    # The last frames decoded, by index, oldest first: what a step back reads.
    self._kept: OrderedDict[int, Any] = OrderedDict()

close 

close() -> None

Close the container (idempotent), waiting for a read in flight to finish.

Source code in src/deeperfly/io/media/pyav.py 
543
544
545
546
547
def close(self) -> None:
    """Close the container (idempotent), waiting for a read in flight to finish."""
    with self._lock:
        self._closed = True
        self._drop()

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

This reader: its decoder is already held open.

Source code in src/deeperfly/io/media/pyav.py 
401
402
403
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """This reader: its decoder is already held open."""
    return self

release 

release() -> None

Let go of the decoder and the kept frames; the next read opens them again.

Waits for a read in flight to finish, as close does.

Source code in src/deeperfly/io/media/pyav.py 
535
536
537
538
539
540
541
def release(self) -> None:
    """Let go of the decoder and the kept frames; the next read opens them again.

    Waits for a read in flight to finish, as `close` does.
    """
    with self._lock:
        self._drop()

PyavVideoReader

Bases: FrameReader

Frame-accurate decode of one video file via PyAV.

A forward read seeks to its start's keyframe and decodes on from there, and a single index or a list seeks per target; count and fps read the container header. Every method opens and closes the file.

Source code in src/deeperfly/io/media/pyav.py 
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
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
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
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
347
348
class PyavVideoReader(FrameReader):
    """Frame-accurate decode of one video file via PyAV.

    A forward read seeks to its start's keyframe and decodes on from there, and a
    single index or a list seeks per target; `count` and `fps` read the container
    header. Every method opens and closes the
    file.
    """

    def __init__(self, path: str | Path) -> None:
        """Open a reader over the video file at `path` (no decode yet)."""
        self.path = Path(path)

    def _decode_stream(
        self, *, start=0, step=1, stop=None, gray_ok=False, thread_count=None
    ) -> Iterator[np.ndarray]:
        """Yield uint8 frames from one forward decode, FFmpeg-multithreaded.

        A `start` past 0 seeks to its keyframe first, as random access does; a file
        that cannot seek there is walked from frame 0 instead. `thread_count` caps the
        pool for a rig decoding many streams at once.
        """
        import av

        for seek in (True, False) if start > 0 else (False,):
            with av.open(str(self.path)) as container:
                stream = container.streams.video[0]
                stream.thread_type = "AUTO"
                if thread_count:
                    # Before the first decode: read when the codec opens.
                    stream.codec_context.thread_count = int(thread_count)
                if seek:
                    frames = _from_keyframe(container, stream, start)
                    if frames is None:
                        log.debug(
                            "could not seek to frame %d of %s; walking instead",
                            start,
                            self.path.name,
                        )
                        continue
                else:
                    frames = enumerate(container.decode(stream))
                for i, frame in frames:
                    if i < start:
                        continue
                    if stop is not None and i >= stop:
                        break
                    if (i - start) % step == 0:
                        yield _to_array(frame, gray_ok=gray_ok, keep_channel_axis=True)
                return

    def _decode_range(self, start, stop, step) -> np.ndarray:
        """Decode `range(start, stop, step)` to a stacked `(T, H, W, 3)` array."""
        out = list(self._decode_stream(start=start, step=step, stop=stop))
        if not out:
            raise ValueError(f"pyav decoded no frames from {str(self.path)!r}")
        return np.stack(out)

    def _decode_indices(self, indices) -> np.ndarray:
        """Random access: seek to each target's keyframe and decode forward to it.

        A frame's index comes from its PTS, and the result is in `indices`' order.
        """
        import av

        picked: dict[int, np.ndarray] = {}
        with av.open(str(self.path)) as container:
            stream = container.streams.video[0]
            stream.thread_type = "AUTO"
            rate = stream.average_rate or stream.guessed_rate
            time_base = stream.time_base
            assert rate is not None and time_base is not None
            for target in sorted(set(indices)):
                ts = int(target / rate / time_base)  # PTS of the target frame
                container.seek(ts, stream=stream, backward=True, any_frame=False)
                for frame in container.decode(stream):
                    assert frame.pts is not None
                    idx = round(float(frame.pts * time_base * rate))
                    if idx >= target:
                        picked[target] = frame.to_ndarray(format="rgb24")
                        break
        try:
            return np.stack([picked[int(i)] for i in indices])
        except KeyError as exc:  # a seek overshot / frame missing
            raise ValueError(
                f"pyav could not seek to frame {exc} of {str(self.path)!r}"
            ) from None

    def _decode_one(self, idx: int) -> np.ndarray:
        """One frame, by seeking to its keyframe, falling back to a forward walk.

        The walk serves a container that cannot seek or has no timestamps.
        """
        if idx >= 0:
            try:
                return self._decode_indices([idx])[0]
            except Exception as exc:  # noqa: BLE001 -- unseekable / no PTS
                log.debug(
                    "could not seek to frame %d of %s (%s); walking instead",
                    idx,
                    self.path.name,
                    exc,
                )
        return self._decode_range(idx, idx + 1, 1)[0]

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames (int / list / slice) into a uint8 RGB array."""
        if isinstance(key, (int, np.integer)):
            return self._decode_one(int(key))
        if isinstance(key, list):
            idx = [int(i) for i in key]
            if not idx:
                raise ValueError("index list must be non-empty")
            return self._decode_indices(idx)
        if isinstance(key, slice):
            return self._decode_slice(key)
        raise TypeError(f"invalid index type {type(key).__name__!r}")

    def _decode_slice(self, key: slice) -> np.ndarray:
        """Decode a slice, resolving negative bounds and steps against the length.

        A plain forward slice is one walk and needs no length; a reverse one is served
        by per-target seeks.
        """
        negative = any(v is not None and v < 0 for v in (key.start, key.stop, key.step))
        if not negative:
            start, stop, step = key.start or 0, key.stop, key.step or 1
            return self._decode_range(int(start), stop, int(step))
        n = self.count()
        if n is None:
            raise ValueError(
                f"cannot resolve slice {key} of {str(self.path)!r}: its container "
                "header carries no frame count to resolve negative indices against"
            )
        start, stop, step = key.indices(n)
        if step < 0:
            indices = list(range(start, stop, step))
            if not indices:
                raise ValueError(f"slice {key} selects no frames of {str(self.path)!r}")
            return self._decode_indices(indices)
        return self._decode_range(start, stop, step)

    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W C"]]:
        """Yield `(H, W, C)` uint8 frames from one forward decode."""
        yield from self._decode_stream(
            start=start,
            stop=stop,
            step=step,
            gray_ok=gray_ok,
            thread_count=thread_count,
        )

    def count(self) -> int | None:
        """The container header's frame count, or `None`; never `duration * fps`."""
        import av

        try:
            with av.open(str(self.path)) as container:
                n = container.streams.video[0].frames
        except Exception:  # noqa: BLE001 -- unreadable / unsupported container
            return None
        return int(n) if n and n > 0 else None

    def fps(self) -> float | None:
        """Average frame rate from the container header, or `None` if absent."""
        import av

        try:
            with av.open(str(self.path)) as container:
                stream = container.streams.video[0]
                rate = stream.average_rate or stream.guessed_rate
        except Exception:  # noqa: BLE001 -- unreadable / unsupported container
            return None
        return float(rate) if rate else None

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """The same file as a [PyavHeldReader][]."""
        return PyavHeldReader(self.path, gray_ok=gray_ok)

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames (int / list / slice) into a uint8 RGB array.

Source code in src/deeperfly/io/media/pyav.py 
268
269
270
271
272
273
274
275
276
277
278
279
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames (int / list / slice) into a uint8 RGB array."""
    if isinstance(key, (int, np.integer)):
        return self._decode_one(int(key))
    if isinstance(key, list):
        idx = [int(i) for i in key]
        if not idx:
            raise ValueError("index list must be non-empty")
        return self._decode_indices(idx)
    if isinstance(key, slice):
        return self._decode_slice(key)
    raise TypeError(f"invalid index type {type(key).__name__!r}")

__init__ 

__init__(path: str | Path) -> None

Open a reader over the video file at path (no decode yet).

Source code in src/deeperfly/io/media/pyav.py 
172
173
174
def __init__(self, path: str | Path) -> None:
    """Open a reader over the video file at `path` (no decode yet)."""
    self.path = Path(path)

count 

count() -> int | None

The container header's frame count, or None; never duration * fps.

Source code in src/deeperfly/io/media/pyav.py 
323
324
325
326
327
328
329
330
331
332
def count(self) -> int | None:
    """The container header's frame count, or `None`; never `duration * fps`."""
    import av

    try:
        with av.open(str(self.path)) as container:
            n = container.streams.video[0].frames
    except Exception:  # noqa: BLE001 -- unreadable / unsupported container
        return None
    return int(n) if n and n > 0 else None

fps 

fps() -> float | None

Average frame rate from the container header, or None if absent.

Source code in src/deeperfly/io/media/pyav.py 
334
335
336
337
338
339
340
341
342
343
344
def fps(self) -> float | None:
    """Average frame rate from the container header, or `None` if absent."""
    import av

    try:
        with av.open(str(self.path)) as container:
            stream = container.streams.video[0]
            rate = stream.average_rate or stream.guessed_rate
    except Exception:  # noqa: BLE001 -- unreadable / unsupported container
        return None
    return float(rate) if rate else None

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

The same file as a PyavHeldReader.

Source code in src/deeperfly/io/media/pyav.py 
346
347
348
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """The same file as a [PyavHeldReader][]."""
    return PyavHeldReader(self.path, gray_ok=gray_ok)

stream_frames 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]

Yield (H, W, C) uint8 frames from one forward decode.

Source code in src/deeperfly/io/media/pyav.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]:
    """Yield `(H, W, C)` uint8 frames from one forward decode."""
    yield from self._decode_stream(
        start=start,
        stop=stop,
        step=step,
        gray_ok=gray_ok,
        thread_count=thread_count,
    )

Several video files read as one stream: ConcatReader.

A long capture split across files is one stream with a global frame index. Its failure mode is silent, a boundary off by one shifting every later frame, so every part's count is exact (decoded when the header has none, never duration * fps) and the parts must agree on frame size and rate.

ConcatReader

Bases: FrameReader

The parts of one video, read as a single stream.

Parameters:

Name Type Description Default
paths

The parts, in decode order, two or more.

 required
factory Callable[[Path], FrameReader]

Builds a part's reader from its path: the video backend.

 required
counts list[int] | None

Each part's exact frame count if already known, which skips the measuring decode.

 None

Attributes:

Name Type Description
paths

The parts, in decode order.
Source code in src/deeperfly/io/media/concat.py 
 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
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
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
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
class ConcatReader(FrameReader):
    """The parts of one video, read as a single stream.

    Args:
        paths: The parts, in decode order, two or more.
        factory: Builds a part's reader from its path: the video backend.
        counts: Each part's exact frame count if already known, which skips the
            measuring decode.

    Attributes:
        paths: The parts, in decode order.
    """

    def __init__(
        self,
        paths,
        *,
        factory: Callable[[Path], FrameReader],
        counts: list[int] | None = None,
    ) -> None:
        """Open the parts and build the global index (measuring counts if needed)."""
        self.paths = [Path(p) for p in paths]
        if len(self.paths) < 2:
            raise ValueError(
                "ConcatReader is for two or more parts; open_reader returns a "
                "plain reader for one"
            )
        self._readers = [factory(p) for p in self.paths]
        self._counts = (
            [int(c) for c in counts] if counts is not None else self._measure()
        )
        # Exclusive prefix sums: `_starts[i]` is the global index of part i's
        # frame 0.
        self._starts: list[int] = []
        total = 0
        for n in self._counts:
            self._starts.append(total)
            total += n
        self._total = total
        # The part a single-frame read landed in last: the one a held-open stream
        # keeps its decoder for (see `__getitem__`).
        self._last_part: int | None = None
        self._check_agreement()

    def _measure(self) -> list[int]:
        """Each part's exact frame count, decoding a part whose header will not say.

        The last part too: `count` reports the total, and a wrong total truncates a
        run.
        """
        out: list[int] = []
        for reader in self._readers:
            n = reader.count()
            if n is None:
                started = time.perf_counter()
                n = sum(1 for _ in reader.stream_frames(gray_ok=True))
                log.warning(
                    "%s carries no frame count in its container header, so it "
                    "was counted by a full decode (%d frames in %.1f s). A "
                    "concatenated stream needs exact lengths -- an estimate "
                    "would shift every frame index after this part.",
                    self.paths[len(out)].name,
                    n,
                    time.perf_counter() - started,
                )
            out.append(int(n))
        return out

    def _check_agreement(self) -> None:
        """Refuse parts that disagree on frame size or frame rate."""
        sizes: dict[tuple[int, int], Path] = {}
        rates: dict[float, Path] = {}
        for reader, path in zip(self._readers, self.paths, strict=True):
            frame = reader[0]
            size = (int(frame.shape[0]), int(frame.shape[1]))
            if size not in sizes and sizes:
                (other_size, other_path) = next(iter(sizes.items()))
                raise ValueError(
                    f"{path.name} is {size[1]}x{size[0]} but {other_path.name} "
                    f"is {other_size[1]}x{other_size[0]}; the parts of one "
                    "video's parts have to agree on frame size -- a "
                    "concatenated stream with two resolutions has no single "
                    "intrinsics"
                )
            sizes[size] = path
            rate = reader.fps()
            if rate is None:
                continue
            rate = round(float(rate), 3)
            if rate not in rates and rates:
                other_rate, other_path = next(iter(rates.items()))
                raise ValueError(
                    f"{path.name} is {rate} fps but {other_path.name} is "
                    f"{other_rate} fps; the parts of one video have to "
                    "agree on frame rate"
                )
            rates[rate] = path

    def _locate(self, idx: int) -> tuple[int, int]:
        """A global frame index -> `(part, local index)`."""
        if idx < 0:
            idx += self._total
        if not 0 <= idx < self._total:
            raise IndexError(
                f"frame {idx} is outside this {self._total}-frame stream "
                f"({len(self.paths)} parts)"
            )
        part = bisect.bisect_right(self._starts, idx) - 1
        return part, idx - self._starts[part]

    def _indices(self, key) -> list[int]:
        if isinstance(key, slice):
            return list(range(*key.indices(self._total)))
        return [int(i) for i in key]

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames (int / list / slice) into a uint8 RGB array."""
        if isinstance(key, (int, np.integer)):
            part, local = self._locate(int(key))
            # A viewer holds one decoder, the part it is reading, not every part it
            # read on the way (a no-op on a stateless part).
            if self._last_part is not None and self._last_part != part:
                self._readers[self._last_part].release()
            self._last_part = part
            return self._readers[part][local]
        wanted = self._indices(key)
        if not wanted:
            raise ValueError("no frames selected (check index list / slice)")
        # One decode per part, then back into the caller's order, so a slice across a
        # boundary never re-opens per frame.
        per_part: dict[int, list[int]] = {}
        for i in wanted:
            part, local = self._locate(i)
            per_part.setdefault(part, []).append(local)
        decoded: dict[int, np.ndarray] = {}
        for part, locals_ in per_part.items():
            block = self._readers[part][locals_]
            for local, frame in zip(locals_, block, strict=True):
                decoded[self._starts[part] + local] = frame
        return np.stack([decoded[i] for i in wanted])

    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W C"]]:
        """Walk the parts in order, forward, never seeking across a boundary.

        The inherited `stream_blocks` then never breaks a block at a part boundary.
        """
        stop = self._total if stop is None else min(stop, self._total)
        wanted = range(start, stop, step)
        for part, reader in enumerate(self._readers):
            base, n = self._starts[part], self._counts[part]
            first = next((i for i in wanted if base <= i < base + n), None)
            if first is None:
                continue
            local_stop = min(stop, base + n) - base
            yield from reader.stream_frames(
                start=first - base,
                stop=local_stop,
                step=step,
                gray_ok=gray_ok,
                thread_count=thread_count,
            )

    def count(self) -> int:
        """The exact total, summed over parts (never `None` -- see `_measure`)."""
        return self._total

    def fps(self) -> float | None:
        """The parts' shared frame rate (they are checked to agree)."""
        return self._readers[0].fps()

    def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
        """This stream with each part held open, and returns itself.

        A part opens its decoder on its first read, so a viewer opens only the parts
        it reads.
        """
        held = []
        for reader in self._readers:
            one = reader.hold_open(gray_ok=gray_ok)
            if one is not reader:
                reader.close()
            held.append(one)
        self._readers = held
        return self

    def release(self) -> None:
        """Let go of every part's held decoder; each reopens on its next read."""
        for reader in self._readers:
            reader.release()

    def close(self) -> None:
        """Close every part's reader."""
        for reader in self._readers:
            reader.close()

    def __repr__(self) -> str:
        """The parts and the total frame count."""
        parts = ", ".join(p.name for p in self.paths)
        return f"ConcatReader({parts}; {self._total} frames)"

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames (int / list / slice) into a uint8 RGB array.

Source code in src/deeperfly/io/media/concat.py 
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
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames (int / list / slice) into a uint8 RGB array."""
    if isinstance(key, (int, np.integer)):
        part, local = self._locate(int(key))
        # A viewer holds one decoder, the part it is reading, not every part it
        # read on the way (a no-op on a stateless part).
        if self._last_part is not None and self._last_part != part:
            self._readers[self._last_part].release()
        self._last_part = part
        return self._readers[part][local]
    wanted = self._indices(key)
    if not wanted:
        raise ValueError("no frames selected (check index list / slice)")
    # One decode per part, then back into the caller's order, so a slice across a
    # boundary never re-opens per frame.
    per_part: dict[int, list[int]] = {}
    for i in wanted:
        part, local = self._locate(i)
        per_part.setdefault(part, []).append(local)
    decoded: dict[int, np.ndarray] = {}
    for part, locals_ in per_part.items():
        block = self._readers[part][locals_]
        for local, frame in zip(locals_, block, strict=True):
            decoded[self._starts[part] + local] = frame
    return np.stack([decoded[i] for i in wanted])

__init__ 

__init__(
    paths,
    *,
    factory: Callable[[Path], FrameReader],
    counts: list[int] | None = None,
) -> None

Open the parts and build the global index (measuring counts if needed).

Source code in src/deeperfly/io/media/concat.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    paths,
    *,
    factory: Callable[[Path], FrameReader],
    counts: list[int] | None = None,
) -> None:
    """Open the parts and build the global index (measuring counts if needed)."""
    self.paths = [Path(p) for p in paths]
    if len(self.paths) < 2:
        raise ValueError(
            "ConcatReader is for two or more parts; open_reader returns a "
            "plain reader for one"
        )
    self._readers = [factory(p) for p in self.paths]
    self._counts = (
        [int(c) for c in counts] if counts is not None else self._measure()
    )
    # Exclusive prefix sums: `_starts[i]` is the global index of part i's
    # frame 0.
    self._starts: list[int] = []
    total = 0
    for n in self._counts:
        self._starts.append(total)
        total += n
    self._total = total
    # The part a single-frame read landed in last: the one a held-open stream
    # keeps its decoder for (see `__getitem__`).
    self._last_part: int | None = None
    self._check_agreement()

__repr__ 

__repr__() -> str

The parts and the total frame count.

Source code in src/deeperfly/io/media/concat.py 
228
229
230
231
def __repr__(self) -> str:
    """The parts and the total frame count."""
    parts = ", ".join(p.name for p in self.paths)
    return f"ConcatReader({parts}; {self._total} frames)"

close 

close() -> None

Close every part's reader.

Source code in src/deeperfly/io/media/concat.py 
223
224
225
226
def close(self) -> None:
    """Close every part's reader."""
    for reader in self._readers:
        reader.close()

count 

count() -> int

The exact total, summed over parts (never None -- see _measure).

Source code in src/deeperfly/io/media/concat.py 
195
196
197
def count(self) -> int:
    """The exact total, summed over parts (never `None` -- see `_measure`)."""
    return self._total

fps 

fps() -> float | None

The parts' shared frame rate (they are checked to agree).

Source code in src/deeperfly/io/media/concat.py 
199
200
201
def fps(self) -> float | None:
    """The parts' shared frame rate (they are checked to agree)."""
    return self._readers[0].fps()

hold_open 

hold_open(*, gray_ok: bool = False) -> FrameReader

This stream with each part held open, and returns itself.

A part opens its decoder on its first read, so a viewer opens only the parts it reads.

Source code in src/deeperfly/io/media/concat.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def hold_open(self, *, gray_ok: bool = False) -> FrameReader:
    """This stream with each part held open, and returns itself.

    A part opens its decoder on its first read, so a viewer opens only the parts
    it reads.
    """
    held = []
    for reader in self._readers:
        one = reader.hold_open(gray_ok=gray_ok)
        if one is not reader:
            reader.close()
        held.append(one)
    self._readers = held
    return self

release 

release() -> None

Let go of every part's held decoder; each reopens on its next read.

Source code in src/deeperfly/io/media/concat.py 
218
219
220
221
def release(self) -> None:
    """Let go of every part's held decoder; each reopens on its next read."""
    for reader in self._readers:
        reader.release()

stream_frames 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]

Walk the parts in order, forward, never seeking across a boundary.

The inherited stream_blocks then never breaks a block at a part boundary.

Source code in src/deeperfly/io/media/concat.py 
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
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W C"]]:
    """Walk the parts in order, forward, never seeking across a boundary.

    The inherited `stream_blocks` then never breaks a block at a part boundary.
    """
    stop = self._total if stop is None else min(stop, self._total)
    wanted = range(start, stop, step)
    for part, reader in enumerate(self._readers):
        base, n = self._starts[part], self._counts[part]
        first = next((i for i in wanted if base <= i < base + n), None)
        if first is None:
            continue
        local_stop = min(stop, base + n) - base
        yield from reader.stream_frames(
            start=first - base,
            stop=local_stop,
            step=step,
            gray_ok=gray_ok,
            thread_count=thread_count,
        )

Image sequences, one frame per file, decoded in parallel by Pillow (ADR-0007).

Frames come back as (T, H, W, 3) uint8 RGB: gray broadcast to three channels, alpha dropped, and samples wider than 8-bit rescaled onto 0..255 (_to_uint8). A sequence has no frame rate.

ImageSequenceReader

Bases: FrameReader

Reads an ordered image sequence (a directory, glob, or explicit file list).

Files decode in parallel on up to workers threads; gray_ok and thread_count are accepted and ignored.

Source code in src/deeperfly/io/media/images.py 
 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
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
class ImageSequenceReader(FrameReader):
    """Reads an ordered image sequence (a directory, glob, or explicit file list).

    Files decode in parallel on up to `workers` threads; `gray_ok` and `thread_count`
    are accepted and ignored.
    """

    def __init__(self, files, *, workers: int | None = None) -> None:
        """Wrap an ordered list of image `files`, decoded on up to `workers` threads."""
        self.files = [Path(f) for f in files]
        self.workers = workers

    @classmethod
    def from_pattern(
        cls, pattern: str | Path, *, workers: int | None = None
    ) -> ImageSequenceReader:
        """A reader for a directory or glob, its files sorted by name."""
        return cls(list_image_files(pattern), workers=workers)

    def _n_workers(self, n: int) -> int:
        return max(1, min(n, self.workers or (os.process_cpu_count() or 4)))

    def _decode(self, files: list[Path]) -> np.ndarray:
        """Parallel CPU decode of `files` to a `(T, H, W, 3)` uint8 array."""
        from PIL import Image

        def decode(f: Path) -> np.ndarray:
            with Image.open(f) as img:
                return _to_rgb_uint8(np.asarray(img))

        with ThreadPoolExecutor(max_workers=self._n_workers(len(files))) as pool:
            frames = list(pool.map(decode, files))
        return np.stack(frames)

    def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
        """Decode frames (int / list / slice) into a uint8 RGB array."""
        if isinstance(key, (int, np.integer)):
            return self._decode([self.files[int(key)]])[0]
        if isinstance(key, list):
            if not key:
                raise ValueError("index list must be non-empty")
            return self._decode([self.files[int(i)] for i in key])
        if isinstance(key, slice):
            files = self.files[key]
            if not files:
                raise ValueError("no frames selected (check slice)")
            return self._decode(files)
        raise TypeError(f"invalid index type {type(key).__name__!r}")

    def stream_frames(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "H W 3"]]:
        """Yield `(H, W, 3)` uint8 frames, one per file, decoded a block at a time."""
        for block in self.stream_blocks(start=start, stop=stop, step=step):
            yield from block

    def stream_blocks(
        self,
        *,
        start: int = 0,
        stop: int | None = None,
        step: int = 1,
        block_size: int = 64,
        gray_ok: bool = False,
        thread_count: int | None = None,
    ) -> Iterator[UInt8[np.ndarray, "T H W 3"]]:
        """Yield `(T, H, W, 3)` uint8 blocks, each decoded in parallel across files."""
        if block_size < 1:
            raise ValueError(f"block_size must be >= 1, got {block_size}")
        files = self.files[start:stop:step]
        for pos in range(0, len(files), block_size):
            yield self._decode(files[pos : pos + block_size])

    def count(self) -> int | None:
        """The number of files -- one frame each."""
        return len(self.files)

__getitem__ 

__getitem__(
    key: int | list[int] | slice,
) -> UInt8[np.ndarray, ...]

Decode frames (int / list / slice) into a uint8 RGB array.

Source code in src/deeperfly/io/media/images.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def __getitem__(self, key: int | list[int] | slice) -> UInt8[np.ndarray, "..."]:
    """Decode frames (int / list / slice) into a uint8 RGB array."""
    if isinstance(key, (int, np.integer)):
        return self._decode([self.files[int(key)]])[0]
    if isinstance(key, list):
        if not key:
            raise ValueError("index list must be non-empty")
        return self._decode([self.files[int(i)] for i in key])
    if isinstance(key, slice):
        files = self.files[key]
        if not files:
            raise ValueError("no frames selected (check slice)")
        return self._decode(files)
    raise TypeError(f"invalid index type {type(key).__name__!r}")

__init__ 

__init__(files, *, workers: int | None = None) -> None

Wrap an ordered list of image files, decoded on up to workers threads.

Source code in src/deeperfly/io/media/images.py 
83
84
85
86
def __init__(self, files, *, workers: int | None = None) -> None:
    """Wrap an ordered list of image `files`, decoded on up to `workers` threads."""
    self.files = [Path(f) for f in files]
    self.workers = workers

count 

count() -> int | None

The number of files -- one frame each.

Source code in src/deeperfly/io/media/images.py 
155
156
157
def count(self) -> int | None:
    """The number of files -- one frame each."""
    return len(self.files)

from_pattern classmethod 

from_pattern(
    pattern: str | Path, *, workers: int | None = None
) -> ImageSequenceReader

A reader for a directory or glob, its files sorted by name.

Source code in src/deeperfly/io/media/images.py 
88
89
90
91
92
93
@classmethod
def from_pattern(
    cls, pattern: str | Path, *, workers: int | None = None
) -> ImageSequenceReader:
    """A reader for a directory or glob, its files sorted by name."""
    return cls(list_image_files(pattern), workers=workers)

stream_blocks 

stream_blocks(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W 3"]]

Yield (T, H, W, 3) uint8 blocks, each decoded in parallel across files.

Source code in src/deeperfly/io/media/images.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def stream_blocks(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    block_size: int = 64,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "T H W 3"]]:
    """Yield `(T, H, W, 3)` uint8 blocks, each decoded in parallel across files."""
    if block_size < 1:
        raise ValueError(f"block_size must be >= 1, got {block_size}")
    files = self.files[start:stop:step]
    for pos in range(0, len(files), block_size):
        yield self._decode(files[pos : pos + block_size])

stream_frames 

stream_frames(
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W 3"]]

Yield (H, W, 3) uint8 frames, one per file, decoded a block at a time.

Source code in src/deeperfly/io/media/images.py 
125
126
127
128
129
130
131
132
133
134
135
136
def stream_frames(
    self,
    *,
    start: int = 0,
    stop: int | None = None,
    step: int = 1,
    gray_ok: bool = False,
    thread_count: int | None = None,
) -> Iterator[UInt8[np.ndarray, "H W 3"]]:
    """Yield `(H, W, 3)` uint8 frames, one per file, decoded a block at a time."""
    for block in self.stream_blocks(start=start, stop=stop, step=step):
        yield from block

list_image_files 

list_image_files(pattern: str | Path) -> list[Path]

Sorted image files for a directory or glob pattern, ordered by name.

Raises:

Type Description
FileNotFoundError

If nothing matches pattern.
Source code in src/deeperfly/io/media/images.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def list_image_files(pattern: str | Path) -> list[Path]:
    """Sorted image files for a directory or glob pattern, ordered by name.

    Raises:
        FileNotFoundError: If nothing matches `pattern`.
    """
    p = Path(pattern)
    if p.is_dir():
        files = sorted(f for f in p.iterdir() if f.suffix.lower() in IMAGE_EXTS)
    else:
        files = sorted(Path(f) for f in glob.glob(str(pattern)))
    if not files:
        raise FileNotFoundError(f"no images matched {pattern!r}")
    return files

MP4 writing: VideoWriter, backed by PyAV.

Frames in, H.264 out, encoded incrementally so a long clip never sits in memory.

with media.VideoWriter("out.mp4", fps=100) as writer:
    for frame in render():          # each an (H, W, 3) uint8 RGB frame
        writer.write_frame(frame)

VideoWriter

Incremental H.264 (libx264) MP4 encoder, backed by PyAV.

The file opens on the first frame, whose size (rounded down to even for yuv420p) sets the encode dimensions; later frames are cropped to it, and non-uint8 input is clipped to [0, 255].

Parameters:

Name Type Description Default
path str | Path

Output file.

 required
fps float

Frame rate written into the container.

 30.0
codec str | None

Encoder, or None for libx264.

 None
pix_fmt str

Pixel format; yuv420p is the widely playable default.

 'yuv420p'
options Mapping[str, str] | None

Private encoder options as FFmpeg spells them ({"crf": "15", "preset": "veryslow"}); None keeps the encoder's defaults (libx264's crf 23). Frames a detector trains on want a lower crf than frames a person watches.

 None
Source code in src/deeperfly/io/media/writer.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
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
class VideoWriter:
    """Incremental H.264 (libx264) MP4 encoder, backed by PyAV.

    The file opens on the first frame, whose size (rounded down to even for `yuv420p`)
    sets the encode dimensions; later frames are cropped to it, and non-uint8 input is
    clipped to `[0, 255]`.

    Args:
        path: Output file.
        fps: Frame rate written into the container.
        codec: Encoder, or `None` for `libx264`.
        pix_fmt: Pixel format; `yuv420p` is the widely playable default.
        options: Private encoder options as FFmpeg spells them
            (`{"crf": "15", "preset": "veryslow"}`); `None` keeps the encoder's
            defaults (libx264's crf 23). Frames a detector trains on want a lower crf
            than frames a person watches.
    """

    def __init__(
        self,
        path: str | Path,
        fps: float = 30.0,
        *,
        codec: str | None = None,
        pix_fmt: str = "yuv420p",
        options: Mapping[str, str] | None = None,
    ) -> None:
        """Prepare a writer for `path` (no file is opened until the first frame)."""
        self.path = Path(path)
        self.fps = fps
        self.codec = codec
        self.pix_fmt = pix_fmt
        self.options = dict(options or {})
        self._container: Any = None  # av OutputContainer (lazy, on the 1st frame)
        self._stream: Any = None  # av VideoStream
        self._size: tuple[int, int] | None = None  # (w, h), even

    def __enter__(self) -> Self:
        """Enter the context, returning the writer."""
        return self

    def __exit__(self, *exc) -> None:
        """Flush and close on context exit."""
        self.close()

    def _open(self, width: int, height: int) -> None:
        import av

        w, h = width & ~1, height & ~1  # yuv420p subsampling needs even dimensions
        rate = Fraction(self.fps).limit_denominator(1_000_000)
        self._container = av.open(str(self.path), mode="w")
        stream = self._container.add_stream(self.codec or "libx264", rate=rate)
        if not isinstance(stream, av.VideoStream):
            raise TypeError(f"{self.codec!r} is not a video codec")
        stream.width = w
        stream.height = h
        stream.pix_fmt = self.pix_fmt
        if self.options:
            stream.options = dict(self.options)
        # Multithreaded encode (frame/slice), several times faster than the
        # single-thread default on a multi-core host; the file is unaffected.
        try:
            stream.thread_type = "AUTO"
        except Exception as exc:  # noqa: BLE001 -- some codecs reject it
            log.debug("multithreaded encode unavailable for %s: %s", self.path, exc)
        self._stream = stream
        self._size = (w, h)
        log.info("writing %s via pyav: %dx%d @ %g fps", self.path.name, w, h, self.fps)

    def write_frame(self, frame) -> None:
        """Append one `(H, W, 3)` RGB frame (NumPy, or a torch or DLPack array)."""
        import av

        frame = _to_numpy(frame)
        if frame.dtype != np.uint8:
            frame = np.clip(frame, 0, 255).astype(np.uint8)
        if self._stream is None:
            self._open(frame.shape[1], frame.shape[0])
        assert self._stream is not None and self._size is not None
        w, h = self._size
        frame = np.ascontiguousarray(frame[:h, :w])
        if self.pix_fmt == "yuv420p":
            # Converted here in the encoder's BT.601 limited range: faster than
            # handing PyAV RGB to convert, and half the bytes to copy.
            vframe = av.VideoFrame.from_ndarray(_i420(frame), format="yuv420p")
        else:
            vframe = av.VideoFrame.from_ndarray(frame, format="rgb24")
        for packet in self._stream.encode(vframe):
            self._container.mux(packet)

    def write_frames(self, frames) -> None:
        """Append a `(T, H, W, 3)` batch, or any iterable of frames or blocks."""
        if isinstance(frames, np.ndarray):
            if frames.ndim == 4:
                for frame in frames:
                    self.write_frame(frame)
            else:
                self.write_frame(frames)
            return
        if hasattr(frames, "detach") or hasattr(frames, "__dlpack__"):  # torch/array
            self.write_frames(_to_numpy(frames))
            return
        for item in frames:  # a list / tuple / generator of frames (or batches)
            self.write_frames(item)

    def close(self) -> None:
        """Flush the encoder and close the file (idempotent)."""
        if self._container is None:
            return
        try:
            for packet in self._stream.encode():  # flush
                self._container.mux(packet)
        finally:
            self._container.close()
            self._container = None
            self._stream = None

__enter__ 

__enter__() -> Self

Enter the context, returning the writer.

Source code in src/deeperfly/io/media/writer.py 
92
93
94
def __enter__(self) -> Self:
    """Enter the context, returning the writer."""
    return self

__exit__ 

__exit__(*exc) -> None

Flush and close on context exit.

Source code in src/deeperfly/io/media/writer.py 
96
97
98
def __exit__(self, *exc) -> None:
    """Flush and close on context exit."""
    self.close()

__init__ 

__init__(
    path: str | Path,
    fps: float = 30.0,
    *,
    codec: str | None = None,
    pix_fmt: str = "yuv420p",
    options: Mapping[str, str] | None = None,
) -> None

Prepare a writer for path (no file is opened until the first frame).

Source code in src/deeperfly/io/media/writer.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    path: str | Path,
    fps: float = 30.0,
    *,
    codec: str | None = None,
    pix_fmt: str = "yuv420p",
    options: Mapping[str, str] | None = None,
) -> None:
    """Prepare a writer for `path` (no file is opened until the first frame)."""
    self.path = Path(path)
    self.fps = fps
    self.codec = codec
    self.pix_fmt = pix_fmt
    self.options = dict(options or {})
    self._container: Any = None  # av OutputContainer (lazy, on the 1st frame)
    self._stream: Any = None  # av VideoStream
    self._size: tuple[int, int] | None = None  # (w, h), even

close 

close() -> None

Flush the encoder and close the file (idempotent).

Source code in src/deeperfly/io/media/writer.py 
160
161
162
163
164
165
166
167
168
169
170
def close(self) -> None:
    """Flush the encoder and close the file (idempotent)."""
    if self._container is None:
        return
    try:
        for packet in self._stream.encode():  # flush
            self._container.mux(packet)
    finally:
        self._container.close()
        self._container = None
        self._stream = None

write_frame 

write_frame(frame) -> None

Append one (H, W, 3) RGB frame (NumPy, or a torch or DLPack array).

Source code in src/deeperfly/io/media/writer.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def write_frame(self, frame) -> None:
    """Append one `(H, W, 3)` RGB frame (NumPy, or a torch or DLPack array)."""
    import av

    frame = _to_numpy(frame)
    if frame.dtype != np.uint8:
        frame = np.clip(frame, 0, 255).astype(np.uint8)
    if self._stream is None:
        self._open(frame.shape[1], frame.shape[0])
    assert self._stream is not None and self._size is not None
    w, h = self._size
    frame = np.ascontiguousarray(frame[:h, :w])
    if self.pix_fmt == "yuv420p":
        # Converted here in the encoder's BT.601 limited range: faster than
        # handing PyAV RGB to convert, and half the bytes to copy.
        vframe = av.VideoFrame.from_ndarray(_i420(frame), format="yuv420p")
    else:
        vframe = av.VideoFrame.from_ndarray(frame, format="rgb24")
    for packet in self._stream.encode(vframe):
        self._container.mux(packet)

write_frames 

write_frames(frames) -> None

Append a (T, H, W, 3) batch, or any iterable of frames or blocks.

Source code in src/deeperfly/io/media/writer.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def write_frames(self, frames) -> None:
    """Append a `(T, H, W, 3)` batch, or any iterable of frames or blocks."""
    if isinstance(frames, np.ndarray):
        if frames.ndim == 4:
            for frame in frames:
                self.write_frame(frame)
        else:
            self.write_frame(frames)
        return
    if hasattr(frames, "detach") or hasattr(frames, "__dlpack__"):  # torch/array
        self.write_frames(_to_numpy(frames))
        return
    for item in frames:  # a list / tuple / generator of frames (or batches)
        self.write_frames(item)

Overlap frame decoding with the work that consumes the frames.

prefetch runs a frame iterator in a background thread and hands its items over through a bounded queue, so the next block decodes while this one is consumed; PyAV releases the GIL, and peak memory is about depth + 2 blocks. A rig prefetches one zipped multi-view iterator.

for block in media.prefetch(reader.stream_blocks(block_size=64, gray_ok=True)):
    model(block)  # the next block decodes during this call

prefetch 

prefetch(
    source: Iterable[T], *, depth: int = 2
) -> Iterator[T]

Yield source's items in order, decoding up to depth ahead in a thread.

An exception raised by source is re-raised here after the items before it. Abandoning the iterator stops the producer and closes source if it can be closed, so no decoder is left running behind a full queue.

Source code in src/deeperfly/io/media/prefetch.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
56
57
58
59
60
61
62
63
64
65
def prefetch[T](source: Iterable[T], *, depth: int = 2) -> Iterator[T]:
    """Yield `source`'s items in order, decoding up to `depth` ahead in a thread.

    An exception raised by `source` is re-raised here after the items before it.
    Abandoning the iterator stops the producer and closes `source` if it can be
    closed, so no decoder is left running behind a full queue.
    """
    if depth < 1:
        raise ValueError(f"depth must be >= 1, got {depth}")
    q: queue.Queue = queue.Queue(maxsize=depth)

    def produce() -> None:
        try:
            for item in source:
                q.put(item)
        except queue.ShutDown:
            pass  # the consumer went away; stop decoding
        except BaseException as exc:  # noqa: BLE001 -- carried to the consumer
            with contextlib.suppress(queue.ShutDown):
                q.put((_ERROR, exc))
        finally:
            # The consumer takes what is queued, then its `get` raises `ShutDown`.
            q.shutdown()
            # Release the source (an open decoder) now rather than at GC.
            close = getattr(source, "close", None)
            if callable(close):
                close()

    threading.Thread(target=produce, daemon=True).start()
    try:
        while True:
            try:
                item = q.get()
            except queue.ShutDown:
                return
            if isinstance(item, tuple) and len(item) == 2 and item[0] is _ERROR:
                raise item[1]
            yield item
    finally:
        # Reached on an early break / close as well as normal exhaustion: a producer
        # blocked on a full queue raises `ShutDown` at once and stops.
        q.shutdown(immediate=True)