Function test_video_decoder_no_padding

dali/test/python/decoder/test_video.py:577–609

(device, batch_size, filenames, stride, sequence_length)

    *[(device, 3, vfr_files, 2, 1000) for device in ["cpu", "mixed"]],
)
def test_video_decoder_no_padding(device, batch_size, filenames, stride, sequence_length):
    skip_if_m60()
    batch = []
    for i in range(batch_size):
        with open(filenames[i % len(filenames)], "rb") as f:
            batch.append(np.frombuffer(f.read(), dtype=np.uint8))

    def get_batch():
        return batch

    @pipeline_def
    def test_pipeline():
        encoded = fn.external_source(source=get_batch, device="cpu")
        # Get full video first as reference
        reference = fn.experimental.decoders.video(encoded, device=device, stride=stride)
        # Request frames that would exceed video length
        decoded = fn.experimental.decoders.video(
            encoded,
            device=device,
            stride=stride,
            sequence_length=sequence_length,
            pad_mode="none",
        )
        return (reference, decoded)

    pipe = test_pipeline(batch_size=batch_size, num_threads=1, device_id=0)
    out_ref, out = (o.as_cpu() for o in pipe.run())

    # Verify each sample in the batch
    for i in range(batch_size):
        ref = out_ref.at(i)
        actual = out.at(i)
        compare_videos(ref, actual)


@params("cpu", "mixed")