Test int resample_chunk(), Refactoring

.github/workflows/run-test.yml

@@ -24,7 +24,7 @@ jobs:
     - name: Set min macOS version
       if: runner.os == 'macOS'
       run: |
-        echo "MACOS_DEPLOYMENT_TARGET=10.14" >> $GITHUB_ENV
+        echo "MACOSX_DEPLOYMENT_TARGET=10.14" >> $GITHUB_ENV
 
     - name: Build and install
       run: |

src/soxr/__init__.py

@@ -26,28 +26,24 @@
 _DTYPE_ERR_STR = 'Data type must be one of [float32, float64, int16, int32], not {}'
 _QUALITY_ERR_STR = "Quality must be one of [QQ, LQ, MQ, HQ, VHQ]"
 
+_QUALITY_ENUM_DICT = {
+    VHQ: VHQ, 'vhq': VHQ, 'soxr_vhq': VHQ,
+    HQ: HQ, 'hq': HQ, 'soxr_hq': HQ,
+    MQ: MQ, 'mq': MQ, 'soxr_mq': MQ,
+    LQ: LQ, 'lq': LQ, 'soxr_lq': LQ,
+    QQ: QQ, 'qq': QQ, 'soxr_qq': QQ,
+}
+
 
 def _quality_to_enum(q):
-    if q in (VHQ, HQ, MQ, LQ, QQ):
-        return q
+    if isinstance(q, str):
+        q = q.lower()
 
-    if type(q) is int:
+    try:
+        return _QUALITY_ENUM_DICT[q]
+    except (KeyError, TypeError):
         raise ValueError(_QUALITY_ERR_STR)
 
-    q = q.lower()
-    if q in ('vhq', 'soxr_vhq'):
-        return VHQ
-    elif q in ('hq', 'soxr_hq'):
-        return HQ
-    elif q in ('mq', 'soxr_mq'):
-        return MQ
-    elif q in ('lq', 'soxr_lq'):
-        return LQ
-    elif q in ('qq', 'soxr_qq'):
-        return QQ
-
-    raise ValueError(_QUALITY_ERR_STR)
-
 
 def _to_soxr_datatype(ntype):
     if ntype == np.float32:

tests/test_resample.py

@@ -84,57 +84,50 @@ def test_channel_match(channels):
     assert np.allclose(y_oneshot, y_split)
 
 
-@pytest.mark.parametrize('in_rate, out_rate', [(44100, 32000), (32000, 44100)])
-@pytest.mark.parametrize('dtype', [np.float32, np.float64])
-@pytest.mark.parametrize('channels', [1, 2])
-def test_stream_match(in_rate, out_rate, dtype, channels):
-    # test resample_chunk()
-    CHUNK_SIZE = 509
-    x = np.random.randn(49999, channels).astype(dtype)
-
-    y_oneshot = soxr._resample_oneshot(x, in_rate, out_rate)
+def stream_resample(x, in_rate, out_rate, chunk_size, dtype):
+    channels = x.shape[1]
 
     rs_stream = soxr.ResampleStream(in_rate, out_rate, channels, dtype=dtype)
 
-    y_list = []
-    for idx in range(0, len(x), CHUNK_SIZE):
-        end = idx + CHUNK_SIZE
+    y_list = [np.ndarray([0, channels], dtype=dtype)]
+    for idx in range(0, len(x), chunk_size):
+        end = idx + chunk_size
         eof = False
         if len(x) <= end:
             eof = True
             end = len(x)
         y_chunk = rs_stream.resample_chunk(x[idx:end], last=eof)
         y_list.append(y_chunk)
 
-    y_stream = np.concatenate(y_list)
-
-    assert np.allclose(y_oneshot, y_stream)
+    return np.concatenate(y_list)
 
 
 @pytest.mark.parametrize('in_rate, out_rate', [(44100, 32000), (32000, 44100)])
-@pytest.mark.parametrize('chunk_size', [7, 50, 101, 44100])
-@pytest.mark.parametrize('length', [0, 1, 100, 101, 31999, 32000, 44100, 44101, 266151])
-def test_stream_length(in_rate, out_rate, chunk_size, length):
+@pytest.mark.parametrize('chunk_size', [7, 509, 44100])
+@pytest.mark.parametrize('length', [0, 100, 31999, 44100, 266151])
+@pytest.mark.parametrize('dtype', ['float32', np.float64])
+def test_stream_length(in_rate, out_rate, chunk_size, length, dtype):
     # test resample_chunk() with various length and chunk size
-    x = np.random.randn(length, 1).astype(np.float32)
+    x = np.random.randn(length, 1).astype(dtype)  # 1ch
 
     y_oneshot = soxr._resample_oneshot(x, in_rate, out_rate)
+    y_stream = stream_resample(x, in_rate, out_rate, chunk_size, dtype)
 
-    rs_stream = soxr.ResampleStream(in_rate, out_rate, 1, dtype=np.float32)
+    assert np.allclose(y_oneshot, y_stream)
 
-    y_list = [np.ndarray([0, 1], dtype=np.float32)]
-    for idx in range(0, len(x), chunk_size):
-        end = idx + chunk_size
-        eof = False
-        if len(x) <= end:
-            eof = True
-            end = len(x)
-        y_chunk = rs_stream.resample_chunk(x[idx:end], last=eof)
-        y_list.append(y_chunk)
 
-    y_stream = np.concatenate(y_list)
+@pytest.mark.parametrize('in_rate, out_rate', [(48000, 22050), (8000, 48000)])
+@pytest.mark.parametrize('chunk_size', [50, 101])
+@pytest.mark.parametrize('length', [1, 101, 32000, 44101, 49999])
+@pytest.mark.parametrize('dtype', ['int32', np.int16])
+def test_stream_int(in_rate, out_rate, chunk_size, length, dtype):
+    # test int resample_chunk() with various length and chunk size
+    x = (np.random.randn(length, 2) * 5000).astype(dtype)   # 2ch
 
-    assert np.allclose(y_oneshot, y_stream)
+    y_oneshot = soxr._resample_oneshot(x, in_rate, out_rate)
+    y_stream = stream_resample(x, in_rate, out_rate, chunk_size, dtype)
+
+    assert np.allclose(y_oneshot, y_stream, atol=2)
 
 
 def make_tone(freq, sr, duration):
@@ -147,7 +140,7 @@ def make_tone(freq, sr, duration):
 
 
 @pytest.mark.parametrize('in_rate,out_rate', [(44100, 22050), (22050, 32000)])
-@pytest.mark.parametrize('quality', ['VHQ', 'HQ', 'MQ', 'LQ', 'QQ'])
+@pytest.mark.parametrize('quality', [soxr.VHQ, 'HQ', 'SOXR_MQ', 'lq', 'soxr_qq'])
 def test_quality_sine(in_rate, out_rate, quality):
     # compare result with reference
     FREQ = 32.0
@@ -175,9 +168,12 @@ def test_int_sine(in_rate, out_rate, dtype):
 
     y_pred = soxr.resample(x, in_rate, out_rate)
     y_split = soxr.resample(np.asfortranarray(x), in_rate, out_rate)
+    y_oneshot = soxr._resample_oneshot(x, in_rate, out_rate)
 
     assert np.allclose(y, y_pred, atol=2)
     assert np.allclose(y, y_split, atol=2)
+    assert np.allclose(y_oneshot, y_pred, atol=2)
+    assert np.allclose(y_oneshot, y_split, atol=2)
 
 
 @pytest.mark.parametrize('num_task', [2, 3, 4, 5, 6, 7, 8, 9, 12, 17, 32])