Fixing stft_data bugs with convert_to_time_domain in duet.
MODIFY
nussl/audio_signal.py
nussl/audio_signal.py
@@ -75,6 +75,7 @@ class AudioSignal(object):
self.path_to_input_file = path_to_input_file
self._audio_data = None
+ self._stft_data = None
self.sample_rate = sample_rate
self._active_start = None
self._active_end = None
@@ -95,7 +96,9 @@ class AudioSignal(object):
self.load_audio_from_array(audio_data_array, sample_rate)
# stft data
+ if stft is not None:
self.stft_data = stft # complex spectrogram data
+
self.stft_params = spectral_utils.StftParams(self.sample_rate) if stft_params is None else stft_params
self.use_librosa_stft = config.USE_LIBROSA_STFT
@@ -220,6 +223,31 @@ class AudioSignal(object):
self.set_active_region_to_default()
+ @property
+ def stft_data(self):
+ """ (:obj:`np.ndarray`): Real-valued, uncompressed, time-domain representation of the audio.
+ 2D numpy array with shape `(n_channels, n_samples)`.
+ ``None`` by default, this can be initialized at instantiation.
+ Usually, this is expected to be floats. Some functions will convert to floats if not already.
+ """
+
+ return self._stft_data
+
+ @stft_data.setter
+ def stft_data(self, value):
+ assert isinstance(value, np.ndarray), 'Type of self.stft_data must be np.ndarray!'
+
+ if value.ndim == 1:
+ raise ValueError('Cannot support arrays with less than 2 dimensions!')
+
+ if value.ndim == 2:
+ value = np.expand_dims(value, axis=constants.STFT_CHAN_INDEX)
+
+ if value.ndim > 3:
+ raise ValueError('Cannot support arrays with more than 3 dimensions!')
+
+ self._stft_data = value
+
@property
def file_name(self):
""" (str): The name of the file wth extension, NOT the full path.
@@ -1182,7 +1210,7 @@ class AudioSignal(object):
# np.array helps with duck typing
return utils._get_axis(np.array(self.magnitude_spectrogram_data), constants.STFT_CHAN_INDEX, n)
- def to_mono(self, overwrite=False):
+ def to_mono(self, overwrite=False, remove_channels=True):
""" Converts :attr:`audio_data` to mono by averaging every sample.
Warning:
@@ -1195,11 +1223,32 @@ class AudioSignal(object):
(:obj:`np.array`): Mono-ed version of :attr:`audio_data`.
"""
+ if remove_channels:
mono = np.mean(self.audio_data, axis=constants.CHAN_INDEX)
+ else:
+ mono = np.mean(self.audio_data, axis=constants.CHAN_INDEX, keepdims=True)
if overwrite:
self.audio_data = mono
return mono
+ def stft_to_mono(self, overwrite=False):
+ """ Converts :attr:`stft_data` to mono by averaging every sample.
+
+ Warning:
+ If overwrite=True (default) this will overwrite any data in :attr:`audio_data`!
+
+ Args:
+ overwrite (bool, optional): If ``True`` this function will overwrite :attr:`stft_data`.
+
+ Returns:
+ (:obj:`np.array`): Mono-ed version of :attr:`stft_data`.
+
+ """
+ mono_stft = np.mean(self.stft_data, axis=constants.CHAN_INDEX)
+ if overwrite:
+ self.stft_data = mono_stft
+ return mono_stft
+
##################################################
# Operator overloading
##################################################

MODIFY
nussl/separation/duet.py
nussl/separation/duet.py
@@ -14,7 +14,6 @@ import nussl.spectral_utils
import nussl.utils
import mask_separation_base
import masks
-from ideal_mask import IdealMask
class Duet(mask_separation_base.MaskSeparationBase):
@@ -46,22 +45,23 @@ class Duet(mask_separation_base.MaskSeparationBase):
p (int): Weight the histogram with the symmetric attenuation estimator
q (int): Weight the histogram with the delay estimator
- Attributes:
- stft_ch0 (np.array):
- stft_ch1 (np.array):
- frequency_matrix (np.array):
- symmetric_atn = None
- delay = None
- num_time_bins = None
- num_frequency_bins = None
- attenuation_bins = None
- delay_bins = None
- normalized_attenuation_delay_histogram (np.array):
- attenuation_delay_histogram (np.array):
- peak_indices (np.array):
- separated_sources (np.array):
-
+ Attributes:
+ stft_ch0 (np.array): A Numpy matrix containing the stft data of channel 0.
+ stft_ch1 (np.array): A Numpy matrix containing the stft data of channel 0.
+ frequency_matrix (np.array): A Numpy matrix containing the frequencies of analysis.
+ symmetric_atn (np.array): A Numpy matrix containing the symmetric attenuation between the two channels.
+ delay (np.array): A Numpy matrix containing the delay between the two channels.
+ num_time_bins (np.array): The number of time bins for the frequency matrix and mask arrays.
+ num_frequency_bins (int): The number of frequency bins for the mask arrays.
+ attenuation_bins (int): A Numpy array containing the attenuation bins for the histogram.
+ delay_bins (np.array): A Numpy array containing the delay bins for the histogram.
+ normalized_attenuation_delay_histogram (np.array): A normalized Numpy matrix containing the
+ attenuation delay histogram, which has peaks for each source.
+ attenuation_delay_histogram (np.array): A non-normalized Numpy matrix containing the
+ attenuation delay histogram, which has peaks for each source.
+ peak_indices (np.array): A Numpy array containing the indices of the peaks for the histogram.
+ separated_sources (np.array): A Numpy array of arrays containing each separated source.
Examples:
:ref:`The DUET Demo Example <duet_demo>`
@@ -157,7 +157,6 @@ class Duet(mask_separation_base.MaskSeparationBase):
self.normalized_attenuation_delay_histogram, self.attenuation_bins, self.delay_bins = self._make_histogram()
# Find the location of peaks in the attenuation-delay plane
-
self.peak_indices = nussl.utils.find_peak_indices(self.normalized_attenuation_delay_histogram, self.num_sources,
threshold=self.peak_threshold,
min_dist=[self.attenuation_min_distance,
@@ -167,9 +166,7 @@ class Duet(mask_separation_base.MaskSeparationBase):
delay_peak, atn_delay_est, atn_peak = self._convert_peaks()
# compute masks for separation
- self._compute_masks(atn_peak, delay_peak)
-
-
+ self._compute_masks(delay_peak, atn_peak)
# demix with ML alignment and convert to time domain
source_estimates = self._convert_to_time_domain(atn_peak, delay_peak)
@@ -209,7 +206,8 @@ class Duet(mask_separation_base.MaskSeparationBase):
wmat += nussl.constants.EPSILON
return stft_ch0, stft_ch1, wmat
- def _compute_atn_delay(self, stft_ch0, stft_ch1, frequency_matrix):
+ @staticmethod
+ def _compute_atn_delay(stft_ch0, stft_ch1, frequency_matrix):
# Calculate the symmetric attenuation (alpha) and delay (delta) for each
# time-freq. point
inter_channel_ratio = (stft_ch1 + nussl.constants.EPSILON) / (stft_ch0 + nussl.constants.EPSILON)
@@ -254,8 +252,7 @@ class Duet(mask_separation_base.MaskSeparationBase):
histogram, atn_bins, delay_bins = np.histogram2d(symmetric_attenuation_vector, delay_vector,
bins=np.array([self.num_attenuation_bins, self.num_delay_bins]),
range=np.array([[self.attenuation_min, self.attenuation_max],
- [self.delay_min, self.delay_max]]), normed=False,
- weights=time_frequency_weights_vector)
+ [self.delay_min, self.delay_max]]), weights=time_frequency_weights_vector)
#Save non-normalized as an option for plotting later
self.attenuation_delay_histogram = histogram
@@ -295,46 +292,40 @@ class Duet(mask_separation_base.MaskSeparationBase):
atn_peak = (symmetric_atn_peak + np.sqrt(symmetric_atn_peak ** 2 + 4)) / 2
return delay_peak, atn_delay_est, atn_peak
- def _compute_masks(self, atn_peak, delay_peak):
+ def _compute_masks(self, delay_peak, atn_peak):
"""Receives the attenuation and delay peaks and computes a mask to be applied to the signal for source
separation.
Parameters:
atn_peak (np.array): Attenuation peaks determined from histogram
delay_peak (np.array): Delay peaks determined from histogram
- Returns:
- best_ind(np.array): The indices for where the masks are applied
- mask (np.array): A binary mask used to separate the sources
"""
# compute masks for separation
best_so_far = np.inf * np.ones((self.num_frequency_bins, self.num_time_bins))
-
for i in range(0, self.num_sources):
- mask_array = np.zeros((self.num_frequency_bins, self.num_time_bins), dtype=int)
- phase = np.exp(-1j * self.frequency_matrix * delay_peak[i])
- score = np.abs(atn_peak[i] * phase * self.stft_ch0 - self.stft_ch1) ** 2 / (1 + atn_peak[i] ** 2)
+ mask_array = np.zeros((self.num_frequency_bins, self.num_time_bins), dtype=bool)
+ score = self._compute_mask_score(delay_peak[i], atn_peak[i])
mask = (score < best_so_far)
- #First pass will be all zeros
- if i == 0:
- mask_array[mask] = 0.0
- else:
- mask_array[mask] = 1.0
+ mask_array[mask] = True
background_mask = masks.BinaryMask(np.array(mask_array))
- best_so_far[mask] = score[mask]
self.masks.append(background_mask)
+ self.masks[0].mask = np.logical_or(self.masks[i].mask, self.masks[0].mask)
+ best_so_far[mask] = score[mask]
+
#Compute first mask based on what the other masks left remaining
- for i in range(1, self.num_sources):
- self.masks[0].mask += self.masks[i].mask
- self.masks[0].mask = abs(self.masks[0].mask - 1)
+ self.masks[0].mask = np.logical_not(self.masks[0].mask)
+
+ def _compute_mask_score(self, delay_peak_val, atn_peak_val):
+ phase = np.exp(-1j * self.frequency_matrix * delay_peak_val)
+ score = np.abs(atn_peak_val * phase * self.stft_ch0 - self.stft_ch1) ** 2 / (1 + atn_peak_val ** 2)
+ return score
def _convert_to_time_domain(self, atn_peak, delay_peak):
"""Receives the attenuation and delay peaks, the mask and best indices and
applies the mask to separate the sources
Parameters:
- best_ind (np.array): The indices for where the masks are applied
- mask (np.array): binary mask used to separate the sources
atn_peak (np.array): Attenuation peaks determined from histogram
delay_peak (np.array): Delay peaks determined from histogram
@@ -343,22 +334,14 @@ class Duet(mask_separation_base.MaskSeparationBase):
"""
source_estimates = np.zeros((self.num_sources, self.audio_signal.signal_length))
- for i in range(0, self.num_sources):
- mask_apply = self.masks[i]
- mask_apply = np.squeeze(mask_apply.mask, axis=2)
+ for i in range(self.num_sources):
#Apply masks to stft channels using equation provided by Rickard
stft_sources = ((self.stft_ch0 + atn_peak[i] * np.exp(1j * self.frequency_matrix * delay_peak[i])
- * self.stft_ch1) / (1 + atn_peak[i] ** 2))*mask_apply
-
- # istft_sources = spectral_utils.f_istft(stft_sources, L, winType, hop, fs)
- istft_sources = nussl.spectral_utils.e_istft(stft_sources, self.stft_params.window_length,
- self.stft_params.hop_length, self.stft_params.window_type)
-
- source_estimates[i, :] = nussl.utils.add_mismatched_arrays(source_estimates[i, ],
- istft_sources)[:self.audio_signal.signal_length]
- # add back to the separated signal a portion of the mixture to eliminate
- # most of the masking artifacts
- # source_estimates=source_estimates+0.05*x[0,:]
+ * self.stft_ch1) / (1 + atn_peak[i] ** 2))
+ new_sig = self.audio_signal.make_copy_with_stft_data(stft_sources, verbose=False)
+ new_sig = new_sig.apply_mask(self.masks[i]) # Ask Ethan if we can have a apply_mask that doesn't return a new mask!
+ new_sig.stft_params = self.stft_params
+ source_estimates[i, :] = new_sig.istft(overwrite=True)
return source_estimates
@staticmethod

MODIFY
nussl/separation/masks/mask_base.py
nussl/separation/masks/mask_base.py
@@ -224,3 +224,9 @@ class MaskBase(object):
def __itruediv__(self, value):
return self.__idiv__(value)
+
+ def __eq__(self, other):
+ return np.all(self.mask == other.mask)
+
+ def __ne__(self, other):
+ return not self.__eq__(other)

MODIFY
tests/test_duet.py
tests/test_duet.py
@@ -12,9 +12,12 @@ import warnings
class DuetUnitTests(unittest.TestCase):
+
+ # Update this if the benchmark file changes and rerun freeze_duet_values() (below)
+ path_to_benchmark_file = os.path.join('..', 'Input', 'dev1_female3_inst_mix.wav')
+
def setUp(self):
- path = os.path.join('..', 'Input', 'dev1_female3_inst_mix.wav')
- self.signal = nussl.AudioSignal(path)
+ self.signal = nussl.AudioSignal(self.path_to_benchmark_file)
# Call benchmarks\
self.benchmark_dict = self.load_benchmarks()
@@ -52,7 +55,7 @@ class DuetUnitTests(unittest.TestCase):
def test_compute_spectrogram_wmat(self):
# Load Duet values to benchmark against
duet = nussl.Duet(self.signal, 3)
- duet_sft0, duet_sft1, duet_wmat = duet._compute_spectrogram(self.duet.sample_rate)
+ duet_sft0, duet_sft1, duet_wmat = duet._compute_spectrogram(duet.sample_rate)
assert np.all(np.array(self.benchmark_dict['benchmark_stft_ch0']) == duet_sft0)
assert np.all(np.array(self.benchmark_dict['benchmark_stft_ch1']) == duet_sft1)
assert np.all(np.array(self.benchmark_dict['benchmark_wmat']) == duet_wmat)
@@ -197,7 +200,31 @@ class DuetUnitTests(unittest.TestCase):
# TODO: fix this function before writing test for it
+def freeze_duet_values():
+ path = DuetUnitTests.path_to_benchmark_file
+
+ signal = nussl.AudioSignal(path)
+ duet = nussl.Duet(signal, 3)
+
+ duet.stft_ch0, duet.stft_ch1, duet.frequency_matrix = duet._compute_spectrogram(duet.sample_rate)
+
+ duet.symmetric_atn, duet.delay = duet._compute_atn_delay(duet.stft_ch0, duet.stft_ch1, duet.frequency_matrix)
+ duet.normalized_attenuation_delay_histogram, duet.attenuation_bins, duet.delay_bins = duet._make_histogram()
+
+ duet.peak_indices = nussl.utils.find_peak_indices(duet.normalized_attenuation_delay_histogram, duet.num_sources,
+ threshold=duet.peak_threshold,
+ min_dist=[duet.attenuation_min_distance,
+ duet.delay_min_distance])
+
+ delay_peak, atn_delay_est, atn_peak = duet._convert_peaks()
+
+ duet._compute_masks(delay_peak, atn_peak)
+
+ source_estimates = duet._convert_to_time_domain(atn_peak, delay_peak)
+
+ duet.separated_sources = source_estimates
if __name__ == '__main__':
unittest.main()
+

Fixing stft_data bugs with convert_to_time_domain in duet.