Merge branch 'ci/batch_mld_script' into 'main'

import argparse

import concurrent.futures

import os

import pathlib

import re

import shutil

import subprocess

import sys

import threading

from itertools import repeat

from pyaudio3dtools.audiofile import readfile

from pyaudio3dtools.audioarray import compare

FILES_EQUAL = "File A = File B"

SNR_EXPR = r"SNR\s+=(.+)dB\s*\(File B Gain = (.+)\)"

SNR_EXPR = r"SNR\s+=(.+)dB\s*\(.+= (.+)\)"

SEG_SNR_EXPR = r"Seg. SNR\s+=(.+)dB"

DIFF_EXPR = r"Max Diff\s+=\s+(\d+)"

DIFF_STR = {

    "CompAudio": "{label} Max. diff (PCM) for file {f}: {diff}",

    "mld": "{label} MLD diff for file {f}: {diff}",

}

def main(args):

    if shutil.which("CompAudio") is None:

        print("CompAudio not in PATH - abort.")

    tool = args.tool

    if shutil.which(tool) is None:

        print(f"{tool} not in PATH - abort.")

        sys.exit(-1)

    num_files_diff = 0

    with OutFileManager(args.out_file) as out_file:

        if args.diffs_only:

            print("Only printing differing files!", file=out_file)

            print("Only printing differing files!")

        fol1, fol2 = os.path.normpath(args.folder1), os.path.normpath(args.folder2)

        common_files = get_common_files(fol1, fol2)

        diff_files = get_diff_files(fol1, fol2)

        num_files_diff = len(diff_files)

        print(f"Comparing {len(common_files)} files...", file=out_file)

        print(f"Comparing {len(common_files)} files...")

        outputs = dict()

        if args.num_threads > 1:

        if args.num_threads == 1:

            # if only one thread is passed, do everything in the main thread

            # to allow for meaningful debugging if needed

            for f in common_files:

                compare_files(f, fol1, fol2, outputs, tool)

        else:

            with concurrent.futures.ThreadPoolExecutor(

                max_workers=args.num_threads

            ) as exc:

                    repeat(fol1),

                    repeat(fol2),

                    repeat(outputs),

                    repeat(tool),

                )

        else:

            # if only one thread is passed, do everything in the main thread

            # to allow for meaningful debugging if needed

            for f in common_files:

                compare_files(f, fol1, fol2, outputs)

        if args.sort:

            out = dict(sorted(outputs.items(), key=lambda item: item[1]))

        else:

            out = outputs

        for f, output_tuple in out.items():

            diff, snr, gain, seg_snr = output_tuple

        # write csv header

        if out_file is not None:

            if tool == "CompAudio":

                out_file.write("filename,diff\n")

            elif tool == "mld":

                out_file.write("filename,mld\n")

        for f, tool_output in out.items():

            if tool == "CompAudio":

                diff, snr, gain, seg_snr = tool_output

            elif tool == "mld":

                diff = tool_output

            if diff > 0:

                num_files_diff = num_files_diff + 1

                    label = "[OKAY]"

                else:

                    label = "[FAIL]"

                result = f"{label} Max. diff (PCM) for file {f}: {diff}"

                result = DIFF_STR[tool].format(label=label, f=f, diff=diff)

                if args.verbose and diff != 0.0:

                if tool == "CompAudio" and args.verbose and diff != 0.0:

                    result += f", SNR = {snr:4.2f} dB (File 2 Gain = {gain:4.3f})"

                    result += f", Seg. SNR = {seg_snr:4.2f} dB"

                print(result, file=out_file)

                print(result)

                if out_file is not None:

                    out_file.write(f"{f},{diff}\n")

        if num_files_diff > 0:

            print(f"{num_files_diff} files differ/don't exist", file=out_file)

            print(f"{num_files_diff} files differ/don't exist")

        else:

            print(f"All files are bitexact", file=out_file)

            print(f"All files are bitexact")

def compare_files(f, fol1, fol2, outputs_dict):

def compare_files(f, fol1, fol2, outputs_dict, tool):

    """

    Compare file f in both folders fol1 and fol2 using CompAudio and

    Compare file f in both folders fol1 and fol2 using the given tool and

    store the parsed difference in outputs_dict.

    """

    f1 = os.path.join(fol1, f)

    f2 = os.path.join(fol2, f)

    cmd = f"CompAudio {f1} {f2}"

    if tool == "CompAudio":

        cmd = f"{tool} {f1} {f2}"

        try:

            output = subprocess.check_output(cmd.split(" "))

        except subprocess.CalledProcessError:

        print("CompAudio returned a non-zero exit status. Check your files.")

            print(f"{tool} returned a non-zero exit status. Check your files.")

            sys.exit(-1)

    output_tuple = _parse_comp_audio(output)

        tool_output = _parse_comp_audio(output)

    elif tool == "mld":

        s1, fs1 = readfile(f1, outdtype="int16")

        s2, fs2 = readfile(f2, outdtype="int16")

        cmp_result = compare(s1, s2, fs1, per_frame=False, get_mld=True)

        tool_output = cmp_result["MLD"]

    with threading.Lock():

        outputs_dict.update({f: output_tuple})

        outputs_dict.update({f: tool_output})

def get_common_files(fol1, fol2):

if __name__ == "__main__":

    parser = argparse.ArgumentParser(

        description="Compare .wav files in two folders using CompAudio"

    )

        "--out_file",

        type=str,

        default=None,

        help="If given, write output diffs to this file",

        help="If given, write output diffs to this file as comma-separated values (csv)",

    )

    parser.add_argument(

        "-t", "--num_threads", type=int, default=None, help="Number of threads to use"

    )

    parser.add_argument(

        "-t", "--num_threads", type=int, default=1, help="Number of threads to use"

        "--tool",

        choices=["mld", "CompAudio"],

        default="CompAudio",

        help="Compare tool to run",

    )

    args = parser.parse_args()