add first infrastructure for splitting

import re

import subprocess

from pathlib import Path

from typing import Optional

from typing import Optional, List

THIS_PATH = os.path.join(os.getcwd(), __file__)

sys.path.append(os.path.join(os.path.dirname(THIS_PATH), "../scripts"))

    ref_jbm_tf: Optional[Path] = None,

    cut_jbm_tf: Optional[Path] = None,

    quiet: Optional[bool] = False,

    scalefac=1,

) -> tuple[int, str]:

    scalefac: int = 1,

    split_idx: np.ndarray = np.empty(0),

) -> tuple[List[int], List[str]]:

    """

    Compare 2 PCM files for bitexactness

    """

    if fs1 != fs2:

        reason = "FAIL: Sampling rate differs."

        return 1, reason

        return [1], [reason]

    # In case number of channels do not match, fail already now. Could happen in case of

    # comparison to input with for a non-passthrough mode.

    if s1.shape[1] != s2.shape[1]:

        reason = "FAIL: Number of channels differ."

        return 1, reason

        return [1], [reason]

    handle_differing_lengths = "fail"

    if allow_differing_lengths:

        if get_mld:

            reason += " - MLD: None"

        return 1, reason

        return [1], [reason]

    output_differs_parts = []

    reason_parts = []

    for s1, s2 in zip(np.split(s1, split_idx), np.split(s2, split_idx)):

        # Apply scalefac if specified. Useful in case scaling has been applied on the input, and the inverse is scaling is supplied in scalefac.

        if scalefac != 1:

            s1 = np.round(s1 * scalefac, 0)  # Need rounding for max abs diff search

        if quiet:

            output_target.close()

    return output_differs, reason

        output_differs_parts.append(output_differs)

        reason_parts.append(reason)

    return output_differs_parts, reason_parts

class OdgParsingFailed(Exception):

    parser.add_argument("--get_odg", action="store_true")

    parser.add_argument("--get_ssnr", action="store_true")

    parser.add_argument("--allow_differing_lengths", action="store_true")

    parser.add_argument("--scalefac", type=float, default=1, dest="scalefac", help="Scale factor to be applied before comparing the output. Useful when input scaling has been applied.")

    parser.add_argument(

        "--scalefac",

        type=float,

        default=1,

        dest="scalefac",

        help="Scale factor to be applied before comparing the output. Useful when input scaling has been applied.",

    )

    parser.add_argument("--quiet", action="store_true")

    args = vars(parser.parse_args())

    parse_properties,

    compare_dmx_signals,

    log_dbg_msg,

    get_split_idx,

)

from tests.testconfig import PARAM_FILE

from tests.constants import (

            ref_file = ref_output_file

        fs = int(sampling_rate) * 1000

        output_differs, reason = cmp_pcm(

        # split_idx = np.empty(0)

        # # TODO:test on --split argument given

        #

        # split_idx = get_split_idx(str(Path(testv_file).stem), int(sampling_rate))

        output_differs_parts, reason_parts = cmp_pcm(

            ref_file,

            dut_output_file,

            out_config_2_nchannels(output_config),

            scalefac=test_info.config.option.scalefac,

        )

        cmp_result_msg += reason

        props = parse_properties(cmp_result_msg, output_differs, props_to_record)

        for output_differs, reason in zip(output_differs_parts, reason_parts):

            props = parse_properties(reason, output_differs, props_to_record)

            for k, v in props.items():

                dut_decoder_frontend.record_property(k, v)

        if enc_test_result:

            pytest.fail("Too high difference in encoder statistics found.")

        at_least_one_output_differs = any(output_differs_parts)

        # TODO

        reason = reason_parts[0]

        if tracefile_last_rtp_numbers_differ:

            pytest.fail(

                "Last RTP sequence num in tracefiles differ for JBM decoding - Not all frames were decoded in both ref and dut."

            )

        elif get_mld and get_mld_lim > 0:

            if output_differs:

            if at_least_one_output_differs:

                pytest.fail(reason)

        else:

            if output_differs or metadata_differs:

            if at_least_one_output_differs or metadata_differs:

                msg = "Difference between ref and dut in "

                if output_differs and metadata_differs:

                if at_least_one_output_differs and metadata_differs:

                    msg += f"output ({reason}) and metadata"

                elif output_differs:

                elif at_least_one_output_differs:

                    msg += f"output only ({reason})"

                elif metadata_differs:

                    msg += "metadata only"

import os

import re

import json

import shutil

from tests import testconfig

import pytest

import platform

from typing import Optional, Union

import numpy as np

from .constants import (

    # MAX_ENC_DIFF_NAME_PATTERN,

    DMX_DIFF,

    DMX_MLD,

    DMX_SSNR,

    MAX_ENC_DIFF_PARAM_NAME,

    MLD_PATTERN,

    MAX_DIFF_PATTERN,

    SPLIT_IDX,

    SSNR_PATTERN,

    ENC_AUX_FILES,

    ODG_PATTERN,

        default=1,

    )

    parser.addoption(

        "--split_comparison",

        action="store_true",

        help="If given, split the output files by the provided indices",

    )

@pytest.fixture(scope="session", autouse=True)

def update_ref(request):

    """

    """

    return request.config.option.odg

@pytest.fixture(scope="session", autouse=True)

def get_odg_bin(request):

    """

            prop_results = parse_properties(reason, dmx_differs, dmx_props)

    return prop_results

def get_split_idx(input_file: str, sampling_rate_khz: int) -> Optional[np.ndarray]:

    """

    Return array for splitting the output file before doing the comparison.

    If no list of indices is available for the given input file, an empty array is returned.

    """

    idx = SPLIT_IDX.get(input_file, np.empty(0))

    if len(idx) > 0 and sampling_rate_khz != 16:

        idx *= sampling_rate_khz // 16

    return idx

    ["total_brate", np.float32, "fs/50"],

    ["vad_flag", np.int16, "fs/50"],

]

# lists of indices for splitting of output files

# values are in SAMPLES for 16kHz (!). For higher rates, need to multiply

SPLIT_IDX_LTV_STEREO_16KHZ = np.asarray(

    [

        302096,

        519280,

        613072,

        690176,

        740992,

        749024,

        782096,

        950064,

        1045408,

        1141408,

        1237424,

        1333392,

        1397712,

        1472912,

        1755840,

        1781680,

        1789712,

        1957664,

        2158112,

        2294704,

        2454688,

        2525872,

        2573888,

        2645504,

        2746096,

        2842096,

        2970080,

        3219792,

    ]

)

SPLIT_IDX = {"ltv16_stereo": SPLIT_IDX_LTV_STEREO_16KHZ}