""" /* Copyright (c) 2023 Amazon Written by Jan Buethe */ /* Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: - Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. - Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ """ import os import multiprocess as multiprocessing import random import subprocess import argparse import shutil import yaml from utils.files import get_wave_file_list from utils.pesq import compute_PESQ from utils.pitch import compute_pitch_error parser = argparse.ArgumentParser() parser.add_argument('setup', type=str, help='setup yaml specifying end to end processing with model under test') parser.add_argument('input_folder', type=str, help='input folder path') parser.add_argument('output_folder', type=str, help='output folder path') parser.add_argument('--num-testitems', type=int, help="number of testitems to be processed (default 100)", default=100) parser.add_argument('--seed', type=int, help='seed for random item selection', default=None) parser.add_argument('--fs', type=int, help="sampling rate at which input is presented as wave file (defaults to 16000)", default=16000) parser.add_argument('--num-workers', type=int, help="number of subprocesses to be used (default=4)", default=4) parser.add_argument('--plc-suffix', type=str, default="_is_lost.txt", help="suffix of plc error pattern file: only relevant if command chain uses PLCFILE (default=_is_lost.txt)") parser.add_argument('--metrics', type=str, default='pesq', help='comma separated string of metrics, supported: {{"pesq", "pitch_error", "voicing_error"}}, default="pesq"') parser.add_argument('--verbose', action='store_true', help='enables printouts of all commands run in the pipeline') def check_for_sox_in_path(): r = subprocess.run("sox -h", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return r.returncode == 0 def run_save_sh(command, verbose=False): if verbose: print(f"[run_save_sh] running command {command}...") r = subprocess.run(command, shell=True) if r.returncode != 0: raise RuntimeError(f"command '{command}' failed with exit code {r.returncode}") def run_processing_chain(input_path, output_path, model_commands, fs, metrics={'pesq'}, plc_suffix="_is_lost.txt", verbose=False): # prepare model input model_input = output_path + ".resamp.wav" run_save_sh(f"sox {input_path} -r {fs} {model_input}", verbose=verbose) plcfile = os.path.splitext(input_path)[0] + plc_suffix if os.path.isfile(plcfile): run_save_sh(f"cp {plcfile} {os.path.dirname(output_path)}") # generate model output for command in model_commands: run_save_sh(command.format(INPUT=model_input, OUTPUT=output_path, PLCFILE=plcfile), verbose=verbose) scores = dict() cache = dict() for metric in metrics: if metric == 'pesq': # run pesq score = compute_PESQ(input_path, output_path, fs=fs) elif metric == 'pitch_error': if metric in cache: score = cache[metric] else: rval = compute_pitch_error(input_path, output_path, fs=fs) score = rval[metric] cache['voicing_error'] = rval['voicing_error'] elif metric == 'voicing_error': if metric in cache: score = cache[metric] else: rval = compute_pitch_error(input_path, output_path, fs=fs) score = rval[metric] cache['pitch_error'] = rval['pitch_error'] else: ValueError(f'error: unknown metric {metric}') scores[metric] = score return (output_path, scores) def get_output_path(root_folder, input, output_folder): input_relpath = os.path.relpath(input, root_folder) os.makedirs(os.path.join(output_folder, 'processing', os.path.dirname(input_relpath)), exist_ok=True) output_path = os.path.join(output_folder, 'processing', input_relpath + '.output.wav') return output_path def add_audio_table(f, html_folder, results, title, metric): item_folder = os.path.join(html_folder, 'items') os.makedirs(item_folder, exist_ok=True) # table with results f.write(f"""
Rank | Name | {metric.upper()} | Audio (out) | Audio (orig) |
---|---|---|---|---|
{i + 1} | {item_name.split('.')[0]} | {score:.3f} |