[Bug]: frames(fps=…) can return pictures from after their sample time, and repeats the last picture across a lost keyframe #731

Version or commit

hflow 0.2.21; reproduced on main at 6cf6e98. #730 does not touch frames().

Environment

macOS 26.6 (arm64), Python 3.11.16, PyAV 18.1.0, ffmpeg 8.1.1 (Homebrew); the uv sync --locked environment plus the packages in step 1

Minimal reproduction

Synthetic data only, the same setup as #727: one 640×360 H.264 camera at 30 fps, a keyframe every 10 frames, 300
frames, each picture drawing its source frame number as 10 black/white blocks. One raw MCAP is complete, the other
lacks source frame 100 (an IDR). Both go through the default TransformConfig(), then frames(fps=…).

1. pip install hflow av pillow mcap-protobuf-support foxglove-schemas-protobuf (and ffmpeg 5.1 or newer on PATH, as HFlow needs)
2. python hflow_frames_fps_repro.py

"""Episode.frames(fps=...): is each picture the source frame its log_time_ns names?

Synthetic data only. Each frame draws its source frame number as 10 black/white blocks, so the number can be read
back from the extracted JPEG. One raw MCAP is complete, the other lacks source frame 100 (an IDR).

    pip install hflow av pillow mcap-protobuf-support foxglove-schemas-protobuf
    python hflow_frames_fps_repro.py
"""
import tempfile
from fractions import Fraction
from pathlib import Path

import av
import numpy as np
from foxglove_schemas_protobuf.CompressedVideo_pb2 import CompressedVideo
from mcap.writer import CompressionType
from mcap_protobuf.writer import Writer
from PIL import Image

from hflow.episode import Episode
from hflow.transform import TransformConfig, write_canonical_episode

W, H, FPS, N, GOP, BITS = 640, 360, 30, 300, 10, 10
T0 = 1_760_000_000_000_000_000
TOPIC = "/camera/cam0/video"


def draw(i):
    y = np.full((H, W), 128, np.uint8)
    y[:40] = 16
    for b in range(BITS):
        if (i >> (BITS - 1 - b)) & 1:
            y[:40, b * (W // BITS):(b + 1) * (W // BITS)] = 235
    return np.stack([y, y, y], -1)


def read_number(path):
    luma = np.asarray(Image.open(path).convert("L"), np.float32)[:40]
    bw = W // BITS
    return int("".join(str(int(luma[:, b * bw + 8:(b + 1) * bw - 8].mean() > 128)) for b in range(BITS)), 2)


def encode():
    ctx = av.CodecContext.create("libx264", "w")
    ctx.width, ctx.height, ctx.pix_fmt = W, H, "yuv420p"
    ctx.framerate, ctx.time_base = FPS, Fraction(1, FPS)
    ctx.gop_size, ctx.max_b_frames = GOP, 0
    ctx.options = {"profile": "baseline", "tune": "zerolatency", "keyint_min": str(GOP), "sc_threshold": "0"}
    packets = []
    for i in range(N):
        frame = av.VideoFrame.from_ndarray(draw(i), format="rgb24").reformat(format="yuv420p")
        frame.pts = i
        packets += [bytes(p) for p in ctx.encode(frame)]
    return packets + [bytes(p) for p in ctx.encode(None)]


def write_raw(path, packets, drop):
    kept = [i for i in range(N) if i != drop]
    with open(path, "wb") as f:
        w = Writer(f, compression=CompressionType.ZSTD)
        for i in kept:
            t = T0 + i * 1_000_000_000 // FPS
            msg = CompressedVideo(frame_id="cam0", data=packets[i], format="h264")
            msg.timestamp.FromNanoseconds(t)
            w.write_message(TOPIC, msg, log_time=t, publish_time=t)
        w.finish()
    return kept                                    # message k carries source frame kept[k]


def source_frame_at(log_time_ns):
    return round((log_time_ns - T0) * FPS / 1e9)


def main():
    packets = encode()
    with tempfile.TemporaryDirectory() as work:
        for drop in (None, 100):
            raw, canonical = Path(work, f"raw_{drop}.mcap"), Path(work, f"canonical_{drop}.mcap")
            write_raw(raw, packets, drop)
            write_canonical_episode(str(raw), str(canonical), TransformConfig())
            print(f"\nsource {'complete' if drop is None else f'missing frame {drop} (an IDR)'}")
            ep = Episode(str(canonical), workdir=Path(work, f"ep_{drop}"))
            for fps in (30, 10, 3, 1):
                frames = ep.frames(TOPIC, fps=fps)
                pairs = [(source_frame_at(f.log_time_ns), read_number(f.path)) for f in frames]
                late = sorted({shown - named for named, shown in pairs})
                print(f"  frames(fps={fps:2d}): {len(frames):3d} frames; picture minus the frame its log_time_ns "
                      f"names: {late}; first pairs (named, shown): {pairs[:3]}")
                if drop is not None:
                    gap = [p for p in pairs if drop < p[0] <= drop + 9]
                    print(f"      samples named inside the damaged run {drop + 1}-{drop + 9}: {gap}")


if __name__ == "__main__":
    main()

Expected behavior

frames() documents: "Each frame's log_time_ns is the log time of the source message it was extracted from,
exact to within one source frame interval even across recording gaps", and
source_log_times_for_sampled_frames labels sample i on the stated assumption that the fps filter "emits the
frame VISIBLE at each output tick (the last frame at-or-before it)", i.e. source frame
floor((start_s + i / fps) * source_fps). So each picture should be that frame, or at most one source frame off.
A sample whose message the decoder could not decode should not be returned as some other message's picture.

Actual behavior

Output on main (6cf6e98), trimmed to the relevant lines; "(named, shown)" pairs are the source frame the
log_time_ns names and the one the JPEG shows:

source complete
  frames(fps=30): 300 frames; picture minus the frame its log_time_ns names: [0]
  frames(fps=10): 100 frames; picture minus the frame its log_time_ns names: [1]
  frames(fps= 3):  30 frames; picture minus the frame its log_time_ns names: [4]
  frames(fps= 1):  10 frames; picture minus the frame its log_time_ns names: [14]; first pairs (named, shown): [(0, 14), (30, 44), (60, 74)]

source missing frame 100 (an IDR)
  frames(fps=30): 299 frames
      samples named inside the damaged run 101-109: [(101, 99), (102, 99), (103, 99), (104, 99), (105, 99), (106, 99), (107, 99), (108, 99), (109, 111)]

Two separate problems:

1. Sampling below half the source rate, the picture comes from after the sample time. At 10 fps from a
   30 fps camera the picture is one frame late, within the documented tolerance; at 3 fps it is 4 frames late and
   at 1 fps 14 frames (0.47 s), well outside it. With its default round=near, the fps filter rounds each input
   timestamp to the nearest output tick, and of the frames that round to the same tick it emits the last one (that
   is what the offsets above show), up to half a sample interval after the tick. The offset depends only on the
   ratio of source rate to sample rate, not on the data: just under half that ratio in source frames (14 at a
   ratio of 30), and none when the ratio is 2 or less (30 fps sampled at 15 fps comes out exact). test_canonical_episode_accessors checks the
   labels against the same floor formula, not the pictures, so it passes.
2. Across a lost keyframe, the last good picture is repeated. The decoder drops most of the frames that depend
   on the lost keyframe; the fps filter fills their ticks with the last picture it has (frame 99 above), and each copy
   gets the log time of a message that was never decoded. The last pair, (109, 111), is the one picture the decoder
   concealed rather than dropped. This is the frames(fps=…) half of [Bug]: frames_at_indices returns frames shifted by several positions, without an error, after a pass-through H.264 camera loses a keyframe #727, which fix(episode): select frames at indices by time, not the decoded counter #730 does not change.

Additional context

A fix I have tested (it does not depend on #730's change):

1. fps=…:round=up makes the filter emit the frame at or before each tick, which is what the labelling, the
   docstring and source_log_times_for_sampled_frames already assume. On the complete source above, every
   picture then matches its label at 1, 3, 10 and 30 fps.
2. The same ffmpeg run gets a second output, a framecrc listing of every picture the decoder produced, with
   timestamps in microseconds (the MP4's nominal rate can differ from the real one, e.g. the remux of a 29.87 fps
   stream reports 179/6, so frame-rate units would drift). A sample whose source message is not in the listing is left out, so
   no picture is returned under the log time of an undecoded message. The file name keeps the sample number, so
   the labels of the other samples do not shift. The one picture the decoder conceals inside a gap still passes,
   as in fix(episode): select frames at indices by time, not the decoded counter #730.

On healthy synthetic episodes (hflow.testing.synthesize_episode), over 120 combinations of source rate (5, 10,
15, 29.97 and 30 fps), sample rate (0.5 to 40 fps) and start_s/end_s, every sample present before keeps its
label; in 10 of them round=up adds one sample at the end of the window (nine at a tick inside the stream, one at
a tick just past it, labelled with the last message as source_log_times_for_sampled_frames already clamps).
That comparison covers labels and counts; picture content was checked on the frame-numbered stream above. Tested
with ffmpeg 8.1.1 only.

Two questions before I open the PR:

• Leaving undecodable samples out keeps frames() usable on a damaged recording, but returns fewer frames than
  ticks. Would you rather it raise, like frames_at_indices after fix(episode): select frames at indices by time, not the decoded counter #730?
• rgb_frames(frames_per_second=…), which mediapipe_hands samples with, uses the same fps filter with its default
  rounding, and the detections are labelled through source_log_times_for_sampled_frames, so it likely has the
  same offset (I have not tested it). Fix it in the same PR, or separately?