Moved convert_to_time_domain under make_audio_signals and modified tests to include freeze_benchmarks functions.
MODIFY
nussl/separation/duet.py
nussl/separation/duet.py
import numpy as np
import matplotlib.pyplot as plt
-
-# noinspection PyUnusedImport
+from mpl_toolkits.mplot3d import axes3d
from scipy import signal
import nussl.audio_signal
@@ -105,6 +104,8 @@ class Duet(mask_separation_base.MaskSeparationBase):
self.normalized_attenuation_delay_histogram = None
self.attenuation_delay_histogram = None
self.peak_indices = None
+ self.delay_peak = None
+ self.atn_peak = None
self.separated_sources = None
def run(self):
@@ -162,17 +163,12 @@ class Duet(mask_separation_base.MaskSeparationBase):
self.delay_min_distance])
# compute delay_peak, attenuation peak, and attenuation/delay estimates
- delay_peak, atn_delay_est, atn_peak = self._convert_peaks()
+ self.delay_peak, atn_delay_est, self.atn_peak = self._convert_peaks()
# compute masks for separation
- 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)
+ masks = self._compute_masks()
- self.separated_sources = source_estimates
- # self._apply_masks()
- return source_estimates, atn_delay_est
+ return masks
def _compute_spectrogram(self, sample_rate):
""" Creates the stfts matrices for channel 0 and 1, and computes the frequency matrix.
@@ -191,17 +187,14 @@ class Duet(mask_separation_base.MaskSeparationBase):
stft_ch0 = self.audio_signal.get_stft_channel(0)
stft_ch1 = self.audio_signal.get_stft_channel(1)
- frequency_vector = self.audio_signal.freq_vector
- time_vector = self.audio_signal.time_bins_vector
# remove dc component to avoid dividing by zero freq. in the delay estimation
stft_ch0 += nussl.constants.EPSILON
stft_ch1 += nussl.constants.EPSILON
- self.num_frequency_bins = len(frequency_vector)
- self.num_time_bins = len(time_vector)
# Compute the freq. matrix for later use in phase calculations
- wmat = np.array(np.tile(np.mat(frequency_vector).T, (1, self.num_time_bins))) * (2 * np.pi / sample_rate)
+ n_time_bins = len(self.audio_signal.time_bins_vector)
+ wmat = np.array(np.tile(np.mat(self.audio_signal.freq_vector).T, (1, n_time_bins))) * (2 * np.pi / sample_rate)
wmat += nussl.constants.EPSILON
return stft_ch0, stft_ch1, wmat
@@ -261,7 +254,6 @@ class Duet(mask_separation_base.MaskSeparationBase):
# smooth the normalized histogram - local average 3-by-3 neighboring bins
histogram = self._smooth_matrix(histogram, np.array([3]))
-
return histogram, atn_bins, delay_bins
def _convert_peaks(self):
@@ -291,7 +283,7 @@ 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, delay_peak, atn_peak):
+ def _compute_masks(self):
"""Receives the attenuation and delay peaks and computes a mask to be applied to the signal for source
separation.
@@ -301,10 +293,12 @@ class Duet(mask_separation_base.MaskSeparationBase):
"""
# compute masks for separation
- best_so_far = np.inf * np.ones((self.num_frequency_bins, self.num_time_bins))
+ best_so_far = np.inf * np.ones_like(self.stft_ch0, dtype=float)
+
for i in range(0, self.num_sources):
- 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_array = np.zeros_like(self.stft_ch0, dtype=bool)
+ phase = np.exp(-1j * self.frequency_matrix * self.delay_peak[i])
+ score = np.abs(self.atn_peak[i] * phase * self.stft_ch0 - self.stft_ch1) ** 2 / (1 + self.atn_peak[i] ** 2)
mask = (score < best_so_far)
mask_array[mask] = True
background_mask = masks.BinaryMask(np.array(mask_array))
@@ -314,34 +308,7 @@ class Duet(mask_separation_base.MaskSeparationBase):
#Compute first mask based on what the other masks left remaining
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:
- atn_peak (np.array): Attenuation peaks determined from histogram
- delay_peak (np.array): Delay peaks determined from histogram
-
- Returns:
- source_estimates (np.array): The arrays of the separated source signals
-
- """
- source_estimates = np.zeros((self.num_sources, self.audio_signal.signal_length))
- 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))
- 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, truncate_to_length=self.audio_signal.signal_length)
- return source_estimates
+ return self.masks
@staticmethod
def _smooth_matrix(matrix, kernel):
@@ -417,7 +384,15 @@ class Duet(mask_separation_base.MaskSeparationBase):
"""
signals = []
for i in range(self.num_sources):
- cur_signal = nussl.audio_signal.AudioSignal(audio_data_array=self.separated_sources[i],
+ # Apply masks to stft channels using equation provided by Rickard
+ stft_sources = ((self.stft_ch0 + self.atn_peak[i] * np.exp(1j * self.frequency_matrix * self.delay_peak[i])
+ * self.stft_ch1) / (1 + self.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_estimate = new_sig.istft(overwrite=True, truncate_to_length=self.audio_signal.signal_length)
+ cur_signal = nussl.audio_signal.AudioSignal(audio_data_array=source_estimate,
sample_rate=self.sample_rate)
signals.append(cur_signal)
return signals

DELETE
tests/duet_reference/duet_benchmarks/benchmark_best_ind.npy.REMOVED.git-id
-f795b9929c6ddab5130048eb54ce93feda9b212d
\ No newline at end of file

MODIFY
tests/duet_reference/duet_benchmarks/benchmark_delay.npy.REMOVED.git-id
tests/duet_reference/duet_benchmarks/benchmark_delay.npy.REMOVED.git-id
-fb06e141efcbf878f10e2e87e0c5dd3f7371b47d
\ No newline at end of file
+cb65959e4786d049fdd174d176be42c1d0f37262
\ No newline at end of file

ADD
tests/duet_reference/duet_benchmarks/benchmark_final_signals.npy.REMOVED.git-id
+9e0eae6d9663764f53b167799acec8f193f136bd
\ No newline at end of file

ADD
tests/duet_reference/duet_benchmarks/benchmark_masks.npy.REMOVED.git-id
+ee842514db30c95b9d6513fd26924c98683cd85f
\ No newline at end of file

DELETE
tests/duet_reference/duet_benchmarks/benchmark_source_estimates.npy.REMOVED.git-id
-85f8d9e53a5f8a2b8d5261a2cbd32c3915580de2
\ No newline at end of file

MODIFY
tests/duet_reference/duet_benchmarks/benchmark_stft_ch0.npy.REMOVED.git-id
tests/duet_reference/duet_benchmarks/benchmark_stft_ch0.npy.REMOVED.git-id
-6d0d15e720bf359d444fd0a3e4b2a10c312ab7cd
\ No newline at end of file
+f003de722b0b05fa5f3bd1c4e2a151c633330fda
\ No newline at end of file

MODIFY
tests/duet_reference/duet_benchmarks/benchmark_stft_ch1.npy.REMOVED.git-id
tests/duet_reference/duet_benchmarks/benchmark_stft_ch1.npy.REMOVED.git-id
-224318577eceffd14466c12dfe4121ceca3ca218
\ No newline at end of file
+1c194e8503fed55edb6753778e632e04abe28f4e
\ No newline at end of file

MODIFY
tests/duet_reference/duet_benchmarks/benchmark_sym_atn.npy.REMOVED.git-id
tests/duet_reference/duet_benchmarks/benchmark_sym_atn.npy.REMOVED.git-id
-e3ee0d75ab4abed4ba9d9a37c6275bd95711f837
\ No newline at end of file
+d8554310a7f3859c86e86cc8c62a1db5e4b4e099
\ No newline at end of file

MODIFY
tests/duet_reference/duet_benchmarks/benchmark_wmat.npy.REMOVED.git-id
tests/duet_reference/duet_benchmarks/benchmark_wmat.npy.REMOVED.git-id
-f8fb481cc282f9c762c8fa40a2514d9c7982e403
\ No newline at end of file
+acadabd7de1cf7149f31795774c472ed57f4ab8a
\ No newline at end of file

MODIFY
tests/test_duet.py
tests/test_duet.py
@@ -18,7 +18,7 @@ class DuetUnitTests(unittest.TestCase):
def setUp(self):
self.signal = nussl.AudioSignal(self.path_to_benchmark_file)
- # Call benchmarks\
+ # Call benchmarks
self.benchmark_dict = self.load_benchmarks()
def load_benchmarks(self):
@@ -33,15 +33,14 @@ class DuetUnitTests(unittest.TestCase):
def test_duet_final_outputs(self):
#Test final outputs
- source_estimates_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_source_estimates.npy')
- atn_delay_est_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_atn_delay_est.npy')
- benchmark_source_estimates = np.load(source_estimates_path)
- benchmark_atn_delay_est = np.load(atn_delay_est_path)
+ mask_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_masks.npy')
+ benchmark_mask = np.load(mask_path)
+
+ duet = nussl.Duet(self.signal, 3)
+ duet_masks = duet.run()
+ for i in range(len(duet_masks)):
+ assert np.all(benchmark_mask[i].mask == duet_masks[i].mask)
- duet = nussl.Duet(signal, 3)
- duet_source_estimates, duet_atn_delay_est = duet.run()
- assert np.all(benchmark_source_estimates == duet_source_estimates)
- assert np.all(benchmark_atn_delay_est == duet_atn_delay_est)
def test_compute_spectrogram_1_channel(self):
# Test with one channel, should throw value error
@@ -67,7 +66,7 @@ class DuetUnitTests(unittest.TestCase):
duet.stft_ch1 = self.benchmark_dict['benchmark_stft_ch1']
duet.frequency_matrix = self.benchmark_dict['benchmark_wmat']
- symmetric_atn, delay = duet._compute_atn_delay()
+ symmetric_atn, delay = duet._compute_atn_delay(duet.stft_ch0, duet.stft_ch1, duet.frequency_matrix)
assert np.all(np.array(self.benchmark_dict['benchmark_sym_atn']) == symmetric_atn)
assert np.all(np.array(self.benchmark_dict['benchmark_delay']) == delay)
@@ -89,7 +88,7 @@ class DuetUnitTests(unittest.TestCase):
benchmark_atn_bins = np.load(atn_bins_path)
benchmark_delay_bins = np.load(delay_bins_path)
- hist, atn_bins, delay_bins = duet.make_histogram(duet.p, duet.q)
+ hist, atn_bins, delay_bins = duet._make_histogram()
assert np.all(benchmark_hist == hist)
assert np.all(benchmark_atn_bins == atn_bins)
@@ -108,16 +107,10 @@ class DuetUnitTests(unittest.TestCase):
benchmark_hist = np.load(hist_path)
benchmark_peak_indices = np.load(peak_indices_path)
- duet_peak_indices = nussl.utils.find_peak_indices(benchmark_hist, self.duet.num_sources,
- threshold=self.duet.peak_threshold,
- min_dist=[self.duet.attenuation_min_distance,
- self.duet.delay_min_distance])
-
- # original DUET returned peaks as [[28 21 44] [24 25 24]] instead of [(28, 24), (21,25), (44,24)]
- # Convert current peaks being tested to original format
- atn_indices = [x[0] for x in duet_peak_indices]
- delay_indices = [x[1] for x in duet_peak_indices]
- duet_peak_indices = np.row_stack((atn_indices, delay_indices))
+ duet_peak_indices = nussl.utils.find_peak_indices(benchmark_hist, duet.num_sources,
+ threshold=duet.peak_threshold,
+ min_dist=[duet.attenuation_min_distance,
+ duet.delay_min_distance])
assert np.all(benchmark_peak_indices == duet_peak_indices)
@@ -128,7 +121,7 @@ class DuetUnitTests(unittest.TestCase):
duet.frequency_matrix = self.benchmark_dict['benchmark_wmat']
duet.symmetric_atn = self.benchmark_dict['benchmark_sym_atn']
duet.delay = self.benchmark_dict['benchmark_delay']
- duet.atn_bins = self.benchmark_dict['benchmark_atn_bins']
+ duet.attenuation_bins = self.benchmark_dict['benchmark_atn_bins']
duet.delay_bins = self.benchmark_dict['benchmark_delay_bins']
duet.peak_indices = self.benchmark_dict['benchmark_peak_indices']
@@ -139,8 +132,7 @@ class DuetUnitTests(unittest.TestCase):
benchmark_atn_peak = np.load(atn_peak_path)
benchmark_atn_delay_est = np.load(atn_delay_est_path)
- duet.peak_indices = zip(duet.peak_indices[0], duet.peak_indices[1])
- delay_peak, atn_delay_est, atn_peak = duet.convert_peaks(duet.atn_bins, duet.delay_bins)
+ delay_peak, atn_delay_est, atn_peak = duet._convert_peaks()
assert np.all(benchmark_delay_peak == delay_peak)
assert np.all(benchmark_atn_delay_est == atn_delay_est)
@@ -153,26 +145,20 @@ class DuetUnitTests(unittest.TestCase):
duet.frequency_matrix = self.benchmark_dict['benchmark_wmat']
duet.symmetric_atn = self.benchmark_dict['benchmark_sym_atn']
duet.delay = self.benchmark_dict['benchmark_delay']
- duet.atn_bins = self.benchmark_dict['benchmark_atn_bins']
+ duet.attenuation_bins = self.benchmark_dict['benchmark_atn_bins']
duet.delay_bins = self.benchmark_dict['benchmark_delay_bins']
duet.peak_indices = self.benchmark_dict['benchmark_peak_indices']
duet.delay_peak = self.benchmark_dict['benchmark_delay_peak']
duet.atn_peak = self.benchmark_dict['benchmark_atn_peak']
- duet.num_frequency_bins = self.benchmark_dict['benchmark_frequency_bins']
- duet.num_time_bins = self.benchmark_dict['benchmark_num_time_bins']
-
- best_ind_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_best_ind.npy')
- mask_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_mask.npy')
- benchmark_best_ind = np.load(best_ind_path)
+ mask_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_masks.npy')
benchmark_mask = np.load(mask_path)
- best_ind, mask = duet.compute_masks(duet.atn_peak, duet.delay_peak)
+ masks = duet._compute_masks()
+ for i in range(len(masks)):
+ assert np.all(benchmark_mask[i].mask == masks[i].mask)
- assert np.all(best_ind == benchmark_best_ind)
- assert np.all(benchmark_mask == mask)
-
- def test_convert_time_domain(self):
+ def test_make_audio_signals(self):
duet = nussl.Duet(self.signal, 3)
duet.stft_ch0 = self.benchmark_dict['benchmark_stft_ch0']
duet.stft_ch1 = self.benchmark_dict['benchmark_stft_ch1']
@@ -184,17 +170,14 @@ class DuetUnitTests(unittest.TestCase):
duet.peak_indices = self.benchmark_dict['benchmark_peak_indices']
duet.delay_peak = self.benchmark_dict['benchmark_delay_peak']
duet.atn_peak = self.benchmark_dict['benchmark_atn_peak']
- duet.num_frequency_bins = self.benchmark_dict['benchmark_frequency_bins']
- duet.num_time_bins = self.benchmark_dict['benchmark_num_time_bins']
- duet.best_ind = self.benchmark_dict['benchmark_best_ind']
- duet.mask = self.benchmark_dict['benchmark_mask']
+ duet.masks = self.benchmark_dict['benchmark_masks']
- source_estimates_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_source_estimates.npy')
- benchmark_source_estimates = np.load(source_estimates_path)
+ final_signals_path = os.path.join('duet_reference', 'duet_benchmarks', 'benchmark_final_signals.npy')
+ benchmark_final_signals = np.load(final_signals_path)
- source_estimates = duet.convert_to_time_domain(duet.best_ind, duet.mask, duet.atn_peak, duet.delay_peak)
+ final_signals = duet.make_audio_signals()
- assert np.all(benchmark_source_estimates == source_estimates)
+ assert np.all(benchmark_final_signals == final_signals)
# test_smooth_matrix
@@ -205,26 +188,41 @@ def freeze_duet_values():
signal = nussl.AudioSignal(path)
duet = nussl.Duet(signal, 3)
+ output_folder = os.path.abspath('duet_reference/duet_benchmarks')
duet.stft_ch0, duet.stft_ch1, duet.frequency_matrix = duet._compute_spectrogram(duet.sample_rate)
+ np.save(os.path.join(output_folder, "benchmark_stft_ch0"), duet.stft_ch0)
+ np.save(os.path.join(output_folder, "benchmark_stft_ch1"), duet.stft_ch1)
+ np.save(os.path.join(output_folder, "benchmark_wmat"), duet.frequency_matrix)
duet.symmetric_atn, duet.delay = duet._compute_atn_delay(duet.stft_ch0, duet.stft_ch1, duet.frequency_matrix)
+ np.save(os.path.join(output_folder, "benchmark_sym_atn"), duet.symmetric_atn)
+ np.save(os.path.join(output_folder, "benchmark_delay"), duet.delay)
duet.normalized_attenuation_delay_histogram, duet.attenuation_bins, duet.delay_bins = duet._make_histogram()
+ np.save(os.path.join(output_folder, "benchmark_hist"), duet.normalized_attenuation_delay_histogram)
+ np.save(os.path.join(output_folder, "benchmark_atn_bins"), duet.attenuation_bins)
+ np.save(os.path.join(output_folder, "benchmark_delay_bins"), duet.delay_bins)
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])
+ np.save(os.path.join(output_folder,"benchmark_peak_indices"), duet.peak_indices)
- delay_peak, atn_delay_est, atn_peak = duet._convert_peaks()
+ duet.delay_peak, duet.atn_delay_est, duet.atn_peak = duet._convert_peaks()
+ np.save(os.path.join(output_folder, "benchmark_delay_peak"), duet.delay_peak)
+ np.save(os.path.join(output_folder, "benchmark_atn_delay_est"), duet.atn_delay_est)
+ np.save(os.path.join(output_folder, "benchmark_atn_peak"), duet.atn_peak)
- duet._compute_masks(delay_peak, atn_peak)
+ duet.masks = duet._compute_masks()
+ np.save(os.path.join(output_folder, "benchmark_masks"), duet.masks)
- source_estimates = duet._convert_to_time_domain(atn_peak, delay_peak)
+ final_signals = duet.make_audio_signals()
+ np.save(os.path.join(output_folder, "benchmark_final_signals"), final_signals)
- duet.separated_sources = source_estimates
if __name__ == '__main__':
unittest.main()
+#freeze_duet_values()

Moved convert_to_time_domain under make_audio_signals and modified tests to include freeze_benchmarks functions.