Function test_optional_build

dali/test/python/test_pipeline.py:2405–2434

()

def test_optional_build():
    bs = 8

    @pipeline_def(batch_size=bs, num_threads=4, device_id=0)
    def pdef_regular():
        enc, _ = fn.readers.file(file_root=jpeg_folder, name="only_reader")
        img = fn.decoders.image(enc, device="mixed")
        return img

    @pipeline_def(batch_size=bs, num_threads=4, device_id=0)
    def pdef_source():
        source = fn.external_source(name="source")
        return source

    pipes = [pdef_regular() for _ in range(5)]
    pipes.append(pdef_source())

    for pipe in pipes:
        pipe.build()

    (res,) = pipes[0].run()
    assert len(res.shape()) == 8
    pipes[1].schedule_run()
    (res_2,) = pipes[1].outputs()
    assert len(res_2.shape()) == 8
    assert pipes[2].epoch_size("only_reader") != 0
    assert pipes[3].executor_statistics() == {}
    assert "shard_id" in pipes[4].reader_meta("only_reader")

    pipes[-1].feed_input("source", np.array([10, 10]))


def test_output_descs():