[dsp] Add AudioProcessor

dsp/__init__.py

@@ -1,6 +1,11 @@
 '''Audio DSP utility classes and functions
 '''
 from .parameter import AudioParameter, AudioParameterBool
+from .processor import AudioProcessor, ProcessorSpec
 
-
-__all__ = ['AudioParameter', 'AudioParameterBool']
+__all__ = [
+ 'AudioParameter',
+ 'AudioParameterBool',
+ 'AudioProcessor',
+ 'ProcessorSpec',
+]

dsp/parameter.py

@@ -51,6 +51,12 @@ def value(self) -> bool:
  '''
  return self._value
 
+ @value.setter
+ def value(self, newValue: bool) -> None:
+ '''Sets the parameter to True or False
+ '''
+ self._value = newValue
+
  @property
  def default_value(self) -> bool:
  '''Returns the default value

dsp/processor.py

@@ -0,0 +1,96 @@
+'''ToDo
+'''
+import abc
+import typing
+
+import numpy as np
+
+from .parameter import AudioParameter
+
+
+class ProcessorSpec:
+ '''ToDo
+ '''
+
+ def __init__(
+ self,
+ sample_rate: float = 44100.0,
+ block_size: int = 1024,
+ channels: int = 1
+ ):
+ self._sample_rate = sample_rate
+ self._block_size = block_size
+ self._channels = channels
+
+ @property
+ def sample_rate(self):
+ '''ToDo
+ '''
+ return self._sample_rate
+
+ @property
+ def block_size(self):
+ '''ToDo
+ '''
+ return self._block_size
+
+ @property
+ def channels(self):
+ '''ToDo
+ '''
+ return self._channels
+
+
+class AudioProcessor(abc.ABC):
+ '''Base class for all audio effects & analyzers
+ '''
+ _spec: ProcessorSpec = ProcessorSpec()
+ _parameters: typing.Dict[(str, AudioParameter)] = {}
+
+ @abc.abstractmethod
+ def prepare(self, spec: ProcessorSpec):
+ '''Needs to be implemented in the child class
+ '''
+
+ @abc.abstractmethod
+ def process(self, buffer: np.ndarray) -> np.ndarray:
+ '''Needs to be implemented in the child class
+ '''
+
+ @abc.abstractmethod
+ def release(self) -> None:
+ '''Needs to be implemented in the child class
+ '''
+
+ @property
+ def spec(self) -> ProcessorSpec:
+ '''Returns the current processor specs
+ '''
+ return self._spec
+
+ @property
+ def parameters(self) -> typing.Dict[(str, AudioParameter)]:
+ '''Returns the current processor specs
+ '''
+ return self._parameters
+
+ def add_parameter(self, parameter: AudioParameter) -> None:
+ '''Adds a parameter to the list of parameters for this processor.
+ '''
+ self._parameters[parameter.identifier] = parameter
+
+ @property
+ def state(self) -> typing.Dict:
+ '''Returns the current processor state
+ '''
+ state = {}
+ for key, param in self.parameters.items():
+ state[key] = param.value
+ return state
+
+ @state.setter
+ def state(self, state: typing.Dict) -> None:
+ '''Sets the current processor state
+ '''
+ for key, value in state.items():
+ self._parameters[key].value = value

examples/parameter.py

@@ -1,3 +1,32 @@
 import dsp
 
-print(dsp.AudioParameter('test', 'Test'))
+
+class PassThruProcessor(dsp.AudioProcessor):
+ '''ToDo
+ '''
+
+ def prepare(self, spec: dsp.ProcessorSpec) -> None:
+ '''ToDo
+ '''
+
+ def process(self, buffer):
+ '''ToDo
+ '''
+ return buffer
+
+ def release(self) -> None:
+ '''ToDo
+ '''
+
+
+effect = PassThruProcessor()
+param = dsp.AudioParameterBool('1', 'name', False)
+effect.add_parameter(parameter=param)
+
+
+state = effect.state
+print(f"state: {state['1']}, fx: {effect.parameters['1'].value}")
+effect.parameters['1'].value = True
+print(f"state: {state['1']}, fx: {effect.parameters['1'].value}")
+effect.state = state
+print(f"state: {state['1']}, fx: {effect.parameters['1'].value}")

tests/test_processor.py

@@ -0,0 +1,77 @@
+# pylint: skip-file
+import pytest
+
+import numpy as np
+
+import dsp
+
+
+@pytest.mark.parametrize("tc", [
+ ({'sr': 44100.0, 'bs': 128, 'ch': 1}),
+ ({'sr': 48000.0, 'bs': 512, 'ch': 1}),
+ ({'sr': 88200.0, 'bs': 1024, 'ch': 2}),
+ ({'sr': 96000.0, 'bs': 32, 'ch': 4}),
+])
+def test_processor_spec(tc):
+ spec = dsp.ProcessorSpec(
+ sample_rate=tc['sr'],
+ block_size=tc['bs'],
+ channels=tc['ch'],
+ )
+ assert spec.sample_rate == tc['sr']
+ assert spec.block_size == tc['bs']
+ assert spec.channels == tc['ch']
+
+
+class TestProcessor(dsp.AudioProcessor):
+ '''Base class for all audio effects & analyzers
+ '''
+
+ def prepare(self, spec: dsp.ProcessorSpec) -> None:
+ '''Needs to be implemented in the child class
+ '''
+
+ def process(self, buffer: np.ndarray) -> np.ndarray:
+ '''Needs to be implemented in the child class
+ '''
+ return buffer
+
+ def release(self) -> None:
+ '''Needs to be implemented in the child class
+ '''
+
+
+def test_audio_processor():
+ effect = TestProcessor()
+ assert effect.spec.sample_rate == 44100.0
+ assert effect.spec.block_size == 1024
+ assert effect.spec.channels == 1
+
+ param = dsp.AudioParameterBool('1', 'name', False)
+ effect.add_parameter(parameter=param)
+
+ assert len(effect.parameters) == 1
+ assert effect.parameters['1'].identifier == '1'
+ assert effect.parameters['1'].name == 'name'
+ assert not effect.parameters['1'].value
+ assert not effect.parameters['1'].default_value
+
+ effect.prepare(dsp.ProcessorSpec())
+ assert effect.process(None) is None
+
+ buffer = np.zeros((1024,), dtype=np.float64)
+ assert type(effect.process(buffer)) == np.ndarray
+
+ state = effect.state
+ assert not effect.parameters['1'].value
+ effect.parameters['1'].value = True
+ assert effect.parameters['1'].value
+
+ effect.state = {'1': False}
+ assert not effect.parameters['1'].value
+
+ effect.parameters['1'].value = True
+ assert effect.parameters['1'].value
+
+ effect.state = state
+ assert not effect.parameters['1'].value