Merge branch '104-characterization-implement-concatenation-of-masa-metadata' into 'main' (560c6395) · Commits · IVAS Codec Public Collaboration / IVAS Processing Scripts

ivas_processing_scripts/init.py

+8 −12

Original line number	Diff line number	Diff line
		@@ -117,12 +117,9 @@ def main(args):
		item_names=cfg.items_list,
		)
		# print info about found and used metadata files
		for i in range(len(metadata_ISM)):
		metadata_str = []
		for o in range(len(metadata_ISM[i])):
		metadata_str.append(str(metadata_ISM[i][o]))
		for audio, meta in zip(cfg.items_list, metadata_ISM):
		logger.debug(
		f" ISM metadata files item {cfg.items_list[i]}: {', '.join(metadata_str)}"
		f" ISM metadata file item {audio}: {', '.join(str(m) for m in meta)}"
		)
		metadata = metadata_ISM

		@@ -134,15 +131,14 @@ def main(args):
		item_names=cfg.items_list,
		)
		# print info about found and used metadata files
		for i in range(len(metadata_MASA)):
		metadata_str = []
		for o in range(len(metadata_MASA[i])):
		metadata_str.append(str(metadata_MASA[i][o]))
		for audio, meta in zip(cfg.items_list, metadata_MASA):
		logger.debug(
		f" MASA metadata file item {cfg.items_list[i]}: {', '.join(metadata_str)}"
		f" MASA metadata file item {audio}: {', '.join(str(m) for m in meta)}"
		)
		for i, meta in enumerate(metadata):
		meta.extend(metadata_MASA[i])
		# extend the list in case ISM metadata already is present
		metadata = [
		meta_list + met for meta_list, met in zip(metadata, metadata_MASA)
		]

		if not cfg.input["fmt"].startswith("ISM") and "MASA" not in cfg.input["fmt"]:
		metadata = [None] * len(cfg.items_list)

ivas_processing_scripts/audiotools/convert/init.py

+3 −3

Original line number	Diff line number	Diff line
		@@ -74,7 +74,7 @@ def convert_file(

		# get audio class object - can be either a regular single audio or scene description .txt
		if not isinstance(in_fmt, PurePath) and in_fmt.startswith("META"):
		input = metadata.Metadata(in_file)
		input = metadata.SceneDescription(in_file)
		else:
		# first check prevents crash on custom_ls setup formats
		if isinstance(in_fmt, str) and in_fmt.startswith("MASA") and in_meta is None:
		@@ -90,7 +90,7 @@ def convert_file(
		out_fs = input.fs

		if not out_fmt:
		if isinstance(input, metadata.Metadata):
		if isinstance(input, metadata.SceneDescription):
		raise ValueError(
		"Output format must be specified for scene description files!"
		)
		@@ -139,7 +139,7 @@ def convert_file(
		output.metadata_files = copy(input.metadata_files)

		# apply actual conversion
		if isinstance(input, metadata.Metadata):
		if isinstance(input, metadata.SceneDescription):
		if logger:
		logger.debug(f"Converting metadata to {out_fmt} : {in_file} -> {out_file}")

ivas_processing_scripts/audiotools/metadata.py

+96 −4

Original line number	Diff line number	Diff line
		@@ -31,7 +31,10 @@
		#

		import csv
		import logging
		from pathlib import Path
		from shutil import copyfile, copyfileobj
		from tempfile import TemporaryDirectory
		from typing import Optional, TextIO, Tuple, Union

		import numpy as np
		@@ -47,7 +50,7 @@ from ivas_processing_scripts.audiotools.constants import (
		)


		class Metadata:
		class SceneDescription:
		def __init__(self, meta_file: Union[str, Path]):
		self.meta_file = Path(meta_file)

		@@ -203,6 +206,8 @@ def write_ISM_metadata_in_file(
		file_names = file_name

		for i, csv_file in enumerate(file_names):
		if not str(csv_file).endswith(".csv"):
		continue # skip MASA metadata
		number_frames = metadata[i].shape[0]
		number_columns = metadata[i].shape[1]
		with open(csv_file, "w", newline="") as file:
		@@ -297,7 +302,7 @@ def trim_meta(
		return


		def concat_meta_from_file(
		def concat_ism_metadata_files(
		audio_files: list[str],
		meta_files: list[list[str]],
		out_file: list[str],
		@@ -376,7 +381,7 @@ def concat_meta_from_file(

		# add preamble
		if preamble:
		concat_meta_all_obj = add_remove_metadata_preamble(
		concat_meta_all_obj = add_remove_ism_metadata_preamble(
		concat_meta_all_obj, preamble
		)

		@@ -623,7 +628,7 @@ def metadata_search_MASA(
		return list_meta


		def add_remove_metadata_preamble(
		def add_remove_ism_metadata_preamble(
		metadata,
		preamble,
		add: Optional[bool] = True,
		@@ -657,3 +662,90 @@ def add_remove_metadata_preamble(
		)

		return metadata


		def concat_masa_metadata_files(
		meta_files: list[list[str]],
		out_file: Union[str, Path],
		) -> None:
		"""
		Concatenate ISM metadata from files

		Parameters
		----------
		meta_files: list[list[str]]
		List of corresponding metadata file names
		out_file: Union[str, Path]
		Name of concatenated output file
		"""
		# meta_files is a list of lists
		# this could be a list of ISM metadata files (OMASA)
		# or just a single MASA metadata file, so we use the last index
		with open(out_file, "wb") as out_file:
		for meta in meta_files:
		with open(meta[-1], "rb") as in_file:
		copyfileobj(in_file, out_file)


		def add_masa_metadata_preamble_repeat(
		masa: audio.MetadataAssistedSpatialAudio,
		metadata_in: Path,
		metadata_out: Path,
		preamble: int,
		repeat: Optional[bool] = False,
		logger: Optional[logging.Logger] = None,
		) -> None:
		preamble_frames = preamble / IVAS_FRAME_LEN_MS
		if not preamble_frames.is_integer():
		raise ValueError(
		f"Application of preamble for MASA metadata is only possible if preamble length is multiple of frame length. "
		f"Frame length: {IVAS_FRAME_LEN_MS}ms"
		)
		preamble_frames = int(preamble_frames)

		from ivas_processing_scripts.audiotools.wrappers.masaAnalyzer import masaAnalyzer

		with TemporaryDirectory() as tmp_dir:
		tmp_dir = Path(tmp_dir)
		tmp_meta = tmp_dir.joinpath("preamble.met")

		# generate zero metadata for preamble
		if preamble > 0:
		frame_smp = 48 * IVAS_FRAME_LEN_MS
		preamble_smp = int(frame_smp * (preamble / IVAS_FRAME_LEN_MS))

		fmt = "HOA2" if masa.dirs == 2 else "FOA"
		sba = audio.fromtype(fmt)
		sba.audio = np.zeros([preamble_smp, sba.num_channels])
		sba.fs = 48000

		num_tc = masa.num_channels - getattr(masa, "num_ism_channels", 0)

		masaAnalyzer(sba, num_tc, masa.dirs, tmp_meta)

		if logger:
		logger.debug(f"Preamble Metadata size {tmp_meta.stat().st_size}")

		# concatenate preamble and metadata
		with open(tmp_meta, "ab") as out_meta:
		# append out the original input metadata
		with open(metadata_in, "rb") as in_meta:
		copyfileobj(in_meta, out_meta)

		if logger:
		logger.debug(f"Input Metadata size {metadata_in.stat().st_size}")
		logger.debug(
		f"Input + Preamble concatenated metadata size {tmp_meta.stat().st_size}"
		)

		if repeat:
		# repeat the entire file
		with open(tmp_meta, "rb") as in_meta:
		md = in_meta.read()
		with open(metadata_out, "wb") as out_meta:
		out_meta.write(md * 2)
		else:
		copyfile(tmp_meta, metadata_out)

		if logger:
		logger.debug(f"Output Metadata size {metadata_out.stat().st_size}")

ivas_processing_scripts/processing/chains.py

+6 −6

Original line number	Diff line number	Diff line
		@@ -54,7 +54,7 @@ from ivas_processing_scripts.processing.tx import get_tx_cfg
		from ivas_processing_scripts.utils import (
		find_binary,
		get_abs_path,
		list_audio,
		list_audio_or_md,
		parse_gain,
		)

		@@ -112,7 +112,7 @@ def init_processing_chains(cfg: TestConfig) -> None:
		)

		# list items in input directory
		cfg.items_list = list_audio(
		cfg.items_list = list_audio_or_md(
		cfg.input_path, select_list=getattr(cfg, "input_select", None)
		)
		if len(cfg.items_list) == 0:
		@@ -659,10 +659,10 @@ def validate_input_files(cfg: TestConfig):
		aligned_output_dir = cfg.output_path / "20ms_aligned_files"
		try:
		aligned_output_dir.mkdir(exist_ok=False, parents=True)
		except FileExistsError:
		raise ValueError(
		"Folder for 20ms aligned files already exists. Please move or delete folder"
		)
		except FileExistsError as e:
		raise FileExistsError(
		f"Folder for 20ms aligned files already exists. Please move or delete folder {aligned_output_dir}"
		) from e

		for i, item in enumerate(cfg.items_list):
		if "fs" in cfg.input:

ivas_processing_scripts/processing/preprocessing_2.py

+14 −14

Original line number	Diff line number	Diff line
		@@ -40,7 +40,8 @@ from ivas_processing_scripts.audiotools import audio
		from ivas_processing_scripts.audiotools.audioarray import trim
		from ivas_processing_scripts.audiotools.audiofile import write
		from ivas_processing_scripts.audiotools.metadata import (
		add_remove_metadata_preamble,
		add_masa_metadata_preamble_repeat,
		add_remove_ism_metadata_preamble,
		write_ISM_metadata_in_file,
		)
		from ivas_processing_scripts.audiotools.wrappers.bs1770 import (
		@@ -64,26 +65,15 @@ class Preprocessing2(Processing):
		self.in_fmt, in_file, fs=self.in_fs, in_meta=in_meta
		)

		if isinstance(
		audio_object, (audio.MetadataAssistedSpatialAudio, audio.OMASAAudio)
		):
		if self.preamble > 0 or self.background_noise or self.repeat_signal:
		raise ValueError(
		"No preprocessing 2 possible for formats including MASA metadata"
		)

		# modify ISM metadata
		if self.in_fmt.startswith("ISM"):
		if not self.preamble:
		preamble = 0
		else:
		preamble = self.preamble
		preamble = self.preamble or 0

		# read out old
		metadata = audio_object.object_pos

		# add preamble
		metadata = add_remove_metadata_preamble(metadata, preamble)
		metadata = add_remove_ism_metadata_preamble(metadata, preamble)

		# repeat signal
		if self.repeat_signal:
		@@ -95,6 +85,16 @@ class Preprocessing2(Processing):
		audio_object.metadata_files = meta_files
		audio_object.object_pos = metadata

		if "MASA" in self.in_fmt:
		add_masa_metadata_preamble_repeat(
		audio_object,
		in_meta[-1],
		out_file.with_suffix(".wav.met"),
		self.preamble or 0,
		self.repeat_signal,
		logger,
		)

		# modify audio signal
		# add preamble
		if self.preamble > 0: