Commit ef555f34 authored by BOHMRR's avatar BOHMRR
Browse files

Merge branch '42-add-sba-pytest-run-to-mr-pipeline' into 'main'

Follow-up to SBA pytest addition

See merge request !46
parents c0f293d4 6e9c92a0
Loading
Loading
Loading
Loading
+1 −1
Original line number Diff line number Diff line
@@ -423,5 +423,5 @@ def pytest_configure(config):
        "markers", "create_ref: mark test capable of producing references"
    )
    config.addinivalue_line(
        "markers", "create_ref_serial: reference creation test that should not run in parallel"
        "markers", "create_ref_part2: reference creation test that depends on create_ref references"
    )
+28 −4
Original line number Diff line number Diff line
@@ -45,6 +45,9 @@ import subprocess
import platform
from pathlib import Path

sys.path.append('scripts/ivas_pytests/tests/')
from cut_pcm import cut_samples

BIN_EXT = ".exe" if platform.system() == "Windows" else ""
HERE = Path(__file__).parent.resolve()
DEFAULT_ENCODER_DUT = str(HERE.joinpath(f"../../IVAS_cod{BIN_EXT}").resolve())
@@ -107,6 +110,24 @@ def build_dut_binaries():
    build_crend_unittest(dut_src_dir)


def create_short_testvectors():
    """
    Create short (5sec) testvectors.
    """
    print("Creating short (5sec) testvectors")
    num_channels = "4" # currently only FOA
    cut_from = "0.0"
    cut_len = "5.0"
    for fs in ['48', '32', '16']:
        in_file = f"{TEST_VECTOR_DIR}/stvFOA{fs}c.pcm"
        cut_gain = "1.0"
        cut_file = f"{TEST_VECTOR_DIR}/stvFOA{fs}c_cut.pcm"
        cut_samples(in_file, cut_file, num_channels, fs + "000", cut_from, cut_len, cut_gain)
        cut_gain = "16.0"
        cut_file = f"{TEST_VECTOR_DIR}/stvFOA{fs}c_cut_{cut_gain}.pcm"
        cut_samples(in_file, cut_file, num_channels, fs + "000", cut_from, cut_len, cut_gain)


def main(argv):
    # check for python >= 3.7
    if sys.version_info[0] < 3 or sys.version_info[1] < 7:
@@ -175,12 +196,15 @@ def main(argv):
    else:
        # create references
        print(f"Creating references within the references directory {REFERENCE_DIR}")
        create_short_testvectors()
        if platform.system() == "Windows":
            base_cmd = ["pytest"]
        else:
            base_cmd = ["python3", "-m", "pytest"]
        base_cmd += [
            "scripts/ivas_pytests/tests",
            "-n",
            args.numprocesses,
            "--update_ref",
            "1",
            "-v",
@@ -207,11 +231,11 @@ def main(argv):
        my_env["TESTVECTOR_PATH_REL_TRUNK"] = "/scripts/ivas_pytests/testv/"  # leading "/" is important
        my_env["CREND_UNIT_TEST_BIN"] = CREND_UNITTEST_REF
        print("pytest command line to be executed from project root folder:")
        print(" ".join(base_cmd + ["-m", "create_ref", "-n", args.numprocesses]))
        subprocess.run(base_cmd + ["-m", "create_ref", "-n", args.numprocesses], check=False, env=my_env)
        print(" ".join(base_cmd + ["-m", "create_ref"]))
        subprocess.run(base_cmd + ["-m", "create_ref"], check=False, env=my_env)
        print("pytest command line to be executed from project root folder:")
        print(" ".join(base_cmd + ["-m", "create_ref_serial"]))
        subprocess.run(base_cmd + ["-m", "create_ref_serial"], check=False, env=my_env)
        print(" ".join(base_cmd + ["-m", "create_ref_part2"]))
        subprocess.run(base_cmd + ["-m", "create_ref_part2"], check=False, env=my_env)

    if args.create_only:
        return
+11 −102
Original line number Diff line number Diff line
@@ -46,11 +46,7 @@ plc_patterns = ['PLperc12mblen5', 'PLperc40mblen50', 'PLperc42mblen2']
dtx_set = ['0', '1']
ivas_br_list = ['32000', '64000', '96000', '256000']
sampling_rate_list = ['48', '32', '16']

# we need signals amplified by 24dB - until those are available, use existing 'stvFOA'
agc_tag_list = ['stvFOA']
# TODO: create and use new 24dB signals
# agc_tag_list = ['stvFOA_24dB_']
agc_list = [0, 1]

ch_count_foa = 4
AbsTol = '3'
@@ -65,63 +61,34 @@ def check_and_makedir(dir_path):
                raise  # raises the error again


@pytest.mark.create_ref_serial
# assumption:
# - the needed reference bitstreams are created by test_spar_foa_enc_system
# -> reference bitstreams are not any longer created as part of this test
# -> the parameters of this test (except additional parameter plc_pattern) need to be a subset of the parameters in test_spar_foa_enc_system
# -> the reference generation for this test (reference decoder output) needs to be done after completion of test_spar_foa_enc_system
#    -> therefore the marker create_ref_part2
@pytest.mark.create_ref_part2
@pytest.mark.parametrize("ivas_br", ivas_br_list)
@pytest.mark.parametrize("dtx", dtx_set)
@pytest.mark.parametrize("tag", tag_list)
@pytest.mark.parametrize("plc_pattern", plc_patterns)
@pytest.mark.parametrize("fs", sampling_rate_list)
@pytest.mark.parametrize("agc", agc_list)
def test_spar_foa_plc_system(
    dut_decoder_frontend: DecoderFrontend,
    data_system_tests_path,
    reference_path,
    dut_base_path,
    ref_encoder_path,
    ref_decoder_path,
    update_ref,
    ivas_br,
    dtx,
    tag,
    plc_pattern,
    fs
):
    tag = tag + fs + 'c'
    agc = 0

    if update_ref == 1 and plc_pattern == plc_patterns[0] and ref_encoder_path:
        # enc
        spar_foa_ref_enc(ref_encoder_path, data_system_tests_path, reference_path, tag, fs, ivas_br, dtx, agc)

    #dec
    spar_foa_dec_plc(dut_decoder_frontend, data_system_tests_path, reference_path, dut_base_path, ref_decoder_path, tag, ch_count_foa, fs, ivas_br, dtx, plc_pattern, update_ref, agc)


@pytest.mark.create_ref_serial
@pytest.mark.parametrize("ivas_br", ivas_br_list)
@pytest.mark.parametrize("dtx", dtx_set)
@pytest.mark.parametrize("tag", agc_tag_list)
@pytest.mark.parametrize("plc_pattern", plc_patterns)
@pytest.mark.parametrize("fs", sampling_rate_list)
def test_spar_foa_agc_plc_system(
    dut_decoder_frontend: DecoderFrontend,
    data_system_tests_path,
    reference_path,
    dut_base_path,
    ref_encoder_path,
    ref_decoder_path,
    update_ref,
    ivas_br,
    dtx,
    tag,
    plc_pattern,
    fs
    fs,
    agc
):
    tag = tag + fs + 'c'
    agc = 1

    if update_ref == 1 and plc_pattern == plc_patterns[0] and ref_encoder_path:
        # enc
        spar_foa_ref_enc(ref_encoder_path, data_system_tests_path, reference_path, tag, fs, ivas_br, dtx, agc)

    #dec
    spar_foa_dec_plc(dut_decoder_frontend, data_system_tests_path, reference_path, dut_base_path, ref_decoder_path, tag, ch_count_foa, fs, ivas_br, dtx, plc_pattern, update_ref, agc)
@@ -129,49 +96,6 @@ def test_spar_foa_agc_plc_system(

#########################################################
############ test function ##############################
def spar_foa_ref_enc(
    ref_encoder_path,
    test_vector_path,
    reference_path,
    tag,
    sampling_rate,
    ivas_br,
    dtx,
    agc
):

    #########  run cmd   #####################################
    ref_out_dir = f"{reference_path}/spar_foa_bs/pkt"
    check_and_makedir(ref_out_dir)

    tag_out = f"{tag}_ivasbr{ivas_br[:-3]}k_DTX{dtx}"
    if agc == 1:
        tag_out += '_AGC1'

    input_path = f"{test_vector_path}/{tag}.pcm"
    cut_file = f"{test_vector_path}/{tag}_cut.pcm"
    # we expect that the reference generation for create_ref did already run and the cut PCM files are available
    assert os.path.exists(cut_file)
    input_path = cut_file

    ref_pkt_file = f"{ref_out_dir}/{tag_out}.pkt"

    dtx_mode = dtx == '1'

    ref_encoder = EncoderFrontend(ref_encoder_path, "REF")
    # call REF encoder
    ref_encoder.run(
        ivas_br,
        sampling_rate,
        input_path,
        ref_pkt_file,
        sba_order="+1",
        max_band="FB",
        agc_op=agc,
        dtx_mode=dtx_mode,
    )


def spar_foa_dec_plc(
    decoder_frontend,
    test_vector_path,
@@ -253,18 +177,3 @@ def spar_foa_dec_plc(

        ##report failure
        assert not test_fail

    if ref_decoder_path:
        # Unclear whether this clean-up is still needed
        # TODO: check temp file generation

        ##File removal##
        for count in range(ch_count, 25):
            ch_id = str(count + 1)
            temp_raw_file = f"{ref_out_dir}/out{ch_id}ch.raw"
            if os.path.isfile(temp_raw_file):
                os.remove(temp_raw_file)

        temp_md_file = f"{ref_out_dir}/MD_OUT_DUMMY.BIN"
        if os.path.isfile(temp_md_file):
            os.remove(temp_md_file)
+17 −111
Original line number Diff line number Diff line
@@ -62,11 +62,7 @@ ivas_br_HOA3 = ['256000', '384000', '512000']

sample_rate_list = ['48', '32', '16']
bypass_list = [1, 2]

# we need signals amplified by 24dB - until those are available, use existing 'stvFOA'
agc_tag_list = ['stvFOA']
# TODO: create and use new 24dB signals
# agc_tag_list = ['stvFOA_24dB_']
agc_list = [0, 1]

sample_rate_bw_idx_list = [('48', 'SWB'), ('48', 'WB'), ('32', 'WB')]

@@ -108,7 +104,6 @@ def test_bypass_enc(
    dtx = '0'
    max_bw = "FB"
    agc = 0
    use_agc_ref = 0
    sba_order = "+1"
    output_config = "FOA"

@@ -126,7 +121,6 @@ def test_bypass_enc(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        sba_order,
        update_ref
    )
@@ -145,7 +139,6 @@ def test_bypass_enc(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        output_config,
        update_ref
    )
@@ -156,7 +149,7 @@ def test_bypass_enc(
@pytest.mark.parametrize("dtx", dtx_set)
@pytest.mark.parametrize("tag", tag_list)
@pytest.mark.parametrize("fs", sample_rate_list)
#@pytest.mark.parametrize("agc", agc_list)
@pytest.mark.parametrize("agc", agc_list)
def test_spar_foa_enc_system(
    dut_encoder_frontend: EncoderFrontend,
    dut_decoder_frontend: DecoderFrontend,
@@ -170,15 +163,17 @@ def test_spar_foa_enc_system(
    dtx,
    tag,
    fs,
#    agc
    agc
):
    tag = tag + fs + 'c'
    max_bw = "FB"
    bypass = -1
    agc = 0
    use_agc_ref = 0
    sba_order = "+1"
    output_config = "FOA"
    if agc == 1:
        cut_gain = "16.0"
    else:
        cut_gain = "1.0"

    # enc
    spar_foa_enc(
@@ -194,9 +189,9 @@ def test_spar_foa_enc_system(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        sba_order,
        update_ref
        update_ref,
        cut_gain=cut_gain
    )

    # dec
@@ -213,7 +208,6 @@ def test_spar_foa_enc_system(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        output_config,
        update_ref
    )
@@ -240,7 +234,6 @@ def test_spar_hoa2_enc_system(
#    tag = tag + '_' + fs + 'k'
    max_bw = "FB"
    bypass = -1
    use_agc_ref = 0
    sba_order = "+2"
    output_config = "HOA2"

@@ -258,7 +251,6 @@ def test_spar_hoa2_enc_system(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        sba_order,
        update_ref,
        '.wav'
@@ -278,7 +270,6 @@ def test_spar_hoa2_enc_system(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        output_config,
        update_ref
    )
@@ -305,7 +296,6 @@ def test_spar_hoa3_enc_system(
#    tag = tag + '_' + fs + 'k'
    max_bw = "FB"
    bypass = -1
    use_agc_ref = 0
    sba_order = "+3"
    output_config = "HOA3"

@@ -323,7 +313,6 @@ def test_spar_hoa3_enc_system(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        sba_order,
        update_ref,
        '.wav'
@@ -343,7 +332,6 @@ def test_spar_hoa3_enc_system(
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        output_config,
        update_ref
    )
@@ -372,7 +360,6 @@ def test_spar_foa_enc_BWforce_system(
    tag = tag + fs + 'c'
    bypass = -1
    agc = 0
    use_agc_ref = 0
    sba_order = "+1"
    output_config = "FOA"

@@ -390,7 +377,6 @@ def test_spar_foa_enc_BWforce_system(
        bw,
        bypass,
        agc,
        use_agc_ref,
        sba_order,
        update_ref
    )
@@ -409,74 +395,6 @@ def test_spar_foa_enc_BWforce_system(
        bw,
        bypass,
        agc,
        use_agc_ref,
        output_config,
        update_ref
    )


@pytest.mark.create_ref
@pytest.mark.parametrize("ivas_br", ivas_br_FOA)
@pytest.mark.parametrize("dtx", dtx_set)
@pytest.mark.parametrize("tag", agc_tag_list)
@pytest.mark.parametrize("fs", sample_rate_list)
def test_spar_foa_enc_agc_system(
    dut_encoder_frontend: EncoderFrontend,
    dut_decoder_frontend: DecoderFrontend,
    data_system_tests_path,
    reference_path,
    dut_base_path,
    ref_encoder_path,
    ref_decoder_path,
    update_ref,
    ivas_br,
    dtx,
    tag,
    fs
):
    tag = tag + fs + 'c'
    max_bw = "FB"
    bypass = -1
    agc = 1
    use_agc_ref = 1
    sba_order = "+1"
    output_config = "FOA"

    # enc
    spar_foa_enc(
        dut_encoder_frontend,
        data_system_tests_path,
        ref_encoder_path,
        reference_path,
        dut_base_path,
        tag,
        fs,
        ivas_br,
        dtx,
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        sba_order,
        update_ref,
        cut_gain="16.0"
    )

    # dec
    spar_foa_dec(
        dut_decoder_frontend,
        ref_decoder_path,
        reference_path,
        dut_base_path,
        tag,
        ch_count_foa,
        fs,
        ivas_br,
        dtx,
        max_bw,
        bypass,
        agc,
        use_agc_ref,
        output_config,
        update_ref
    )
@@ -497,7 +415,6 @@ def spar_foa_enc(
    ivas_max_bw,
    bypass,
    agc,
    use_agc_ref,
    sba_order,
    update_ref,
    in_extension = '.pcm',
@@ -520,7 +437,7 @@ def spar_foa_enc(

    tag_out = f"{tag}_ivasbr{ivas_br[:-3]}k_DTX{dtx}"
    short_tag_ext = ""
    if agc == 1 and use_agc_ref == 1:
    if agc == 1:
        short_tag_ext += '_AGC1'
    # we update only bypass = 0/2 (bypass 1 is the same as the baseline)
    if bypass in [0, 2]:
@@ -536,7 +453,8 @@ def spar_foa_enc(
    dtx_mode = dtx == '1'

    if in_extension == '.pcm':
        # cut input PCM file - currently with fixed (i.e. not test dependant) values
        # use shortened and potentially gain adjusted input PCM file - create if not present
        # cut input PCM file: currently with mostly fixed (i.e. not test dependant) values
        num_channels = "4" # currently only FOA inputs end up, here
        cut_from = "0.0"
        cut_len = "5.0"
@@ -591,7 +509,6 @@ def spar_foa_dec(
    ivas_max_bw,
    bypass,
    agc,
    use_agc_ref,
    output_config,
    update_ref,
    keep_files=False
@@ -607,7 +524,7 @@ def spar_foa_dec(
    tag_out = f"{tag}_ivasbr{ivas_br[:-3]}k_DTX{dtx}"

    short_tag_ext = ""
    if agc == 1 and use_agc_ref == 1:
    if agc == 1:
        short_tag_ext += '_AGC1'
    # we update only bypass = 0/2 (bypass 1 is the same as the baseline)
    if bypass in [0, 2]:
@@ -658,6 +575,10 @@ def spar_foa_dec(
        for count in range(ch_count):
            ch_id = str(count + 1)

            # TEST
            fsize1 = os.path.getsize(f"{dut_out_dir}/out{ch_id}ch.raw")
            fsize2 = os.path.getsize(f"{ref_out_dir}/out{ch_id}ch.raw")
            print(f"Want to compare {dut_out_dir}/out{ch_id}ch.raw ({fsize1} bytes) with {ref_out_dir}/out{ch_id}ch.raw ({fsize2} bytes)")
            if cmp_custom(
                f"{dut_out_dir}/out{ch_id}ch.raw",
                f"{ref_out_dir}/out{ch_id}ch.raw",
@@ -674,18 +595,3 @@ def spar_foa_dec(

        ##report failure
        assert not test_fail

    if ref_decoder_path:
        # Unclear whether this clean-up is still needed
        # TODO: check temp file generation

        ##File removal##
        for count in range(ch_count, 25):
            ch_id = str(count + 1)
            temp_raw_file = f"{ref_out_dir}/out{ch_id}ch.raw"
            if os.path.isfile(temp_raw_file):
                os.remove(temp_raw_file)

        temp_md_file = f"{ref_out_dir}/MD_OUT_DUMMY.BIN"
        if os.path.isfile(temp_md_file):
            os.remove(temp_md_file)
+1 −1

File changed.

Contains only whitespace changes.