[cm] Separating delay into buffering and modeling (#39)

* [cm] Splitting delay into buffering and model delay

* [cm] Updating version to 1.3.0

* [cm] Removing unnecessary method calls

.gitignore

@@ -199,8 +199,9 @@ pip-selfcheck.json
 #   git rm -r .ipynb_checkpoints/
 
 # User added
-.idea
-exports
+.idea/
+exports/
+export_model/
+models/
+out/
 scratch.py
-export_model
-models

examples/example_delayed_passthrough.py

@@ -0,0 +1,109 @@
+import logging
+import os
+import pathlib
+from argparse import ArgumentParser
+from typing import Dict, List
+
+import torch as tr
+import torch.nn as nn
+from torch import Tensor
+
+from neutone_sdk import WaveformToWaveformBase, NeutoneParameter
+from neutone_sdk.utils import save_neutone_model
+
+logging.basicConfig()
+log = logging.getLogger(__name__)
+log.setLevel(level=os.environ.get("LOGLEVEL", "INFO"))
+
+
+class DelayedPassthroughModel(nn.Module):
+    def __init__(self, delay_n_samples: int, in_ch: int = 2) -> None:
+        super().__init__()
+        self.delay_n_samples = delay_n_samples
+        self.delay_buf = tr.zeros((in_ch, delay_n_samples))
+
+    def forward(self, x: Tensor) -> Tensor:
+        x = tr.cat([self.delay_buf, x], dim=-1)
+        self.delay_buf[:, :] = x[:, -self.delay_n_samples:]
+        x = x[:, :-self.delay_n_samples]
+        return x
+
+
+class DelayedPassthroughModelWrapper(WaveformToWaveformBase):
+    def get_model_name(self) -> str:
+        return "delayed.passthrough"
+
+    def get_model_authors(self) -> List[str]:
+        return ["Christopher Mitcheltree"]
+
+    def get_model_short_description(self) -> str:
+        return "Delayed passthrough model."
+
+    def get_model_long_description(self) -> str:
+        return "Delays the input audio by some number of samples. Should be tested with 50/50 dry/wet."
+
+    def get_technical_description(self) -> str:
+        return "Delays the input audio by some number of samples. Should be tested with 50/50 dry/wet."
+
+    def get_technical_links(self) -> Dict[str, str]:
+        return {}
+
+    def get_tags(self) -> List[str]:
+        return []
+
+    def get_model_version(self) -> str:
+        return "1.0.0"
+
+    def is_experimental(self) -> bool:
+        return True
+
+    def get_neutone_parameters(self) -> List[NeutoneParameter]:
+        return []
+
+    @tr.jit.export
+    def is_input_mono(self) -> bool:
+        return False
+
+    @tr.jit.export
+    def is_output_mono(self) -> bool:
+        return False
+
+    @tr.jit.export
+    def get_native_sample_rates(self) -> List[int]:
+        return [44100]  # Change this to test different scenarios
+
+    @tr.jit.export
+    def get_native_buffer_sizes(self) -> List[int]:
+        return [2048]  # Change this to test different scenarios
+
+    @tr.jit.export
+    def reset_model(self) -> bool:
+        self.model.delay_buf.fill_(0)
+        return True
+
+    @tr.jit.export
+    def calc_model_delay_samples(self) -> int:
+        return self.model.delay_n_samples
+
+    @tr.jit.export
+    def get_wet_default_value(self) -> float:
+        return 0.5
+
+    @tr.jit.export
+    def get_dry_default_value(self) -> float:
+        return 0.5
+
+    def do_forward_pass(self, x: Tensor, params: Dict[str, Tensor]) -> Tensor:
+        x = self.model.forward(x)
+        return x
+
+
+if __name__ == "__main__":
+    parser = ArgumentParser()
+    parser.add_argument("-o", "--output", default="export_model")
+    args = parser.parse_args()
+    root_dir = pathlib.Path(args.output)
+
+    model = DelayedPassthroughModel(delay_n_samples=500)  # Change delay_n_samples to test different scenarios
+    wrapper = DelayedPassthroughModelWrapper(model)
+    save_neutone_model(wrapper, root_dir, dump_samples=True, submission=True)

examples/example_rave_prefilter.py

@@ -1,23 +1,21 @@
-import json
 import logging
 import os
 from argparse import ArgumentParser
 from pathlib import Path
 from typing import Dict, List
 
 import torch
+import torchaudio
 from torch import Tensor, nn
 
-import torchaudio
+from neutone_sdk import WaveformToWaveformBase, NeutoneParameter
 from neutone_sdk.audio import (
     AudioSample,
     AudioSamplePair,
     render_audio_sample,
 )
-
-from neutone_sdk import WaveformToWaveformBase, NeutoneParameter
+from neutone_sdk.filters import FIRFilter, FilterType
 from neutone_sdk.utils import save_neutone_model
-from neutone_sdk.filters import FIRFilter, IIRFilter, FilterType
 
 logging.basicConfig()
 log = logging.getLogger(__name__)
@@ -103,7 +101,7 @@ def get_native_sample_rates(self) -> List[int]:
     def get_native_buffer_sizes(self) -> List[int]:
         return [2048]
 
-    def calc_min_delay_samples(self) -> int:
+    def calc_model_delay_samples(self) -> int:
         # model latency should also be added if non-causal
         return self.pre_filter.delay
 

examples/example_rave_v1_prefilter.py

@@ -1,23 +1,21 @@
-import json
 import logging
 import os
 from argparse import ArgumentParser
 from pathlib import Path
 from typing import Dict, List
 
 import torch
+import torchaudio
 from torch import Tensor, nn
 
-import torchaudio
+from neutone_sdk import WaveformToWaveformBase, NeutoneParameter
 from neutone_sdk.audio import (
     AudioSample,
     AudioSamplePair,
     render_audio_sample,
 )
-
-from neutone_sdk import WaveformToWaveformBase, NeutoneParameter
+from neutone_sdk.filters import FIRFilter, FilterType
 from neutone_sdk.utils import save_neutone_model
-from neutone_sdk.filters import FIRFilter, IIRFilter, FilterType
 
 logging.basicConfig()
 log = logging.getLogger(__name__)
@@ -103,7 +101,7 @@ def get_native_sample_rates(self) -> List[int]:
     def get_native_buffer_sizes(self) -> List[int]:
         return [2048]
 
-    def calc_min_delay_samples(self) -> int:
+    def calc_model_delay_samples(self) -> int:
         # model latency should also be added if non-causal
         return self.pre_filter.delay
 

examples/example_spectral_filter.py

@@ -134,7 +134,7 @@ def __init__(
         if use_debug_mode:
             log.info(f"Supported buffer sizes = {self.get_native_buffer_sizes()}")
             log.info(f"Supported sample rate = {self.get_native_sample_rates()}")
-            log.info(f"STFT delay = {self.calc_min_delay_samples()}")
+            log.info(f"STFT delay = {self.calc_model_delay_samples()}")
 
     def get_model_name(self) -> str:
         return "spectral.filter"
@@ -201,9 +201,9 @@ def get_native_buffer_sizes(self) -> List[int]:
         )  # Possible buffer sizes are determined by the STFT parameters
 
     @tr.jit.export
-    def calc_min_delay_samples(self) -> int:
+    def calc_model_delay_samples(self) -> int:
         # TODO(cm): make a model specific version of this method?
-        return self.stft.calc_min_delay_samples()  # This is equal to `fade_n_samples`
+        return self.stft.calc_model_delay_samples()  # This is equal to `fade_n_samples`
 
     def set_model_buffer_size(self, n_samples: int) -> bool:
         self.stft.set_buffer_size(n_samples)

neutone_sdk/constants.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 
-SDK_VERSION = "1.2.1"
+SDK_VERSION = "1.3.0"
 
 MAX_N_PARAMS = 4
 MAX_N_AUDIO_SAMPLES = 3

neutone_sdk/realtime_stft.py

@@ -213,7 +213,7 @@ def set_buffer_size(self, io_n_samples: int) -> None:
         self.reset()
 
     @tr.jit.export
-    def calc_min_delay_samples(self) -> int:
+    def calc_model_delay_samples(self) -> int:
         return self.fade_n_samples
 
     @tr.jit.export

neutone_sdk/sqw.py

@@ -321,14 +321,18 @@ def is_resampling(self) -> bool:
         return self.resample_sandwich.is_resampling()
 
     @tr.jit.export
-    def calc_min_delay_samples(self) -> int:
-        model_min_delay = self.w2w_base.calc_min_delay_samples()
-        wrapper_min_delay = self.calc_delay_samples(self.io_bs, self.model_bs)
-        min_delay = model_min_delay + wrapper_min_delay
+    def calc_buffering_delay_samples(self) -> int:
+        delay_samples = self.calc_delay_samples(self.io_bs, self.model_bs)
         if self.is_resampling():
-            min_delay = int(min_delay * self.daw_bs / self.io_bs)
+            delay_samples = int(delay_samples * self.daw_bs / self.io_bs)
+        return delay_samples
 
-        return min_delay
+    @tr.jit.export
+    def calc_model_delay_samples(self) -> int:
+        delay_samples = self.w2w_base.calc_model_delay_samples()
+        if self.is_resampling():
+            delay_samples = int(delay_samples * self.daw_bs / self.io_bs)
+        return delay_samples
 
     @tr.jit.export
     def set_daw_sample_rate_and_buffer_size(
@@ -433,7 +437,8 @@ def get_preserved_attributes(self) -> List[str]:
             "get_input_gain_default_value",
             "get_output_gain_default_value",
             "is_resampling",
-            "calc_min_delay_samples",
+            "calc_buffering_delay_samples",
+            "calc_model_delay_samples",
             "set_daw_sample_rate_and_buffer_size",
             "reset",
             "get_preserved_attributes",

neutone_sdk/utils.py

@@ -1,11 +1,15 @@
-import copy
 import io
 import json
 import logging
 import os
 import random
 from pathlib import Path
 from typing import Tuple, Dict, List
+
+import torch as tr
+from torch import Tensor
+from torch.jit import ScriptModule
+
 from neutone_sdk.audio import (
     AudioSample,
     AudioSamplePair,
@@ -16,10 +20,6 @@
 from neutone_sdk.core import NeutoneModel
 from neutone_sdk.metadata import validate_metadata
 
-import torch as tr
-from torch import Tensor
-from torch.jit import ScriptModule
-
 logging.basicConfig()
 log = logging.getLogger(__name__)
 log.setLevel(level=os.environ.get("LOGLEVEL", "INFO"))
@@ -157,13 +157,16 @@ def save_neutone_model(
             loaded_model, loaded_model.get_preserved_attributes()
         )
         log.info("Testing methods used by the VST...")
-        loaded_model.calc_min_delay_samples()
         loaded_model.set_daw_sample_rate_and_buffer_size(48000, 512)
         loaded_model.reset()
         loaded_model.is_resampling()
         log.info(
-            f"Delay reported to the DAW for 48000 Hz sampling rate and 512 buffer size: "
-            f"{loaded_model.calc_min_delay_samples()}"
+            f"Buffering delay reported to the DAW for 48000 Hz sampling rate and 512 buffer size: "
+            f"{loaded_model.calc_buffering_delay_samples()}"
+        )
+        log.info(
+            f"Model delay reported to the DAW for 48000 Hz sampling rate and 512 buffer size: "
+            f"{loaded_model.calc_model_delay_samples()}"
         )
 
         if submission:  # Do extra checks

neutone_sdk/wavform_to_wavform.py

@@ -256,9 +256,9 @@ def is_resampling(self) -> bool:
         return False
 
     @tr.jit.export
-    def calc_min_delay_samples(self) -> int:
+    def calc_model_delay_samples(self) -> int:
         """
-        If the model introduces a minimum amount of delay to the output audio,
+        If the model introduces an amount of delay to the output audio,
         for example due to a lookahead buffer or cross-fading, return it here
         so that it can be forwarded to the DAW to compensate. Defaults to 0.
 
@@ -344,7 +344,7 @@ def get_preserved_attributes(self) -> List[str]:
                 "get_native_sample_rates",
                 "get_native_buffer_sizes",
                 "is_resampling",
-                "calc_min_delay_samples",
+                "calc_model_delay_samples",
                 "set_sample_rate_and_buffer_size",
                 "set_daw_sample_rate_and_buffer_size",
                 "reset",

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "neutone_sdk"
-version = "1.2.1"
+version = "1.3.0"
 description = "SDK for wrapping deep learning models for usage in the Neutone plugin"
 readme = "README.md"
 authors = ["Qosmo <[email protected]>"]