[fix] implement MASA metadata concatenation (7cc489c0) · Commits · IVAS Codec Public Collaboration / IVAS Processing Scripts

ivas_processing_scripts/init.py

+22 −12

Original line number	Diff line number	Diff line
		@@ -117,13 +117,17 @@ 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" MASA metadata file item {audio}: {', '.join(str(m) for m in meta)}"
		)
		# 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]))
		# logger.debug(
		# f" ISM metadata files item {cfg.items_list[i]}: {', '.join(metadata_str)}"
		# )
		metadata = metadata_ISM

		# check for MASA metadata
		@@ -134,15 +138,21 @@ 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])
		# 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]))
		# logger.debug(
		# f" MASA metadata file item {cfg.items_list[i]}: {', '.join(metadata_str)}"
		# )
		# 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/metadata.py

+92 −2

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
		@@ -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:
		@@ -376,7 +381,7 @@ def concat_ism_metadata_files(

		# 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,88 @@ 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"Input Metadata size {metadata_in.stat().st_size}")
		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 + Preamble 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

+2 −2

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:

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:

ivas_processing_scripts/processing/processing.py

+42 −24

Original line number	Diff line number	Diff line
		@@ -44,13 +44,19 @@ from ivas_processing_scripts.audiotools.audiofile import concat, trim
		from ivas_processing_scripts.audiotools.constants import IVAS_FRAME_LEN_MS
		from ivas_processing_scripts.audiotools.convert.__init__ import convert
		from ivas_processing_scripts.audiotools.metadata import (
		add_remove_metadata_preamble,
		add_remove_ism_metadata_preamble,
		concat_ism_metadata_files,
		concat_masa_metadata_files,
		)
		from ivas_processing_scripts.constants import LOGGER_DATEFMT, LOGGER_FORMAT
		from ivas_processing_scripts.processing.config import TestConfig
		from ivas_processing_scripts.processing.tx import get_timescaled_splits
		from ivas_processing_scripts.utils import apply_func_parallel, list_audio, pairwise
		from ivas_processing_scripts.utils import (
		ALLOWED_INPUT_EXT_MD,
		apply_func_parallel,
		list_audio_or_md,
		pairwise,
		)


		class Processing(ABC):
		@@ -120,30 +126,41 @@ def concat_setup(cfg: TestConfig, chain, logger: logging.Logger):
		except AttributeError:
		input_format = cfg.input["fmt"]

		# concatenation of met files not possible -> do not concatenate MASA and OMASA
		if "MASA" in input_format:
		raise ValueError(
		"Concatenation of formats including MASA metadata not possible"
		concat_md_prefix = cfg.tmp_dirs[0].joinpath(
		f"{cfg.input_path.name}_concatenated.wav"
		)

		# concatenate ISM metadata
		if input_format.startswith("ISM"):
		cfg.concat_meta = []
		for obj_idx in range(len(cfg.metadata_path[0])):
		cfg.concat_meta.append(
		cfg.tmp_dirs[0].joinpath(
		f"{cfg.input_path.name}_concatenated.wav.{obj_idx}.csv"
		)
		)
		for i, md in enumerate(cfg.metadata_path[0]):
		if md.suffix == ".csv":
		cfg.concat_meta.append(Path(f"{concat_md_prefix}.{i}.csv"))
		elif md.suffix == ".met":
		cfg.concat_meta.append(Path(f"{concat_md_prefix}.met"))

		concat_ism_metadata_files(
		cfg.items_list,
		cfg.metadata_path,
		cfg.concat_meta,
		input_format,
		)
		if "MASA" in input_format:
		concat_masa_metadata_files(
		cfg.metadata_path,
		cfg.concat_meta[-1],
		)

		# set input to the concatenated file we have just written to the output dir
		cfg.metadata_path = [cfg.concat_meta]
		elif input_format.startswith("MASA"):
		cfg.concat_meta = Path(f"{concat_md_prefix}.met")
		concat_masa_metadata_files(
		cfg.metadata_path,
		cfg.concat_meta,
		)

		# set input to the concatenated file we have just written to the output dir
		cfg.metadata_path = [[cfg.concat_meta]]

		# concatenate audio
		cfg.concat_file = cfg.tmp_dirs[0].joinpath(
		@@ -317,7 +334,7 @@ def preprocess(cfg, logger):
		)

		# update the configuration to use preprocessing outputs as new inputs
		cfg.items_list = list_audio(
		cfg.items_list = list_audio_or_md(
		cfg.out_dirs[0], select_list=getattr(cfg, "input_select", None)
		)

		@@ -394,17 +411,18 @@ def preprocess_2(cfg, logger):
		)

		# update the configuration to use preprocessing 2 outputs as new inputs
		cfg.items_list = list_audio(
		cfg.items_list = list_audio_or_md(
		cfg.out_dirs[0], select_list=getattr(cfg, "input_select", None)
		)

		if cfg.metadata_path[0] is not None:
		for item_idx in range(len(cfg.metadata_path)):
		for obj_idx in range(len(cfg.metadata_path[item_idx])):
		if cfg.metadata_path[item_idx][obj_idx]:
		cfg.metadata_path[item_idx][obj_idx] = cfg.out_dirs[0] / Path(
		f"{cfg.items_list[item_idx].stem}.wav.{obj_idx}.csv"
		cfg.metadata_path = [
		list_audio_or_md(
		cfg.out_dirs[0],
		select_list=getattr(cfg, "input_select", None),
		allowed_ext=ALLOWED_INPUT_EXT_MD,
		)
		]

		# remove already applied processing stage
		cfg.proc_chains = cfg.proc_chains[1:]
		cfg.tmp_dirs = cfg.tmp_dirs[1:]
		@@ -536,7 +554,7 @@ def remove_preamble(x, out_fmt, fs, repeat_signal, preamble_len_ms, meta, logger

		# remove preamble
		if preamble_len_ms > 0:
		meta = add_remove_metadata_preamble(meta, preamble_len_ms, add=False)
		meta = add_remove_ism_metadata_preamble(meta, preamble_len_ms, add=False)

		# cut first half of signal
		if repeat_signal: