Function test_decoded_vs_generated

dali/test/python/reader/test_nemo_asr.py:204–283

()

def test_decoded_vs_generated():
    batch_size = 3
    pipeline = NemoAsrReaderPipeline(batch_size=batch_size)

    for iter in range(1):
        out = pipeline.run()
        for idx in range(batch_size):
            audio_plain_i = out[0].at(idx)
            audio_plain_f = out[1].at(idx)
            audio_downmix_i = out[2].at(idx)
            audio_downmix_f = out[3].at(idx)
            audio_resampled1_i = out[4].at(idx)
            audio_resampled1_f = out[5].at(idx)
            audio_resampled2_i = out[6].at(idx)
            audio_resampled2_f = out[7].at(idx)
            text = out[8].at(idx)
            text_non_ascii = out[9].at(idx)

            ref_plain_i = ref_i[idx]
            np.testing.assert_allclose(audio_plain_i, ref_plain_i, rtol=1e-7)

            ref_plain_f = ref_i[idx].astype(np.float32) / 32767
            np.testing.assert_allclose(audio_plain_f, ref_plain_f, rtol=1e-4)

            ref_downmix_i_float = ref_i[idx].astype(np.float32).mean(axis=1, keepdims=1)

            ref_downmix_i = ref_downmix_i_float.astype(np.int16).flatten()
            np.testing.assert_allclose(audio_downmix_i, ref_downmix_i, atol=1)

            ref_downmix_f = (ref_downmix_i_float / 32767).flatten()
            np.testing.assert_allclose(audio_downmix_f, ref_downmix_f, rtol=1e-4)

            ref_resampled1_float = generate_waveforms(
                lengths[idx] * rate1 / rates[idx], freqs[idx] * (rates[idx] / rate1)
            )
            ref_resampled1_downmix = ref_resampled1_float.astype(np.float32).mean(
                axis=1, keepdims=1
            )
            ref_resampled1_i = (ref_resampled1_downmix * 32767).astype(np.int16).flatten()
            # resampling - allow for 1e-3 dynamic range error
            np.testing.assert_allclose(
                audio_resampled1_i, ref_resampled1_i, atol=round(32767 * 1e-3)
            )

            ref_resampled1_f = ref_resampled1_downmix.flatten()
            # resampling - allow for 1e-3 dynamic range error
            np.testing.assert_allclose(audio_resampled1_f, ref_resampled1_f, atol=1e-3)

            ref_resampled2_float = generate_waveforms(
                lengths[idx] * rate2 / rates[idx], freqs[idx] * (rates[idx] / rate2)
            )
            ref_resampled2_downmix = ref_resampled2_float.astype(np.float32).mean(
                axis=1, keepdims=1
            )
            ref_resampled2_i = (ref_resampled2_downmix * 32767).astype(np.int16).flatten()
            # resampling - allow for 1e-3 dynamic range error
            np.testing.assert_allclose(
                audio_resampled2_i, ref_resampled2_i, atol=round(32767 * 1e-3)