[BREAKING] Remove support for non-integer latency compensation (#547)

* Remove support for non-integer delay, Black

* Remove scipy dependency, clean up environment definitions, make pytest verbose on workflow

Author: sdatkinson

.github/workflows/python-package.yml

@@ -42,4 +42,4 @@ jobs:
       - name: Test with pytest
         run: |
           python -m pip install pytest pytest-mock
-          xvfb-run -a pytest
+          xvfb-run -a pytest -v

environments/environment_cpu_apple.yml

@@ -8,6 +8,7 @@ name: nam
 channels:
   - conda-forge # pytest-mock
   - pytorch
+  - defaults
 dependencies:
   - python>=3.9
   - black
@@ -24,7 +25,6 @@ dependencies:
   # Performance note:
   # https://github.com/sdatkinson/neural-amp-modeler/issues/505
   - pytorch
-  - scipy
   - semver
   - tensorboard
   - tqdm

environments/environment_gpu.yml

@@ -7,6 +7,7 @@ channels:
   - conda-forge # pytest-mock
   - pytorch
   - nvidia  # GPU
+  - defaults
 dependencies:
   - python>=3.9
   - black
@@ -25,7 +26,6 @@ dependencies:
   # You're going to need to look at Table 3 here: 
   # https://docs.nvidia.com/cuda/cuda-toolkit-release-notes/index.html#cuda-major-component-versions__table-cuda-toolkit-driver-versions
   - pytorch::pytorch-cuda=12.1  # GPU
-  - scipy
   - semver
   - tensorboard
   - tqdm

nam/cli.py

@@ -92,6 +92,7 @@ def nam_hello_world():
     was installed successfully
     """
     from nam import __version__
+
     msg = f"""
     Neural Amp Modeler
 

nam/data.py

@@ -26,7 +26,6 @@
 import numpy as _np
 import torch as _torch
 import wavio as _wavio
-from scipy.interpolate import interp1d as _interp1d
 from torch.utils.data import Dataset as _Dataset
 from tqdm import tqdm as _tqdm
 
@@ -176,37 +175,6 @@ def __getitem__(self, idx: int):
         pass
 
 
-class _DelayInterpolationMethod(_Enum):
-    """
-    :param LINEAR: Linear interpolation
-    :param CUBIC: Cubic spline interpolation
-    """
-
-    # Note: these match scipy.interpolate.interp1d kwarg "kind"
-    LINEAR = "linear"
-    CUBIC = "cubic"
-
-
-def _interpolate_delay(
-    x: _torch.Tensor, delay: float, method: _DelayInterpolationMethod
-) -> _np.ndarray:
-    """
-    NOTE: This breaks the gradient tape!
-    """
-    if delay == 0.0:
-        return x
-    t_in = _np.arange(len(x))
-    n_out = len(x) - int(_np.ceil(_np.abs(delay)))
-    if delay > 0:
-        t_out = _np.arange(n_out) + delay
-    elif delay < 0:
-        t_out = _np.arange(len(x) - n_out, len(x)) - _np.abs(delay)
-
-    return _torch.Tensor(
-        _interp1d(t_in, x.detach().cpu().numpy(), kind=method.value)(t_out)
-    )
-
-
 class XYError(ValueError, DataError):
     """
     Exceptions related to invalid x and y provided for data sets
@@ -268,9 +236,6 @@ def __init__(
         start_seconds: _Optional[_Union[int, float]] = None,
         stop_seconds: _Optional[_Union[int, float]] = None,
         delay: _Optional[_Union[int, float]] = None,
-        delay_interpolation_method: _Union[
-            str, _DelayInterpolationMethod
-        ] = _DelayInterpolationMethod.CUBIC,
         y_scale: float = 1.0,
         x_path: _Optional[_Union[str, _Path]] = None,
         y_path: _Optional[_Union[str, _Path]] = None,
@@ -305,8 +270,7 @@ def __init__(
             the end of the audio. Requires providing `sample_rate`.
         :param delay: In samples. Positive means we get rid of the start of x, end of y
             (i.e. we are correcting for an alignment error in which y is delayed behind
-            x). If a non-integer delay is provided, then y is interpolated, with
-            the extra sample removed.
+            x). Only integer delays are supported.
         :param y_scale: Multiplies the output signal by a factor (e.g. if the data are
             too quiet).
         :param input_gain: In dB. If the input signal wasn't fed to the amp at unity
@@ -335,17 +299,13 @@ def __init__(
             stop_seconds,
             self.sample_rate,
         )
-        if not isinstance(delay_interpolation_method, _DelayInterpolationMethod):
-            delay_interpolation_method = _DelayInterpolationMethod(
-                delay_interpolation_method
-            )
         if require_input_pre_silence is not None:
             self._validate_preceding_silence(
                 x, start, require_input_pre_silence, self.sample_rate
             )
         x, y = [z[start:stop] for z in (x, y)]
         if delay is not None and delay != 0:
-            x, y = self._apply_delay(x, y, delay, delay_interpolation_method)
+            x, y = self._apply_delay(x, y, delay)
         x_scale = 10.0 ** (input_gain / 20.0)
         x = x * x_scale
         y = y * y_scale
@@ -477,15 +437,12 @@ def _apply_delay(
         x: _torch.Tensor,
         y: _torch.Tensor,
         delay: _Union[int, float],
-        method: _DelayInterpolationMethod,
     ) -> _Tuple[_torch.Tensor, _torch.Tensor]:
         # Check for floats that could be treated like ints (simpler algorithm)
         if isinstance(delay, float) and int(delay) == delay:
             delay = int(delay)
         if isinstance(delay, int):
             return cls._apply_delay_int(x, y, delay)
-        elif isinstance(delay, float):
-            return cls._apply_delay_float(x, y, delay, method)
         else:
             raise TypeError(type(delay))
 
@@ -501,22 +458,6 @@ def _apply_delay_int(
             y = y[:delay]
         return x, y
 
-    @classmethod
-    def _apply_delay_float(
-        cls,
-        x: _torch.Tensor,
-        y: _torch.Tensor,
-        delay: float,
-        method: _DelayInterpolationMethod,
-    ) -> _Tuple[_torch.Tensor, _torch.Tensor]:
-        n_out = len(y) - int(_np.ceil(_np.abs(delay)))
-        if delay > 0:
-            x = x[:n_out]
-        elif delay < 0:
-            x = x[-n_out:]
-        y = _interpolate_delay(y, delay, method)
-        return x, y
-
     @classmethod
     def _validate_start_stop(
         cls,

nam/models/recurrent.py

@@ -212,6 +212,7 @@ def _export_config(self):
             "hidden_size": self._core.hidden_size,
             "num_layers": self._core.num_layers,
         }
+
     def _export_weights(self):
         """
         * Loop over cells:

pyproject.toml

@@ -17,7 +17,6 @@ dependencies = [
     "numpy<2",
     "pydantic>=2.0.0",
     "pytorch_lightning",
-    "scipy",
     "sounddevice",
     "tensorboard",
     "torch",

tests/test_nam/test_data.py

@@ -36,30 +36,10 @@ def test_apply_delay_zero(self):
         no change.
         """
         x, y = self._create_xy()
-        x_out, y_out = data.Dataset._apply_delay(
-            x, y, 0, data._DelayInterpolationMethod.CUBIC
-        )
+        x_out, y_out = data.Dataset._apply_delay(x, y, 0)
         assert torch.all(x == x_out)
         assert torch.all(y == y_out)
 
-    @pytest.mark.parametrize("method", (data._DelayInterpolationMethod))
-    def test_apply_delay_float_negative(self, method):
-        n = 7
-        delay = -2.5
-        x_out, y_out = self._t_apply_delay_float(n, delay, method)
-
-        assert torch.all(x_out == torch.Tensor([3, 4, 5, 6]))
-        assert torch.all(y_out == torch.Tensor([0.5, 1.5, 2.5, 3.5]))
-
-    @pytest.mark.parametrize("method", (data._DelayInterpolationMethod))
-    def test_apply_delay_float_positive(self, method):
-        n = 7
-        delay = 2.5
-        x_out, y_out = self._t_apply_delay_float(n, delay, method)
-
-        assert torch.all(x_out == torch.Tensor([0, 1, 2, 3]))
-        assert torch.all(y_out == torch.Tensor([2.5, 3.5, 4.5, 5.5]))
-
     def test_apply_delay_int_negative(self):
         """
         Assert proper function of Dataset._apply_delay() when a positive integer delay
@@ -299,29 +279,12 @@ def _create_xy(
                 torch.tile((torch.linspace(0.0, 1.0, n) > 0.5)[None, :], (2, 1))
             )
 
-    def _t_apply_delay_float(
-        self, n: int, delay: int, method: data._DelayInterpolationMethod
-    ):
-        x, y = self._create_xy(
-            n=n, method=_XYMethod.ARANGE, must_be_in_valid_range=False
-        )
-
-        x_out, y_out = data.Dataset._apply_delay(x, y, delay, method)
-        # 7, +/-2.5 -> 4
-        n_out = n - int(np.ceil(np.abs(delay)))
-        assert len(x_out) == n_out
-        assert len(y_out) == n_out
-
-        return x_out, y_out
-
     def _t_apply_delay_int(self, n: int, delay: int):
         x, y = self._create_xy(
             n=n, method=_XYMethod.ARANGE, must_be_in_valid_range=False
         )
 
-        x_out, y_out = data.Dataset._apply_delay(
-            x, y, delay, data._DelayInterpolationMethod.CUBIC
-        )
+        x_out, y_out = data.Dataset._apply_delay(x, y, delay)
         n_out = n - np.abs(delay)
         assert len(x_out) == n_out
         assert len(y_out) == n_out

tests/test_nam/test_models/test_base.py

@@ -31,7 +31,6 @@ def pad_start_default(self) -> bool:
     def receptive_field(self) -> int:
         return 1
 
-
     def _export_config(self):
         pass