pytest: add network simulator call to fix new JBM tests (7638cd85) · Commits · IVAS Codec Public Collaboration / IVAS Codec

tests/test_param_file.py

+134 −8

Original line number	Diff line number	Diff line
		@@ -31,11 +31,29 @@ the United Nations Convention on Contracts on the International Sales of Goods.
		import os
		import errno
		import pytest
		import platform
		from subprocess import run
		from cmp_custom import cmp_custom
		from conftest import EncoderFrontend, DecoderFrontend
		from testconfig import PARAM_FILE


		VALID_DEC_OUTPUT_CONF = [
		"MONO",
		"STEREO",
		"5_1",
		"7_1",
		"5_1_2",
		"5_1_4",
		"7_1_4",
		"FOA",
		"HOA2",
		"HOA3",
		"BINAURAL",
		"BINAURAL_ROOM",
		"EXT",
		]

		param_file_test_dict = {}
		with open(PARAM_FILE, "r", encoding="UTF-8") as fp:
		data = fp.read()
		@@ -44,6 +62,7 @@ with open(PARAM_FILE, "r", encoding="UTF-8") as fp:
		tag = ""
		enc_opts = ""
		dec_opts = ""
		sim_opts = ""
		for line in block.split("\n"):
		if line.startswith("// "):
		tag = line[3:]
		@@ -51,13 +70,15 @@ with open(PARAM_FILE, "r", encoding="UTF-8") as fp:
		enc_opts = line[12:]
		if line.startswith("../IVAS_dec "):
		dec_opts = line[12:]
		if line.startswith("networkSimulator_g192 "):
		sim_opts = line[22:]
		if tag == "" or enc_opts == "" or dec_opts == "":
		# no complete parameter set
		continue
		if tag in param_file_test_dict:
		print("non-unique tag found - ignoring new entry")
		continue
		param_file_test_dict[tag] = (enc_opts, dec_opts)
		param_file_test_dict[tag] = (enc_opts, dec_opts, sim_opts)


		def check_and_makedir(dir_path):
		@@ -101,9 +122,10 @@ def test_param_file_tests(
		dut_base_path,
		test_vector_path,
		update_ref,
		rootdir,
		test_tag,
		):
		enc_opts, dec_opts = param_file_test_dict[test_tag]
		enc_opts, dec_opts, sim_opts = param_file_test_dict[test_tag]

		tag_str = convert_test_string_to_tag(test_tag)

		@@ -148,6 +170,36 @@ def test_param_file_tests(
		update_ref,
		)

		# check for networkSimulator_g192 command line
		if sim_opts != "":
		sim_split = sim_opts.split()
		assert len(sim_split) == 6, "networkSimulator_g192 binary expects 6 parameters"
		# [sim_profile, sim_input, sim_output, sim_trace, sim_nFPP, sim_offset] = sim_split
		if sim_split[0].startswith(("../")):
		# remove leading "../"
		sim_split[0] = sim_split[0][3:]
		assert sim_split[1] == "bit"
		# in the parameter file, only "bit" is used as bitstream file name
		# -> re-use bitstream filename from encoder call
		sim_split[1] = bitstream_file
		assert sim_split[2] == "netsimoutput"
		# in the parameter file, only "netsimoutput" is used as netsim output file name
		# -> construct netsim output file name
		netsim_outfile = f"{testv_base}_{tag_str}.netsimout"
		sim_split[2] = netsim_outfile
		assert sim_split[3] == "tracefile_sim"
		# in the parameter file, only "tracefile_sim" is used as trace output file name
		# -> construct trace output file name
		netsim_trace_outfile = f"{testv_base}_{tag_str}.netsimtrace"
		sim_split[3] = netsim_trace_outfile
		simulate(
		reference_path,
		dut_base_path,
		sim_split,
		update_ref,
		rootdir,
		)

		# evaluate decoder options
		dec_split = dec_opts.split()
		assert len(dec_split) >= 3
		@@ -165,6 +217,11 @@ def test_param_file_tests(
		sampling_rate = int(dec_split.pop())
		if len(dec_split) > 0:
		output_config = dec_split.pop()
		if output_config.upper() not in VALID_DEC_OUTPUT_CONF:
		if not output_config.endswith(".txt"):
		# must be EVS tests with additional parameters - put param back
		dec_split.append(output_config)
		output_config = ""
		else:
		output_config = ""

		@@ -173,6 +230,14 @@ def test_param_file_tests(
		# the output config is a file
		output_config_name = os.path.splitext(os.path.basename(output_config))[0]

		tracefile_dec = ""
		if sim_opts != "":
		assert bitstream_file_dec == "netsimoutput"
		# in the parameter file, only "netsimoutput" is used as bitstream file name
		# -> re-use netsim_outfile
		bitstream_file = netsim_outfile
		tracefile_dec = f"{testv_base}_{tag_str}.dectrace"
		else:
		assert bitstream_file_dec == "bit"
		# in the parameter file, only "bit" is used as bitstream file name
		# -> re-use bitstream filename from encoder call
		@@ -196,6 +261,7 @@ def test_param_file_tests(
		output_file,
		dec_split,
		update_ref,
		tracefile_dec,
		)

		# compare
		@@ -222,7 +288,7 @@ def encode(
		update_ref,
		):
		"""
		Call REF and/or DUT decoder.
		Call REF and/or DUT encoder.
		"""
		# directories
		dut_out_dir = f"{dut_base_path}/param_file/enc"
		@@ -256,6 +322,53 @@ def encode(
		)


		def simulate(
		reference_path,
		dut_base_path,
		sim_opts_list,
		update_ref,
		rootdir,
		):
		"""
		Call network simulator on REF and/or DUT encoder output.
		"""
		# directories
		dut_out_dir = f"{dut_base_path}/param_file/enc"
		ref_out_dir = f"{reference_path}/param_file/enc"

		netsim_infile = sim_opts_list[1]
		netsim_outfile = sim_opts_list[2]
		netsim_tracefile = sim_opts_list[3]
		ref_out_file = f"{ref_out_dir}/{netsim_outfile}"

		if platform.system() == "Windows":
		netsim = [os.path.join(rootdir, "scripts", "tools", "Win32", "networkSimulator_g192.exe")]
		elif platform.system() == "Linux":
		# there is no Linux binary available -> use the Win32 binary via wine
		netsim = [
		"wine",
		os.path.join(rootdir, "scripts", "tools", "Win32", "networkSimulator_g192.exe"),
		]
		elif platform.system() == "Darwin":
		netsim = [os.path.join(rootdir, "scripts", "tools", "Darwin", "networkSimulator_g192")]

		if update_ref == 1 or update_ref == 2 and not os.path.exists(ref_out_file):
		# call network simulator on REF encoder output
		cmd_opts = sim_opts_list
		cmd_opts[1] = f"{ref_out_dir}/{netsim_infile}"
		cmd_opts[2] = f"{ref_out_dir}/{netsim_outfile}" # ref_out_file
		cmd_opts[3] = f"{ref_out_dir}/{netsim_tracefile}"
		run(netsim + cmd_opts, check=False)

		if update_ref in [0, 2]:
		# call network simulator on DUT encoder output
		cmd_opts = sim_opts_list
		cmd_opts[1] = f"{dut_out_dir}/{netsim_infile}"
		cmd_opts[2] = f"{dut_out_dir}/{netsim_outfile}" # dut_out_file
		cmd_opts[3] = f"{dut_out_dir}/{netsim_tracefile}"
		run(netsim + cmd_opts, check=False)


		def decode(
		decoder_frontend,
		ref_decoder_path,
		@@ -267,6 +380,7 @@ def decode(
		output_file,
		dec_opts_list,
		update_ref,
		tracefile_dec,
		):
		"""
		Call REF and/or DUT decoder.
		@@ -282,6 +396,12 @@ def decode(

		if update_ref == 1 or update_ref == 2 and not os.path.exists(ref_out_file):
		check_and_makedir(ref_out_dir)
		add_option_list = dec_opts_list
		if tracefile_dec != "":
		add_option_list = [
		x if x != "tracefile_dec" else f"{ref_out_dir}/{tracefile_dec}"
		for x in dec_opts_list
		]
		# call REF decoder
		assert ref_decoder_path
		ref_decoder = DecoderFrontend(ref_decoder_path, "REF")
		@@ -290,18 +410,24 @@ def decode(
		sampling_rate,
		ref_in_file,
		ref_out_file,
		add_option_list=dec_opts_list,
		add_option_list=add_option_list,
		)

		if update_ref in [0, 2]:
		check_and_makedir(dut_out_dir)
		add_option_list = dec_opts_list
		if tracefile_dec != "":
		add_option_list = [
		x if x != "tracefile_dec" else f"{dut_out_dir}/{tracefile_dec}"
		for x in dec_opts_list
		]
		# call DUT decoder
		decoder_frontend.run(
		output_config,
		sampling_rate,
		dut_in_file,
		dut_out_file,
		add_option_list=dec_opts_list,
		add_option_list=add_option_list,
		)