Source code for compiam.utils.augment

import os
from joblib import Parallel, delayed

from compiam.utils import create_if_not_exists
from compiam.utils.augment.augment_data_ar import augment_data_attack_remix

# from compiam.utils.augment.augment_data_ps import augment_data_pitch_shift
from compiam.utils.augment.augment_data_sf import augment_data_spectral_shape
from compiam.utils.augment.augment_data_sr import augment_data_stroke_remix

# from compiam.utils.augment.augment_data_ts import augment_data_time_scale

file_dir = os.path.dirname(__file__)


# def pitch_shift(in_path, out_dir, shifts=-1.0, sr=16000, n_jobs=4):
#    """Pitch shift audio at <in_path>, by shifts in <shifts>. Output to <out_dir>
#
#    :param in_path: Path to input audio
#    :type in_path: str
#    :param out_dir: Directory to output pitch shifted audios
#    :type out_dir: str
#    :param shifts: Pitch shifts value in semitones (or list of values for multiple outputs)
#    :type shifts: float or list
#    :param sr: Sampling rate of input audio
#    :type sr: float
#    :param n_jobs: n jobs for parrelelization
#    :type n_jobs: int
#    """
#    create_if_not_exists(out_dir)
#
#    if not isinstance(shifts, list):
#        shifts = [shifts]
#
#    _ = Parallel(n_jobs=n_jobs)(
#        delayed(augment_data_pitch_shift)(in_path, out_dir, sr, ps) for ps in shifts
#    )


[docs]def spectral_shape( in_path, out_dir, gain_factors=(0.6, 2, 0.2), winDur=46.4, hopDur=5e-3, sr=16000, n_jobs=4, ): """Augmenting data by perturbin 'nuisance attributes' that are unimportant in the specific discrimination task. :param in_path: Path to input audio :type in_path: str :param out_dir: Directory to output pitch shifted audios :type out_dir: str :param gain_factors: 3-tuple (or list of) with gain factors for remixing. Tuple entries correspond to each of bass, treble, & damped components. :type gain_factors: float or list :param winDur: Window size in milliseconds :type winDur: float :param hopDur: Hop size in milliseconds :type hopDur: float :param sr: Sampling rate of input audio :type sr: float :param n_jobs: n jobs for parrelelization :type n_jobs: int """ create_if_not_exists(out_dir) if not isinstance(gain_factors, list): gain_factors = [gain_factors] _ = Parallel(n_jobs=n_jobs)( delayed(augment_data_spectral_shape)( in_path, out_dir, sr, gain_set, winDur=winDur, hopDur=hopDur ) for gain_set in gain_factors )
[docs]def stroke_remix( in_path, out_dir, gain_factors=(0.6, 2, 0.2), templates=os.path.join(file_dir, "augmentation", "templates.npy"), winDur=46.4, hopDur=5e-3, sr=16000, n_jobs=4, ): """Simulate the expected variations of relative strengths of drums in a mix using non-negative matrix factorization (NMF). :param in_path: Path to input audio :type in_path: str :param out_dir: Directory to output pitch shifted audios :type out_dir: str :param gain_factors: 3-tuple (or list of) with gain factors for remixing. Tuple entries correspond to each of bass, treble, & damped components. :type gain_factors: float or list :param templates: path to saved nmf templates :type templates: str :param winDur: Window size in milliseconds :type winDur: float :param hopDur: Hop size in milliseconds :type hopDur: float :param sr: Sampling rate of input audio :type sr: float :param n_jobs: n jobs for parrelelization :type n_jobs: int """ create_if_not_exists(out_dir) if not isinstance(gain_factors, list): gain_factors = [gain_factors] _ = Parallel(n_jobs=n_jobs)( delayed(augment_data_stroke_remix)( in_path, out_dir, sr, gain_set, templates, winDur, hopDur ) for gain_set in gain_factors )
# def time_scale(in_path, out_dir, time_shifts=0.8, sr=16000, n_jobs=4): # """Time scaling of input audio, pitch maintained. # # :param in_path: Path to input audio # :type in_path: str # :param out_dir: Directory to output pitch shifted audios # :type out_dir: str # :param time_shifts: time scale value (or list of) # :type time_shifts: float or list # :param sr: Sampling rate of input audio # :type sr: float # :param n_jobs: n jobs for parrelelization # :type n_jobs: int # """ # create_if_not_exists(out_dir) # # if not isinstance(time_shifts, list): # time_shifts = [time_shifts] # # _ = Parallel(n_jobs=n_jobs)( # delayed(augment_data_time_scale)(in_path, out_dir, sr, ts) for ts in time_shifts # )
[docs]def attack_remix( in_path, out_dir, gain_factors=0.3, winDur=46.4, hopDur=5, sr=16000, n_jobs=4 ): """Modifying the relative levels of attack and decay regions of an audio :param in_path: Path to input audio :type in_path: str :param out_dir: Directory to output pitch shifted audios :type out_dir: str :param gain_factors: gain factor (or list of) to scale attack portion with :type gain_factors: float or list :param winDur: Window size in milliseconds :type winDur: float :param hopDur: Hop size in milliseconds :type hopDur: float :param sr: Sampling rate of input audio :type sr: float :param n_jobs: n jobs for parrelelization :type n_jobs: int """ create_if_not_exists(out_dir) if not isinstance(gain_factors, list): gain_factors = [gain_factors] _ = Parallel(n_jobs=n_jobs)( delayed(augment_data_attack_remix)(in_path, out_dir, sr, G, winDur, hopDur) for G in gain_factors )