refactor tx configuration for processing chains (7091fd04) · Commits · IVAS Codec Public Collaboration / IVAS Processing Scripts

examples/TEMPLATE.yml

+1 −1

Original line number	Diff line number	Diff line
		@@ -154,7 +154,7 @@ input:
		### REQUIRED: either error_pattern (and errpatt_late_loss_rate or errpatt_delay) or error_profile
		### delay error profile file
		# error_pattern: ".../dly_error_profile.dat"
		### Late loss rate in precent or EVS
		### Late loss rate in precent for EVS
		# errpatt_late_loss_rate: 1
		### Constant JBM delay in milliseconds for EVS
		# errpatt_delay: 200

ivas_processing_scripts/processing/chains.py

+11 −73

Original line number	Diff line number	Diff line
		@@ -41,6 +41,7 @@ from ivas_processing_scripts.audiotools.audiofile import read, write
		from ivas_processing_scripts.processing.config import TestConfig
		from ivas_processing_scripts.processing.evs import EVS
		from ivas_processing_scripts.processing.ivas import IVAS, IVAS_rend
		from ivas_processing_scripts.processing.tx import get_tx_cfg
		from ivas_processing_scripts.processing.postprocessing import Postprocessing
		from ivas_processing_scripts.processing.preprocessing import Preprocessing
		from ivas_processing_scripts.processing.preprocessing_2 import Preprocessing2
		@@ -312,52 +313,17 @@ def get_processing_chain(
		evs_lfe_9k6bps_nb = cond_cfg.get("evs_lfe_9k6bps_nb", None)

		# Frame error pattern bitstream modification
		tx_cfg = None
		if "tx" in cond_cfg.keys() or hasattr(cfg, "tx"):
		# postprocess also signal without error if there is loudness scaling
		if post_cfg.get("loudness"):
		tx_condition = True
		# local specification overwrites global one
		if "tx" in cond_cfg.keys():
		tx_cfg_tmp = cond_cfg["tx"]
		else:
		tx_cfg_tmp = cfg.tx

		if tx_cfg_tmp.get("type", None) == "FER":
		tx_cfg = {
		"type": tx_cfg_tmp.get("type", None),
		"error_pattern": get_abs_path(
		tx_cfg_tmp.get("error_pattern", None)
		),
		"error_rate": tx_cfg_tmp.get("error_rate", None),
		"master_seed": cfg.master_seed,
		"prerun_seed": cfg.prerun_seed,
		}
		elif tx_cfg_tmp.get("type", None) == "JBM":
		tx_cfg = {
		"type": tx_cfg_tmp.get("type", None),
		"error_pattern": get_abs_path(
		tx_cfg_tmp.get("error_pattern", None)
		),
		"errpatt_late_loss_rate": tx_cfg_tmp.get(
		"errpatt_late_loss_rate", None
		),
		"errpatt_delay": tx_cfg_tmp.get("errpatt_delay", None),
		"errpatt_seed": tx_cfg_tmp.get("errpatt_seed", None),
		"error_profile": tx_cfg_tmp.get("error_profile", None),
		"n_frames_per_packet": tx_cfg_tmp.get("n_frames_per_packet", None),
		"master_seed": cfg.master_seed,
		}
		else:
		raise ValueError(
		"Type of bitstream procesing either missing or not valid"
		)
		else:
		tx_cfg = None
		tx_cfg = get_tx_cfg(cfg, cond_cfg, is_EVS=True)

		preamble = 0
		if hasattr(cfg, "preprocessing_2"):
		preamble = cfg.preprocessing_2.get("preamble", 0)
		else:
		preamble = 0

		# if the encoding format differs from the format after the preprocessing, add format conversion stuff
		if (cod_fmt := cod_cfg.get("fmt", tmp_in_fmt)) != tmp_in_fmt:
		@@ -423,47 +389,18 @@ def get_processing_chain(
		)

		# Frame error pattern bitstream modification
		tx_cfg = None
		if "tx" in cond_cfg.keys() or hasattr(cfg, "tx"):
		# postprocess also signal without error if there is loudness scaling
		if post_cfg.get("loudness"):
		tx_condition = True
		# local specification overwrites global one
		if "tx" in cond_cfg.keys():
		tx_cfg_tmp = cond_cfg["tx"]
		else:
		tx_cfg_tmp = cfg.tx

		if tx_cfg_tmp.get("type", None) == "FER":
		tx_cfg = {
		"type": tx_cfg_tmp.get("type", None),
		"error_pattern": get_abs_path(
		tx_cfg_tmp.get("error_pattern", None)
		),
		"error_rate": tx_cfg_tmp.get("error_rate", None),
		"master_seed": cfg.master_seed,
		"prerun_seed": cfg.prerun_seed,
		}
		elif tx_cfg_tmp.get("type", None) == "JBM":
		tx_cfg = {
		"type": tx_cfg_tmp.get("type", None),
		"error_pattern": tx_cfg_tmp.get("error_pattern", None),
		"error_profile": tx_cfg_tmp.get("error_profile", None),
		"n_frames_per_packet": tx_cfg_tmp.get("n_frames_per_packet", None),
		"master_seed": cfg.master_seed,
		"errpatt_seed": tx_cfg_tmp.get("errpatt_seed", None),
		}
		ivas_jbm = True
		else:
		raise ValueError(
		"Type of bitstream procesing either missing or not valid"
		)
		else:
		tx_cfg = None
		tx_cfg = get_tx_cfg(cfg, cond_cfg)
		ivas_jbm = tx_cfg.get("type") == "JBM"

		preamble = 0
		if hasattr(cfg, "preprocessing_2"):
		preamble = cfg.preprocessing_2.get("preamble", 0)
		else:
		preamble = 0

		# if the encoding format differs from the format after the preprocessing, add format conversion stuff
		if (cod_fmt := cod_cfg.get("fmt", tmp_in_fmt)) != tmp_in_fmt:
		@@ -591,6 +528,7 @@ def get_processing_chain(
		}
		)
		)

		# add splitting and scaling for all conditions
		chain["processes"].append(
		Processing_splitting_scaling(

ivas_processing_scripts/processing/tx.py

0 → 100755

+72 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3

		#
		# (C) 2022-2025 IVAS codec Public Collaboration with portions copyright Dolby International AB, Ericsson AB,
		# Fraunhofer-Gesellschaft zur Foerderung der angewandten Forschung e.V., Huawei Technologies Co. LTD.,
		# Koninklijke Philips N.V., Nippon Telegraph and Telephone Corporation, Nokia Technologies Oy, Orange,
		# Panasonic Holdings Corporation, Qualcomm Technologies, Inc., VoiceAge Corporation, and other
		# contributors to this repository. All Rights Reserved.
		#
		# This software is protected by copyright law and by international treaties.
		# The IVAS codec Public Collaboration consisting of Dolby International AB, Ericsson AB,
		# Fraunhofer-Gesellschaft zur Foerderung der angewandten Forschung e.V., Huawei Technologies Co. LTD.,
		# Koninklijke Philips N.V., Nippon Telegraph and Telephone Corporation, Nokia Technologies Oy, Orange,
		# Panasonic Holdings Corporation, Qualcomm Technologies, Inc., VoiceAge Corporation, and other
		# contributors to this repository retain full ownership rights in their respective contributions in
		# the software. This notice grants no license of any kind, including but not limited to patent
		# license, nor is any license granted by implication, estoppel or otherwise.
		#
		# Contributors are required to enter into the IVAS codec Public Collaboration agreement before making
		# contributions.
		#
		# This software is provided "AS IS", without any express or implied warranties. The software is in the
		# development stage. It is intended exclusively for experts who have experience with such software and
		# solely for the purpose of inspection. All implied warranties of non-infringement, merchantability
		# and fitness for a particular purpose are hereby disclaimed and excluded.
		#
		# Any dispute, controversy or claim arising under or in relation to providing this software shall be
		# submitted to and settled by the final, binding jurisdiction of the courts of Munich, Germany in
		# accordance with the laws of the Federal Republic of Germany excluding its conflict of law rules and
		# the United Nations Convention on Contracts on the International Sales of Goods.
		#

		from typing import Optional

		from ivas_processing_scripts.processing.config import TestConfig
		from ivas_processing_scripts.utils import get_abs_path


		def get_tx_cfg(cfg: TestConfig, cond_cfg: dict, is_EVS: bool = False) -> Optional[dict]:
		# local specification overwrites global one
		if "tx" in cond_cfg.keys():
		tx_cfg_tmp = cond_cfg["tx"]
		else:
		tx_cfg_tmp = cfg.tx

		if tx_cfg_tmp.get("type", None) == "FER":
		tx_cfg = {
		"type": tx_cfg_tmp.get("type"),
		"error_pattern": get_abs_path(tx_cfg_tmp.get("error_pattern", None)),
		"error_rate": tx_cfg_tmp.get("error_rate", None),
		"master_seed": cfg.master_seed,
		"prerun_seed": cfg.prerun_seed,
		}
		elif tx_cfg_tmp.get("type", None) == "JBM":
		tx_cfg = {
		"type": tx_cfg_tmp.get("type"),
		"error_pattern": get_abs_path(tx_cfg_tmp.get("error_pattern", None)),
		"errpatt_seed": tx_cfg_tmp.get("errpatt_seed", None),
		"error_profile": tx_cfg_tmp.get("error_profile", None),
		"n_frames_per_packet": tx_cfg_tmp.get("n_frames_per_packet", None),
		"master_seed": cfg.master_seed,
		}
		# EVS specific keys
		if is_EVS:
		tx_cfg["errpatt_late_loss_rate"] = tx_cfg_tmp.get(
		"errpatt_late_loss_rate", None
		)
		tx_cfg["errpatt_delay"] = tx_cfg_tmp.get("errpatt_delay", None)
		else:
		raise ValueError("Type of bitstream procesing either missing or not valid")

		return tx_cfg