Commit f3f3bcf0 authored by Anika Treffehn's avatar Anika Treffehn
Browse files

added validation step for bitstream processing

parent 6d51bf09
Loading
Loading
Loading
Loading
Loading
+45 −0
Original line number Diff line number Diff line
@@ -30,6 +30,7 @@
#  the United Nations Convention on Contracts on the International Sales of Goods.
#

import os.path
from pathlib import Path
from typing import Optional, Union

@@ -139,3 +140,47 @@ def create_and_apply_error_pattern(
    eid_xor(error_pattern, in_bitstream, out_bitstream)

    return


def validate_error_pattern_application(
    error_pattern: Optional[Union[Path, str]] = None,
    error_rate: Optional[int] = None,
) -> None:
    """
    Validate settings for the network simulator

    Parameters
    ----------
    error_pattern: Optional[Union[Path, str]]
        Path to existing error pattern
    error_rate: Optional[int]
        Frame error rate
    """

    if find_binary("gen-patt") is None:
        raise FileNotFoundError(
            "The binary gen-patt for error pattern generation was not found! Please check the configuration."
        )
    if find_binary("eid-xor") is None:
        raise FileNotFoundError(
            "The binary eid-xor for error patter application was not found! Please check the configuration."
        )
    if error_pattern is not None:
        if not os.path.exists(os.path.realpath(error_pattern)):
            raise FileNotFoundError(
                f"The frame error profile file {error_pattern} was not found! Please check the configuration."
            )
        if error_rate is not None:
            raise ValueError(
                "Frame error pattern and error rate are specified for bitstream processing. Can't use both! Please check the configuration."
            )
    else:
        if error_rate is None:
            raise ValueError(
                "Either error rate or error pattern has to be specified for FER bitstream processing."
            )
        elif error_rate < 0 or error_rate > 100:
            raise ValueError(
                f"Specified error rate of {error_rate}% is either too large or too small."
            )
    return
+1 −3
Original line number Diff line number Diff line
@@ -30,8 +30,6 @@
#  the United Nations Convention on Contracts on the International Sales of Goods.
#

import shutil
import warnings
from os import getcwd
from pathlib import Path
from tempfile import TemporaryDirectory
@@ -129,7 +127,7 @@ def create_error_pattern(
    with TemporaryDirectory() as tmp_dir:
        tmp_dir = Path(tmp_dir)

        sta_file = ERROR_PATTERNS_DIR.joinpath(f"sta_template")
        sta_file = ERROR_PATTERNS_DIR.joinpath("sta_template")
        tmp_sta_file = tmp_dir.joinpath("sta")

        # compute seed
+0 −1
Original line number Diff line number Diff line
@@ -34,7 +34,6 @@ import os.path
from pathlib import Path
from typing import Optional, Union

from ivas_processing_scripts.audiotools.wrappers.gen_patt import create_error_pattern
from ivas_processing_scripts.utils import find_binary, run

LIST_JBM_PROFILES = range(12)
+14 −0
Original line number Diff line number Diff line
@@ -48,6 +48,7 @@ from ivas_processing_scripts.audiotools.audiofile import (
from ivas_processing_scripts.audiotools.constants import IVAS_FRAME_LEN_MS
from ivas_processing_scripts.audiotools.wrappers.eid_xor import (
    create_and_apply_error_pattern,
    validate_error_pattern_application,
)
from ivas_processing_scripts.audiotools.wrappers.networkSimulator import (
    apply_network_simulator,
@@ -99,6 +100,19 @@ class EVS(Processing):
                self.bitrate.extend(
                    [self.bitrate[-1]] * (self.in_fmt.num_channels - len(self.bitrate))
                )
        # existence of error pattern files (if given) already here
        if self.tx is not None:
            if self.tx.get("type", None) == "JBM":
                validate_network_simulator(
                    self.tx["error_pattern"],
                    self.tx["error_profile"],
                    self.tx["n_frames_per_packet"],
                )
            elif self.tx.get("type", None) == "FER":
                validate_error_pattern_application(
                    self.tx["error_pattern"],
                    self.tx["error_rate"],
                )

    def process(
        self, in_file: Path, out_file: Path, in_meta, logger: logging.Logger
+6 −3
Original line number Diff line number Diff line
@@ -41,6 +41,7 @@ from ivas_processing_scripts.audiotools.audiofile import parse_wave_header, read
from ivas_processing_scripts.audiotools.constants import IVAS_FRAME_LEN_MS
from ivas_processing_scripts.audiotools.wrappers.eid_xor import (
    create_and_apply_error_pattern,
    validate_error_pattern_application,
)
from ivas_processing_scripts.audiotools.wrappers.networkSimulator import (
    apply_network_simulator,
@@ -79,7 +80,7 @@ class IVAS(Processing):
                raise FileNotFoundError(
                    "The IVAS decoder binary was not found! Please check the configuration."
                )
        # TODO: if tx_fer or tx_jbm test for existence of bitstream processing binaries and

        # existence of error pattern files (if given) already here
        if self.tx is not None:
            if self.tx.get("type", None) == "JBM":
@@ -89,8 +90,10 @@ class IVAS(Processing):
                    self.tx["n_frames_per_packet"],
                )
            elif self.tx.get("type", None) == "FER":
                # TODO
                pass
                validate_error_pattern_application(
                    self.tx["error_pattern"],
                    self.tx["error_rate"],
                )

    def process(
        self, in_file: Path, out_file: Path, in_meta, logger: logging.Logger
Loading