5 from optparse import OptionGroup
11 from digress.cli import Dispatcher as _Dispatcher
12 from digress.errors import ComparisonError, FailedTestError, DisabledTestError
13 from digress.testing import depends, comparer, Fixture, Case
14 from digress.comparers import compare_pass
15 from digress.scm import git as x264git
17 from subprocess import Popen, PIPE, STDOUT
24 from random import randrange, seed
27 from itertools import imap, izip
29 os.chdir(os.path.join(os.path.dirname(__file__), ".."))
34 [ "--tune %s" % t for t in ("film", "zerolatency") ],
35 ("", "--intra-refresh"),
38 ("", "--slice-max-size 1000"),
39 ("", "--frame-packing 5"),
40 [ "--preset %s" % p for p in ("ultrafast",
54 def compare_yuv_output(width, height):
55 def _compare_yuv_output(file_a, file_b):
56 size_a = os.path.getsize(file_a)
57 size_b = os.path.getsize(file_b)
60 raise ComparisonError("%s is not the same size as %s" % (
69 with open(file_a) as f_a:
70 with open(file_b) as f_b:
71 for chunk_a, chunk_b in izip(
73 lambda i: f_a.read(BUFFER_SIZE),
74 xrange(size_a // BUFFER_SIZE + 1)
77 lambda i: f_b.read(BUFFER_SIZE),
78 xrange(size_b // BUFFER_SIZE + 1)
81 chunk_size = len(chunk_a)
83 if chunk_a != chunk_b:
84 for i in xrange(chunk_size):
85 if chunk_a[i] != chunk_b[i]:
86 # calculate the macroblock, plane and frame from the offset
89 y_plane_area = width * height
90 u_plane_area = y_plane_area + y_plane_area * 0.25
91 v_plane_area = u_plane_area + y_plane_area * 0.25
93 pixel = offs % v_plane_area
94 frame = offs // v_plane_area
96 if pixel < y_plane_area:
99 pixel_x = pixel % width
100 pixel_y = pixel // width
102 macroblock = (ceil(pixel_x / 16.0), ceil(pixel_y / 16.0))
103 elif pixel < u_plane_area:
106 pixel -= y_plane_area
108 pixel_x = pixel % width
109 pixel_y = pixel // width
111 macroblock = (ceil(pixel_x / 8.0), ceil(pixel_y / 8.0))
115 pixel -= u_plane_area
117 pixel_x = pixel % width
118 pixel_y = pixel // width
120 macroblock = (ceil(pixel_x / 8.0), ceil(pixel_y / 8.0))
122 macroblock = tuple([ int(x) for x in macroblock ])
124 raise ComparisonError("%s differs from %s at frame %d, " \
125 "macroblock %s on the %s plane (offset %d)" % (
136 return _compare_yuv_output
138 def program_exists(program):
140 return os.path.exists(fpath) and os.access(fpath, os.X_OK)
142 fpath, fname = os.path.split(program)
148 for path in os.environ["PATH"].split(os.pathsep):
149 exe_file = os.path.join(path, program)
159 @comparer(compare_pass)
160 def test_configure(self):
164 ], stdout=PIPE, stderr=STDOUT).communicate()
166 configure_proc = Popen([
168 ] + self.fixture.dispatcher.configure, stdout=PIPE, stderr=STDOUT)
170 output = configure_proc.communicate()[0]
171 if configure_proc.returncode != 0:
172 raise FailedTestError("configure failed: %s" % output.replace("\n", " "))
174 @depends("configure")
175 @comparer(compare_pass)
180 ], stdout=PIPE, stderr=STDOUT)
182 output = make_proc.communicate()[0]
183 if make_proc.returncode != 0:
184 raise FailedTestError("make failed: %s" % output.replace("\n", " "))
186 _dimension_pattern = re.compile(r"\w+ [[]info[]]: (\d+)x(\d+)[pi] \d+:\d+ @ \d+/\d+ fps [(][vc]fr[)]")
188 def _YUVOutputComparisonFactory():
189 class YUVOutputComparison(Case):
190 _dimension_pattern = _dimension_pattern
192 depends = [ Compile ]
196 for name, meth in inspect.getmembers(self):
197 if name[:5] == "test_" and name[5:] not in self.fixture.dispatcher.yuv_tests:
198 delattr(self.__class__, name)
204 "%s.264" % self.fixture.dispatcher.video,
208 self.fixture.dispatcher.video
209 ], stdout=PIPE, stderr=STDOUT)
211 output = x264_proc.communicate()[0]
212 if x264_proc.returncode != 0:
213 raise FailedTestError("x264 did not complete properly: %s" % output.replace("\n", " "))
215 matches = _dimension_pattern.match(output)
217 return (int(matches.group(1)), int(matches.group(2)))
219 @comparer(compare_pass)
221 if not program_exists("ldecod"): raise DisabledTestError("jm unavailable")
224 runres = self._run_x264()
229 "%s.264" % self.fixture.dispatcher.video,
232 ], stdout=PIPE, stderr=STDOUT)
234 output = jm_proc.communicate()[0]
235 if jm_proc.returncode != 0:
236 raise FailedTestError("jm did not complete properly: %s" % output.replace("\n", " "))
239 compare_yuv_output(*runres)("x264-output.yuv", "jm-output.yuv")
240 except ComparisonError, e:
241 raise FailedTestError(e)
243 try: os.remove("x264-output.yuv")
246 try: os.remove("%s.264" % self.fixture.dispatcher.video)
249 try: os.remove("jm-output.yuv")
252 try: os.remove("log.dec")
255 try: os.remove("dataDec.txt")
258 @comparer(compare_pass)
259 def test_ffmpeg(self):
260 if not program_exists("ffmpeg"): raise DisabledTestError("ffmpeg unavailable")
262 runres = self._run_x264()
264 ffmpeg_proc = Popen([
268 "%s.264" % self.fixture.dispatcher.video,
270 ], stdout=PIPE, stderr=STDOUT)
272 output = ffmpeg_proc.communicate()[0]
273 if ffmpeg_proc.returncode != 0:
274 raise FailedTestError("ffmpeg did not complete properly: %s" % output.replace("\n", " "))
277 compare_yuv_output(*runres)("x264-output.yuv", "ffmpeg-output.yuv")
278 except ComparisonError, e:
279 raise FailedTestError(e)
281 try: os.remove("x264-output.yuv")
284 try: os.remove("%s.264" % self.fixture.dispatcher.video)
287 try: os.remove("ffmpeg-output.yuv")
290 return YUVOutputComparison
292 class Regression(Case):
293 depends = [ Compile ]
295 _psnr_pattern = re.compile(r"x264 [[]info[]]: PSNR Mean Y:\d+[.]\d+ U:\d+[.]\d+ V:\d+[.]\d+ Avg:\d+[.]\d+ Global:(\d+[.]\d+) kb/s:\d+[.]\d+")
296 _ssim_pattern = re.compile(r"x264 [[]info[]]: SSIM Mean Y:(\d+[.]\d+) [(]\d+[.]\d+db[)]")
299 if self.fixture.dispatcher.x264:
300 self.__class__.__name__ += " %s" % " ".join(self.fixture.dispatcher.x264)
307 "%s.264" % self.fixture.dispatcher.video,
309 ] + self.fixture.dispatcher.x264 + [
310 self.fixture.dispatcher.video
311 ], stdout=PIPE, stderr=STDOUT)
313 output = x264_proc.communicate()[0]
315 if x264_proc.returncode != 0:
316 raise FailedTestError("x264 did not complete properly: %s" % output.replace("\n", " "))
318 for line in output.split("\n"):
319 if line.startswith("x264 [info]: PSNR Mean"):
320 return float(self._psnr_pattern.match(line).group(1))
322 raise FailedTestError("no PSNR output caught from x264")
324 try: os.remove("%s.264" % self.fixture.dispatcher.video)
332 "%s.264" % self.fixture.dispatcher.video,
334 ] + self.fixture.dispatcher.x264 + [
335 self.fixture.dispatcher.video
336 ], stdout=PIPE, stderr=STDOUT)
338 output = x264_proc.communicate()[0]
340 if x264_proc.returncode != 0:
341 raise FailedTestError("x264 did not complete properly: %s" % output.replace("\n", " "))
343 for line in output.split("\n"):
344 if line.startswith("x264 [info]: SSIM Mean"):
345 return float(self._ssim_pattern.match(line).group(1))
347 raise FailedTestError("no PSNR output caught from x264")
349 try: os.remove("%s.264" % self.fixture.dispatcher.video)
352 def _generate_random_commandline():
355 for suboptions in OPTIONS:
356 commandline.append(suboptions[randrange(0, len(suboptions))])
358 return filter(None, reduce(operator.add, [ shlex.split(opt) for opt in commandline ]))
363 fixture.register_case(Compile)
365 fixture.register_case(Regression)
367 class Dispatcher(_Dispatcher):
368 video = "akiyo_qcif.y4m"
374 def _populate_parser(self):
375 super(Dispatcher, self)._populate_parser()
377 # don't do a whole lot with this
378 tcase = _YUVOutputComparisonFactory()
380 yuv_tests = [ name[5:] for name, meth in filter(lambda pair: pair[0][:5] == "test_", inspect.getmembers(tcase)) ]
382 group = OptionGroup(self.optparse, "x264 testing-specific options")
391 callback=lambda option, opt, value, parser: setattr(self, "video", value),
392 help="yuv video to perform testing on (default: %s)" % self.video
402 callback=lambda option, opt, value, parser: setattr(self, "seed", value),
403 help="seed for the random number generator (default: unix timestamp)"
413 callback=lambda option, opt, value, parser: setattr(self, "products", value),
414 help="number of cartesian products to generate for yuv comparison testing (default: %d)" % self.products
423 callback=lambda option, opt, value, parser: setattr(self, "configure", shlex.split(value)),
424 help="options to run ./configure with"
432 callback=lambda option, opt, value, parser: setattr(self, "yuv_tests", [
433 val.strip() for val in value.split(",")
435 help="select tests to run with yuv comparisons (default: %s, available: %s)" % (
436 ", ".join(self.yuv_tests),
447 callback=lambda option, opt, value, parser: setattr(self, "x264", shlex.split(value)),
448 help="additional options to run ./x264 with"
451 self.optparse.add_option_group(group)
453 def pre_dispatch(self):
454 if not hasattr(self, "seed"):
455 self.seed = int(time())
457 print "Using seed: %d" % self.seed
460 for i in xrange(self.products):
461 YUVOutputComparison = _YUVOutputComparisonFactory()
463 commandline = _generate_random_commandline()
467 while commandline in _generated:
469 commandline = _generate_random_commandline()
472 print >>sys.stderr, "Maximum command-line regeneration exceeded. " \
473 "Try a different seed or specify fewer products to generate."
476 commandline += self.x264
478 _generated.append(commandline)
480 YUVOutputComparison.options = commandline
481 YUVOutputComparison.__name__ = ("%s %s" % (YUVOutputComparison.__name__, " ".join(commandline)))
483 fixture.register_case(YUVOutputComparison)
485 Dispatcher(fixture).dispatch()