remove AFsP dependency in python scripts (e3ea151c) · Commits · IVAS Codec Public Collaboration / IVAS Codec

scripts/pyaudio3dtools/audiofile.py

100644 → 100755

+10 −1

Original line number	Diff line number	Diff line
		@@ -124,6 +124,7 @@ def convertfile(
		out_nchans: Optional[int] = None,
		in_fs: Optional[int] = None,
		out_fs: Optional[int] = None,
		out_len_samples: Optional[int] = None,
		verbose: bool = False,
		) -> None:
		"""Convert audio file, can convert wav from/to pcm, change nchannels and sampling rate
		@@ -142,6 +143,9 @@ def convertfile(
		Input sampling rate, required for .pcm input file
		out_fs: Optional[int]
		Output sampling rate, default out_fs = in_fs
		out_len_samples: Optional[int]
		Cut file to this length in samples.
		Adds zeros at the end if bigger than file length.

		Returns
		-------
		@@ -155,6 +159,7 @@ def convertfile(
		in_nchans = 1
		x, in_fs = readfile(in_file, nchannels=in_nchans, fs=in_fs)
		in_nchans = x.shape[1]
		in_len_samples = x.shape[0]

		# Configure output file
		y = x
		@@ -167,12 +172,16 @@ def convertfile(
		print(f"Input file: {in_file}, sampling rate {str(in_fs)} size {str(x.shape)}")

		# Process
		if in_file == out_file and in_nchans == out_nchans and in_fs == out_fs:
		if in_file == out_file and in_nchans == out_nchans and in_fs == out_fs and in_len_samples == out_len_samples:
		if verbose:
		print("Convert file: nothing to be done")
		else:
		y = audioarray.convert(x, out_nchans=out_nchans, in_fs=in_fs, out_fs=out_fs)

		if out_len_samples is None:
		out_len_samples = y.shape[0]
		y = audioarray.cut(y, (0, out_len_samples))

		# write/convert wav format
		writefile(out_file, y, fs=out_fs)
		if verbose:

scripts/pyivastest/IvasModeRunner.py

100644 → 100755

+31 −168

Original line number	Diff line number	Diff line
		@@ -47,6 +47,7 @@ from pyivastest import IvasModeCollector
		from pyivastest import constants
		from pyaudio3dtools.spatialaudioformat import Format as spformat
		import pyaudio3dtools.audiofile as af
		import pyaudio3dtools.audioarray as ar
		BW_TO_SR = {"nb": 8, "wb": 16, "swb": 32, "fb": 48}
		IN_CONFIG_TO_COPY_AUDIO = {
		"SBA": [],
		@@ -583,107 +584,13 @@ class IvasModeRunner(IvasModeCollector.IvasModeCollector):
		pcm_name_res_tmp = pcm_name + ".res.wav"
		pcm_name_cpy_tmp = pcm_name + ".cpy.wav"
		in_file_name_transformed = self.transform_path(in_file_name)
		pcm_name_res_transformed = self.transform_path(pcm_name_res_tmp)
		pcm_name_cpy_transformed = self.transform_path(pcm_name_cpy_tmp)

		# check if the given length with -U is longer than the file itself and avoid cutting then
		if do_limit_duration:
		in_file_info = af.get_wav_file_info(
		in_file_name_transformed
		)
		cut_len_samples = int(
		float(self.max_duration) * in_file_info["fs"]
		)

		get_in_len_cmd = [
		os.path.join(self.config["afspPath"], "InfoAudio"),
		in_file_name_transformed,
		]
		info_audio = subprocess.check_output(get_in_len_cmd).decode(
		"utf8"
		)

		# if there is a problem with parsing the length, stick with
		# the old behaviour, hence the +1
		in_len = cut_len_samples + 1
		search_for = "No. frames: "
		for line in info_audio.splitlines():
		if line.startswith(search_for):
		try:
		in_len = int(
		line.strip().replace(search_for, "")
		)
		except:
		self.logger.console(
		"Problem with parsing length from InfoAudio",
		logging.ERROR,
		)
		break

		do_limit_duration = cut_len_samples < in_len

		if do_limit_duration:
		# TODO: is this + 8000 still needed?
		out_len = int(
		float(self.max_duration)
		* float(int(in_file_info["fs"]))
		+ 8000
		)
		limit_cpy_cmd = [
		os.path.join(self.config["afspPath"], "CopyAudio"),
		"-l",
		"0:{}".format(out_len),
		in_file_name_transformed,
		pcm_name_cpy_transformed,
		]
		pcm_log.write(" ".join(limit_cpy_cmd))
		limit_cpy_return_code = subprocess.run(
		limit_cpy_cmd, capture_output=True, text=True
		)
		if limit_cpy_return_code.returncode:
		self.logger.console(
		"Problem with time limiting copying {} before resampling!".format(
		in_file_name
		),
		logging.ERROR,
		)
		pcm_log.write(limit_cpy_return_code.stderr)
		pcm_log.write(limit_cpy_return_code.stdout)
		resamp_cmd = [
		os.path.join(self.config["afspPath"], "ResampAudio"),
		"--srate=" + str(int(sample_rate_in) * 1000),
		pcm_name_cpy_transformed,
		pcm_name_res_transformed,
		]
		else:
		resamp_cmd = [
		os.path.join(self.config["afspPath"], "ResampAudio"),
		"--srate=" + str(int(sample_rate_in) * 1000),
		in_file_name_transformed,
		pcm_name_res_transformed,
		]
		resamp_in_path = in_file_name_transformed

		pcm_log.write(" ".join(resamp_cmd))
		resamp_return_code = subprocess.run(
		resamp_cmd, capture_output=True, text=True
		)
		if resamp_return_code.returncode:
		self.logger.console(
		"Problem with resampling {} before encoding!".format(
		in_file_name
		),
		logging.ERROR,
		)
		pcm_log.write(resamp_return_code.stderr)
		pcm_log.write(resamp_return_code.stdout)
		# get number of channels from ResampAudio output
		if config["cmd"]["in_config"] == "SBA":
		nchannels = 0
		for line in resamp_return_code.stdout.split("\n"):
		if re.findall("Number of channels", line):
		nchannels = int(line.split()[3])
		in_format = spformat.detect_format(nchannels)
		break
		wav_info = af.get_wav_file_info(in_file_name_transformed)
		in_format = spformat.detect_format(wav_info["channels"])

		# save in config as json file
		in_format_dict = spformat.get_format_dict(in_format)
		@@ -692,56 +599,45 @@ class IvasModeRunner(IvasModeCollector.IvasModeCollector):

		self.pcm_info.update({pcm_name: in_format_dict})

		# read signal and sampling rate from file
		sig, fs = af.readfile(in_file_name_transformed)

		# check if the given length with -U is longer than the file itself and avoid cutting then
		if do_limit_duration:
		cut_len_samples = int(float(self.max_duration) * fs)

		in_len = sig.shape[0]

		# no need to cut anything if given length is bigger than signal length
		if cut_len_samples < in_len:
		out_len = int(float(self.max_duration) * fs)
		sig = ar.cut(sig, (0, out_len))

		pcm_log.write("Limit signal length to {} samples".format(out_len))
		af.writefile(pcm_name_cpy_transformed, sig, fs)

		resamp_in_path = pcm_name_cpy_transformed

		pcm_name_transformed = self.transform_path(pcm_name)

		out_fs = int(sample_rate_in) * 1000
		out_len = int(float(self.max_duration) * out_fs)
		pcm_log.write("Resampling to {} with length: {}".format(out_fs, out_len))
		af.convertfile(resamp_in_path, pcm_name_transformed, out_fs=out_fs, out_len_samples=out_len)

		in_config = config["cmd"]["in_config"].upper()
		chn_arg = []
		if in_config in IN_CONFIG_TO_COPY_AUDIO:
		chn_arg = IN_CONFIG_TO_COPY_AUDIO[
		config["cmd"]["in_config"].upper()
		]
		pcm_name_transformed = self.transform_path(pcm_name)

		limit_cmd = []

		if do_limit_duration:
		# avoid padding with zeros at the end when given length is bigger than the initial file length
		out_len = int(
		float(self.max_duration)
		* float(int(sample_rate_in) * 1000)
		)
		limit_cmd = ["-l", "0:{}".format(out_len)]

		cpy_cmd = (
		[
		os.path.join(self.config["afspPath"], "CopyAudio"),
		"-F",
		"noheader",
		"-D",
		"integer16",
		]
		+ limit_cmd
		+ chn_arg
		+ [pcm_name_res_transformed, pcm_name_transformed]
		)
		pcm_log.write(" ".join(cpy_cmd))
		cpy_return_code = subprocess.run(
		list(filter(None, cpy_cmd)), capture_output=True, text=True
		)
		if cpy_return_code.returncode:
		self.logger.console(
		"Problem with copying to PCM {} before encoding!".format(
		pcm_name
		),
		logging.ERROR,
		)

		pcm_log.write(cpy_return_code.stderr)
		pcm_log.write(cpy_return_code.stdout)
		pcm_log.flush()
		pcm_log.close()
		# remove file lock
		self.lock.acquire()
		os.remove(pcm_name_lock)
		os.remove(pcm_name_res_tmp)
		# os.remove(pcm_name_res_tmp)
		if do_limit_duration:
		os.remove(pcm_name_cpy_tmp)
		self.logger.info(
		@@ -1550,39 +1446,6 @@ class IvasModeRunner(IvasModeCollector.IvasModeCollector):
		"Binaries: Encoder {}, Decoder {}".format(self.encoder, self.decoder)
		)

		# check for AFSp binaries...
		bin_ext = ".exe" if platform.system() == "Windows" else ""
		if self.run_encoder:
		if self.config["afspPath"] != "not_needed":
		if not os.path.exists(self.config["afspPath"]):
		self.logger.console(
		"AFSp binaries path {} does not exist!".format(
		self.config["afspPath"]
		),
		logging.CRITICAL,
		)
		raise BinaryNotFoundError(
		"AFSp binaries path {} does not exist!".format(
		self.config["afspPath"]
		)
		)
		else:

		afsp_bin = "".join(["CopyAudio", bin_ext])
		if not os.path.exists(
		os.path.join(self.config["afspPath"], afsp_bin)
		):
		self.logger.console(
		"AFSp binary {} in path {} does not exist!".format(
		afsp_bin, self.config["afspPath"]
		),
		logging.CRITICAL,
		)
		raise BinaryNotFoundError(
		"AFSp binary {} in path {} does not exist!".format(
		afsp_bin, self.config["afspPath"]
		)
		)

		def check_and_create_out_dirs(self):
		if not os.path.exists(self.dir_name):