Add preliminary corridor based on pyTestTags (9e5a036a) · Commits · IVAS Codec Public Collaboration / IVAS Codec

scripts/ivas_conformance/runConformance.py

+109 −50

Original line number	Diff line number	Diff line
		@@ -195,9 +195,17 @@ IVAS_Bins = {
		"ISAR": "ISAR_post_rend",
		}

		ReferenceMldFiles = {
		"ENC": "mld_ref_ENC.csv",
		"DEC": "mld_ref_DEC.csv",
		"REND": "mld_ref_REND.csv",
		"ISAR_ENC": "mld_ref_ISAR_ENC.csv",
		"ISAR": "mld_ref_ISAR.csv",
		}


		@dataclass
		class TestDesciptor:
		class TestDescriptor:
		dutOutput: str = ""
		refOutput: str = ""
		rawCmdline: str = ""
		@@ -206,7 +214,7 @@ class TestDesciptor:


		@dataclass
		class BitstrmTestDescriptor(TestDesciptor):
		class BitstreamTestDescriptor(TestDescriptor):
		rawDecCmdline: str = ""
		dutDecCmdline: str = ""
		refDecCmdline: str = ""
		@@ -216,7 +224,7 @@ class MLDConformance:
		def setupCommon(self):
		self.Commands = dict()
		self.TestDesc = dict[
		str, dict[str, Union[TestDesciptor, BitstrmTestDescriptor]]
		str, dict[str, Union[TestDescriptor, BitstreamTestDescriptor]]
		]()
		self.EncoderToDecoderCmdMap = dict()
		self.IsarEncoderToDecoderCmdMap = dict()
		@@ -325,13 +333,13 @@ class MLDConformance:

		def createTestDescriptors(
		self,
		) -> dict[str, dict[str, Union[TestDesciptor, BitstrmTestDescriptor]]]:
		) -> dict[str, dict[str, Union[TestDescriptor, BitstreamTestDescriptor]]]:
		testDesciptor = dict[
		str, dict[str, Union[TestDesciptor, BitstrmTestDescriptor]]
		str, dict[str, Union[TestDescriptor, BitstreamTestDescriptor]]
		]()
		for tag in IVAS_Bins.keys():
		testDesciptor[tag] = dict[
		str, Union[TestDesciptor, BitstrmTestDescriptor]
		str, Union[TestDescriptor, BitstreamTestDescriptor]
		]()

		IvasDecCmdMap = dict()
		@@ -346,7 +354,7 @@ class MLDConformance:
		)
		refOutput = self.getRendOutputFile(refCmdline)
		dutOutput = self.getRendOutputFile(dutCmdline)
		testDesciptor["REND"][pyTestTag] = TestDesciptor(
		testDesciptor["REND"][pyTestTag] = TestDescriptor(
		rawCmdline=command,
		refOutput=refOutput,
		dutOutput=dutOutput,
		@@ -363,7 +371,7 @@ class MLDConformance:
		)
		refOutput = self.getOutputFile(refCmdline)
		dutOutput = self.getOutputFile(dutCmdline)
		testDesciptor["DEC"][pyTestTag] = TestDesciptor(
		testDesciptor["DEC"][pyTestTag] = TestDescriptor(
		rawCmdline=command,
		refOutput=refOutput,
		dutOutput=dutOutput,
		@@ -381,7 +389,7 @@ class MLDConformance:
		)
		refOutput = self.getRendOutputFile(refCmdline)
		dutOutput = self.getRendOutputFile(dutCmdline)
		testDesciptor["ISAR"][pyTestTag] = TestDesciptor(
		testDesciptor["ISAR"][pyTestTag] = TestDescriptor(
		rawCmdline=command,
		refOutput=refOutput,
		dutOutput=dutOutput,
		@@ -411,7 +419,7 @@ class MLDConformance:
		dutDecCmdline = self.setCommandExec(
		tag="DEC", command=self.reformatCommand(rawDecCmdline)
		)
		testDesciptor["ENC"][pyTestTag] = BitstrmTestDescriptor(
		testDesciptor["ENC"][pyTestTag] = BitstreamTestDescriptor(
		rawCmdline=command,
		rawDecCmdline=rawDecCmdline,
		refOutput=refOutput,
		@@ -445,7 +453,7 @@ class MLDConformance:
		dutDecCmdline = self.setCommandExec(
		tag="ISAR", command=self.reformatCommand(rawDecCmdline)
		)
		testDesciptor["ISAR_ENC"][pyTestTag] = BitstrmTestDescriptor(
		testDesciptor["ISAR_ENC"][pyTestTag] = BitstreamTestDescriptor(
		rawCmdline=command,
		rawDecCmdline=rawDecCmdline,
		refOutput=refOutput,
		@@ -467,7 +475,7 @@ class MLDConformance:
		# RUN ENCODER'S OUTPUT DECODED WITH REF DECODER
		testDesc = self.TestDesc[tag][encPytestTag]
		assert isinstance(
		testDesc, BitstrmTestDescriptor
		testDesc, BitstreamTestDescriptor
		), f"Expected bitstream test descriptor for {tag}"

		# Decode the encoded output with Reference IVAS decoder
		@@ -503,13 +511,20 @@ class MLDConformance:
		def analyseWavOutputTest(self, tag: str, dutPytestTag: str):
		testDesc = self.TestDesc[tag][dutPytestTag]
		assert isinstance(
		testDesc, TestDesciptor
		testDesc, TestDescriptor
		), f"Expected pcm test descriptor for {tag}"

		if self.args.be_test:
		DUTmdFiles = self.getMDfileList(outFile=testDesc.dutOutput)
		REFmdFiles = self.getMDfileList(outFile=testDesc.refOutput)
		self.beTest(tag, dutPytestTag, refFile=testDesc.refOutput, dutFile=testDesc.dutOutput, DUTmdFileList=DUTmdFiles, REFmdFileList=REFmdFiles )
		self.beTest(
		tag,
		dutPytestTag,
		refFile=testDesc.refOutput,
		dutFile=testDesc.dutOutput,
		DUTmdFileList=DUTmdFiles,
		REFmdFileList=REFmdFiles,
		)
		else:
		##### skip MLD verification for files with only 1 frame as MLD does not run with such files. Possible solution: append 0s and then compare #####
		if testDesc.rawCmdline.find("_cut.192.fer") == -1:
		@@ -523,11 +538,16 @@ class MLDConformance:
		def analyseOneEncoderTest(self, tag: str, encPytestTag: str):
		testDesc = self.TestDesc[tag][encPytestTag]
		assert isinstance(
		testDesc, BitstrmTestDescriptor
		testDesc, BitstreamTestDescriptor
		), f"Expected bitstream test descriptor for {tag}"

		if self.args.be_test:
		self.beTest(tag, encPytestTag, refFile=testDesc.refOutput, dutFile=testDesc.dutOutput )
		self.beTest(
		tag,
		encPytestTag,
		refFile=testDesc.refOutput,
		dutFile=testDesc.dutOutput,
		)
		else:
		refDecOutputFile = testDesc.refOutput.replace(".192", "_REFDECODED.wav")

		@@ -540,16 +560,20 @@ class MLDConformance:
		dutDecCmd = ["" if x == "-VOIP" else x for x in dutDecCmd]
		dutDecCmd = " ".join(dutDecCmd)
		self.process(command=dutDecCmd)
		self.mld(tag, encPytestTag, refFile=refDecOutputFile, dutFile=dutDecOutputFile)
		self.mld(
		tag, encPytestTag, refFile=refDecOutputFile, dutFile=dutDecOutputFile
		)

		def analyseOneIsarEncoderTest(self, tag: str, pytestTag: str):
		testDesc = self.TestDesc[tag][pytestTag]
		assert isinstance(
		testDesc, BitstrmTestDescriptor
		testDesc, BitstreamTestDescriptor
		), f"Expected bitstream test descriptor for {tag}"

		if self.args.be_test:
		self.beTest(tag, pytestTag, refFile=testDesc.refOutput, dutFile=testDesc.dutOutput )
		self.beTest(
		tag, pytestTag, refFile=testDesc.refOutput, dutFile=testDesc.dutOutput
		)
		else:
		refDecOutputFile = testDesc.refOutput.replace(".splt.bit", ".wav")
		dutDecOutputFile = testDesc.dutOutput.replace(".splt.bit", ".wav")
		@@ -576,11 +600,10 @@ class MLDConformance:
		def getMDfileList(self, outFile: str):
		MDfiles = []
		for i in range(0, 3):
		MDfiles.append(outFile + '.' + str(i) + '.csv')
		MDfiles.append(outFile + '.met')
		MDfiles.append(outFile + "." + str(i) + ".csv")
		MDfiles.append(outFile + ".met")
		return MDfiles


		def setCommandExec(self, tag: str, command, ref: bool = False):
		exec = self.RefBins[tag] if ref else self.CutBins[tag]
		commands = command.split()
		@@ -717,7 +740,7 @@ class MLDConformance:
		if self.args.be_test:
		self.doBEanalysis(selectTag=tag)
		else:
		self.doAnalysis(selectTag=tag)
		self.doAnalysis(selectTag=tag, corridor=True)

		def process(self, command) -> int:
		if self.args.verbose:
		@@ -820,21 +843,27 @@ class MLDConformance:
		mldThisFile = np.maximum(mldThisFile, mldThisChan)

		if mldThisFile.size > 0:
		mldWithTags = np.column_stack(
		(
		mldThisFile,
		np.array([f"{pytestTag}-FRM{x}" for x in range(mldThisFile.size)]),
		)
		)
		with open(self.mldcsv[tag], "ab") as f:
		np.savetxt(f, mldThisFile, delimiter=",")
		np.savetxt(f, mldWithTags, fmt="%s", delimiter=",")
		with open(self.sampleStats[tag], "a") as f:
		f.write(
		f"{pytestTag}, {maxDiff}, {rmsdB}, {beSamplesPercent}, {mldThisFile.max()}\n"
		)

		def beTest(self, tag, pytestTag, refFile, dutFile, DUTmdFileList=[], REFmdFileList=[]):
		def beTest(
		self, tag, pytestTag, refFile, dutFile, DUTmdFileList=[], REFmdFileList=[]
		):
		BE_flag = 0
		if not filecmp.cmp(refFile, dutFile):
		BE_flag = 1
		with open(self.BEcsv[tag], "a") as f:
		f.write(
		f"{pytestTag}, {BE_flag}\n"
		)
		f.write(f"{pytestTag}, {BE_flag}\n")

		for i in range(0, len(DUTmdFileList)):
		if os.path.exists(DUTmdFileList[i]):
		@@ -842,10 +871,7 @@ class MLDConformance:
		if not filecmp.cmp(REFmdFileList[i], DUTmdFileList[i]):
		BE_flag = 1
		with open(self.BEcsv[tag], "a") as f:
		f.write(
		f"{DUTmdFileList[i]}, {BE_flag}\n"
		)

		f.write(f"{DUTmdFileList[i]}, {BE_flag}\n")

		def doBEanalysis(self, selectTag="all"):
		keys = IVAS_Bins.keys() if selectTag == "all" else [selectTag]
		@@ -863,11 +889,32 @@ class MLDConformance:
		else:
		print(f"<{tag}> PASSED BE TEST")

		def doAnalysis(self, selectTag="all"):
		def computeCorridor(self, mldRefWithTags, mldCutWithTags, tag, threshold=1.0):
		mldDict = dict[str, list]()
		failed = False

		intersection, indRef, indDut = np.intersect1d(
		mldRefWithTags["pyTestTag"],
		mldCutWithTags["pyTestTag"],
		return_indices=True,
		)
		refMLD = mldRefWithTags["MLD"][indRef]
		dutMLD = mldCutWithTags["MLD"][indDut]
		diff = dutMLD - refMLD
		if diff.max() > threshold:
		print(f"MLD Corridor failed for {tag} with max MLD diff of {diff.max()}")
		else:
		print(f"MLD Corridor passed for {tag} with max MLD diff of {diff.max()}")

		def doAnalysis(self, selectTag="all", corridor=False):
		keys = IVAS_Bins.keys() if selectTag == "all" else [selectTag]
		for tag in keys:
		if os.path.exists(self.mldcsv[tag]):
		mdlValues = np.loadtxt(self.mldcsv[tag], delimiter=" ", dtype=float)
		mdlCutWithTags = np.loadtxt(
		self.mldcsv[tag],
		delimiter=",",
		dtype=[("MLD", "f8"), ("pyTestTag", "<U256")],
		)
		bePercent = np.loadtxt(
		self.sampleStats[tag],
		delimiter=",",
		@@ -884,6 +931,7 @@ class MLDConformance:
		)
		bePercentAvg = np.average(bePercent)
		maxDiffmax = np.max(maxDiff) * 32768.0
		mdlValues = mdlCutWithTags["MLD"]
		N = mdlValues.shape[0]
		if N == 0:
		continue
		@@ -908,6 +956,17 @@ class MLDConformance:
		)
		print("##########################################################\n")

		if corridor:
		refMldFile = os.path.join(
		self.testvDir, "mld_ref", ReferenceMldFiles[tag]
		)
		mldRefWithTags = np.loadtxt(
		refMldFile,
		delimiter=",",
		dtype=[("MLD", "f8"), ("pyTestTag", "<U90")],
		)
		self.computeCorridor(mldRefWithTags, mdlCutWithTags, tag)


		if __name__ == "__main__":
		parser = argparse.ArgumentParser(
		@@ -973,7 +1032,7 @@ if __name__ == "__main__":
		"--be-test",
		default=False,
		action="store_true",
		help='runs only BE tests',
		help="runs only BE tests",
		)
		parser.add_argument(
		"--no-multi-processing",
		@@ -1013,7 +1072,7 @@ if __name__ == "__main__":
		testTags = IVAS_Bins.keys() if args.test_mode == "ALL" else [args.test_mode]
		for tag in testTags:
		if args.report_only:
		conformance.doAnalysis(selectTag=tag)
		conformance.doAnalysis(selectTag=tag, corridor=True)
		elif not args.analyse:
		conformance.runTag(tag)
		else: