modules/audio_processing/test/py_quality_assessment/apm_quality_assessment_export.py

#!/usr/bin/env python
# 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.

"""Export the scores computed by the apm_quality_assessment.py script into an
   HTML file.
"""

import argparse
import logging
import glob
import os
import re
import sys

try:
  import pandas as pd
except ImportError:
  logging.critical('Cannot import the third-party Python package pandas')
  sys.exit(1)

import quality_assessment.data_access as data_access
import quality_assessment.export as export
import quality_assessment.simulation as sim

# Compiled regular expressions used to extract score descriptors.
RE_CONFIG_NAME = re.compile(
    sim.ApmModuleSimulator.GetPrefixApmConfig() + r'(.+)')
RE_CAPTURE_NAME = re.compile(
    sim.ApmModuleSimulator.GetPrefixCapture() + r'(.+)')
RE_RENDER_NAME = re.compile(
    sim.ApmModuleSimulator.GetPrefixRender() + r'(.+)')
RE_ECHO_SIM_NAME = re.compile(
    sim.ApmModuleSimulator.GetPrefixEchoSimulator() + r'(.+)')
RE_TEST_DATA_GEN_NAME = re.compile(
    sim.ApmModuleSimulator.GetPrefixTestDataGenerator() + r'(.+)')
RE_TEST_DATA_GEN_PARAMS = re.compile(
    sim.ApmModuleSimulator.GetPrefixTestDataGeneratorParameters() + r'(.+)')
RE_SCORE_NAME = re.compile(
    sim.ApmModuleSimulator.GetPrefixScore() + r'(.+)(\..+)')


def _InstanceArgumentsParser():
  """Arguments parser factory.
  """
  parser = argparse.ArgumentParser(description=(
      'Exports pre-computed APM module quality assessment results into HTML '
      'tables.'))

  parser.add_argument('-o', '--output_dir', required=True,
                      help=('the same base path used with the '
                            'apm_quality_assessment tool'))

  parser.add_argument('-f', '--filename_suffix',
                      help=('suffix of the exported file'))

  parser.add_argument('-c', '--config_names', type=re.compile,
                      help=('regular expression to filter the APM configuration'
                            ' names'))

  parser.add_argument('-i', '--capture_names', type=re.compile,
                      help=('regular expression to filter the capture signal '
                            'names'))

  parser.add_argument('-r', '--render_names', type=re.compile,
                      help=('regular expression to filter the render signal '
                            'names'))

  parser.add_argument('-e', '--echo_simulator_names', type=re.compile,
                      help=('regular expression to filter the echo simulator '
                            'names'))

  parser.add_argument('-t', '--test_data_generators', type=re.compile,
                      help=('regular expression to filter the test data '
                            'generator names'))

  parser.add_argument('-s', '--eval_scores', type=re.compile,
                      help=('regular expression to filter the evaluation score '
                            'names'))

  return parser


def _GetScoreDescriptors(score_filepath):
  """Extracts a score descriptor from the given score file path.

  Args:
    score_filepath: path to the score file.

  Returns:
    A tuple of strings (APM configuration name, capture audio track name,
    render audio track name, echo simulator name, test data generator name,
    test data generator parameters as string, evaluation score name).
  """
  fields = score_filepath.split(os.sep)[-7:]
  extract_name = lambda index, reg_expr: (
      reg_expr.match(fields[index]).groups(0)[0])
  return (
      extract_name(0, RE_CONFIG_NAME),
      extract_name(1, RE_CAPTURE_NAME),
      extract_name(2, RE_RENDER_NAME),
      extract_name(3, RE_ECHO_SIM_NAME),
      extract_name(4, RE_TEST_DATA_GEN_NAME),
      extract_name(5, RE_TEST_DATA_GEN_PARAMS),
      extract_name(6, RE_SCORE_NAME),
  )


def _ExcludeScore(config_name, capture_name, render_name, echo_simulator_name,
                  test_data_gen_name, score_name, args):
  """Decides whether excluding a score.

  A set of optional regular expressions in args is used to determine if the
  score should be excluded (depending on its |*_name| descriptors).

  Args:
    config_name: APM configuration name.
    capture_name: capture audio track name.
    render_name: render audio track name.
    echo_simulator_name: echo simulator name.
    test_data_gen_name: test data generator name.
    score_name: evaluation score name.
    args: parsed arguments.

  Returns:
    A boolean.
  """
  value_regexpr_pairs = [
      (config_name, args.config_names),
      (capture_name, args.capture_names),
      (render_name, args.render_names),
      (echo_simulator_name, args.echo_simulator_names),
      (test_data_gen_name, args.test_data_generators),
      (score_name, args.eval_scores),
  ]

  # Score accepted if each value matches the corresponding regular expression.
  for value, regexpr in value_regexpr_pairs:
    if regexpr is None:
      continue
    if not regexpr.match(value):
      return True

  return False


def _BuildOutputFilename(filename_suffix):
  """Builds the filename for the exported file.

  Args:
    filename_suffix: suffix for the output file name.

  Returns:
    A string.
  """
  if filename_suffix is None:
    return 'results.html'
  return 'results-{}.html'.format(filename_suffix)


def _FindScores(src_path, args):
  """Given a search path, find scores and return a DataFrame object.

  Args:
    src_path: Search path pattern.
    args: parsed arguments.

  Returns:
    A DataFrame object.
  """
  # Get scores.
  scores = []
  for score_filepath in glob.iglob(src_path):
    # Extract score descriptor fields from the path.
    (config_name,
     capture_name,
     render_name,
     echo_simulator_name,
     test_data_gen_name,
     test_data_gen_params,
     score_name) = _GetScoreDescriptors(score_filepath)

    # Ignore the score if required.
    if _ExcludeScore(
        config_name,
        capture_name,
        render_name,
        echo_simulator_name,
        test_data_gen_name,
        score_name,
        args):
      logging.info(
          'ignored score: %s %s %s %s %s %s',
          config_name,
          capture_name,
          render_name,
          echo_simulator_name,
          test_data_gen_name,
          score_name)
      continue

    # Read metadata and score.
    metadata = data_access.Metadata.LoadAudioTestDataPaths(
        os.path.split(score_filepath)[0])
    score = data_access.ScoreFile.Load(score_filepath)

    # Add a score with its descriptor fields.
    scores.append((
        metadata['clean_capture_input_filepath'],
        metadata['echo_free_capture_filepath'],
        metadata['echo_filepath'],
        metadata['render_filepath'],
        metadata['capture_filepath'],
        metadata['apm_output_filepath'],
        metadata['apm_reference_filepath'],
        config_name,
        capture_name,
        render_name,
        echo_simulator_name,
        test_data_gen_name,
        test_data_gen_params,
        score_name,
        score,
    ))

  return pd.DataFrame(
      data=scores,
      columns=(
          'clean_capture_input_filepath',
          'echo_free_capture_filepath',
          'echo_filepath',
          'render_filepath',
          'capture_filepath',
          'apm_output_filepath',
          'apm_reference_filepath',
          'apm_config',
          'capture',
          'render',
          'echo_simulator',
          'test_data_gen',
          'test_data_gen_params',
          'eval_score_name',
          'score',
      ))


def main():
  # Init.
  logging.basicConfig(level=logging.DEBUG)  # TODO(alessio): INFO once debugged.
  parser = _InstanceArgumentsParser()
  args = parser.parse_args()

  # Get the scores.
  src_path = os.path.join(
      args.output_dir,
      sim.ApmModuleSimulator.GetPrefixApmConfig() + '*',
      sim.ApmModuleSimulator.GetPrefixCapture() + '*',
      sim.ApmModuleSimulator.GetPrefixRender() + '*',
      sim.ApmModuleSimulator.GetPrefixEchoSimulator() + '*',
      sim.ApmModuleSimulator.GetPrefixTestDataGenerator() + '*',
      sim.ApmModuleSimulator.GetPrefixTestDataGeneratorParameters() + '*',
      sim.ApmModuleSimulator.GetPrefixScore() + '*')
  logging.debug(src_path)
  scores_data_frame = _FindScores(src_path, args)

  # Export.
  output_filepath = os.path.join(args.output_dir, _BuildOutputFilename(
      args.filename_suffix))
  exporter = export.HtmlExport(output_filepath)
  exporter.Export(scores_data_frame)

  logging.info('output file successfully written in %s', output_filepath)
  sys.exit(0)


if __name__ == '__main__':
  main()