Class DecoderPipeline

dali/test/python/decoder/test_audio.py:47–99

class DecoderPipeline(Pipeline):
    def __init__(self):
        super().__init__(
            batch_size=8,
            num_threads=3,
            device_id=0,
            exec_async=True,
            exec_pipelined=True,
            output_dtype=[
                types.INT16,
                types.INT16,
                types.INT16,
                types.FLOAT,
                types.FLOAT,
                types.FLOAT,
                types.FLOAT,
                types.FLOAT,
            ],
            output_ndim=[2, 2, 1, 1, 0, 0, 0, 0],
        )
        self.file_source = ops.ExternalSource()
        self.plain_decoder = ops.decoders.Audio(dtype=types.INT16)
        self.resampling_decoder = ops.decoders.Audio(sample_rate=rate1, dtype=types.INT16)
        self.downmixing_decoder = ops.decoders.Audio(downmix=True, dtype=types.INT16)
        self.resampling_downmixing_decoder = ops.decoders.Audio(
            sample_rate=rate2, downmix=True, quality=50, dtype=types.FLOAT
        )

    def define_graph(self):
        self.raw_file = self.file_source()
        dec_plain, rates_plain = self.plain_decoder(self.raw_file)
        dec_res, rates_res = self.resampling_decoder(self.raw_file)
        dec_mix, rates_mix = self.downmixing_decoder(self.raw_file)
        dec_res_mix, rates_res_mix = self.resampling_downmixing_decoder(self.raw_file)
        out = [
            dec_plain,
            dec_res,
            dec_mix,
            dec_res_mix,
            rates_plain,
            rates_res,
            rates_mix,
            rates_res_mix,
        ]
        return out

    def iter_setup(self):
        list = []
        for i in range(self.batch_size):
            idx = i % len(names)
            with open(names[idx], mode="rb") as f:
                list.append(np.array(bytearray(f.read()), np.uint8))
        self.feed_input(self.raw_file, list)


@attr("sanitizer_skip")