Function test_source_info

dali/test/python/decoder/test_video.py:346–380

(device)

@params("cpu", "gpu")
def test_source_info(device):
    skip_if_m60()
    filenames = glob.glob(f"{get_dali_extra_path()}/db/video/[cv]fr/*.mp4")
    # filter out HEVC because some GPUs do not support it
    filenames = filter(lambda filename: "hevc" not in filename, filenames)
    # filter out AV1 because some GPUs do not support it
    filenames = filter(lambda filename: "av1" not in filename, filenames)
    if device == "cpu":
        # some formats are not yet supported in the CPU operator itself
        filenames = filter(lambda filename: "mpeg4" not in filename, filenames)
        excluded = {"test_1.mp4", "test_2.mp4"}
        filenames = filter(lambda f: os.path.basename(f) not in excluded, filenames)

    files = list(filenames)

    @pipeline_def
    def test_pipeline():
        videos = fn.experimental.readers.video(
            device=device,
            filenames=files,
            sequence_length=1,
            step=10000000,  # make sure that each video has only one valid sequence
        )
        return videos

    batch_size = 4
    device_id = None if device == "cpu" else 0
    p = test_pipeline(batch_size=batch_size, num_threads=1, device_id=device_id)

    samples_read = 0
    while samples_read < len(files):
        o = p.run()
        for idx, t in enumerate(o[0]):
            assert t.source_info() == files[(samples_read + idx) % len(files)]
        samples_read += batch_size


@params("cpu", "mixed")