effect_chains and effects_applied are now lists of filter functions
MODIFY
nussl/core/audio_signal.py
nussl/core/audio_signal.py
@@ -21,7 +21,7 @@ from . import effects
__all__ = ['AudioSignal', 'STFTParams', 'AudioSignalException']
STFTParams = namedtuple('STFTParams',
- ['window_length', 'hop_length', 'window_type'],
+ ['window_length', 'hop_length', 'window_type']
)
STFTParams.__new__.__defaults__ = (None,) * len(STFTParams._fields)
"""
@@ -160,8 +160,10 @@ class AudioSignal(object):
path_to_input_file (``str``): Path to the input file. ``None`` if this AudioSignal never
loaded a file, i.e., initialized with a ``np.ndarray``.
label (``str``): A user-definable label for this :class:`AudioSignal` object.
- applied_effects (``list`` of ``str``): Effects applied to this :class:`AudioSignal`
- object. For more information, see apply_effects.
+ applied_effects (``list`` of ``effects.FilterFunction``): Effects applied to this
+ :class:`AudioSignal` object. For more information, see apply_effects.
+ effects_chain (``list`` of ``effects.FilterFunction``): Effects queues to be applied to
+ this :class:`AudioSignal` object. For more information, see apply_effects.
"""
@@ -202,7 +204,7 @@ class AudioSignal(object):
# Effects
self._effects_chain = []
- self.effects_applied = []
+ self._effects_applied = []
def __str__(self):
dur = f'{self.signal_duration:0.3f}' if self.signal_duration else '[unknown]'
@@ -714,6 +716,28 @@ class AudioSignal(object):
'because self.stft_data is None')
return 20 * np.log10(np.abs(self.stft_data) + 1e-8)
+ @property
+ def effects_chain(self):
+ """
+ (``list`` of ``nussl.core.FilterFunction``): Returns a copy of the AudioSignal's
+ effect chain.
+
+ See Also:
+ * :func:`apply_effects`
+ """
+ return self._effects_chain.copy()
+
+ @property
+ def effects_applied(self):
+ """
+ (``list`` of ``nussl.core.FilterFunction``): Returns a list of effects applied to
+ the AudioSignal.
+
+ See Also:
+ * :func:`apply_effects`
+ """
+ return self._effects_applied.copy()
+
##################################################
# I/O
##################################################
@@ -1135,6 +1159,7 @@ class AudioSignal(object):
return ipd, ild
+
##################################################
# Utilities
##################################################
@@ -1672,7 +1697,7 @@ class AudioSignal(object):
Notes:
The effects will be added in the order that they are added to the effects chain, unless
- `user_order=False`, in case the order is not gauranteed to be preserved. Setting
+ `user_order=False`, in case the order is not guaranteed to be preserved. Setting
`user_order=False` will apply all SoX effects first, then FFMpeg effects, which can
sped up processing time by ~30% in our experiments.
@@ -1698,24 +1723,40 @@ class AudioSignal(object):
>>> signal.signal_duration
10.0
+ You can find this effect in the AudioSignal's effects chain.
+
+ >>> effect = signal.effects_chain[0]
+ >>> str(effect)
+ "time_stretch (params: {factor=0.5})"
+
However, the signal's duration hasn't changed! You will need to call `apply_effects()`
- to apply the changes in the signal's effects chains.
+ to apply the changes in the signal's effects chains. Applied effects can be found in
+ `effects_applied`.
>>> new_signal = signal.apply_effects()
>>> new_signal.signal_duration
5.0
+ >>> str(new_signal.effects_applied[0])
+ "time_stretch (params: {factor=0.5})"
+
>>> # This doesn't change the original signal
>>> signal.signal_duration
10.0
+ You can iterate through effects_chain to use the properties of FilterFunction
+ objects as arguments to `make_effect`:
+
+ >>> for effect in signal1.effects_applied:
+ >>> filter_ = effect.filter
+ >>> params = effect.params
+ >>> signal2.make_effect(filter_, **params)
+
Using `apply_effects()` will clear out the current effects chain. This behavior can be
- avoided by setting `reset` to False. Applied effects can be found in `effects_applied`:
+ avoided by setting `reset` to False.
>>> another_signal = signal.apply_effects()
>>> another_signal.signal_duration
10.0
- >>> new_signal.effects_applied
- ["time_stretch"]
To clear out the current effects chain without applying effect, use
`reset_effects_chain()`. It will not revert effects already applied (i.e., your audio
@@ -1779,7 +1820,7 @@ class AudioSignal(object):
self.reset_effects_chain()
if overwrite:
self.audio_data = new_signal.audio_data
- self.effects_applied += new_signal.effects_applied
+ self._effects_applied += new_signal.effects_applied
self.stft_data = None
return self
@@ -1828,6 +1869,14 @@ class AudioSignal(object):
Is the same as
>>> signal.make_effect("time_stretch", factor=1.5)
+ The attributes of a FilterFunction in the lists effects_applied or effects_chain may
+ used with `make_effect`.
+
+ >>> for effect in signal1.effects_applied:
+ >>> filter_ = effect.filter
+ >>> params = effect.params
+ >>> signal2.make_effect(filter_, **params)
+
Notes:
This effect won't be applied until you call `apply_effect()`!
@@ -1888,10 +1937,11 @@ class AudioSignal(object):
self._effects_chain.append(effects.time_stretch(factor))
return self
- def pitch_shift(self, shift):
+ def pitch_shift(self, n_semitones):
"""
Add pitch shift effect to AudioSignal's effect chain.
- A positive shift will raise the pitch of the signal by `shift` semitones.
+ A positive shift will raise the pitch of the signal by `n_semitones`
+ semitones.
This is a SoX effect. Please see:
https://pysox.readthedocs.io/en/latest/_modules/sox/transform.html#Transformer.pitch
@@ -1901,7 +1951,7 @@ class AudioSignal(object):
This effect won't be applied until you call `apply_effect()`!
Args:
- shift (float): The number of semitones to shift the audio.
+ n_semitones (integer): The number of semitones to shift the audio.
Positive values increases the frequency of the signal
Returns:
self: Initial AudioSignal with updated effect chains
@@ -1911,7 +1961,7 @@ class AudioSignal(object):
* :func:`make_effect`: Syntactic sugar for adding an effect to the chain by name.
* :func:`reset_effects_chain`: Empties the effects chain without applying any effects.
"""
- self._effects_chain.append(effects.pitch_shift(shift))
+ self._effects_chain.append(effects.pitch_shift(n_semitones))
return self
def low_pass(self, freq, poles=2, width_type="h", width=0.707):

MODIFY
nussl/core/effects.py
nussl/core/effects.py
@@ -24,6 +24,7 @@ import numpy as np
import warnings
import ffmpeg
import tempfile
+import copy
try:
import soxbindings as sox
except Exception:
@@ -36,12 +37,17 @@ from .utils import _close_temp_files
class FilterFunction:
"""
The FilterFunction class is an abstract class for functions that take
- audio processing streams, such as ffmpeg-python and pysox.
+ audio processing streams, such as ffmpeg-python and pysox effects.
- Don't call this class. It will not do anything.
+ Don't call this class. It will not do anything. Please use either FFmpegFilter
+ or SoXFilter.
+
+ Attributes:
+ filter: Name of filter used
+ params: Parameters for filter
"""
- def __init__(self, _filter, **kwargs):
- self.filter = _filter
+ def __init__(self, filter_, **kwargs):
+ self.filter = filter_
self.params = kwargs
def __str__(self):
@@ -61,9 +67,10 @@ class FFmpegFilter(FilterFunction):
To use them, apply_effects_ffmpeg can take a list of effects, and apply them onto an
AudioSignal object.
"""
- def __init__(self, _filter, **filter_kwargs):
- super().__init__(_filter, **filter_kwargs)
- self.func = lambda stream: stream.filter(_filter, **filter_kwargs)
+ def __init__(self, filter_, ffmpeg_name=None, **filter_kwargs):
+ super().__init__(filter_, **filter_kwargs)
+ ffmpeg_name = self.filter if ffmpeg_name is None else ffmpeg_name
+ self.func = lambda stream: stream.filter(ffmpeg_name, **filter_kwargs)
def apply_effects_ffmpeg(audio_signal, filters, silent=False):
@@ -99,8 +106,8 @@ def apply_effects_ffmpeg(audio_signal, filters, silent=False):
tmpfiles.append(curr_tempfile)
stream = ffmpeg.input(curr_tempfile.name, **input_args)
- for _filter in filters:
- stream = _filter(stream)
+ for filter_ in filters:
+ stream = filter_(stream)
(stream
.output(out_tempfile.name)
.overwrite_output()
@@ -109,22 +116,20 @@ def apply_effects_ffmpeg(audio_signal, filters, silent=False):
augmented_signal = AudioSignal(path_to_input_file=out_tempfile.name)
augmented_signal.label = audio_signal.label
- augmented_signal.effects_applied = (audio_signal.effects_applied
- + [str(f) for f in filters])
+ augmented_signal._effects_applied = audio_signal.effects_applied + filters
return augmented_signal
-
class SoXFilter(FilterFunction):
"""
- SoXFilter is an object returned by FFmpeg effects in effects.py
+ SoXFilter is an object returned by SoX effects in effects.py
To use them, apply_effects_sox can take a list of effects, and apply them onto an
AudioSignal object.
"""
- def __init__(self, _filter, **filter_kwargs):
- super().__init__(_filter, **filter_kwargs)
- if _filter == "time_stretch":
+ def __init__(self, filter_, **filter_kwargs):
+ super().__init__(filter_, **filter_kwargs)
+ if filter_== "time_stretch":
self.func = lambda tfm: tfm.tempo(**filter_kwargs)
- elif _filter == "pitch_shift":
+ elif filter_== "pitch_shift":
self.func = lambda tfm: tfm.pitch(**filter_kwargs)
else:
raise ValueError("Unknown SoX effect passed")
@@ -147,8 +152,8 @@ def apply_effects_sox(audio_signal, filters):
audio_data = audio_signal.audio_data
tfm = sox.Transformer()
- for _filter in filters:
- tfm = _filter(tfm)
+ for filter_ in filters:
+ tfm = filter_(tfm)
augmented_data = tfm.build_array(
input_array=np.transpose(audio_data),
sample_rate_in=audio_signal.sample_rate
@@ -156,8 +161,7 @@ def apply_effects_sox(audio_signal, filters):
augmented_signal = AudioSignal(audio_data_array=np.transpose(augmented_data))
augmented_signal.label = audio_signal.label
- augmented_signal.effects_applied = (audio_signal.effects_applied
- + [str(f) for f in filters])
+ augmented_signal._effects_applied = audio_signal.effects_applied + filters
return augmented_signal
@@ -187,11 +191,11 @@ def time_stretch(factor, **kwargs):
return SoXFilter("time_stretch", factor=factor, **kwargs)
-def pitch_shift(shift, **kwargs):
+def pitch_shift(n_semitones, **kwargs):
"""
Returns a SoXFilter, when called on an pysox stream, will increase the pitch
- of the audio by a number of semitones, denoted in shift. A positive shift will
- raise the pitch of the signal.
+ of the audio by a number of semitones, denoted in n_semitones. A positive n_semitones
+ will raise the pitch of the signal.
It is recommended that users use `AudioSignal.pitch_shift` rather than this function.
@@ -200,15 +204,15 @@ def pitch_shift(shift, **kwargs):
for details.
Args:
- shift (float): The number of semitones to shift the audio.
+ n_semitones (integer): The number of semitones to shift the audio.
Positive values increases the frequency of the signal
Returns:
filter (SoxFilter): A SoXFilter object, to be called on an pysndfx stream
"""
- if not np.issubdtype(type(shift), np.integer):
+ if not np.issubdtype(type(n_semitones), np.integer):
raise ValueError("shift must be an integer.")
- return SoXFilter("pitch_shift", n_semitones=shift, **kwargs)
+ return SoXFilter("pitch_shift", n_semitones=n_semitones, **kwargs)
def _pass_arg_check(freq, poles, width_type, width):
@@ -247,8 +251,11 @@ def low_pass(freq, poles=2, width_type="h", width=.707):
"""
_pass_arg_check(freq, poles, width_type, width)
- return FFmpegFilter("lowpass", f=freq, p=poles,
+ filter_ = FFmpegFilter("low_pass", ffmpeg_name="lowpass", f=freq, p=poles,
t=width_type, w=width)
+ filter_.params = {"freq": freq, "poles": poles,
+ "width_type": width_type, "width":width}
+ return filter_
def high_pass(freq, poles=2, width_type="h", width=.707):
@@ -276,7 +283,11 @@ def high_pass(freq, poles=2, width_type="h", width=.707):
"""
_pass_arg_check(freq, poles, width_type, width)
- return FFmpegFilter("highpass", f=freq, p=poles, t=width_type, w=width)
+ filter_ = FFmpegFilter("high_pass", ffmpeg_name="highpass", f=freq, p=poles,
+ t=width_type, w=width)
+ filter_.params = {"freq": freq, "poles": poles,
+ "width_type": width_type, "width":width}
+ return filter_
def tremolo(mod_freq, mod_depth):
@@ -301,7 +312,9 @@ def tremolo(mod_freq, mod_depth):
if not np.issubdtype(type(mod_depth), np.number) or mod_depth < 0 or mod_depth > 1:
raise ValueError("mod_depth should be positve scalar between 0 and 1.")
- return FFmpegFilter("tremolo", f=mod_freq, d=mod_depth)
+ filter_ = FFmpegFilter("tremolo", f=mod_freq, d=mod_depth)
+ filter_.params = {"mod_freq": mod_freq, "mod_depth": mod_depth}
+ return filter_
def vibrato(mod_freq, mod_depth):
@@ -326,7 +339,9 @@ def vibrato(mod_freq, mod_depth):
if not np.issubdtype(type(mod_depth), np.number) or mod_depth < 0 or mod_depth > 1:
raise ValueError("mod_depth should be positve scalar between 0 and 1.")
- return FFmpegFilter("vibrato", f=mod_freq, d=mod_depth)
+ filter_ = FFmpegFilter("vibrato", f=mod_freq, d=mod_depth)
+ filter_.params = {"mod_freq": mod_freq, "mod_depth": mod_depth}
+ return filter_
def chorus(delays, decays, speeds, depths,
@@ -363,14 +378,20 @@ def chorus(delays, decays, speeds, depths,
!= len(speeds) or len(speeds) != len(depths)):
raise ValueError("Delays, decays, depths, and speeds must all be the same length.")
- delays = make_arglist_ffmpeg(delays)
- speeds = make_arglist_ffmpeg(speeds)
- decays = make_arglist_ffmpeg(decays)
- depths = make_arglist_ffmpeg(depths)
+ ffmpeg_delays = make_arglist_ffmpeg(delays)
+ ffmpeg_speeds = make_arglist_ffmpeg(speeds)
+ ffmpeg_decays = make_arglist_ffmpeg(decays)
+ ffmpeg_depths = make_arglist_ffmpeg(depths)
- return FFmpegFilter("chorus", in_gain=in_gain,
- out_gain=out_gain, delays=delays,
- speeds=speeds, decays=decays, depths=depths)
+ filter_ = FFmpegFilter("chorus", in_gain=in_gain,
+ out_gain=out_gain, delays=ffmpeg_delays,
+ speeds=ffmpeg_speeds, decays=ffmpeg_decays,
+ depths=ffmpeg_depths)
+ filter_.params["delays"] = delays
+ filter_.params["speeds"] = speeds
+ filter_.params["decays"] = decays
+ filter_.params["depths"] = depths
+ return filter_
def phaser(in_gain=.4, out_gain=.74, delay=3,
@@ -412,9 +433,14 @@ def phaser(in_gain=.4, out_gain=.74, delay=3,
"type": type_
}
- return FFmpegFilter("aphaser", in_gain=in_gain, out_gain=out_gain,
+ filter_ = FFmpegFilter("phaser", ffmpeg_name="aphaser", in_gain=in_gain, out_gain=out_gain,
delay=delay, speed=speed, decay=decay, **type_kwarg)
+ filter_.params = copy.deepcopy(filter_.params)
+ del filter_.params["type"]
+ filter_.params["type_"] = type_
+ return filter_
+
def _flanger_argcheck(delay, depth, regen, width,
speed, phase, shape, interp):
@@ -528,8 +554,13 @@ def emphasis(level_in, level_out, type_="col", mode='production'):
'type': type_
}
- return FFmpegFilter("aemphasis", level_in=level_in, level_out=level_out,
- mode=mode, **type_kwarg)
+ filter_ = FFmpegFilter("emphasis", ffmpeg_name="aemphasis", level_in=level_in,
+ level_out=level_out, mode=mode, **type_kwarg)
+
+ filter_.params = copy.deepcopy(filter_.params)
+ del filter_.params["type"]
+ filter_.params["type_"] = type_
+ return filter_
def _compressor_argcheck(level_in, mode, reduction_ratio,
@@ -611,10 +642,14 @@ def compressor(level_in, mode="downward", reduction_ratio=2,
_compressor_argcheck(level_in, mode, reduction_ratio,
attack, release, makeup, knee, link, detection, mix, threshold)
- # TODO: for some reason the mode arg doesn't work in ffmpeg. figure out why
- return FFmpegFilter("acompressor", level_in=level_in,
+ filter_ = FFmpegFilter("compressor", ffmpeg_name="acompressor", level_in=level_in,
ratio=reduction_ratio, attack=attack, release=release, makeup=makeup,
knee=knee, link=link, detection=detection, mix=mix, threshold=threshold)
+ filter_.params = copy.deepcopy(filter_.params)
+ del filter_.params["ratio"]
+ filter_.params["reduction_ratio"] = reduction_ratio
+ return filter_
+
def equalizer(bands):
@@ -662,4 +697,7 @@ def equalizer(bands):
for band in bands
])
- return FFmpegFilter("anequalizer", params=params)
+ filter_ = FFmpegFilter("equalizer", ffmpeg_name="anequalizer", params=params)
+ filter_.params = {"bands": bands}
+ return filter_
+

MODIFY
tests/core/test_effects.py
tests/core/test_effects.py
@@ -8,6 +8,8 @@ import pytest
REGRESSION_PATH = "tests/core/regression/augmentation"
os.makedirs(REGRESSION_PATH, exist_ok=True)
+# Note: When changing these tests, please do not call `mix_and_sources["mix"].apply_effects`
+# with overwrite=True
def fx_regression(audio_data, reg_path, check_against_regression_data):
scores = {
@@ -364,19 +366,19 @@ def test_hooks(mix_and_sources, check_against_regression_data):
}])
)
- assert len(signal._effects_chain) == 12
+ assert len(signal.effects_chain) == 12
augmented_signal = signal.apply_effects(reset=False, user_order=False)
- assert len(signal._effects_chain) == 12
+ assert len(signal.effects_chain) == 12
assert len(augmented_signal.effects_applied) == 12
- assert len(augmented_signal._effects_chain) == 0
+ assert len(augmented_signal.effects_chain) == 0
reg_path = path.join(REGRESSION_PATH, "hooks.json")
fx_regression(augmented_signal.audio_data, reg_path, check_against_regression_data)
augmented_signal.time_stretch(.7).apply_effects(overwrite=True, user_order=False)
- assert len(augmented_signal._effects_chain) == 0
+ assert len(augmented_signal.effects_chain) == 0
assert len(augmented_signal.effects_applied) == 25
assert str(augmented_signal.effects_applied[-1]) == "time_stretch (params: factor=0.7)"
@@ -391,7 +393,7 @@ def test_make_effect(mix_and_sources, check_against_regression_data):
(signal
.make_effect("time_stretch", factor=3)
- .make_effect("pitch_shift", shift=2)
+ .make_effect("pitch_shift", n_semitones=2)
.make_effect("low_pass", freq=512)
.make_effect("high_pass", freq=512)
.make_effect("tremolo", mod_freq=5, mod_depth=.4)
@@ -409,24 +411,39 @@ def test_make_effect(mix_and_sources, check_against_regression_data):
't': 0
}])
)
- signal.apply_effects(reset=False)
- augmented_signal = signal.apply_effects(reset=False, overwrite=True, user_order=False)
+
+ augmented_signal = signal.apply_effects(user_order=False)
reg_path = path.join(REGRESSION_PATH, "hooks.json")
# This should result in the same signal in test_hooks
fx_regression(augmented_signal.audio_data, reg_path, check_against_regression_data)
+ for effect in augmented_signal.effects_applied:
+ _filter = effect.filter
+ params = effect.params
+ signal.make_effect(_filter, **params)
+
+ augmented_signal2 = signal.apply_effects(user_order=False)
+
with pytest.raises(AudioSignalException):
signal.make_effect("fail")
+ assert np.allclose(augmented_signal.audio_data, augmented_signal2.audio_data)
def test_order(mix_and_sources):
mix, _ = mix_and_sources
mix.reset_effects_chain().time_stretch(1.5).tremolo(.3,.4).pitch_shift(5).vibrato(.3, .4)
## User order
- signal0 = mix.apply_effects(reset=False)
+ signal_user = mix.apply_effects(reset=False)
## SoX-FFmpeg Order
- signal1 = mix.apply_effects(reset=False, user_order=False)
+ signal_sf = mix.apply_effects(reset=False, user_order=False)
+
+ assert len(signal_user.effects_applied) == len(signal_sf.effects_applied)
+
+ user_order = ["time_stretch", "tremolo", "pitch_shift", "vibrato"]
+ for effect, _filter in zip(signal_user.effects_applied, user_order):
+ assert effect.filter == _filter
- assert signal0.effects_applied != signal1.effects_applied
- assert len(signal0.effects_applied) == len(signal1.effects_applied)
+ sox_ffmpeg_order = ["time_stretch", "pitch_shift", "tremolo", "vibrato"]
+ for effect, _filter in zip(signal_sf.effects_applied, sox_ffmpeg_order):
+ assert effect.filter == _filter

effect_chains and effects_applied are now lists of filter functions