Function test_inflate

dali/test/python/test_dali_variable_batch_size.py:1446–1480

()

@has_operator("decoders.inflate")
@restrict_platform(min_compute_cap=6.0)
def test_inflate():
    import lz4.block

    def sample_to_lz4(sample):
        deflated_buf = lz4.block.compress(sample, store_size=False)
        return np.frombuffer(deflated_buf, dtype=np.uint8)

    def inflate_pipline(max_batch_size, inputs, device):
        input_data = [[sample_to_lz4(sample) for sample in batch] for batch in inputs]
        input_shape = [
            [np.array(sample.shape, dtype=np.int32) for sample in batch] for batch in inputs
        ]

        @pipeline_def
        def piepline():
            defalted = fn.external_source(source=input_data)
            shape = fn.external_source(source=input_shape)
            return fn.decoders.inflate(defalted.gpu(), shape=shape)

        return piepline(batch_size=max_batch_size, num_threads=4, device_id=0)

    def sample_gen():
        j = 42
        while True:
            yield np.full((13, 7), j)
            j += 1

    sample = sample_gen()
    batches = [
        [next(sample) for _ in range(5)],
        [next(sample) for _ in range(13)],
        [next(sample) for _ in range(2)],
    ]

    check_pipeline(batches, inflate_pipline, devices=["gpu"])


@params(fn.experimental.debayer, fn.debayer)