Merge branch 'ci/ssnr' into ci/check-ssnr-in-basop-ci-branch

  script:

    - *print-common-info

    - *update-ltv-repo

    # Temporary fix to test only SID start config files in this job

    # this rm makes check-for-testvectors only check for the signals we actually need in this test

    - rm scripts/config/ci_linux_ltv.json scripts/config/ci_linux.json

    - *check-for-testvectors

    - cmake .

    - make -j

    - exit_code_no_sba=0

    - exit_code_hoa=0

    - exit_code_foa=0

    # run all modes and cut bitstream to start with an SID. Use stereo output to limit runtime, test is only about decoding

    - ism_md_cmd="--ism_metadata_files /usr/local/ltv/ltvISM1.csv /usr/local/ltv/ltvISM2.csv /usr/local/ltv/ltvISM3.csv /usr/local/ltv/ltvISM4.csv"

    - modes=$(scripts/runIvasCodec.py -l | grep dtx | grep -vE "FOA|HOA" )

    - scripts/runIvasCodec.py -z console -p scripts/config/ci_linux_sidstart_test.json -m $modes -s --bs_length 200 -U 0:20 --oc stereo $ism_md_cmd --timeout 10 || exit_code_no_sba=$?

    - modes=$(scripts/runIvasCodec.py -l | grep dtx | grep -E "HOA")

    - scripts/runIvasCodec.py -z console -p scripts/config/ci_linux_sidstart_test.json -m $modes -s --bs_length 100 -U 70:80 --oc stereo --timeout 10 || exit_code_hoa=$?

    - modes=$(scripts/runIvasCodec.py -l | grep dtx | grep "FOA")

    - scripts/runIvasCodec.py -z console -p scripts/config/ci_linux_sidstart_test.json -m $modes -s --bs_length 100 -U 75:110 --oc stereo --timeout 10 || exit_code_foa=$?

    - bash ci/run-first-frame-is-sid-test.sh

    - if [ $exit_code_no_sba -ne 0 ] || [ $exit_code_hoa -ne 0 ] || [ $exit_code_foa -ne 0 ]; then exit 1; fi

  artifacts:

    paths:

      - out/logs

      - logs_enc

      - logs_dec_msan

      - logs_dec_asan

      - logs_dec_usan

    when: always

    name: "$CI_JOB_NAME--$CI_MERGE_REQUEST_IID--sha-$CI_COMMIT_SHORT_SHA--sidstart"

    expose_as: "logs-sidstart"

# expected columns. actual columns are filtered from the incoming data later, this

# is mainly for controlling the order in the output table

COLUMNS = ["testcase", "Result", "MLD", "MAXIMUM ABS DIFF"]

COLUMNS = ["testcase", "Result", "MLD", "MAXIMUM ABS DIFF", "MIN_SSNR"]

COLUMNS_GLOBAL = COLUMNS[:1]

COLUMNS_DIFFERENTIAL = COLUMNS[1:]

COLUMNS_DIFFERENTIAL_NOT_MLD = COLUMNS_DIFFERENTIAL[2:]

def create_subpage(

    html_out,

    csv_out,

    )

    write_out_csv(merged_reports, merged_reports[0].keys(), csv_out)

    table_header_a = "".join([TH_TMPL_GLOBAL.format(c) for c in COLUMNS_GLOBAL] + [TH_TMPL_DIFFERENTIAL.format(c) for c in COLUMNS_DIFFERENTIAL])

    table_header_a = "".join(

        [TH_TMPL_GLOBAL.format(c) for c in COLUMNS_GLOBAL]

        + [TH_TMPL_DIFFERENTIAL.format(c) for c in COLUMNS_DIFFERENTIAL]

    )

    table_header_b = list()

    for c in COLUMNS_DIFFERENTIAL:

        table_header_b.append(TH_TMPL_SECOND_ROW.format(f"Previous Run<br>ID: {id_previous}"))

        table_header_b.append(TH_TMPL_SECOND_ROW.format(f"Current Run<br>ID: {id_current}"))

        table_header_b.append(

            TH_TMPL_SECOND_ROW.format(f"Previous Run<br>ID: {id_previous}")

        )

        table_header_b.append(

            TH_TMPL_SECOND_ROW.format(f"Current Run<br>ID: {id_current}")

        )

    table_header_b = "".join(table_header_b)

    table_body = "\n".join(

        tr_from_row(row, id_current, id_previous) for row in merged_reports

        return diff

    other_col_pairs = [(f"{col}-{id_previous}", f"{col}-{id_current}") for col in COLUMNS_DIFFERENTIAL_NOT_MLD]

    merged = sorted(merged, key=partial(sort_func, other_col_pairs=other_col_pairs), reverse=True)

    other_col_pairs = [

        (f"{col}-{id_previous}", f"{col}-{id_current}")

        for col in COLUMNS_DIFFERENTIAL_NOT_MLD

    ]

    merged = sorted(

        merged, key=partial(sort_func, other_col_pairs=other_col_pairs), reverse=True

    )

    # remove the unecessary whole path from the testcase names

    for row in merged:

#! /usr/bin/bash

# build encoder without sanitizers for faster runtime

make clean

make -j IVAS_cod

mv IVAS_cod IVAS_cod_nosan

# run all modes and cut bitstream to start with an SID. Use mono output to limit runtime, test is only about decoding the first frame

modes_no_sba=$(scripts/runIvasCodec.py -l | grep dtx | grep -vE "FOA|HOA" )

modes_hoa=$(scripts/runIvasCodec.py -l | grep dtx | grep -E "HOA")

modes_foa=$(scripts/runIvasCodec.py -l | grep dtx | grep "FOA")

# config vars

testcase_timeout=20

bitstream_cut_length=5

common_args="-z console -p scripts/config/ci_linux_sidstart_test.json -s --oc mono --timeout $testcase_timeout --bs_length $bitstream_cut_length"

# first encoder + MSAN decoder

# hack to use the encoder with no sanitizers

mkdir CLANG1

make clean

make IVAS_dec -j CLANG=1

cp IVAS_dec CLANG1/IVAS_dec

cp IVAS_cod_nosan CLANG1/IVAS_cod

exit_code_msan=0

echo "-------------- 1. Encoder + Msan decoder -------------- "

echo "-------------- 1.1 all DTX modes except SBA -------------- "

scripts/IvasBuildAndRunChecks.py --checks CLANG1 -m $modes_no_sba -U 0:20 $common_args || exit_code_msan=$?

echo "-------------- 1.2 HOA2 + HOA3 DTX modes -------------- "

scripts/IvasBuildAndRunChecks.py --checks CLANG1 -m $modes_hoa -U 70:80 $common_args || exit_code_msan=$?

echo "-------------- 1.3 FOA DTX modes -------------- "

scripts/IvasBuildAndRunChecks.py --checks CLANG1 -m $modes_foa -U 75:110 $common_args || exit_code_msan=$?

# archive encoder logs separately

mkdir logs_enc logs_dec_msan

mv CLANG1/logs/*.enc.txt logs_enc/

mv CLANG1/logs/*.dec*.txt logs_dec_msan/

# ASAN and USAN can be done in one go and decoder only

# copy encoder output from CLANG1 dir

mkdir CLANG2 CLANG3

cp -r CLANG1/enc CLANG2/enc

cp -r CLANG1/enc CLANG3/enc

exit_code_asan_usan=0

echo "-------------- 2. Asan + Usan decoder -------------- "

echo "-------------- 2.1 all DTX modes except SBA -------------- "

scripts/IvasBuildAndRunChecks.py --checks CLANG2 CLANG3 --decoder_only -m $modes_no_sba -U 0:20 $common_args || exit_code_asan_usan=$?

echo "-------------- 2.2 HOA2 + HOA3 DTX modes -------------- "

scripts/IvasBuildAndRunChecks.py --checks CLANG2 CLANG3 --decoder_only -m $modes_hoa -U 70:80 $common_args || exit_code_asan_usan=$?

echo "-------------- 2.3 FOA DTX modes -------------- "

scripts/IvasBuildAndRunChecks.py --checks CLANG2 CLANG3 --decoder_only -m $modes_foa -U 75:110 $common_args || exit_code_asan_usan=$?

mv CLANG2/logs logs_dec_asan

mv CLANG3/logs logs_dec_usan

if [ $exit_code_msan -ne 0 ] || [ $exit_code_asan_usan -ne 0 ]; then echo "There was either a crash or a sanitizer error encountered when decoding a bitstream that starts with an SID. Check the artifacts for the logfiles."; exit 1; fi

Parse a junit report and create an MLD summary report.

"""

PROPERTIES = ["MLD", "MAXIMUM ABS DIFF"]

PROPERTIES = ["MLD", "MAXIMUM ABS DIFF", "MIN_SSNR"]

# Main routine

"""

import logging

import warnings

import math

import multiprocessing as mp

import platform

    fs: int,

    per_frame: bool = True,

    get_mld: bool = False,

    get_ssnr: bool = False,

    ssnr_thresh_low: float = -np.inf,

    ssnr_thresh_high: float = np.inf,

) -> dict:

    """Compare two audio arrays

        "first_diff_pos_sample": -1,

        "first_diff_pos_channel": -1,

        "first_diff_pos_frame": -1,

        "MLD": 0 if get_mld else None,

    }

    if get_mld:

        result["MLD"] = 0

    if get_ssnr:

        result["SSNR"] = np.asarray([np.inf] * ref.shape[1])

    if per_frame:

        result["max_abs_diff_pos_frame"] = 0

        result["nframes_diff"] = 0

            result["nframes_diff_percentage"] = nframes_diff_percentage

        if get_mld:

            mld_max = 0

            toolsdir = Path(__file__).parent.parent.joinpath("tools")

            if platform.system() == "Windows":

                for i in range(nchannels):

                    tmpfile_ref = Path(tmpdir).joinpath(f"ref_ch{i+1}.wav")

                    tmpfile_test = Path(tmpdir).joinpath(f"test_ch{i+1}.wav")

                    r48 = np.clip( resample(ref[:, i].astype(float), fs, 48000), -32768, 32767 ).astype(np.int16) # Convert to float for resample, then to int16 for wavfile.write

                    t48 = np.clip( resample(test[:, i].astype(float), fs, 48000), -32768, 32767 ).astype(np.int16)

                    r48 = np.clip(

                        resample(ref[:, i].astype(float), fs, 48000), -32768, 32767

                    ).astype(

                        np.int16

                    )  # Convert to float for resample, then to int16 for wavfile.write

                    t48 = np.clip(

                        resample(test[:, i].astype(float), fs, 48000), -32768, 32767

                    ).astype(np.int16)

                    wavfile.write(str(tmpfile_ref), 48000, r48)

                    wavfile.write(str(tmpfile_test), 48000, t48)

                    out = subprocess.check_output([mld, tmpfile_ref, tmpfile_test])

            result["MLD"] = mld_max

        if get_ssnr:

            # length of segment is always 20ms

            len_seg = int(0.02 * fs)

            print(len_seg, ref.shape, test.shape)

            result["SSNR"] = ssnr(

                ref,

                test,

                len_seg,

                thresh_low=ssnr_thresh_low,

                thresh_high=ssnr_thresh_high,

            )

    return result

    for r in results:

        r.get()

    return results

def ssnr(

    ref_sig: np.ndarray,

    test_sig: np.ndarray,

    len_seg: int,

    thresh_low: float = -200,

    thresh_high: float = 0,

) -> np.ndarray:

    """

    Calculate Segmental SNR for test_sig to ref_sig as defined in ISO/IEC 14496-4

    """

    ss = list()

    ref_sig_norm = ref_sig / -np.iinfo(np.int16).min

    test_sig_norm = test_sig / -np.iinfo(np.int16).min

    # check if diff of signal is zero already, then SNR is infinite, since no noise

    diff_sig_norm = ref_sig_norm - test_sig_norm

    if np.all(diff_sig_norm == 0):

        return np.asarray([np.inf] * ref_sig_norm.shape[1])

    denom_add = 10**-13 * len_seg

    segment_counter = np.zeros(ref_sig.shape[1])

    for ref_seg, diff_seg in zip(

        get_framewise(ref_sig_norm, len_seg, zero_pad=True),

        get_framewise(diff_sig_norm, len_seg, zero_pad=True),

    ):

        nrg_ref = np.sum(ref_seg**2, axis=0)

        nrg_diff = np.sum(diff_seg**2, axis=0)

        ss_seg = np.log10(1 + nrg_ref / (denom_add + nrg_diff))

        # only sum up segments that fall inside the thresholds

        # add small eps to nrg_ref to prevent RuntimeWarnings from numpy

        ref_power = 10 * np.log10((nrg_ref + 10**-7) / len_seg)

        zero_mask = np.logical_or(ref_power < thresh_low, ref_power > thresh_high)

        ss_seg[zero_mask] = 0

        # increase segment counter only for channels that were not zeroed

        segment_counter += np.logical_not(zero_mask)

        ss.append(ss_seg)

    # if the reference signal was outside the thresholds for all segments in a channel, segment_counter is zero

    # for that channel and the division here would trigger a warning. We supress the warning and later

    # set the SSNR for those channels to -inf manually instead (overwriting later is simply easier than adding ifs here)

    with warnings.catch_warnings():

        ssnr = np.round(

            10 * np.log10(10 ** (np.sum(ss, axis=0) / segment_counter) - 1), 2

        )

    ssnr[segment_counter == 0] = -np.inf

    return ssnr