fixup! Add README

Author: mthrok

test/torchaudio_unittest/common_utils/kaldi_utils.py

@@ -0,0 +1,39 @@
+import subprocess
+
+import torch
+
+
+def convert_args(**kwargs):
+    args = []
+    for key, value in kwargs.items():
+        if key == 'sample_rate':
+            key = 'sample_frequency'
+        key = '--' + key.replace('_', '-')
+        value = str(value).lower() if value in [True, False] else str(value)
+        args.append('%s=%s' % (key, value))
+    return args
+
+
+def run_kaldi(command, input_type, input_value):
+    """Run provided Kaldi command, pass a tensor and get the resulting tensor
+
+    Args:
+        input_type: str
+            'ark' or 'scp'
+        input_value:
+            Tensor for 'ark'
+            string for 'scp' (path to an audio file)
+    """
+    import kaldi_io
+
+    key = 'foo'
+    process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+    if input_type == 'ark':
+        kaldi_io.write_mat(process.stdin, input_value.cpu().numpy(), key=key)
+    elif input_type == 'scp':
+        process.stdin.write(f'{key} {input_value}'.encode('utf8'))
+    else:
+        raise NotImplementedError('Unexpected type')
+    process.stdin.close()
+    result = dict(kaldi_io.read_mat_ark(process.stdout))['foo']
+    return torch.from_numpy(result.copy())  # copy supresses some torch warning

test/torchaudio_unittest/functional/kaldi_compatibility_cpu_test.py

@@ -0,0 +1,7 @@
+from torchaudio_unittest.common_utils import PytorchTestCase
+from .kaldi_compatibility_test_impl import KaldiCPUOnly
+
+
+class TestKaldiCPUOnly(KaldiCPUOnly, PytorchTestCase):
+    dtype = torch.float32
+    device = torch.device('cpu')

test/torchaudio_unittest/functional/kaldi_compatibility_test_imp.py

@@ -0,0 +1,37 @@
+from parameterized import parameterized
+import torchaudio.functional as F
+
+from torchaudio_unittest.common_utils import (
+    get_sinusoid,
+    load_params,
+    save_wav,
+    skipIfNoExec,
+    TempDirMixin,
+    TestBaseMixin,
+)
+from torchaudio_unittest.common_utils.kaldi_utils import (
+    convert_args,
+    run_kaldi,
+)
+
+
+class KaldiCPUOnly(TempDirMixin, TestBaseMixin):
+    def assert_equal(self, output, *, expected, rtol=None, atol=None):
+        expected = expected.to(dtype=self.dtype, device=self.device)
+        self.assertEqual(output, expected, rtol=rtol, atol=atol)
+
+    @parameterized.expand(load_params('kaldi_test_pitch_args.json'))
+    @skipIfNoExec('compute-kaldi-pitch-feats')
+    def test_pitch_feats(self, kwargs):
+        """compute_kaldi_pitch produces numerically compatible result with compute-kaldi-pitch-feats"""
+        sample_rate = kwargs['sample_rate']
+        waveform = get_sinusoid(dtype='float32', sample_rate=sample_rate)
+        result = F.compute_kaldi_pitch(waveform[0], **kwargs)
+
+        waveform = get_sinusoid(dtype='int16', sample_rate=sample_rate)
+        wave_file = self.get_temp_path('test.wav')
+        save_wav(wave_file, waveform, sample_rate)
+
+        command = ['compute-kaldi-pitch-feats'] + convert_args(**kwargs) + ['scp:-', 'ark:-']
+        kaldi_result = run_kaldi(command, 'scp', wave_file)
+        self.assert_equal(result, expected=kaldi_result)

test/torchaudio_unittest/kaldi_compatibility_cpu_test.py

@@ -1,7 +1,7 @@
 import torch
 
 from torchaudio_unittest import common_utils
-from .kaldi_compatibility_impl import Kaldi, KaldiCPUOnly
+from .kaldi_compatibility_impl import Kaldi
 
 
 class TestKaldiFloat32(Kaldi, common_utils.PytorchTestCase):
@@ -12,8 +12,3 @@ class TestKaldiFloat32(Kaldi, common_utils.PytorchTestCase):
 class TestKaldiFloat64(Kaldi, common_utils.PytorchTestCase):
     dtype = torch.float64
     device = torch.device('cpu')
-
-
-class TestKaldiCPUOnly(KaldiCPUOnly, common_utils.PytorchTestCase):
-    dtype = torch.float32
-    device = torch.device('cpu')

test/torchaudio_unittest/kaldi_compatibility_impl.py

@@ -1,7 +1,4 @@
 """Test suites for checking numerical compatibility against Kaldi"""
-import subprocess
-
-import kaldi_io
 import torch
 import torchaudio.functional as F
 import torchaudio.compliance.kaldi
@@ -14,52 +11,18 @@
     skipIfNoExec,
     get_asset_path,
     load_wav,
-    save_wav,
-    get_sinusoid,
+)
+from torchaudio_unittest.common_utils.kaldi_utils import (
+    convert_args,
+    run_kaldi,
 )
 
 
-def _convert_args(**kwargs):
-    args = []
-    for key, value in kwargs.items():
-        if key == 'sample_rate':
-            key = 'sample_frequency'
-        key = '--' + key.replace('_', '-')
-        value = str(value).lower() if value in [True, False] else str(value)
-        args.append('%s=%s' % (key, value))
-    return args
-
-
-def _run_kaldi(command, input_type, input_value):
-    """Run provided Kaldi command, pass a tensor and get the resulting tensor
-
-    Args:
-        input_type: str
-            'ark' or 'scp'
-        input_value:
-            Tensor for 'ark'
-            string for 'scp' (path to an audio file)
-    """
-    key = 'foo'
-    process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
-    if input_type == 'ark':
-        kaldi_io.write_mat(process.stdin, input_value.cpu().numpy(), key=key)
-    elif input_type == 'scp':
-        process.stdin.write(f'{key} {input_value}'.encode('utf8'))
-    else:
-        raise NotImplementedError('Unexpected type')
-    process.stdin.close()
-    result = dict(kaldi_io.read_mat_ark(process.stdout))['foo']
-    return torch.from_numpy(result.copy())  # copy supresses some torch warning
-
-
-class KaldiTestBase(TempDirMixin, TestBaseMixin):
+class Kaldi(TempDirMixin, TestBaseMixin):
     def assert_equal(self, output, *, expected, rtol=None, atol=None):
         expected = expected.to(dtype=self.dtype, device=self.device)
         self.assertEqual(output, expected, rtol=rtol, atol=atol)
 
-
-class Kaldi(KaldiTestBase):
     @skipIfNoExec('apply-cmvn-sliding')
     def test_sliding_window_cmn(self):
         """sliding_window_cmn should be numerically compatible with apply-cmvn-sliding"""
@@ -72,8 +35,8 @@ def test_sliding_window_cmn(self):
 
         tensor = torch.randn(40, 10, dtype=self.dtype, device=self.device)
         result = F.sliding_window_cmn(tensor, **kwargs)
-        command = ['apply-cmvn-sliding'] + _convert_args(**kwargs) + ['ark:-', 'ark:-']
-        kaldi_result = _run_kaldi(command, 'ark', tensor)
+        command = ['apply-cmvn-sliding'] + convert_args(**kwargs) + ['ark:-', 'ark:-']
+        kaldi_result = run_kaldi(command, 'ark', tensor)
         self.assert_equal(result, expected=kaldi_result)
 
     @parameterized.expand(load_params('kaldi_test_fbank_args.json'))
@@ -83,8 +46,8 @@ def test_fbank(self, kwargs):
         wave_file = get_asset_path('kaldi_file.wav')
         waveform = load_wav(wave_file, normalize=False)[0].to(dtype=self.dtype, device=self.device)
         result = torchaudio.compliance.kaldi.fbank(waveform, **kwargs)
-        command = ['compute-fbank-feats'] + _convert_args(**kwargs) + ['scp:-', 'ark:-']
-        kaldi_result = _run_kaldi(command, 'scp', wave_file)
+        command = ['compute-fbank-feats'] + convert_args(**kwargs) + ['scp:-', 'ark:-']
+        kaldi_result = run_kaldi(command, 'scp', wave_file)
         self.assert_equal(result, expected=kaldi_result, rtol=1e-4, atol=1e-8)
 
     @parameterized.expand(load_params('kaldi_test_spectrogram_args.json'))
@@ -94,8 +57,8 @@ def test_spectrogram(self, kwargs):
         wave_file = get_asset_path('kaldi_file.wav')
         waveform = load_wav(wave_file, normalize=False)[0].to(dtype=self.dtype, device=self.device)
         result = torchaudio.compliance.kaldi.spectrogram(waveform, **kwargs)
-        command = ['compute-spectrogram-feats'] + _convert_args(**kwargs) + ['scp:-', 'ark:-']
-        kaldi_result = _run_kaldi(command, 'scp', wave_file)
+        command = ['compute-spectrogram-feats'] + convert_args(**kwargs) + ['scp:-', 'ark:-']
+        kaldi_result = run_kaldi(command, 'scp', wave_file)
         self.assert_equal(result, expected=kaldi_result, rtol=1e-4, atol=1e-8)
 
     @parameterized.expand(load_params('kaldi_test_mfcc_args.json'))
@@ -105,24 +68,6 @@ def test_mfcc(self, kwargs):
         wave_file = get_asset_path('kaldi_file.wav')
         waveform = load_wav(wave_file, normalize=False)[0].to(dtype=self.dtype, device=self.device)
         result = torchaudio.compliance.kaldi.mfcc(waveform, **kwargs)
-        command = ['compute-mfcc-feats'] + _convert_args(**kwargs) + ['scp:-', 'ark:-']
-        kaldi_result = _run_kaldi(command, 'scp', wave_file)
+        command = ['compute-mfcc-feats'] + convert_args(**kwargs) + ['scp:-', 'ark:-']
+        kaldi_result = run_kaldi(command, 'scp', wave_file)
         self.assert_equal(result, expected=kaldi_result, rtol=1e-4, atol=1e-8)
-
-
-class KaldiCPUOnly(KaldiTestBase):
-    @parameterized.expand(load_params('kaldi_test_pitch_args.json'))
-    @skipIfNoExec('compute-kaldi-pitch-feats')
-    def test_pitch_feats(self, kwargs):
-        """compute_kaldi_pitch produces numerically compatible result with compute-kaldi-pitch-feats"""
-        sample_rate = kwargs['sample_rate']
-        waveform = get_sinusoid(dtype='float32', sample_rate=sample_rate)
-        result = F.compute_kaldi_pitch(waveform[0], **kwargs)
-
-        waveform = get_sinusoid(dtype='int16', sample_rate=sample_rate)
-        wave_file = self.get_temp_path('test.wav')
-        save_wav(wave_file, waveform, sample_rate)
-
-        command = ['compute-kaldi-pitch-feats'] + _convert_args(**kwargs) + ['scp:-', 'ark:-']
-        kaldi_result = _run_kaldi(command, 'scp', wave_file)
-        self.assert_equal(result, expected=kaldi_result)