Merge branch '1378_conformance_script_isar' into rtyag_ivas_nonbe_conf (abae3fe9) · Commits · IVAS Codec Public Collaboration / IVAS Codec

scripts/ivas_conformance/runConformance.py

+152 −56

Original line number	Diff line number	Diff line
		@@ -191,12 +191,14 @@ class MLDConformance:
		"ENC": "IVAS_cod",
		"DEC": "IVAS_dec",
		"REND": "IVAS_rend",
		"ISAR_ENC": "IVAS_dec",
		"ISAR": "ISAR_post_rend",
		}

		def setupCommon(self):
		self.Commands = dict()
		self.EncoderToDecoderCmdMap = dict()
		self.IsarEncoderToDecoderCmdMap = dict()
		for tag in MLDConformance.IVAS_Bins.keys():
		self.Commands[tag] = list()

		@@ -207,7 +209,7 @@ class MLDConformance:
		os.makedirs(self.outputDir, exist_ok=True)
		subdirs = ["enc", "dec", "renderer_short", "split_rendering"]
		for odir in subdirs:
		os.makedirs(os.path.join(self.outputDir, "ref", odir), exist_ok=True)
		os.makedirs(os.path.join(self.testvDir, "ref", odir), exist_ok=True)
		os.makedirs(os.path.join(self.outputDir, "dut", odir), exist_ok=True)

		self.logFile = os.path.join(self.outputDir, "runlog.txt")
		@@ -261,25 +263,23 @@ class MLDConformance:
		for root, _, files in os.walk(self.testvecDir):
		for file_name in files:
		basename, ext = os.path.splitext(file_name)
		if (
		("Readme_IVAS_" in basename)
		and ext == ".txt"
		and not ("ISAR" in basename)
		):
		if ("Readme_IVAS_" in basename) and ext == ".txt":
		print(f"Accumulating commands from {file_name}")
		file = os.path.join(root, file_name)
		self.parseCommandsFile(file)
		self.mapEncoderToDecoderCommands()
		self.mapIsarEncToDecCommands()
		print("No of tests :")
		for key in self.Commands.keys():
		print(f" {key} : {len(self.Commands[key])}")

		def parseCommandsFile(self, filePath):
		isISAREnc = "IVAS_ISAR_dec" in filePath
		with open(filePath) as fp:
		for line in fp.readlines():
		m = re.search(r"^\$(CUT_.+_BIN) ", line)
		if m:
		tag = m.group(1).split("_")[1]
		tag = m.group(1).split("_")[1] if not isISAREnc else "ISAR_ENC"
		if tag in self.Commands.keys():
		self.Commands[tag].append(line)

		@@ -291,9 +291,48 @@ class MLDConformance:
		)
		return decInput.split(".")[-2]

		def getIsarDecPytestTag(self, command: str) -> str:
		getName = False
		for command in command.split():
		if getName:
		return os.path.basename(command).split(".")[-3]
		getName = True if command == "-i" else getName
		assert False, f"No match found for {command}"

		def getEncPytestTag(self, command: str) -> str:
		return os.path.basename(command.split()[-1]).split(".")[-2]

		def getIsarEncPytestTag(self, command: str) -> str:
		return os.path.basename(command.split()[-1]).split(".")[-3]

		def mapIsarEncToDecCommands(self):
		decoderPyTestTags = dict()
		encoderPyTestTags = dict()
		for idx, command in enumerate(self.Commands["ISAR"]):
		decoderPyTestTags[self.getIsarDecPytestTag(command)] = idx
		for idx, command in enumerate(self.Commands["ISAR_ENC"]):
		encoderPyTestTags[self.getIsarEncPytestTag(command)] = idx

		for encTag in encoderPyTestTags.keys():
		if encTag in decoderPyTestTags.keys():
		self.IsarEncoderToDecoderCmdMap[encoderPyTestTags[encTag]] = (
		decoderPyTestTags[encTag]
		)
		if self.args.verbose:
		print(
		f"{encTag} {encoderPyTestTags[encTag]} -> {decoderPyTestTags[encTag]}"
		)
		print(f"{self.Commands['ISAR_ENC'][encoderPyTestTags[encTag]]}")
		print(f"{self.Commands['ISAR'][decoderPyTestTags[encTag]]}")
		else:
		print(f"{encTag} not fount in decoder")
		print(
		f"Mapped decoder tests for {len(self.IsarEncoderToDecoderCmdMap)} encoder tests out of {len(self.Commands['ISAR_ENC'])} tests"
		)
		assert len(self.IsarEncoderToDecoderCmdMap) == len(
		self.Commands["ISAR_ENC"]
		), "Failed to Map Encoder Commands to Decoder Commands"

		def mapEncoderToDecoderCommands(self):
		decoderPyTestTags = dict()
		encoderPyTestTags = dict()
		@@ -322,40 +361,52 @@ class MLDConformance:
		self.Commands["ENC"]
		), "Failed to Map Encoder Commands to Decoder Commands"

		def genEncoderReferences(self, command: str, encCommandIdx: int):
		def genEncoderReferences(
		self, command: str, encCommandIdx: int, encTag: str = "ENC"
		):
		# RUN ENCODER COMMAND LINE WITH REFERENCE ENCODER
		refCommand = self.reformatCommand(command=command, ref=True)
		refCommand = self.reformatCommand(command, ref=True)
		refEncOutput = self.getOutputFile(refCommand)
		if not os.path.exists(refEncOutput):
		# if not os.path.exists(refEncOutput):
		self.process(
		command=self.setCommandExec(tag="ENC", command=refCommand, ref=True)
		command=self.setCommandExec(tag=encTag, command=refCommand, ref=True)
		)

		# FIND CORRESPONDING DECODER COMMAND
		if encTag == "ISAR_ENC":
		decTag = "ISAR"
		decCommandIdx = self.IsarEncoderToDecoderCmdMap[encCommandIdx]
		refDecOutputFile = refEncOutput.replace(".splt.bit", ".splt.REFDECODED.wav")
		else:
		decTag = "DEC"
		decCommandIdx = self.EncoderToDecoderCmdMap[encCommandIdx]
		refDecOutputFile = refEncOutput.replace(".192", "_REFDECODED.wav")

		command = self.reformatCommand(
		command=self.Commands["DEC"][decCommandIdx], ref=True
		command=self.Commands[decTag][decCommandIdx], ref=True
		)
		command = command.replace("-VOIP", "")
		if encTag == "ISAR_ENC":
		refDecCmd = [self.RefBins[decTag]] + command.split()[1:]
		else:
		refDecCmd = (
		[self.RefBins["DEC"]]
		[self.RefBins[decTag]]
		+ command.split()[1:-2]
		+ [refEncOutput, refDecOutputFile]
		)

		self.process(command=" ".join(refDecCmd))
		self.executedTests.value += 1
		self.stats()

		def runReferenceGeneration(self):
		def runReferenceGeneration(self, encTag="ENC"):
		processes = list() # Multiprocess list
		commands = conformance.Commands["ENC"]
		commands = conformance.Commands[encTag]
		self.totalTests = len(commands)
		if not self.args.no_multi_processing:
		for commandIdx, command in enumerate(commands):
		p = Process(
		target=self.genEncoderReferences, args=(command, commandIdx)
		target=self.genEncoderReferences, args=(command, commandIdx, encTag)
		)
		processes.append(p)
		p.start()
		@@ -363,7 +414,7 @@ class MLDConformance:
		p.join()
		else:
		for commandIdx, command in enumerate(commands):
		conformance.genEncoderReferences(command, commandIdx)
		conformance.genEncoderReferences(command, commandIdx, encTag)

		def runOneEncoderTest(self, command: str):
		encPytestTag = self.getEncPytestTag(command)
		@@ -380,18 +431,16 @@ class MLDConformance:

		# Run CUT Encoder
		encCommandIdx = self.Commands["ENC"].index(command)
		command = self.reformatCommand(command=command, ref=False)
		command = self.setCommandExec(tag="ENC", command=command, ref=False)
		command = self.reformatCommand(command)
		command = self.setCommandExec(tag="ENC", command=command)
		dutEncOutput = self.getOutputFile(command)
		self.process(command=command)
		self.process(command)
		assert ".192" in dutEncOutput, "Output file not identified"

		# Decode the encoded output with Reference decoder
		dutDecOutputFile = dutEncOutput.replace(".192", "_CUT_REFDECODED.wav")
		decCommandIdx = self.EncoderToDecoderCmdMap[encCommandIdx]
		command = self.reformatCommand(
		command=self.Commands["DEC"][decCommandIdx], ref=False
		)
		command = self.reformatCommand(command=self.Commands["DEC"][decCommandIdx])
		command = command.replace("-VOIP", "")
		dutDecCmd = (
		[self.RefBins["DEC"]]
		@@ -413,7 +462,7 @@ class MLDConformance:
		refDecOutput = self.getOutputFile(command).replace(
		"$CUT_PATH/ref", f"{self.testvDir}/ref"
		)
		command = self.reformatCommand(command=command, ref=False)
		command = self.reformatCommand(command)
		# command = command.replace("-VOIP", "")
		dutDecOutputFile = self.getOutputFile(command)
		dutDecCmd = (
		@@ -441,7 +490,7 @@ class MLDConformance:
		"$CUT_PATH/renderer_short", f"{self.testvDir}/renderer_short"
		)
		rendPytestTag = os.path.basename(refRendOutputFile).split(".")[-2]
		command = self.reformatCommand(command=command, ref=False)
		command = self.reformatCommand(command)
		dutRendCmd = " ".join([self.CutBins["REND"]] + command.split()[1:])
		dutRendOutputFile = self.getRendOutputFile(dutRendCmd)
		self.process(command=dutRendCmd)
		@@ -449,32 +498,67 @@ class MLDConformance:
		"REND", rendPytestTag, refFile=refRendOutputFile, dutFile=dutRendOutputFile
		)

		def runOneIsarEncoderTest(self, command: str):
		encCommandIdx = self.Commands["ISAR_ENC"].index(command)
		decCommandIdx = self.IsarEncoderToDecoderCmdMap[encCommandIdx]

		isarEncPytestTag = self.getIsarEncPytestTag(command)
		refEncCommand = self.reformatCommand(command, ref=True)
		refEncOutput = self.getOutputFile(refEncCommand)
		refDecOutputFile = refEncOutput.replace(".splt.bit", ".wav")

		# Run CUT Encoder
		dutEncCommand = self.reformatCommand(command)
		dutEncOutput = self.getOutputFile(dutEncCommand)
		self.process(command=self.setCommandExec(tag="ISAR_ENC", command=dutEncCommand))

		# Decode the encoded output with Reference decoder
		dutDecCommand = self.reformatCommand(
		command=self.Commands["ISAR"][decCommandIdx]
		)
		dutDecOutputFile = self.getRendOutputFile(dutDecCommand)
		self.process(
		command=self.setCommandExec(tag="ISAR", command=dutDecCommand, ref=True)
		)
		self.mld(
		"ISAR_ENC",
		isarEncPytestTag,
		refFile=refDecOutputFile,
		dutFile=dutDecOutputFile,
		)

		def runOneIsarDecoderTest(self, command: str):
		isarEncPytestTag = self.getIsarDecPytestTag(command)
		refDecCommand = self.reformatCommand(command, ref=True)
		refDecOutputFile = self.getRendOutputFile(refDecCommand)

		# Decode the encoded output with Reference decoder
		dutDecCommand = self.reformatCommand(command)
		dutDecOutputFile = self.getRendOutputFile(dutDecCommand)
		self.process(
		command=self.setCommandExec(tag="ISAR", command=dutDecCommand)
		)
		self.mld(
		"ISAR", isarEncPytestTag, refFile=refDecOutputFile, dutFile=dutDecOutputFile
		)

		def getOutputFile(self, command: str):
		return command.split()[-1]

		def setCommandExec(self, tag: str, command, ref: bool):
		def setCommandExec(self, tag: str, command, ref: bool = False):
		exec = self.RefBins[tag] if ref else self.CutBins[tag]
		commands = command.split()
		return " ".join([exec, *commands[1:]])

		def reformatCommand(self, command: str, ref: bool) -> str:
		def reformatCommand(self, command: str, ref: bool = False) -> str:
		command = command.replace("$TESTV_PATH", self.scriptsDir)
		command = command.replace(
		"$REF_PATH/split_rendering", f"{self.testvecDir}/testv/split_rendering"
		)

		################ HACKS #########################
		command = command.replace("_cut.192.fer", ".192")
		command = command.replace("_cut.192", ".192")
		command = command.replace(".fer.192", ".192")
		command = command.replace(".192.fer", ".192")
		##################################################
		command = command.replace(
		"$REF_PATH/ref/param_file/", f"{self.testvDir}/ref/param_file/"
		)
		command = command.replace(
		"$REF_PATH/ref/sba_bs/pkt/", f"{self.testvDir}/ref/sba_bs/pkt/"
		)

		if ref:
		command = command.replace(
		"$CUT_PATH/dut/sba_bs/pkt/", f"{self.testvDir}/ref/sba_bs/pkt/"
		@@ -489,16 +573,13 @@ class MLDConformance:
		"$CUT_PATH/renderer_short/ref/", f"{self.testvDir}/ref/renderer_short/"
		)
		command = command.replace(
		"$CUT_PATH/split_rendering/cut/",
		f"{self.testvDir}/ref/split_rendering/",
		"$CUT_PATH/split_rendering/ref",
		f"{self.testvDir}/ref/split_rendering",
		)
		command = command.replace(
		"$CUT_PATH/ref/sba_bs/", f"{self.testvDir}/ref/sba_bs/"
		)
		else:
		#command = command.replace(
		# "$CUT_PATH/dut/sba_bs/pkt/", f"{self.outputDir}/dut/enc/"
		#)
		command = command.replace(
		"$CUT_PATH/ref/param_file/enc/", f"{self.outputDir}/dut/enc/"
		)
		@@ -512,6 +593,10 @@ class MLDConformance:
		"$CUT_PATH/split_rendering/cut/",
		f"{self.outputDir}/dut/split_rendering/",
		)
		command = command.replace(
		"$CUT_PATH/split_rendering/ref",
		f"{self.outputDir}/dut/split_rendering",
		)
		command = command.replace(
		"$CUT_PATH/ref/sba_bs/pkt/", f"{self.outputDir}/dut/enc/"
		)
		@@ -519,6 +604,15 @@ class MLDConformance:
		"$CUT_PATH/ref/sba_bs/raw/", f"{self.outputDir}/dut/dec/"
		)

		command = command.replace(
		"$REF_PATH/split_rendering", f"{self.testvecDir}/testv/split_rendering"
		)
		command = command.replace(
		"$REF_PATH/ref/param_file/", f"{self.testvDir}/ref/param_file/"
		)
		command = command.replace(
		"$REF_PATH/ref/sba_bs/pkt/", f"{self.testvDir}/ref/sba_bs/pkt/"
		)
		return command

		def runOneCommand(self, tag: str, command: str):
		@@ -528,6 +622,10 @@ class MLDConformance:
		self.runOneDecoderTest(tag, command)
		elif tag == "REND":
		self.runOneRendererTest(tag, command)
		elif tag == "ISAR_ENC":
		self.runOneIsarEncoderTest(command)
		elif tag == "ISAR":
		self.runOneIsarDecoderTest(command)
		else:
		assert False, f"Un-implemented Tag {tag}"
		self.executedTests.value += 1
		@@ -632,10 +730,10 @@ class MLDConformance:
		tmpdir, f"{tempfile.gettempprefix()}_ch{ch}_MLD2.txt"
		)
		refFileMono = os.path.join(
		tmpdir, os.path.basename(refFile).replace(".wav", f"_ch{ch}.wav")
		tmpdir, os.path.basename(refFile).replace(".wav", f"_REF_ch{ch}.wav")
		)
		dutFileMono = os.path.join(
		tmpdir, os.path.basename(dutFile).replace(".wav", f"_ch{ch}.wav")
		tmpdir, os.path.basename(dutFile).replace(".wav", f"_DUT_ch{ch}.wav")
		)
		writefile(refFileMono, refSamples[:, ch], 48000)
		writefile(dutFileMono, dutSamples[:, ch], 48000)
		@@ -777,16 +875,14 @@ if __name__ == "__main__":
		conformance.accumulateCommands()

		if args.regenerate_enc_refs:
		conformance.runReferenceGeneration()
		conformance.runReferenceGeneration(encTag="ISAR_ENC")
		conformance.runReferenceGeneration(encTag="ENC")
		sys.exit(0)

		testTags = (
		MLDConformance.IVAS_Bins.keys() if args.test_mode == "ALL" else [args.test_mode]
		)
		for tag in testTags:
		if tag == "ISAR":
		# Not implemented yet
		continue
		if not args.analyse_only:
		conformance.runTag(tag)
		conformance.doAnalysis(selectTag=tag)