| # 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. |
| |
| """APM module simulator. |
| """ |
| |
| import logging |
| import os |
| |
| from . import annotations |
| from . import data_access |
| from . import echo_path_simulation |
| from . import echo_path_simulation_factory |
| from . import eval_scores |
| from . import exceptions |
| from . import input_mixer |
| from . import input_signal_creator |
| from . import signal_processing |
| from . import test_data_generation |
| |
| |
| class ApmModuleSimulator(object): |
| """Audio processing module (APM) simulator class. |
| """ |
| |
| _TEST_DATA_GENERATOR_CLASSES = ( |
| test_data_generation.TestDataGenerator.REGISTERED_CLASSES) |
| _EVAL_SCORE_WORKER_CLASSES = eval_scores.EvaluationScore.REGISTERED_CLASSES |
| |
| _PREFIX_APM_CONFIG = 'apmcfg-' |
| _PREFIX_CAPTURE = 'capture-' |
| _PREFIX_RENDER = 'render-' |
| _PREFIX_ECHO_SIMULATOR = 'echosim-' |
| _PREFIX_TEST_DATA_GEN = 'datagen-' |
| _PREFIX_TEST_DATA_GEN_PARAMS = 'datagen_params-' |
| _PREFIX_SCORE = 'score-' |
| |
| def __init__(self, test_data_generator_factory, evaluation_score_factory, |
| ap_wrapper, evaluator, external_vads=None): |
| if external_vads is None: |
| external_vads = {} |
| self._test_data_generator_factory = test_data_generator_factory |
| self._evaluation_score_factory = evaluation_score_factory |
| self._audioproc_wrapper = ap_wrapper |
| self._evaluator = evaluator |
| self._annotator = annotations.AudioAnnotationsExtractor( |
| annotations.AudioAnnotationsExtractor.VadType.ENERGY_THRESHOLD | |
| annotations.AudioAnnotationsExtractor.VadType.WEBRTC_COMMON_AUDIO | |
| annotations.AudioAnnotationsExtractor.VadType.WEBRTC_APM, |
| external_vads |
| ) |
| |
| # Init. |
| self._test_data_generator_factory.SetOutputDirectoryPrefix( |
| self._PREFIX_TEST_DATA_GEN_PARAMS) |
| self._evaluation_score_factory.SetScoreFilenamePrefix( |
| self._PREFIX_SCORE) |
| |
| # Properties for each run. |
| self._base_output_path = None |
| self._output_cache_path = None |
| self._test_data_generators = None |
| self._evaluation_score_workers = None |
| self._config_filepaths = None |
| self._capture_input_filepaths = None |
| self._render_input_filepaths = None |
| self._echo_path_simulator_class = None |
| |
| @classmethod |
| def GetPrefixApmConfig(cls): |
| return cls._PREFIX_APM_CONFIG |
| |
| @classmethod |
| def GetPrefixCapture(cls): |
| return cls._PREFIX_CAPTURE |
| |
| @classmethod |
| def GetPrefixRender(cls): |
| return cls._PREFIX_RENDER |
| |
| @classmethod |
| def GetPrefixEchoSimulator(cls): |
| return cls._PREFIX_ECHO_SIMULATOR |
| |
| @classmethod |
| def GetPrefixTestDataGenerator(cls): |
| return cls._PREFIX_TEST_DATA_GEN |
| |
| @classmethod |
| def GetPrefixTestDataGeneratorParameters(cls): |
| return cls._PREFIX_TEST_DATA_GEN_PARAMS |
| |
| @classmethod |
| def GetPrefixScore(cls): |
| return cls._PREFIX_SCORE |
| |
| def Run(self, config_filepaths, capture_input_filepaths, |
| test_data_generator_names, eval_score_names, output_dir, |
| render_input_filepaths=None, echo_path_simulator_name=( |
| echo_path_simulation.NoEchoPathSimulator.NAME)): |
| """Runs the APM simulation. |
| |
| Initializes paths and required instances, then runs all the simulations. |
| The render input can be optionally added. If added, the number of capture |
| input audio tracks and the number of render input audio tracks have to be |
| equal. The two lists are used to form pairs of capture and render input. |
| |
| Args: |
| config_filepaths: set of APM configuration files to test. |
| capture_input_filepaths: set of capture input audio track files to test. |
| test_data_generator_names: set of test data generator names to test. |
| eval_score_names: set of evaluation score names to test. |
| output_dir: base path to the output directory for wav files and outcomes. |
| render_input_filepaths: set of render input audio track files to test. |
| echo_path_simulator_name: name of the echo path simulator to use when |
| render input is provided. |
| """ |
| assert render_input_filepaths is None or ( |
| len(capture_input_filepaths) == len(render_input_filepaths)), ( |
| 'render input set size not matching input set size') |
| assert render_input_filepaths is None or echo_path_simulator_name in ( |
| echo_path_simulation.EchoPathSimulator.REGISTERED_CLASSES), ( |
| 'invalid echo path simulator') |
| self._base_output_path = os.path.abspath(output_dir) |
| |
| # Output path used to cache the data shared across simulations. |
| self._output_cache_path = os.path.join(self._base_output_path, '_cache') |
| |
| # Instance test data generators. |
| self._test_data_generators = [self._test_data_generator_factory.GetInstance( |
| test_data_generators_class=( |
| self._TEST_DATA_GENERATOR_CLASSES[name])) for name in ( |
| test_data_generator_names)] |
| |
| # Instance evaluation score workers. |
| self._evaluation_score_workers = [ |
| self._evaluation_score_factory.GetInstance( |
| evaluation_score_class=self._EVAL_SCORE_WORKER_CLASSES[name]) for ( |
| name) in eval_score_names] |
| |
| # Set APM configuration file paths. |
| self._config_filepaths = self._CreatePathsCollection(config_filepaths) |
| |
| # Set probing signal file paths. |
| if render_input_filepaths is None: |
| # Capture input only. |
| self._capture_input_filepaths = self._CreatePathsCollection( |
| capture_input_filepaths) |
| self._render_input_filepaths = None |
| else: |
| # Set both capture and render input signals. |
| self._SetTestInputSignalFilePaths( |
| capture_input_filepaths, render_input_filepaths) |
| |
| # Set the echo path simulator class. |
| self._echo_path_simulator_class = ( |
| echo_path_simulation.EchoPathSimulator.REGISTERED_CLASSES[ |
| echo_path_simulator_name]) |
| |
| self._SimulateAll() |
| |
| def _SimulateAll(self): |
| """Runs all the simulations. |
| |
| Iterates over the combinations of APM configurations, probing signals, and |
| test data generators. This method is mainly responsible for the creation of |
| the cache and output directories required in order to call _Simulate(). |
| """ |
| without_render_input = self._render_input_filepaths is None |
| |
| # Try different APM config files. |
| for config_name in self._config_filepaths: |
| config_filepath = self._config_filepaths[config_name] |
| |
| # Try different capture-render pairs. |
| for capture_input_name in self._capture_input_filepaths: |
| # Output path for the capture signal annotations. |
| capture_annotations_cache_path = os.path.join( |
| self._output_cache_path, |
| self._PREFIX_CAPTURE + capture_input_name) |
| data_access.MakeDirectory(capture_annotations_cache_path) |
| |
| # Capture. |
| capture_input_filepath = self._capture_input_filepaths[ |
| capture_input_name] |
| if not os.path.exists(capture_input_filepath): |
| # If the input signal file does not exist, try to create using the |
| # available input signal creators. |
| self._CreateInputSignal(capture_input_filepath) |
| assert os.path.exists(capture_input_filepath) |
| self._ExtractCaptureAnnotations( |
| capture_input_filepath, capture_annotations_cache_path) |
| |
| # Render and simulated echo path (optional). |
| render_input_filepath = None if without_render_input else ( |
| self._render_input_filepaths[capture_input_name]) |
| render_input_name = '(none)' if without_render_input else ( |
| self._ExtractFileName(render_input_filepath)) |
| echo_path_simulator = ( |
| echo_path_simulation_factory.EchoPathSimulatorFactory.GetInstance( |
| self._echo_path_simulator_class, render_input_filepath)) |
| |
| # Try different test data generators. |
| for test_data_generators in self._test_data_generators: |
| logging.info('APM config preset: <%s>, capture: <%s>, render: <%s>,' |
| 'test data generator: <%s>, echo simulator: <%s>', |
| config_name, capture_input_name, render_input_name, |
| test_data_generators.NAME, echo_path_simulator.NAME) |
| |
| # Output path for the generated test data. |
| test_data_cache_path = os.path.join( |
| capture_annotations_cache_path, |
| self._PREFIX_TEST_DATA_GEN + test_data_generators.NAME) |
| data_access.MakeDirectory(test_data_cache_path) |
| logging.debug('test data cache path: <%s>', test_data_cache_path) |
| |
| # Output path for the echo simulator and APM input mixer output. |
| echo_test_data_cache_path = os.path.join( |
| test_data_cache_path, 'echosim-{}'.format( |
| echo_path_simulator.NAME)) |
| data_access.MakeDirectory(echo_test_data_cache_path) |
| logging.debug('echo test data cache path: <%s>', |
| echo_test_data_cache_path) |
| |
| # Full output path. |
| output_path = os.path.join( |
| self._base_output_path, |
| self._PREFIX_APM_CONFIG + config_name, |
| self._PREFIX_CAPTURE + capture_input_name, |
| self._PREFIX_RENDER + render_input_name, |
| self._PREFIX_ECHO_SIMULATOR + echo_path_simulator.NAME, |
| self._PREFIX_TEST_DATA_GEN + test_data_generators.NAME) |
| data_access.MakeDirectory(output_path) |
| logging.debug('output path: <%s>', output_path) |
| |
| self._Simulate(test_data_generators, capture_input_filepath, |
| render_input_filepath, test_data_cache_path, |
| echo_test_data_cache_path, output_path, |
| config_filepath, echo_path_simulator) |
| |
| @staticmethod |
| def _CreateInputSignal(input_signal_filepath): |
| """Creates a missing input signal file. |
| |
| The file name is parsed to extract input signal creator and params. If a |
| creator is matched and the parameters are valid, a new signal is generated |
| and written in |input_signal_filepath|. |
| |
| Args: |
| input_signal_filepath: Path to the input signal audio file to write. |
| |
| Raises: |
| InputSignalCreatorException |
| """ |
| filename = os.path.splitext(os.path.split(input_signal_filepath)[-1])[0] |
| filename_parts = filename.split('-') |
| |
| if len(filename_parts) < 2: |
| raise exceptions.InputSignalCreatorException( |
| 'Cannot parse input signal file name') |
| |
| signal, metadata = input_signal_creator.InputSignalCreator.Create( |
| filename_parts[0], filename_parts[1].split('_')) |
| |
| signal_processing.SignalProcessingUtils.SaveWav( |
| input_signal_filepath, signal) |
| data_access.Metadata.SaveFileMetadata(input_signal_filepath, metadata) |
| |
| def _ExtractCaptureAnnotations(self, input_filepath, output_path, |
| annotation_name=""): |
| self._annotator.Extract(input_filepath) |
| self._annotator.Save(output_path, annotation_name) |
| |
| def _Simulate(self, test_data_generators, clean_capture_input_filepath, |
| render_input_filepath, test_data_cache_path, |
| echo_test_data_cache_path, output_path, config_filepath, |
| echo_path_simulator): |
| """Runs a single set of simulation. |
| |
| Simulates a given combination of APM configuration, probing signal, and |
| test data generator. It iterates over the test data generator |
| internal configurations. |
| |
| Args: |
| test_data_generators: TestDataGenerator instance. |
| clean_capture_input_filepath: capture input audio track file to be |
| processed by a test data generator and |
| not affected by echo. |
| render_input_filepath: render input audio track file to test. |
| test_data_cache_path: path for the generated test audio track files. |
| echo_test_data_cache_path: path for the echo simulator. |
| output_path: base output path for the test data generator. |
| config_filepath: APM configuration file to test. |
| echo_path_simulator: EchoPathSimulator instance. |
| """ |
| # Generate pairs of noisy input and reference signal files. |
| test_data_generators.Generate( |
| input_signal_filepath=clean_capture_input_filepath, |
| test_data_cache_path=test_data_cache_path, |
| base_output_path=output_path) |
| |
| # Extract metadata linked to the clean input file (if any). |
| apm_input_metadata = None |
| try: |
| apm_input_metadata = data_access.Metadata.LoadFileMetadata( |
| clean_capture_input_filepath) |
| except IOError as e: |
| apm_input_metadata = {} |
| apm_input_metadata['test_data_gen_name'] = test_data_generators.NAME |
| apm_input_metadata['test_data_gen_config'] = None |
| |
| # For each test data pair, simulate a call and evaluate. |
| for config_name in test_data_generators.config_names: |
| logging.info(' - test data generator config: <%s>', config_name) |
| apm_input_metadata['test_data_gen_config'] = config_name |
| |
| # Paths to the test data generator output. |
| # Note that the reference signal does not depend on the render input |
| # which is optional. |
| noisy_capture_input_filepath = ( |
| test_data_generators.noisy_signal_filepaths[config_name]) |
| reference_signal_filepath = ( |
| test_data_generators.reference_signal_filepaths[config_name]) |
| |
| # Output path for the evaluation (e.g., APM output file). |
| evaluation_output_path = test_data_generators.apm_output_paths[ |
| config_name] |
| |
| # Paths to the APM input signals. |
| echo_path_filepath = echo_path_simulator.Simulate( |
| echo_test_data_cache_path) |
| apm_input_filepath = input_mixer.ApmInputMixer.Mix( |
| echo_test_data_cache_path, noisy_capture_input_filepath, |
| echo_path_filepath) |
| |
| # Extract annotations for the APM input mix. |
| apm_input_basepath, apm_input_filename = os.path.split( |
| apm_input_filepath) |
| self._ExtractCaptureAnnotations( |
| apm_input_filepath, apm_input_basepath, |
| os.path.splitext(apm_input_filename)[0] + '-') |
| |
| # Simulate a call using APM. |
| self._audioproc_wrapper.Run( |
| config_filepath=config_filepath, |
| capture_input_filepath=apm_input_filepath, |
| render_input_filepath=render_input_filepath, |
| output_path=evaluation_output_path) |
| |
| try: |
| # Evaluate. |
| self._evaluator.Run( |
| evaluation_score_workers=self._evaluation_score_workers, |
| apm_input_metadata=apm_input_metadata, |
| apm_output_filepath=self._audioproc_wrapper.output_filepath, |
| reference_input_filepath=reference_signal_filepath, |
| render_input_filepath=render_input_filepath, |
| output_path=evaluation_output_path, |
| ) |
| |
| # Save simulation metadata. |
| data_access.Metadata.SaveAudioTestDataPaths( |
| output_path=evaluation_output_path, |
| clean_capture_input_filepath=clean_capture_input_filepath, |
| echo_free_capture_filepath=noisy_capture_input_filepath, |
| echo_filepath=echo_path_filepath, |
| render_filepath=render_input_filepath, |
| capture_filepath=apm_input_filepath, |
| apm_output_filepath=self._audioproc_wrapper.output_filepath, |
| apm_reference_filepath=reference_signal_filepath, |
| apm_config_filepath=config_filepath, |
| ) |
| except exceptions.EvaluationScoreException as e: |
| logging.warning('the evaluation failed: %s', e.message) |
| continue |
| |
| def _SetTestInputSignalFilePaths(self, capture_input_filepaths, |
| render_input_filepaths): |
| """Sets input and render input file paths collections. |
| |
| Pairs the input and render input files by storing the file paths into two |
| collections. The key is the file name of the input file. |
| |
| Args: |
| capture_input_filepaths: list of file paths. |
| render_input_filepaths: list of file paths. |
| """ |
| self._capture_input_filepaths = {} |
| self._render_input_filepaths = {} |
| assert len(capture_input_filepaths) == len(render_input_filepaths) |
| for capture_input_filepath, render_input_filepath in zip( |
| capture_input_filepaths, render_input_filepaths): |
| name = self._ExtractFileName(capture_input_filepath) |
| self._capture_input_filepaths[name] = os.path.abspath( |
| capture_input_filepath) |
| self._render_input_filepaths[name] = os.path.abspath( |
| render_input_filepath) |
| |
| @classmethod |
| def _CreatePathsCollection(cls, filepaths): |
| """Creates a collection of file paths. |
| |
| Given a list of file paths, makes a collection with one item for each file |
| path. The value is absolute path, the key is the file name without |
| extenstion. |
| |
| Args: |
| filepaths: list of file paths. |
| |
| Returns: |
| A dict. |
| """ |
| filepaths_collection = {} |
| for filepath in filepaths: |
| name = cls._ExtractFileName(filepath) |
| filepaths_collection[name] = os.path.abspath(filepath) |
| return filepaths_collection |
| |
| @classmethod |
| def _ExtractFileName(cls, filepath): |
| return os.path.splitext(os.path.split(filepath)[-1])[0] |
