Add reference generation for encoder (4725761d) · Commits · IVAS Codec Public Collaboration / IVAS Codec

scripts/ivas_conformance/runConformance.py

+124 −92

Original line number	Diff line number	Diff line
		@@ -55,64 +55,67 @@ class MLDConformance:
		"ISAR": "ISAR_post_rend",
		}

		def __init__(self, args) -> None:
		self.RefBins = dict()
		self.CutBins = dict()
		def setupCommon(self):
		self.Commands = dict()
		self.multiprocessing = not args.no_multi_processing
		self.regenEncRefs = args.regenerate_enc_refs
		self.dryrun = args.dryrun
		self.verbose = args.verbose
		self.executedTests = Value("i", 0)
		self.failedTests = Value("i", 0)
		self.testvecDir = args.testvecDir
		self.ref_build_path = args.ref_build_path
		self.cut_build_path = args.cut_build_path
		self.filter = args.filter
		self.EncoderToDecoderCmdMap = dict()
		for tag in MLDConformance.IVAS_Bins.keys():
		self.Commands[tag] = list()

		self.scriptsDir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
		# CREATE OUTPUT DIRECTORY STRUCTURE : CLEAN PREV OUTPUT
		self.outputDir = os.path.join(self.scriptsDir, "CUT_OUTPUTS")
		if os.path.exists(self.outputDir):
		shutil.rmtree(self.outputDir, ignore_errors=False)
		os.mkdir(self.outputDir)
		self.toolsdir = os.path.join(self.scriptsDir, "tools")
		self.testvDir = os.path.join(self.testvecDir, "testv")
		self.refDir = self.testvDir
		self.mldbin = os.path.join(self.toolsdir, platform.system(), "mld")
		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.outputDir, "dut", odir), exist_ok=True)

		self.logFile = os.path.join(self.outputDir, "runlog.txt")
		self.failedCmdsFile = os.path.join(self.outputDir, "failedCmds.txt")
		open(self.logFile, "w").close()
		open(self.failedCmdsFile, "w").close()

		def setupDUT(self):
		self.cut_build_path = args.cut_build_path
		self.filter = args.filter
		self.mldbin = os.path.join(self.toolsdir, platform.system(), "mld")
		self.CutBins = dict()
		self.mldcsv = dict()
		self.sampleStats = dict()

		for tag in MLDConformance.IVAS_Bins.keys():
		self.CutBins[tag] = os.path.join(
		self.cut_build_path, MLDConformance.IVAS_Bins[tag]
		)
		self.mldcsv[tag] = os.path.join(self.outputDir, f"mld_{tag}.csv")
		self.sampleStats[tag] = os.path.join(
		self.outputDir, f"sampleStats_{tag}.csv"
		)

		self.setup()

		def createDirs(self):
		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.outputDir, "dut", odir), exist_ok=True)

		def setup(self):
		self.createDirs()
		def setupRef(self):
		self.RefBins = dict()
		self.ref_build_path = self.args.ref_build_path
		for tag in MLDConformance.IVAS_Bins.keys():
		self.RefBins[tag] = os.path.join(
		self.ref_build_path, MLDConformance.IVAS_Bins[tag]
		)
		self.CutBins[tag] = os.path.join(
		self.cut_build_path, MLDConformance.IVAS_Bins[tag]
		)
		self.Commands[tag] = list()

		def setup(self):
		self.setupCommon()
		self.setupRef()
		if not self.args.regenerate_enc_refs:
		self.setupDUT()

		def __init__(self, args) -> None:
		self.args = args
		self.scriptsDir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
		self.testvecDir = args.testvecDir
		self.toolsdir = os.path.join(self.scriptsDir, "tools")
		self.testvDir = os.path.join(self.testvecDir, "testv")
		self.executedTests = Value("i", 0)
		self.failedTests = Value("i", 0)
		self.setup()

		def accumulateCommands(self):
		for root, _, files in os.walk(self.testvecDir):
		@@ -164,7 +167,7 @@ class MLDConformance:
		self.EncoderToDecoderCmdMap[encoderPyTestTags[encTag]] = (
		decoderPyTestTags[encTag]
		)
		if self.verbose:
		if self.args.verbose:
		print(
		f"{encTag} {encoderPyTestTags[encTag]} -> {decoderPyTestTags[encTag]}"
		)
		@@ -179,17 +182,51 @@ class MLDConformance:
		self.Commands["ENC"]
		), "Failed to Map Encoder Commands to Decoder Commands"

		def runOneEncoderTest(self, command: str):

		encPytestTag = self.getEncPytestTag(command)

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

		# FIND CORRESPONDING DECODER COMMAND
		decCommandIdx = self.EncoderToDecoderCmdMap[encCommandIdx]
		refDecOutputFile = refEncOutput.replace(".192", "_REFDECODED.wav")

		command = self.reformatCommand(
		command=self.Commands["DEC"][decCommandIdx], ref=True
		)
		command = command.replace("-VOIP", "")
		refDecCmd = (
		[self.RefBins["DEC"]]
		+ command.split()[1:-2]
		+ [refEncOutput, refDecOutputFile]
		)
		self.process(command=" ".join(refDecCmd))
		self.executedTests.value += 1
		self.stats()

		def runReferenceGeneration(self):
		processes = list() # Multiprocess list
		commands = conformance.Commands["ENC"]
		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)
		)
		processes.append(p)
		p.start()
		for p in processes:
		p.join()
		else:
		for commandIdx, command in enumerate(commands):
		conformance.genEncoderReferences(command, commandIdx)

		def runOneEncoderTest(self, command: str):
		encPytestTag = self.getEncPytestTag(command)
		refEncOutput = self.getOutputFile(command)
		refEncOutput = refEncOutput.replace(
		"$CUT_PATH/ref/param_file/enc/",
		@@ -199,8 +236,9 @@ class MLDConformance:
		"$CUT_PATH/ref/sba_bs/pkt/",
		f"{self.testvecDir}/testv/ref/sba_bs/pkt/",
		)
		refDecOutputFile = refEncOutput.replace(".192", "_REFDECODED.wav")

		# Run reference Encoder
		# 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)
		@@ -210,7 +248,6 @@ class MLDConformance:

		# Decode the encoded output with Reference decoder
		dutDecOutputFile = dutEncOutput.replace(".192", "_CUT_REFDECODED.wav")
		refDecOutputFile = dutEncOutput.replace(".192", "_REF_REFDECODED.wav")
		decCommandIdx = self.EncoderToDecoderCmdMap[encCommandIdx]
		command = self.reformatCommand(
		command=self.Commands["DEC"][decCommandIdx], ref=False
		@@ -221,13 +258,7 @@ class MLDConformance:
		+ command.split()[1:-2]
		+ [dutEncOutput, dutDecOutputFile]
		)
		refDecCmd = (
		[self.RefBins["DEC"]]
		+ command.split()[1:-2]
		+ [refEncOutput, refDecOutputFile]
		)
		self.process(command=" ".join(dutDecCmd))
		self.process(command=" ".join(refDecCmd))
		self.mld(
		"ENC", encPytestTag, refFile=refDecOutputFile, dutFile=dutDecOutputFile
		)
		@@ -254,7 +285,9 @@ class MLDConformance:

		##### 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 refInputFile.find("_cut.192.fer") == -1:
		self.mld("DEC", dutPytestTag, refFile=refDecOutput, dutFile=dutDecOutputFile)
		self.mld(
		"DEC", dutPytestTag, refFile=refDecOutput, dutFile=dutDecOutputFile
		)

		def getRendOutputFile(self, command: str):
		cmds = command.split()
		@@ -285,7 +318,7 @@ class MLDConformance:
		return " ".join([exec, *commands[1:]])

		def reformatCommand(self, command: str, ref: bool) -> str:
		command = command.replace("$TESTV_PATH", self.testvecDir)
		command = command.replace("$TESTV_PATH", self.scriptsDir)
		command = command.replace(
		"$REF_PATH/split_rendering", f"{self.testvecDir}/testv/split_rendering"
		)
		@@ -296,38 +329,36 @@ class MLDConformance:
		command = command.replace(".fer.192", ".192")
		command = command.replace(".192.fer", ".192")
		##################################################

		command = command.replace(
		"$REF_PATH/ref/param_file/enc/", f"{self.outputDir}/ref/enc/"
		)
		command = command.replace(
		"$REF_PATH/ref/param_file/dec/", f"{self.outputDir}/ref/dec/"
		"$REF_PATH/ref/param_file/", f"{self.testvDir}/ref/param_file/"
		)
		command = command.replace(
		"$REF_PATH/ref/sba_bs/pkt/", f"{self.outputDir}/ref/enc/"
		"$REF_PATH/ref/sba_bs/pkt/", f"{self.testvDir}/ref/sba_bs/pkt/"
		)

		if ref:
		command = command.replace(
		"$CUT_PATH/ref/param_file/enc/", f"{self.outputDir}/ref/enc/"
		"$CUT_PATH/dut/sba_bs/pkt/", f"{self.testvDir}/ref/sba_bs/pkt/"
		)
		command = command.replace(
		"$CUT_PATH/ref/param_file/dec/", f"{self.outputDir}/ref/dec/"
		"$CUT_PATH/dut/param_file/", f"{self.testvDir}/ref/param_file/"
		)
		command = command.replace(
		"$CUT_PATH/renderer_short/ref/", f"{self.outputDir}/ref/renderer_short/"
		"$CUT_PATH/ref/param_file/", f"{self.testvDir}/ref/param_file/"
		)
		command = command.replace(
		"$CUT_PATH/split_rendering/cut/",
		f"{self.outputDir}/ref/split_rendering/",
		"$CUT_PATH/renderer_short/ref/", f"{self.testvDir}/ref/renderer_short/"
		)
		command = command.replace(
		"$CUT_PATH/ref/sba_bs/pkt/", f"{self.outputDir}/ref/enc/"
		"$CUT_PATH/split_rendering/cut/",
		f"{self.testvDir}/ref/split_rendering/",
		)
		command = command.replace(
		"$CUT_PATH/ref/sba_bs/raw/", f"{self.outputDir}/ref/dec/"
		"$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/"
		)
		@@ -350,7 +381,7 @@ class MLDConformance:

		return command

		def runOneCommand(self, tag: str, command: str, ref: bool):
		def runOneCommand(self, tag: str, command: str):
		if tag == "ENC":
		self.runOneEncoderTest(command)
		elif tag == "DEC":
		@@ -362,7 +393,7 @@ class MLDConformance:
		self.executedTests.value += 1
		self.stats()

		def runTag(self, tag: str, ref: bool = False):
		def runTag(self, tag: str):
		self.executedTests.value = 0
		self.failedTests.value = 0
		# reset MLD, Sample Stats
		@@ -383,11 +414,11 @@ class MLDConformance:
		print(
		f"Executing tests for {tag} {'Filter='+self.filter if self.filter else ''} ({self.totalTests} tests)"
		)
		if self.multiprocessing:
		if not self.args.no_multi_processing:
		for command in commands:
		p = Process(
		target=self.runOneCommand,
		args=(tag, command, ref),
		args=(tag, command),
		)
		processes.append(p)
		p.start()
		@@ -395,16 +426,16 @@ class MLDConformance:
		p.join()
		else:
		for command in commands:
		self.runOneCommand(tag, command, ref)
		self.runOneCommand(tag, command)

		def process(self, command) -> int:
		if self.verbose:
		if self.args.verbose:
		print(command)
		with open(self.logFile, "a") as fd:
		fd.write(command + "\n")

		with open(self.logFile, "a") as fd:
		if not self.dryrun:
		if not self.args.dryrun:
		c = subprocess.run(
		command, stdout=fd, stderr=subprocess.STDOUT, text=True, shell=True
		)
		@@ -526,20 +557,18 @@ if __name__ == "__main__":
		parser.add_argument(
		"--ref_build_path",
		type=str,
		required=True,
		help="Path to the reference build folder containing IVAS Encoder, Decoder, Renderer and Post Render binaries",
		)
		parser.add_argument(
		"--cut_build_path",
		type=str,
		required=True,
		help="Path to the CUT build folder containing IVAS Encoder, Decoder, Renderer and Post Render binaries",
		)
		parser.add_argument(
		"--regenerate_enc_refs",
		"--regenerate-enc-refs",
		default=False,
		action="store_true",
		help="Enable verbose printing",
		help="Regenerate the encoder reference bitstreams and decoded outputs",
		)
		parser.add_argument(
		"--verbose",
		@@ -567,7 +596,7 @@ if __name__ == "__main__":
		help='Choose tests to run ["ENC", "DEC", "REND", "ISAR", "ALL"]',
		)
		parser.add_argument(
		"--no_multi_processing",
		"--no-multi-processing",
		default=False,
		action="store_true",
		help="Disable multi-processing for sequential test run (debugging)",
		@@ -584,8 +613,11 @@ if __name__ == "__main__":
		conformance = MLDConformance(args)

		conformance.accumulateCommands()
		# import sys
		# sys.exit(0)

		if args.regenerate_enc_refs:
		conformance.runReferenceGeneration()
		sys.exit(0)

		testTags = (
		MLDConformance.IVAS_Bins.keys() if args.test_mode == "ALL" else [args.test_mode]
		)
		@@ -594,5 +626,5 @@ if __name__ == "__main__":
		# Not implemented yet
		continue
		if not args.analyse_only:
		conformance.runTag(tag, ref=True)
		conformance.runTag(tag)
		conformance.doAnalysis(selectTag=tag)