cleaned up for pep8
MODIFY
nussl/core/effects.py
nussl/core/effects.py
import numpy as np
-import numpy.random as random
-from librosa import load
-import os
-import tempfile
import warnings
import ffmpeg
import tempfile
from pysndfx import AudioEffectsChain
+from abc import abstractmethod
# import av
# from av.filter import Filter, Graph
from .constants import LEVEL_MIN, LEVEL_MAX
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 pysndfx
"""
- def __init__(self, _filter, **filter_kwargs):
- raise NotImplementedError
def __call__(self, stream):
return self.func(stream)
+ def func(self, stream):
+ pass
+
+
class FFmpegFilter(FilterFunction):
# @staticmethod
# def _dict_to_ffmpeg_kwargs(dictionary):
@@ -40,6 +40,7 @@ class FFmpegFilter(FilterFunction):
# FFmpegFilter._dict_to_ffmpeg_kwargs(filter_kwargs))
self.func = lambda stream: stream.filter(_filter, **filter_kwargs)
+
class SoXFilter(FilterFunction):
def __init__(self, _filter, **filter_kwargs):
self.filter = _filter
@@ -50,6 +51,7 @@ class SoXFilter(FilterFunction):
else:
raise ValueError
+
def build_effects_ffmpeg(audio_signal, filters, silent=False):
"""
build_effects_ffmpeg takes an AudioSignal object and a list of FFmpegFilter objects
@@ -131,6 +133,7 @@ def build_effects_ffmpeg(audio_signal, filters, silent=False):
augmented_signal = AudioSignal(path_to_input_file=curr_tempfile.name)
return augmented_signal
+
def build_effects_sox(audio_signal, filters):
"""
build_effects_sox takes an AudioSignal object and a list of SoXFilter objects
@@ -151,9 +154,11 @@ def build_effects_sox(audio_signal, filters):
return audio_signal.make_copy_with_audio_data(augmented_data)
+
def make_arglist_ffmpeg(lst, sep="|"):
return sep.join([str(s) for s in lst])
+
def time_stretch(factor):
"""
Returns a SoXFilter, when called on an pysndfx stream, will multiply the
@@ -168,6 +173,7 @@ def time_stretch(factor):
return SoXFilter("tempo", factor=factor)
+
def pitch_shift(shift):
"""
Returns a SoXFilter, when called on an pysndfx stream, will increase the pitch
@@ -183,6 +189,20 @@ def pitch_shift(shift):
return SoXFilter("pitch", shift=shift)
+def _pass_arg_check(freq, poles, width_type, width):
+ if not np.issubdtype(type(freq), np.number) or freq <= 0:
+ raise ValueError("lowest_freq should be positive scalar")
+ if poles not in {1, 2}:
+ raise ValueError("poles must be either 1 or 2")
+ if width_type not in {"h", "q", "o", "s", "k"}:
+ raise ValueError("width_type must be either h, q, o, s, or k.")
+ if not np.issubdtype(type(width), np.number) or width <= 0:
+ raise ValueError("width should be positive scalar")
+ # TODO: mix does not work for some reason, despite being listed in ffmpeg docs
+ # if mix < 0 or mix > 1:
+ # raise ValueError("mix must be between 0 and 1")
+
+
def low_pass(freq, poles=2, width_type="h", width=.707):
"""
https://ffmpeg.org/ffmpeg-all.html#lowpass
@@ -201,22 +221,12 @@ def low_pass(freq, poles=2, width_type="h", width=.707):
Returns:
filter: A FFmpegFilter object, to be called on an ffmpeg stream
"""
- if not np.issubdtype(type(freq), np.number) or freq <= 0:
- raise ValueError("lowest_freq should be positive scalar")
- if poles not in {1, 2}:
- raise ValueError("poles must be either 1 or 2")
- if width_type not in {"h", "q", "o", "s", "k"}:
- raise ValueError("width_type must be either h, q, o, s, or k.")
- if not np.issubdtype(type(width), np.number) or width <= 0:
- raise ValueError("width should be positive scalar")
- # TODO: mix does not work for some reason, despite being listed in ffmpeg docs
- # if mix < 0 or mix > 1:
- # raise ValueError("mix must be between 0 and 1")
-
+ _pass_arg_check(freq, poles, width_type, width)
return FFmpegFilter("lowpass", f=freq, p=poles,
t=width_type, w=width)
+
def high_pass(freq, poles=2, width_type="h", width=.707):
"""
https://ffmpeg.org/ffmpeg-all.html#highpass
@@ -235,22 +245,11 @@ def high_pass(freq, poles=2, width_type="h", width=.707):
Returns:
filter: A FFmpegFilter object, to be called on an ffmpeg stream
"""
- if not np.issubdtype(type(freq), np.number) or freq <= 0:
- raise ValueError("lowest_freq should be positive scalar")
- if poles not in {1, 2}:
- raise ValueError("poles must be either 1 or 2")
- if width_type not in {"h", "q", "o", "s", "k"}:
- raise ValueError("width_type must be either h, q, o, s, or k.")
- if not np.issubdtype(type(width), np.number) or width <= 0:
- raise ValueError("width should be positive scalar")
-
- # TODO: mix does not work for some reason, despite being listed in ffmpeg docs
- # if mix < 0 or mix > 1:
- # raise ValueError("mix must be between 0 and 1")
-
+ _pass_arg_check(freq, poles, width_type, width)
return FFmpegFilter("highpass", f=freq, p=poles, t=width_type, w=width)
+
def tremolo(mod_freq, mod_depth):
"""
https://ffmpeg.org/ffmpeg-all.html#tremolo
@@ -271,6 +270,7 @@ def tremolo(mod_freq, mod_depth):
return FFmpegFilter("tremolo", f=mod_freq, d=mod_depth)
+
def vibrato(mod_freq, mod_depth):
"""
https://ffmpeg.org/ffmpeg-all.html#vibrato
@@ -291,6 +291,7 @@ def vibrato(mod_freq, mod_depth):
return FFmpegFilter("vibrato", f=mod_freq, d=mod_depth)
+
def chorus(delays, decays, speeds, depths,
in_gain=.4, out_gain=.4):
"""
@@ -299,12 +300,12 @@ def chorus(delays, decays, speeds, depths,
Creates a FFmpegFilter object, when called on an ffmpeg stream,
applies a vibrato filter to the audio signal
Args:
- input_gain: Proportion of input gain
- output_gain: Proportion of output gain
delays: list of delays in ms. Typical Delay is 40ms-6ms
decays: list of decays. Must be between 0 and 1
speeds: list of speeds. Must be between 0 and 1
depths: list of depths. Must be between 0 and 1
+ in_gain: Proportion of input gain
+ out_gain: Proportion of output gain
Returns:
filter: A FFmpegFilter object, to be called on an ffmpeg stream
"""
@@ -330,6 +331,7 @@ def chorus(delays, decays, speeds, depths,
return FFmpegFilter("chorus", in_gain=in_gain,
out_gain=out_gain, delays=delays, speeds=speeds, decays=decays, depths=depths)
+
def phaser(in_gain=.4, out_gain=.74, delay=3,
decay=.4, speed=.5, _type="triangular"):
"""
@@ -338,16 +340,23 @@ def phaser(in_gain=.4, out_gain=.74, delay=3,
Creates a FFmpegFilter object, when called on an ffmpeg stream,
applies a phaser filter to the audio signal
Args:
- input_gain: Proportion of input gain. Must be between 0 and 1
- output_gain: Proportion of output gain. Must be between 0 and 1.
+ in_gain: Proportion of input gain. Must be between 0 and 1
+ out_gain: Proportion of output gain. Must be between 0 and 1.
delay: Delay of chorus filter in ms. (Time between original signal and delayed)
+ decay: Decay of copied signal. Must be between 0 and 1.
speed: Modulation speed of the delayed filter.
- _type: modulation type
+ _type: modulation type. Either Triangular or Sinusoidal
+ "triangular" or "t" for Triangular
+ "sinusoidal" of "s" for sinusoidal
Returns:
filter: A FFmpegFilter object, to be called on an ffmpeg stream
"""
- if (in_gain > 1 or in_gain < 0 or out_gain > 1 or out_gain < 0):
- raise ValueError("in_gain and out_gain must be between 0 and 1.")
+ if in_gain > 1 or in_gain < 0:
+ raise ValueError("in_gain must be between 0 and 1.")
+ if out_gain > 1 or out_gain < 1:
+ raise ValueError("out_gain must be between 0 and 1.")
+ if delay > 1 or delay < 1:
+ raise ValueError("delay must be between 0 and 1.")
allowed_mod_types = {"triangular", "sinusoidal", "t", "s"}
if _type not in allowed_mod_types:
@@ -364,7 +373,6 @@ def phaser(in_gain=.4, out_gain=.74, delay=3,
def _flanger_argcheck(delay, depth, regen, width,
speed, phase, shape, interp):
-
allowed_shape_types = {"triangular", "sinusoidal"}
if shape not in allowed_shape_types:
raise ValueError(f"shape must be one of the following:\n{allowed_shape_types}")
@@ -390,6 +398,7 @@ def _flanger_argcheck(delay, depth, regen, width,
".1 < speed < 10\n"
)
+
def flanger(delay=0, depth=2, regen=0, width=71,
speed=.5, phase=25, shape="sinusoidal", interp="linear"):
"""
@@ -419,6 +428,7 @@ def flanger(delay=0, depth=2, regen=0, width=71,
depth=depth, regen=regen, width=width, speed=speed, phase=phase, shape=shape,
interp=interp)
+
def emphasis(level_in, level_out, _type="col", mode='production'):
"""
https://ffmpeg.org/ffmpeg-all.html#aemphasis
@@ -453,10 +463,10 @@ def emphasis(level_in, level_out, _type="col", mode='production'):
return FFmpegFilter("aemphasis", level_in=level_in, level_out=level_out, mode=mode, **type_kwarg)
+
def _compressor_argcheck(level_in, mode, reduction_ratio,
attack, release, makeup, knee, link,
detection, mix, threshold):
-
# The following values are taken from ffmpeg documentation
if (level_in < LEVEL_MIN or level_in > LEVEL_MAX or
mode not in {"upward", "downward"} or
@@ -493,6 +503,7 @@ def _compressor_argcheck(level_in, mode, reduction_ratio,
" .000097563 < threshold < 1\n"
)
+
def compressor(level_in, mode="downward", reduction_ratio=2,
attack=20, release=250, makeup=1, knee=2.8284, link="average",
detection="rms", mix=1, threshold=.125):
@@ -538,7 +549,6 @@ def equalizer(bands):
Creates a FFmpegFilter object, when called on an ffmpeg stream,
applies a equalizer filter to the audio signal.
Args:
- audio_signal: An AudioSignal object
bands: A list of dictionaries, for each band. The required values for each dictionary:
'chn': List of channel numbers to apply filter. Must be list of ints.
'f': central freqency of band

MODIFY
nussl/datasets/base_dataset.py
nussl/datasets/base_dataset.py
@@ -327,7 +327,6 @@ class BaseDataset(Dataset):
else:
self.num_channels = audio_signal.num_channels
-
class DataSetException(Exception):
"""
Exception class for errors when working with data sets in nussl.

MODIFY
tests/core/test_effects.py
tests/core/test_effects.py
@@ -271,6 +271,10 @@ def test_misc_param_check():
with pytest.raises(ValueError):
effects.phaser(in_gain=-1)
+ with pytest.raises(ValueError):
+ effects.phaser(out_gain=-1)
+ with pytest.raises(ValueError):
+ effects.phaser(decay=-1)
with pytest.raises(ValueError):
effects.phaser(_type="fail")

cleaned up for pep8