path: root/third_party/libwebrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/simulation.py

# 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]