diff --git a/dsp/__init__.py b/dsp/__init__.py index 806246e..1914fc8 100644 --- a/dsp/__init__.py +++ b/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', +] diff --git a/dsp/parameter.py b/dsp/parameter.py index a06f9d3..d89cea0 100644 --- a/dsp/parameter.py +++ b/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 diff --git a/dsp/processor.py b/dsp/processor.py new file mode 100644 index 0000000..7c238fc --- /dev/null +++ b/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 diff --git a/examples/parameter.py b/examples/parameter.py index 294467b..9660f32 100644 --- a/examples/parameter.py +++ b/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}") diff --git a/tests/test_processor.py b/tests/test_processor.py new file mode 100644 index 0000000..6aef5e1 --- /dev/null +++ b/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