Loading ci/remove_unsupported_testcases.py +4 −0 Original line number Diff line number Diff line Loading @@ -61,9 +61,13 @@ TESTCASES = [ "Multi-channel 5_1 at 512 kbps, 48kHz in 48kHz out, BINAURAL_ROOM_REVERB out custom acoustic environment with a sequence (CREND)", "Multi-channel 5_1 at 64 kbps, 48kHz in 48kHz out, BINAURAL_ROOM_REVERB out custom acoustic environment with a sequence (FastConv)", "Multi-channel 5_1 at 32 kbps, 48kHz in 48kHz out, BINAURAL_ROOM_REVERB out custom acoustic environment with a sequence (ParamBin)", "MASA 1dir 1TC at 256 kbps, 48kHz in, 48kHz out, BINAURAL_ROOM_REVERB out, HR custom configuration", "MASA 1dir 1 TC at 256kbps, 48kHz in, 48 kHz out, BINAURAL_ROOM_REVERB out custom configuration", "MASA 1dir 1TC at 256kbps, 48kHz in, 48 kHz out, BINAURAL_ROOM_REVERB out custom configuration", ] def remove_testcases(cfg: Path, testcases: list): """ Go through file line by line and copy all testcases except the given ones Loading tests/cmp_stats_files.py 0 → 100644 +146 −0 Original line number Diff line number Diff line #!/usr/bin/env python3 import argparse import os import sys import json import numpy as np def str2num(s): """ Convert string either to integer or float """ try: return int(s) except ValueError: return float(s) def cmp_stats_files( ref_stats_file, dut_stats_file, min_enc_file_length_diff=0.1, min_enc_stats_diff=0.1 ) -> (int, str): """ Compare two .stats files containing encoder statistics (extracted from binary files) """ print(f"Comparing {os.path.basename(ref_stats_file)} between Ref and Dut ... \n") # open and read the .stats files with open(ref_stats_file, "r") as f_aux: ref_stats = json.load(f_aux) # create dictionary to map "name" to the corresponding dictionaries ref_stats_names = {d["name"]: d for d in ref_stats} with open(dut_stats_file, "r") as f_aux: dut_stats = json.load(f_aux) # create dictionary to map "name" to the corresponding dictionaries dut_stats_names = {d["name"]: d for d in dut_stats} # loop over all common aux files enc_test_result = 0 enc_test_result_msg = "" max_total_num_diff = 0 max_total_num_diff_ratio = 0 for name in ref_stats_names: if name in dut_stats_names: # retrieve the dictionaries ref_stats_dict = ref_stats_names[name] dut_stats_dict = dut_stats_names[name] msg = f"File {name}" # compare the file lengths result_len_check = 0 file_length = max(ref_stats_dict["length"], dut_stats_dict["length"]) if ref_stats_dict["length"] != dut_stats_dict["length"]: msg += f" has different length between Ref {ref_stats_dict['length']} and DuT {dut_stats_dict['length']}" # check if threshold has been exceeded if ( abs(ref_stats_dict["length"] - dut_stats_dict["length"]) / file_length > min_enc_file_length_diff ): result_len_check = 1 msg += ", " # remove the "name" and "length" keys for further processing del ref_stats_dict["name"] del dut_stats_dict["name"] del ref_stats_dict["length"] del dut_stats_dict["length"] # convert keys and values from string to float ref_hist = {str2num(i): str2num(j) for i, j in ref_stats_dict.items()} cut_hist = {str2num(i): str2num(j) for i, j in dut_stats_dict.items()} delta_ref = set(cut_hist) - set(ref_hist) delta_cut = set(ref_hist) - set(cut_hist) # append missing keys for item in delta_cut: cut_hist[item] = 0 for item in delta_ref: ref_hist[item] = 0 ref_hist = dict(sorted(ref_hist.items())) cut_hist = dict(sorted(cut_hist.items())) # caculate difference of statistics diff_hist = {k: cut_hist[k] - ref_hist[k] for k in ref_hist.keys()} # calculate the total number of differences total_num_diff = sum(np.abs(list(diff_hist.values()))) total_num_diff_ratio = total_num_diff / ( sum(ref_hist.values()) + sum(cut_hist.values()) ) msg += f"the total number of differences is {total_num_diff} ({(total_num_diff_ratio*100):.2f}%)" if total_num_diff_ratio > min_enc_stats_diff: result_diff_check = 1 msg += "! " else: result_diff_check = 0 msg += ". " # check if the maximum difference has been exceeded if total_num_diff_ratio > max_total_num_diff_ratio: max_total_num_diff = total_num_diff max_total_num_diff_ratio = total_num_diff_ratio # update test result if result_len_check or result_diff_check: enc_test_result = 1 enc_test_result_msg += msg print(msg) if enc_test_result and max_total_num_diff > 0: msg = f"MAXIMUM ENC DIFF: {max_total_num_diff} ({(max_total_num_diff_ratio*100):.2f}%) " enc_test_result_msg += msg print(msg) return enc_test_result, enc_test_result_msg if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("ref_stats_file", type=str) parser.add_argument("dut_stats_file", type=str) parser.add_argument( "--min_enc_file_length_diff", type=float, default=0.1, dest="min_enc_file_length_diff", ) parser.add_argument( "--min_enc_stats_diff", type=float, default=0.1, dest="min_enc_stats_diff" ) args = parser.parse_args() enc_test_result, enc_test_result_msg = cmp_stats_files(**vars(args)) print(enc_test_result_msg) sys.exit(enc_test_result) tests/codec_be_on_mr_nonselection/test_param_file.py +80 −5 Original line number Diff line number Diff line Loading @@ -42,8 +42,15 @@ import pytest import numpy as np from tests.cmp_pcm import cmp_pcm from tests.cmp_stats_files import cmp_stats_files from tests.conftest import DecoderFrontend, EncoderFrontend, parse_properties from tests.testconfig import PARAM_FILE from tests.constants import ( MIN_ENC_FILE_LENGTH_DIFF, MIN_ENC_STATS_DIFF, SCRIPTS_DIR, ) VALID_DEC_OUTPUT_CONF = [ "MONO", Loading Loading @@ -134,6 +141,7 @@ def convert_test_string_to_tag(test_string): def test_param_file_tests( record_property, props_to_record, encoder_only, decoder_only, dut_encoder_frontend: EncoderFrontend, dut_decoder_frontend: DecoderFrontend, Loading @@ -151,6 +159,7 @@ def test_param_file_tests( get_mld_lim, abs_tol, get_ssnr, get_enc_stats, get_odg, ): enc_opts, dec_opts, sim_opts, eid_opts = param_file_test_dict[test_tag] Loading @@ -175,7 +184,7 @@ def test_param_file_tests( # bitrate can be a filename: remove leading "../" if bitrate.startswith("../"): bitrate = bitrate[3:] bitrate = Path(bitrate[3:]).absolute() testv_base = testv_file.split("/")[-1] if testv_base.endswith(".pcm"): Loading @@ -186,6 +195,9 @@ def test_param_file_tests( # -> construct bitstream filename bitstream_file = f"{testv_base}_{tag_str}.192" cmp_result_msg = "" enc_test_result = 0 if not decoder_only: encode( dut_encoder_frontend, Loading @@ -198,8 +210,42 @@ def test_param_file_tests( bitstream_file, enc_split, update_ref, get_enc_stats, ) # compare binary files extracted from the encoder if update_ref in [0, 2] and get_enc_stats: print("Comparing encoder files") print("=======================\n") stats_file = bitstream_file.replace(".192", ".stats") ref_stats_file = f"{reference_path}/param_file/enc/{stats_file}" dut_stats_file = f"{dut_base_path}/param_file/enc/{stats_file}" # compare ref and dut .stats files enc_test_result, enc_test_result_msg = cmp_stats_files( ref_stats_file, dut_stats_file, min_enc_file_length_diff=MIN_ENC_FILE_LENGTH_DIFF, min_enc_stats_diff=MIN_ENC_STATS_DIFF, ) print("") cmp_result_msg += enc_test_result_msg props = parse_properties(cmp_result_msg, False, props_to_record) for k, v in props.items(): record_property(k, v) if enc_test_result: pytest.fail("Too high difference in encoder statistics found.") else: # remove DUT stats file when test result is OK (to save disk space) if not keep_files: os.remove(dut_stats_file) if encoder_only: return # don't proceed with the decoder if user specified --encoder_only on the command line # check for networkSimulator_g192 command line if sim_opts != "": sim_split = sim_opts.split() Loading Loading @@ -232,6 +278,7 @@ def test_param_file_tests( ) # check for eid-xor command line if eid_opts != "": eid_split = eid_opts.split() assert len(eid_split) >= 3, "eid-xor expects at least 3 parameters" Loading @@ -248,6 +295,7 @@ def test_param_file_tests( # -> construct netsim output file name eid_xor_outfile = f"{testv_base}_{tag_str}.fer.192" eid_split[-1] = eid_xor_outfile error_insertion( reference_path, dut_base_path, Loading @@ -268,6 +316,11 @@ def test_param_file_tests( ] # remove leading "../" dec_split = [x[3:] if x.startswith("../") else x for x in dec_split] # convert "scripts/" paths into absolute ones dec_split = [ str(SCRIPTS_DIR.joinpath(x[8:])) if x.startswith("scripts/") else x for x in dec_split ] output_file = dec_split.pop() bitstream_file_dec = dec_split.pop() Loading Loading @@ -330,8 +383,12 @@ def test_param_file_tests( ref_tracefile_dec = f"{reference_path}/param_file/dec/{tracefile_dec}" # check for same RTP sequence number in last line of tracefile dut_rtp_num_last = np.genfromtxt(dut_tracefile_dec, delimiter=";", usecols=[0])[-1] ref_rtp_num_last = np.genfromtxt(ref_tracefile_dec, delimiter=";", usecols=[0])[-1] dut_rtp_num_last = np.genfromtxt( dut_tracefile_dec, delimiter=";", usecols=[0] )[-1] ref_rtp_num_last = np.genfromtxt( ref_tracefile_dec, delimiter=";", usecols=[0] )[-1] tracefile_last_rtp_numbers_differ = dut_rtp_num_last != ref_rtp_num_last # same sequence number -> likely no crash, assume length difference is due to difference in TSM Loading @@ -354,7 +411,9 @@ def test_param_file_tests( ) md_out_files = get_expected_md_files(ref_output_file, enc_split, output_config) props = parse_properties(reason, output_differs, props_to_record) cmp_result_msg += reason props = parse_properties(cmp_result_msg, output_differs, props_to_record) for k, v in props.items(): record_property(k, v) Loading Loading @@ -391,6 +450,9 @@ def test_param_file_tests( msg += "metadata only" pytest.fail(msg) if enc_test_result: pytest.fail("Too high difference in encoder statistics found.") # remove DUT output files when test result is OK (to save disk space) if not keep_files: os.remove(f"{dut_base_path}/param_file/dec/{output_file}") Loading Loading @@ -419,6 +481,7 @@ def encode( bitstream_file, enc_opts_list, update_ref, get_enc_stats=False, ): """ Call REF and/or DUT encoder. Loading @@ -430,8 +493,17 @@ def encode( ref_out_file = f"{ref_out_dir}/{bitstream_file}" dut_out_file = f"{dut_out_dir}/{bitstream_file}" if update_ref == 1 or update_ref == 2 and not os.path.exists(ref_out_file): if get_enc_stats: stats_file = bitstream_file.replace(".192", ".stats") ref_stats_file = f"{ref_out_dir}/{stats_file}" dut_stats_file = f"{dut_out_dir}/{stats_file}" else: ref_stats_file = None dut_stats_file = None if update_ref in [1, 2] and not os.path.exists(ref_out_file): check_and_makedir(ref_out_dir) # call REF encoder ref_encoder_frontend.run( bitrate, Loading @@ -439,10 +511,12 @@ def encode( testv_file, ref_out_file, add_option_list=enc_opts_list, stats_file=ref_stats_file, ) if update_ref in [0, 2]: check_and_makedir(dut_out_dir) # call DUT encoder dut_encoder_frontend.run( bitrate, Loading @@ -450,6 +524,7 @@ def encode( testv_file, dut_out_file, add_option_list=enc_opts_list, stats_file=dut_stats_file, ) Loading tests/codec_be_on_mr_nonselection/test_sba.py +607 −381 File changed.Preview size limit exceeded, changes collapsed. Show changes tests/conftest.py +207 −44 Original line number Diff line number Diff line Loading @@ -35,23 +35,37 @@ Pytest customization (configuration and fixtures) for the IVAS codec test suite. import logging import os import re import json from tests import testconfig import pytest import platform import textwrap from pathlib import Path from subprocess import TimeoutExpired, run import tempfile from typing import Optional, Union from .constants import MLD_PATTERN, MAX_DIFF_PATTERN, SSNR_PATTERN, ODG_PATTERN import numpy as np from .constants import ( MLD_PATTERN, MAX_DIFF_PATTERN, SSNR_PATTERN, ENC_AUX_FILES, ODG_PATTERN, MLD, MAX_ABS_DIFF, SSNR, ODG, MAX_ENC_DIFF, MAX_ENC_DIFF_PATTERN, SCRIPTS_DIR, ) logger = logging.getLogger(__name__) USE_LOGGER_FOR_DBG = False # current tests do not make use of the logger feature HERE = Path(__file__).parent SCRIPTS_DIR = str(HERE.parent.joinpath("scripts").absolute()) import sys sys.path.append(SCRIPTS_DIR) sys.path.append(str(SCRIPTS_DIR)) import prepare_combined_format_inputs Loading Loading @@ -180,6 +194,12 @@ def pytest_addoption(parser): help="Compute Segmental SNR (SSNR) between ref and dut output instead of just comparing for bitexactness", ) parser.addoption( "--enc_stats", action="store_true", help="Activate logging and comparison of statistics from the encoder.", ) parser.addoption( "--odg", action="store_true", Loading Loading @@ -219,6 +239,13 @@ def pytest_addoption(parser): default=20, ) parser.addoption( "--encoder_only", help="Only run encoder parts of tests in 'codec_be_on_mr_nonselection'.", action="store_true", default=False, ) parser.addoption( "--decoder_only", help="Only run decoder parts of tests in 'codec_be_on_mr_nonselection'. Use ref encoder output bitstreams.", Loading Loading @@ -278,6 +305,14 @@ def get_ssnr(request): return request.config.option.ssnr @pytest.fixture(scope="session", autouse=True) def get_enc_stats(request) -> bool: """ Return value of cmdl param --get_enc_stats """ return request.config.getoption("--enc_stats") @pytest.fixture(scope="session", autouse=True) def get_odg(request): """ Loading Loading @@ -338,16 +373,90 @@ def dut_encoder_path(request) -> str: class EncoderFrontend: def __init__(self, path, enc_type, timeout=None) -> None: self._path = path self._path = Path(path).absolute() self._type = enc_type self.returncode = None self.stdout = None self.stderr = None self.timeout = timeout def extract_enc_stats( self, dbg_tweak_folder, stats_file, sampling_rate, ): """ Extract statistics from auxiliary encoder files generated by running the encoder with DEBUG_MODE_INFO. Write the statistics to a text file """ hist_dicts = [] if not os.path.exists(dbg_tweak_folder): print(f"No statistics have been extracted from the res/ folder to the {stats_file} file!") else: for f in ENC_AUX_FILES: filename = f[0] dtype = f[1] fs = int(sampling_rate) * 1000 if isinstance(f[2], str): nsamples_per_frame = np.int16(eval(f[2])) else: nsamples_per_frame = np.int16(f[2]) # aux_files = glob.glob(os.path.join(dbg_tweak_folder, filename + '\.*')) aux_files = [ f for f in os.listdir(dbg_tweak_folder) if re.search(rf"^{filename}(\..*)?$", f) ] # aux_files = [os.path.basename(f) for f in aux_files] for aux_file in aux_files: # extract statistics from the aux file based on histogram of values print( f"Extracting statistics from {os.path.basename(aux_file)} ... ", end="", ) # read the aux file with open(os.path.join(dbg_tweak_folder, aux_file), "r") as f_aux: data = np.fromfile(f_aux, dtype=dtype) # get file length data_len = data.shape[0] # remove the duplicates of each value per frame if nsamples_per_frame > 1: data = data[::nsamples_per_frame] # calculate histogram from data unique_values = np.sort(np.unique(data)) hist, _ = np.histogram( data, bins=np.append(unique_values, unique_values[-1] + 10) ) # convert to dict and sort by absolute value of difference hist_dict = {"name": os.path.basename(aux_file), "length": data_len} dict_values = { str(unique_values[i]): str(hist[i]) for i in range(len(unique_values)) } hist_dict.update(dict_values) hist_dicts.append(hist_dict) print("DONE") print("") with open(stats_file, "w") as f_stats: # append the statistics to the output file in text format f_stats.write(json.dumps(hist_dicts, indent=2)) def run( self, bitrate: int, bitrate: Union[int, Path], input_sampling_rate: int, input_path: Path, output_bitstream_path: Path, Loading @@ -358,8 +467,9 @@ class EncoderFrontend: quiet_mode: Optional[bool] = True, add_option_list: Optional[list] = None, run_dir: Optional[Path] = None, stats_file: Optional[Path] = None, ) -> None: command = [self._path] command = [str(self._path)] # add optional parameters if sba_order is not None: Loading Loading @@ -392,13 +502,25 @@ class EncoderFrontend: log_dbg_msg(f"{self._type} encoder command:\n{cmd_str}") try: with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() result = run( command, capture_output=True, check=False, timeout=self.timeout, cwd=run_dir, cwd=cwd, ) if stats_file is not None: self.extract_enc_stats( cwd.joinpath("res"), stats_file, input_sampling_rate ) except TimeoutExpired: pytest.fail(f"{self._type} encoder run timed out after {self.timeout}s.") Loading Loading @@ -519,7 +641,7 @@ def dut_decoder_path(request) -> str: class DecoderFrontend: def __init__(self, path, dec_type, timeout=None, fr=20) -> None: self._path = path self._path = str(Path(path).absolute()) self._type = dec_type self.returncode = None self.stdout = None Loading Loading @@ -551,18 +673,18 @@ class DecoderFrontend: system = platform.system() if system == "Windows": eid_path = "./scripts/tools/Win32/eid-xor.exe" eid_path = SCRIPTS_DIR.joinpath("tools/Win32/eid-xor.exe") elif system == "Linux": eid_path = "./scripts/tools/Linux/eid-xor" eid_path = SCRIPTS_DIR.joinpath("tools/Linux/eid-xor") elif system == "Darwin": eid_path = "./scripts/tools/Darwin/eid-xor" eid_path = SCRIPTS_DIR.joinpath("tools/Darwin/eid-xor") else: raise ValueError(f'Wrong system "{system}"!') if not os.path.isfile(eid_path): if not eid_path.exists(): raise FileNotFoundError(f"eid-xor binary {eid_path} not found!\n") eid_command = [eid_path] eid_command = [str(eid_path)] eid_command.extend(["-fer", "-vbr", "-bs", "g192", "-ep", "g192"]) eid_output_suffix = ".fer" eid_command += [ Loading @@ -573,7 +695,12 @@ class DecoderFrontend: try: if not os.path.exists(str(input_bitstream_path) + eid_output_suffix): result = run(eid_command, check=True, cwd=run_dir) with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() result = run(eid_command, check=True, cwd=cwd) except Exception as e: pytest.fail(f"eid-xor operation failed! - {e}") Loading @@ -584,11 +711,13 @@ class DecoderFrontend: # TODO: centralize this in a utils file if system == "Windows": netsim_path = "./scripts/tools/Win32/networkSimulator_g192.exe" netsim_path = SCRIPTS_DIR.joinpath( "tools/Win32/networkSimulator_g192.exe" ) elif system == "Linux": netsim_path = "./scripts/tools/Linux/networkSimulator_g192" netsim_path = SCRIPTS_DIR.joinpath("tools/Linux/networkSimulator_g192") elif system == "Darwin": netsim_path = "./scripts/tools/Darwin/networkSimulator_g192" netsim_path = SCRIPTS_DIR.joinpath("tools/Darwin/networkSimulator_g192") else: raise ValueError(f'Wrong system "{system}"!') Loading @@ -609,7 +738,12 @@ class DecoderFrontend: ] print(netsim_command) try: run(netsim_command, check=True, cwd=run_dir) with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() run(netsim_command, check=True, cwd=cwd) except Exception as e: pytest.fail(f"netsim operation failed! - {e}") Loading Loading @@ -637,12 +771,17 @@ class DecoderFrontend: log_dbg_msg(f"{self._type} decoder command:\n{cmd_str}") try: with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() result = run( command, capture_output=True, check=False, timeout=self.timeout, cwd=run_dir, cwd=cwd, ) except TimeoutExpired: pytest.fail(f"{self._type} decoder run timed out after {self.timeout}s.") Loading Loading @@ -822,6 +961,14 @@ def decoder_only(request) -> bool: return request.config.getoption("--decoder_only") @pytest.fixture(scope="session", autouse=True) def encoder_only(request) -> bool: """ Return value of cmdl param --encoder_only """ return request.config.getoption("--encoder_only") def pytest_configure(config): config.addinivalue_line("markers", "serial: mark test to run only in serial") if config.option.param_file: Loading @@ -841,14 +988,22 @@ def pytest_configure(config): @pytest.fixture(scope="session") def props_to_record(request, get_mld, get_ssnr, get_odg) -> str: props = ["MAXIMUM ABS DIFF"] def props_to_record( request, get_mld, get_ssnr, get_odg, get_enc_stats, encoder_only ) -> str: props = [] if get_enc_stats: props.append(MAX_ENC_DIFF) if not encoder_only: props.append(MAX_ABS_DIFF) if get_mld: props.append("MLD") props.append(MLD) if get_ssnr: props.append("SSNR") props.append(SSNR) if get_odg: props.append("ODG") props.append(ODG) return props Loading @@ -861,10 +1016,10 @@ def parse_properties(text_to_parse: str, output_differs: bool, props_to_record: props = dict() for prop in props_to_record: if prop == "MLD": if prop == MLD: mld = float(re.search(MLD_PATTERN, text_to_parse).groups(1)[0]) props[prop] = mld elif prop == "MAXIMUM ABS DIFF": elif prop == MAX_ABS_DIFF: max_diff = 0 if output_differs: if (match := re.search(MAX_DIFF_PATTERN, text_to_parse)) is not None: Loading @@ -872,18 +1027,26 @@ def parse_properties(text_to_parse: str, output_differs: bool, props_to_record: else: raise MaxDiffPatternNotFound() props[prop] = max_diff elif prop == "SSNR": elif prop == SSNR: ssnrs = re.findall(SSNR_PATTERN, text_to_parse) min_ssnr = min(ssnrs) min_ssnr_channel = ssnrs.index(min_ssnr) props["MIN_SSNR"] = min_ssnr props["MIN_SSNR_CHANNEL"] = min_ssnr_channel elif prop == "ODG": elif prop == ODG: odgs = re.findall(ODG_PATTERN, text_to_parse) min_odg = min(odgs) min_odg_channel = odgs.index(min_odg) props["MIN_ODG"] = min_odg props["MIN_ODG_CHANNEL"] = min_odg_channel elif prop == MAX_ENC_DIFF: search_result = re.search(MAX_ENC_DIFF_PATTERN, text_to_parse) max_enc_diff = 0 if search_result: max_enc_diff, max_enc_diff_ratio = search_result.groups(0) if max_enc_diff: max_enc_diff = float(max_enc_diff) props[MAX_ENC_DIFF] = max_enc_diff return props Loading Loading
ci/remove_unsupported_testcases.py +4 −0 Original line number Diff line number Diff line Loading @@ -61,9 +61,13 @@ TESTCASES = [ "Multi-channel 5_1 at 512 kbps, 48kHz in 48kHz out, BINAURAL_ROOM_REVERB out custom acoustic environment with a sequence (CREND)", "Multi-channel 5_1 at 64 kbps, 48kHz in 48kHz out, BINAURAL_ROOM_REVERB out custom acoustic environment with a sequence (FastConv)", "Multi-channel 5_1 at 32 kbps, 48kHz in 48kHz out, BINAURAL_ROOM_REVERB out custom acoustic environment with a sequence (ParamBin)", "MASA 1dir 1TC at 256 kbps, 48kHz in, 48kHz out, BINAURAL_ROOM_REVERB out, HR custom configuration", "MASA 1dir 1 TC at 256kbps, 48kHz in, 48 kHz out, BINAURAL_ROOM_REVERB out custom configuration", "MASA 1dir 1TC at 256kbps, 48kHz in, 48 kHz out, BINAURAL_ROOM_REVERB out custom configuration", ] def remove_testcases(cfg: Path, testcases: list): """ Go through file line by line and copy all testcases except the given ones Loading
tests/cmp_stats_files.py 0 → 100644 +146 −0 Original line number Diff line number Diff line #!/usr/bin/env python3 import argparse import os import sys import json import numpy as np def str2num(s): """ Convert string either to integer or float """ try: return int(s) except ValueError: return float(s) def cmp_stats_files( ref_stats_file, dut_stats_file, min_enc_file_length_diff=0.1, min_enc_stats_diff=0.1 ) -> (int, str): """ Compare two .stats files containing encoder statistics (extracted from binary files) """ print(f"Comparing {os.path.basename(ref_stats_file)} between Ref and Dut ... \n") # open and read the .stats files with open(ref_stats_file, "r") as f_aux: ref_stats = json.load(f_aux) # create dictionary to map "name" to the corresponding dictionaries ref_stats_names = {d["name"]: d for d in ref_stats} with open(dut_stats_file, "r") as f_aux: dut_stats = json.load(f_aux) # create dictionary to map "name" to the corresponding dictionaries dut_stats_names = {d["name"]: d for d in dut_stats} # loop over all common aux files enc_test_result = 0 enc_test_result_msg = "" max_total_num_diff = 0 max_total_num_diff_ratio = 0 for name in ref_stats_names: if name in dut_stats_names: # retrieve the dictionaries ref_stats_dict = ref_stats_names[name] dut_stats_dict = dut_stats_names[name] msg = f"File {name}" # compare the file lengths result_len_check = 0 file_length = max(ref_stats_dict["length"], dut_stats_dict["length"]) if ref_stats_dict["length"] != dut_stats_dict["length"]: msg += f" has different length between Ref {ref_stats_dict['length']} and DuT {dut_stats_dict['length']}" # check if threshold has been exceeded if ( abs(ref_stats_dict["length"] - dut_stats_dict["length"]) / file_length > min_enc_file_length_diff ): result_len_check = 1 msg += ", " # remove the "name" and "length" keys for further processing del ref_stats_dict["name"] del dut_stats_dict["name"] del ref_stats_dict["length"] del dut_stats_dict["length"] # convert keys and values from string to float ref_hist = {str2num(i): str2num(j) for i, j in ref_stats_dict.items()} cut_hist = {str2num(i): str2num(j) for i, j in dut_stats_dict.items()} delta_ref = set(cut_hist) - set(ref_hist) delta_cut = set(ref_hist) - set(cut_hist) # append missing keys for item in delta_cut: cut_hist[item] = 0 for item in delta_ref: ref_hist[item] = 0 ref_hist = dict(sorted(ref_hist.items())) cut_hist = dict(sorted(cut_hist.items())) # caculate difference of statistics diff_hist = {k: cut_hist[k] - ref_hist[k] for k in ref_hist.keys()} # calculate the total number of differences total_num_diff = sum(np.abs(list(diff_hist.values()))) total_num_diff_ratio = total_num_diff / ( sum(ref_hist.values()) + sum(cut_hist.values()) ) msg += f"the total number of differences is {total_num_diff} ({(total_num_diff_ratio*100):.2f}%)" if total_num_diff_ratio > min_enc_stats_diff: result_diff_check = 1 msg += "! " else: result_diff_check = 0 msg += ". " # check if the maximum difference has been exceeded if total_num_diff_ratio > max_total_num_diff_ratio: max_total_num_diff = total_num_diff max_total_num_diff_ratio = total_num_diff_ratio # update test result if result_len_check or result_diff_check: enc_test_result = 1 enc_test_result_msg += msg print(msg) if enc_test_result and max_total_num_diff > 0: msg = f"MAXIMUM ENC DIFF: {max_total_num_diff} ({(max_total_num_diff_ratio*100):.2f}%) " enc_test_result_msg += msg print(msg) return enc_test_result, enc_test_result_msg if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("ref_stats_file", type=str) parser.add_argument("dut_stats_file", type=str) parser.add_argument( "--min_enc_file_length_diff", type=float, default=0.1, dest="min_enc_file_length_diff", ) parser.add_argument( "--min_enc_stats_diff", type=float, default=0.1, dest="min_enc_stats_diff" ) args = parser.parse_args() enc_test_result, enc_test_result_msg = cmp_stats_files(**vars(args)) print(enc_test_result_msg) sys.exit(enc_test_result)
tests/codec_be_on_mr_nonselection/test_param_file.py +80 −5 Original line number Diff line number Diff line Loading @@ -42,8 +42,15 @@ import pytest import numpy as np from tests.cmp_pcm import cmp_pcm from tests.cmp_stats_files import cmp_stats_files from tests.conftest import DecoderFrontend, EncoderFrontend, parse_properties from tests.testconfig import PARAM_FILE from tests.constants import ( MIN_ENC_FILE_LENGTH_DIFF, MIN_ENC_STATS_DIFF, SCRIPTS_DIR, ) VALID_DEC_OUTPUT_CONF = [ "MONO", Loading Loading @@ -134,6 +141,7 @@ def convert_test_string_to_tag(test_string): def test_param_file_tests( record_property, props_to_record, encoder_only, decoder_only, dut_encoder_frontend: EncoderFrontend, dut_decoder_frontend: DecoderFrontend, Loading @@ -151,6 +159,7 @@ def test_param_file_tests( get_mld_lim, abs_tol, get_ssnr, get_enc_stats, get_odg, ): enc_opts, dec_opts, sim_opts, eid_opts = param_file_test_dict[test_tag] Loading @@ -175,7 +184,7 @@ def test_param_file_tests( # bitrate can be a filename: remove leading "../" if bitrate.startswith("../"): bitrate = bitrate[3:] bitrate = Path(bitrate[3:]).absolute() testv_base = testv_file.split("/")[-1] if testv_base.endswith(".pcm"): Loading @@ -186,6 +195,9 @@ def test_param_file_tests( # -> construct bitstream filename bitstream_file = f"{testv_base}_{tag_str}.192" cmp_result_msg = "" enc_test_result = 0 if not decoder_only: encode( dut_encoder_frontend, Loading @@ -198,8 +210,42 @@ def test_param_file_tests( bitstream_file, enc_split, update_ref, get_enc_stats, ) # compare binary files extracted from the encoder if update_ref in [0, 2] and get_enc_stats: print("Comparing encoder files") print("=======================\n") stats_file = bitstream_file.replace(".192", ".stats") ref_stats_file = f"{reference_path}/param_file/enc/{stats_file}" dut_stats_file = f"{dut_base_path}/param_file/enc/{stats_file}" # compare ref and dut .stats files enc_test_result, enc_test_result_msg = cmp_stats_files( ref_stats_file, dut_stats_file, min_enc_file_length_diff=MIN_ENC_FILE_LENGTH_DIFF, min_enc_stats_diff=MIN_ENC_STATS_DIFF, ) print("") cmp_result_msg += enc_test_result_msg props = parse_properties(cmp_result_msg, False, props_to_record) for k, v in props.items(): record_property(k, v) if enc_test_result: pytest.fail("Too high difference in encoder statistics found.") else: # remove DUT stats file when test result is OK (to save disk space) if not keep_files: os.remove(dut_stats_file) if encoder_only: return # don't proceed with the decoder if user specified --encoder_only on the command line # check for networkSimulator_g192 command line if sim_opts != "": sim_split = sim_opts.split() Loading Loading @@ -232,6 +278,7 @@ def test_param_file_tests( ) # check for eid-xor command line if eid_opts != "": eid_split = eid_opts.split() assert len(eid_split) >= 3, "eid-xor expects at least 3 parameters" Loading @@ -248,6 +295,7 @@ def test_param_file_tests( # -> construct netsim output file name eid_xor_outfile = f"{testv_base}_{tag_str}.fer.192" eid_split[-1] = eid_xor_outfile error_insertion( reference_path, dut_base_path, Loading @@ -268,6 +316,11 @@ def test_param_file_tests( ] # remove leading "../" dec_split = [x[3:] if x.startswith("../") else x for x in dec_split] # convert "scripts/" paths into absolute ones dec_split = [ str(SCRIPTS_DIR.joinpath(x[8:])) if x.startswith("scripts/") else x for x in dec_split ] output_file = dec_split.pop() bitstream_file_dec = dec_split.pop() Loading Loading @@ -330,8 +383,12 @@ def test_param_file_tests( ref_tracefile_dec = f"{reference_path}/param_file/dec/{tracefile_dec}" # check for same RTP sequence number in last line of tracefile dut_rtp_num_last = np.genfromtxt(dut_tracefile_dec, delimiter=";", usecols=[0])[-1] ref_rtp_num_last = np.genfromtxt(ref_tracefile_dec, delimiter=";", usecols=[0])[-1] dut_rtp_num_last = np.genfromtxt( dut_tracefile_dec, delimiter=";", usecols=[0] )[-1] ref_rtp_num_last = np.genfromtxt( ref_tracefile_dec, delimiter=";", usecols=[0] )[-1] tracefile_last_rtp_numbers_differ = dut_rtp_num_last != ref_rtp_num_last # same sequence number -> likely no crash, assume length difference is due to difference in TSM Loading @@ -354,7 +411,9 @@ def test_param_file_tests( ) md_out_files = get_expected_md_files(ref_output_file, enc_split, output_config) props = parse_properties(reason, output_differs, props_to_record) cmp_result_msg += reason props = parse_properties(cmp_result_msg, output_differs, props_to_record) for k, v in props.items(): record_property(k, v) Loading Loading @@ -391,6 +450,9 @@ def test_param_file_tests( msg += "metadata only" pytest.fail(msg) if enc_test_result: pytest.fail("Too high difference in encoder statistics found.") # remove DUT output files when test result is OK (to save disk space) if not keep_files: os.remove(f"{dut_base_path}/param_file/dec/{output_file}") Loading Loading @@ -419,6 +481,7 @@ def encode( bitstream_file, enc_opts_list, update_ref, get_enc_stats=False, ): """ Call REF and/or DUT encoder. Loading @@ -430,8 +493,17 @@ def encode( ref_out_file = f"{ref_out_dir}/{bitstream_file}" dut_out_file = f"{dut_out_dir}/{bitstream_file}" if update_ref == 1 or update_ref == 2 and not os.path.exists(ref_out_file): if get_enc_stats: stats_file = bitstream_file.replace(".192", ".stats") ref_stats_file = f"{ref_out_dir}/{stats_file}" dut_stats_file = f"{dut_out_dir}/{stats_file}" else: ref_stats_file = None dut_stats_file = None if update_ref in [1, 2] and not os.path.exists(ref_out_file): check_and_makedir(ref_out_dir) # call REF encoder ref_encoder_frontend.run( bitrate, Loading @@ -439,10 +511,12 @@ def encode( testv_file, ref_out_file, add_option_list=enc_opts_list, stats_file=ref_stats_file, ) if update_ref in [0, 2]: check_and_makedir(dut_out_dir) # call DUT encoder dut_encoder_frontend.run( bitrate, Loading @@ -450,6 +524,7 @@ def encode( testv_file, dut_out_file, add_option_list=enc_opts_list, stats_file=dut_stats_file, ) Loading
tests/codec_be_on_mr_nonselection/test_sba.py +607 −381 File changed.Preview size limit exceeded, changes collapsed. Show changes
tests/conftest.py +207 −44 Original line number Diff line number Diff line Loading @@ -35,23 +35,37 @@ Pytest customization (configuration and fixtures) for the IVAS codec test suite. import logging import os import re import json from tests import testconfig import pytest import platform import textwrap from pathlib import Path from subprocess import TimeoutExpired, run import tempfile from typing import Optional, Union from .constants import MLD_PATTERN, MAX_DIFF_PATTERN, SSNR_PATTERN, ODG_PATTERN import numpy as np from .constants import ( MLD_PATTERN, MAX_DIFF_PATTERN, SSNR_PATTERN, ENC_AUX_FILES, ODG_PATTERN, MLD, MAX_ABS_DIFF, SSNR, ODG, MAX_ENC_DIFF, MAX_ENC_DIFF_PATTERN, SCRIPTS_DIR, ) logger = logging.getLogger(__name__) USE_LOGGER_FOR_DBG = False # current tests do not make use of the logger feature HERE = Path(__file__).parent SCRIPTS_DIR = str(HERE.parent.joinpath("scripts").absolute()) import sys sys.path.append(SCRIPTS_DIR) sys.path.append(str(SCRIPTS_DIR)) import prepare_combined_format_inputs Loading Loading @@ -180,6 +194,12 @@ def pytest_addoption(parser): help="Compute Segmental SNR (SSNR) between ref and dut output instead of just comparing for bitexactness", ) parser.addoption( "--enc_stats", action="store_true", help="Activate logging and comparison of statistics from the encoder.", ) parser.addoption( "--odg", action="store_true", Loading Loading @@ -219,6 +239,13 @@ def pytest_addoption(parser): default=20, ) parser.addoption( "--encoder_only", help="Only run encoder parts of tests in 'codec_be_on_mr_nonselection'.", action="store_true", default=False, ) parser.addoption( "--decoder_only", help="Only run decoder parts of tests in 'codec_be_on_mr_nonselection'. Use ref encoder output bitstreams.", Loading Loading @@ -278,6 +305,14 @@ def get_ssnr(request): return request.config.option.ssnr @pytest.fixture(scope="session", autouse=True) def get_enc_stats(request) -> bool: """ Return value of cmdl param --get_enc_stats """ return request.config.getoption("--enc_stats") @pytest.fixture(scope="session", autouse=True) def get_odg(request): """ Loading Loading @@ -338,16 +373,90 @@ def dut_encoder_path(request) -> str: class EncoderFrontend: def __init__(self, path, enc_type, timeout=None) -> None: self._path = path self._path = Path(path).absolute() self._type = enc_type self.returncode = None self.stdout = None self.stderr = None self.timeout = timeout def extract_enc_stats( self, dbg_tweak_folder, stats_file, sampling_rate, ): """ Extract statistics from auxiliary encoder files generated by running the encoder with DEBUG_MODE_INFO. Write the statistics to a text file """ hist_dicts = [] if not os.path.exists(dbg_tweak_folder): print(f"No statistics have been extracted from the res/ folder to the {stats_file} file!") else: for f in ENC_AUX_FILES: filename = f[0] dtype = f[1] fs = int(sampling_rate) * 1000 if isinstance(f[2], str): nsamples_per_frame = np.int16(eval(f[2])) else: nsamples_per_frame = np.int16(f[2]) # aux_files = glob.glob(os.path.join(dbg_tweak_folder, filename + '\.*')) aux_files = [ f for f in os.listdir(dbg_tweak_folder) if re.search(rf"^{filename}(\..*)?$", f) ] # aux_files = [os.path.basename(f) for f in aux_files] for aux_file in aux_files: # extract statistics from the aux file based on histogram of values print( f"Extracting statistics from {os.path.basename(aux_file)} ... ", end="", ) # read the aux file with open(os.path.join(dbg_tweak_folder, aux_file), "r") as f_aux: data = np.fromfile(f_aux, dtype=dtype) # get file length data_len = data.shape[0] # remove the duplicates of each value per frame if nsamples_per_frame > 1: data = data[::nsamples_per_frame] # calculate histogram from data unique_values = np.sort(np.unique(data)) hist, _ = np.histogram( data, bins=np.append(unique_values, unique_values[-1] + 10) ) # convert to dict and sort by absolute value of difference hist_dict = {"name": os.path.basename(aux_file), "length": data_len} dict_values = { str(unique_values[i]): str(hist[i]) for i in range(len(unique_values)) } hist_dict.update(dict_values) hist_dicts.append(hist_dict) print("DONE") print("") with open(stats_file, "w") as f_stats: # append the statistics to the output file in text format f_stats.write(json.dumps(hist_dicts, indent=2)) def run( self, bitrate: int, bitrate: Union[int, Path], input_sampling_rate: int, input_path: Path, output_bitstream_path: Path, Loading @@ -358,8 +467,9 @@ class EncoderFrontend: quiet_mode: Optional[bool] = True, add_option_list: Optional[list] = None, run_dir: Optional[Path] = None, stats_file: Optional[Path] = None, ) -> None: command = [self._path] command = [str(self._path)] # add optional parameters if sba_order is not None: Loading Loading @@ -392,13 +502,25 @@ class EncoderFrontend: log_dbg_msg(f"{self._type} encoder command:\n{cmd_str}") try: with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() result = run( command, capture_output=True, check=False, timeout=self.timeout, cwd=run_dir, cwd=cwd, ) if stats_file is not None: self.extract_enc_stats( cwd.joinpath("res"), stats_file, input_sampling_rate ) except TimeoutExpired: pytest.fail(f"{self._type} encoder run timed out after {self.timeout}s.") Loading Loading @@ -519,7 +641,7 @@ def dut_decoder_path(request) -> str: class DecoderFrontend: def __init__(self, path, dec_type, timeout=None, fr=20) -> None: self._path = path self._path = str(Path(path).absolute()) self._type = dec_type self.returncode = None self.stdout = None Loading Loading @@ -551,18 +673,18 @@ class DecoderFrontend: system = platform.system() if system == "Windows": eid_path = "./scripts/tools/Win32/eid-xor.exe" eid_path = SCRIPTS_DIR.joinpath("tools/Win32/eid-xor.exe") elif system == "Linux": eid_path = "./scripts/tools/Linux/eid-xor" eid_path = SCRIPTS_DIR.joinpath("tools/Linux/eid-xor") elif system == "Darwin": eid_path = "./scripts/tools/Darwin/eid-xor" eid_path = SCRIPTS_DIR.joinpath("tools/Darwin/eid-xor") else: raise ValueError(f'Wrong system "{system}"!') if not os.path.isfile(eid_path): if not eid_path.exists(): raise FileNotFoundError(f"eid-xor binary {eid_path} not found!\n") eid_command = [eid_path] eid_command = [str(eid_path)] eid_command.extend(["-fer", "-vbr", "-bs", "g192", "-ep", "g192"]) eid_output_suffix = ".fer" eid_command += [ Loading @@ -573,7 +695,12 @@ class DecoderFrontend: try: if not os.path.exists(str(input_bitstream_path) + eid_output_suffix): result = run(eid_command, check=True, cwd=run_dir) with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() result = run(eid_command, check=True, cwd=cwd) except Exception as e: pytest.fail(f"eid-xor operation failed! - {e}") Loading @@ -584,11 +711,13 @@ class DecoderFrontend: # TODO: centralize this in a utils file if system == "Windows": netsim_path = "./scripts/tools/Win32/networkSimulator_g192.exe" netsim_path = SCRIPTS_DIR.joinpath( "tools/Win32/networkSimulator_g192.exe" ) elif system == "Linux": netsim_path = "./scripts/tools/Linux/networkSimulator_g192" netsim_path = SCRIPTS_DIR.joinpath("tools/Linux/networkSimulator_g192") elif system == "Darwin": netsim_path = "./scripts/tools/Darwin/networkSimulator_g192" netsim_path = SCRIPTS_DIR.joinpath("tools/Darwin/networkSimulator_g192") else: raise ValueError(f'Wrong system "{system}"!') Loading @@ -609,7 +738,12 @@ class DecoderFrontend: ] print(netsim_command) try: run(netsim_command, check=True, cwd=run_dir) with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() run(netsim_command, check=True, cwd=cwd) except Exception as e: pytest.fail(f"netsim operation failed! - {e}") Loading Loading @@ -637,12 +771,17 @@ class DecoderFrontend: log_dbg_msg(f"{self._type} decoder command:\n{cmd_str}") try: with tempfile.TemporaryDirectory() as tmp_dir: if run_dir is None: cwd = Path(tmp_dir).absolute() else: cwd = Path(run_dir).absolute() result = run( command, capture_output=True, check=False, timeout=self.timeout, cwd=run_dir, cwd=cwd, ) except TimeoutExpired: pytest.fail(f"{self._type} decoder run timed out after {self.timeout}s.") Loading Loading @@ -822,6 +961,14 @@ def decoder_only(request) -> bool: return request.config.getoption("--decoder_only") @pytest.fixture(scope="session", autouse=True) def encoder_only(request) -> bool: """ Return value of cmdl param --encoder_only """ return request.config.getoption("--encoder_only") def pytest_configure(config): config.addinivalue_line("markers", "serial: mark test to run only in serial") if config.option.param_file: Loading @@ -841,14 +988,22 @@ def pytest_configure(config): @pytest.fixture(scope="session") def props_to_record(request, get_mld, get_ssnr, get_odg) -> str: props = ["MAXIMUM ABS DIFF"] def props_to_record( request, get_mld, get_ssnr, get_odg, get_enc_stats, encoder_only ) -> str: props = [] if get_enc_stats: props.append(MAX_ENC_DIFF) if not encoder_only: props.append(MAX_ABS_DIFF) if get_mld: props.append("MLD") props.append(MLD) if get_ssnr: props.append("SSNR") props.append(SSNR) if get_odg: props.append("ODG") props.append(ODG) return props Loading @@ -861,10 +1016,10 @@ def parse_properties(text_to_parse: str, output_differs: bool, props_to_record: props = dict() for prop in props_to_record: if prop == "MLD": if prop == MLD: mld = float(re.search(MLD_PATTERN, text_to_parse).groups(1)[0]) props[prop] = mld elif prop == "MAXIMUM ABS DIFF": elif prop == MAX_ABS_DIFF: max_diff = 0 if output_differs: if (match := re.search(MAX_DIFF_PATTERN, text_to_parse)) is not None: Loading @@ -872,18 +1027,26 @@ def parse_properties(text_to_parse: str, output_differs: bool, props_to_record: else: raise MaxDiffPatternNotFound() props[prop] = max_diff elif prop == "SSNR": elif prop == SSNR: ssnrs = re.findall(SSNR_PATTERN, text_to_parse) min_ssnr = min(ssnrs) min_ssnr_channel = ssnrs.index(min_ssnr) props["MIN_SSNR"] = min_ssnr props["MIN_SSNR_CHANNEL"] = min_ssnr_channel elif prop == "ODG": elif prop == ODG: odgs = re.findall(ODG_PATTERN, text_to_parse) min_odg = min(odgs) min_odg_channel = odgs.index(min_odg) props["MIN_ODG"] = min_odg props["MIN_ODG_CHANNEL"] = min_odg_channel elif prop == MAX_ENC_DIFF: search_result = re.search(MAX_ENC_DIFF_PATTERN, text_to_parse) max_enc_diff = 0 if search_result: max_enc_diff, max_enc_diff_ratio = search_result.groups(0) if max_enc_diff: max_enc_diff = float(max_enc_diff) props[MAX_ENC_DIFF] = max_enc_diff return props Loading