| # Copyright (c) 2017 The WebRTC project authors. All Rights Reserved. |
| # |
| # Use of this source code is governed by a BSD-style license |
| # that can be found in the LICENSE file in the root of the source |
| # tree. An additional intellectual property rights grant can be found |
| # in the file PATENTS. All contributing project authors may |
| # be found in the AUTHORS file in the root of the source tree. |
| |
| """Unit tests for the input mixer module. |
| """ |
| |
| import logging |
| import os |
| import shutil |
| import sys |
| import tempfile |
| import unittest |
| |
| SRC = os.path.abspath(os.path.join( |
| os.path.dirname((__file__)), os.pardir, os.pardir, os.pardir, os.pardir)) |
| sys.path.append(os.path.join(SRC, 'third_party', 'pymock')) |
| |
| import mock |
| |
| from . import exceptions |
| from . import input_mixer |
| from . import signal_processing |
| |
| |
| class TestApmInputMixer(unittest.TestCase): |
| """Unit tests for the ApmInputMixer class. |
| """ |
| |
| # Audio track file names created in setUp(). |
| _FILENAMES = ['capture', 'echo_1', 'echo_2', 'shorter', 'longer'] |
| |
| # Target peak power level (dBFS) of each audio track file created in setUp(). |
| # These values are hand-crafted in order to make saturation happen when |
| # capture and echo_2 are mixed and the contrary for capture and echo_1. |
| # None means that the power is not changed. |
| _MAX_PEAK_POWER_LEVELS = [-10.0, -5.0, 0.0, None, None] |
| |
| # Audio track file durations in milliseconds. |
| _DURATIONS = [1000, 1000, 1000, 800, 1200] |
| |
| _SAMPLE_RATE = 48000 |
| |
| def setUp(self): |
| """Creates temporary data.""" |
| self._tmp_path = tempfile.mkdtemp() |
| |
| # Create audio track files. |
| self._audio_tracks = {} |
| for filename, peak_power, duration in zip( |
| self._FILENAMES, self._MAX_PEAK_POWER_LEVELS, self._DURATIONS): |
| audio_track_filepath = os.path.join(self._tmp_path, '{}.wav'.format( |
| filename)) |
| |
| # Create a pure tone with the target peak power level. |
| template = signal_processing.SignalProcessingUtils.GenerateSilence( |
| duration=duration, sample_rate=self._SAMPLE_RATE) |
| signal = signal_processing.SignalProcessingUtils.GeneratePureTone( |
| template) |
| if peak_power is not None: |
| signal = signal.apply_gain(-signal.max_dBFS + peak_power) |
| |
| signal_processing.SignalProcessingUtils.SaveWav( |
| audio_track_filepath, signal) |
| self._audio_tracks[filename] = { |
| 'filepath': audio_track_filepath, |
| 'num_samples': signal_processing.SignalProcessingUtils.CountSamples( |
| signal) |
| } |
| |
| def tearDown(self): |
| """Recursively deletes temporary folders.""" |
| shutil.rmtree(self._tmp_path) |
| |
| def testCheckMixSameDuration(self): |
| """Checks the duration when mixing capture and echo with same duration.""" |
| mix_filepath = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['echo_1']['filepath']) |
| self.assertTrue(os.path.exists(mix_filepath)) |
| |
| mix = signal_processing.SignalProcessingUtils.LoadWav(mix_filepath) |
| self.assertEqual(self._audio_tracks['capture']['num_samples'], |
| signal_processing.SignalProcessingUtils.CountSamples(mix)) |
| |
| def testRejectShorterEcho(self): |
| """Rejects echo signals that are shorter than the capture signal.""" |
| try: |
| _ = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['shorter']['filepath']) |
| self.fail('no exception raised') |
| except exceptions.InputMixerException: |
| pass |
| |
| def testCheckMixDurationWithLongerEcho(self): |
| """Checks the duration when mixing an echo longer than the capture.""" |
| mix_filepath = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['longer']['filepath']) |
| self.assertTrue(os.path.exists(mix_filepath)) |
| |
| mix = signal_processing.SignalProcessingUtils.LoadWav(mix_filepath) |
| self.assertEqual(self._audio_tracks['capture']['num_samples'], |
| signal_processing.SignalProcessingUtils.CountSamples(mix)) |
| |
| def testCheckOutputFileNamesConflict(self): |
| """Checks that different echo files lead to different output file names.""" |
| mix1_filepath = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['echo_1']['filepath']) |
| self.assertTrue(os.path.exists(mix1_filepath)) |
| |
| mix2_filepath = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['echo_2']['filepath']) |
| self.assertTrue(os.path.exists(mix2_filepath)) |
| |
| self.assertNotEqual(mix1_filepath, mix2_filepath) |
| |
| def testHardClippingLogExpected(self): |
| """Checks that hard clipping warning is raised when occurring.""" |
| logging.warning = mock.MagicMock(name='warning') |
| _ = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['echo_2']['filepath']) |
| logging.warning.assert_called_once_with( |
| input_mixer.ApmInputMixer.HardClippingLogMessage()) |
| |
| def testHardClippingLogNotExpected(self): |
| """Checks that hard clipping warning is not raised when not occurring.""" |
| logging.warning = mock.MagicMock(name='warning') |
| _ = input_mixer.ApmInputMixer.Mix( |
| self._tmp_path, |
| self._audio_tracks['capture']['filepath'], |
| self._audio_tracks['echo_1']['filepath']) |
| self.assertNotIn( |
| mock.call(input_mixer.ApmInputMixer.HardClippingLogMessage()), |
| logging.warning.call_args_list) |
