skia2/tools/find_bad_images_in_skps.py

#!/usr/bin/env python

# Copyright 2013 Google Inc. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""
This script will take as an argument either a list of skp files or a
set of directories that contains skp files.  It will then test each
skp file with the `render_pictures` program. If that program either
spits out any unexpected output or doesn't return 0, I will flag that
skp file as problematic.  We then extract all of the embedded images
inside the skp and test each one of them against the
SkImageDecoder::DecodeFile function.  Again, we consider any
extraneous output or a bad return value an error.  In the event of an
error, we retain the image and print out information about the error.
The output (on stdout) is formatted as a csv document.

A copy of each bad image is left in a directory created by
tempfile.mkdtemp().
"""

import glob
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading

import test_rendering  # skia/trunk/tools.  reuse FindPathToProgram()

USAGE = """
Usage:
    {command} SKP_FILE [SKP_FILES]
    {command} SKP_DIR [SKP_DIRS]\n
Environment variables:
    To run multiple worker threads, set NUM_THREADS.
    To use a different temporary storage location, set TMPDIR.

"""

def execute_program(args, ignores=None):
    """
    Execute a process and waits for it to complete.  Returns all
    output (stderr and stdout) after (optional) filtering.

    @param args is passed into subprocess.Popen().

    @param ignores (optional) is a list of regular expression strings
    that will be ignored in the output.

    @returns a tuple (returncode, output)
    """
    if ignores is None:
        ignores = []
    else:
        ignores = [re.compile(ignore) for ignore in ignores]
    proc = subprocess.Popen(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT)
    output = ''.join(
        line for line in proc.stdout
        if not any(bool(ignore.match(line)) for ignore in ignores))
    returncode = proc.wait()
    return (returncode, output)


def list_files(paths):
    """
    Accepts a list of directories or filenames on the command line.
    We do not choose to recurse into directories beyond one level.
    """
    class NotAFileException(Exception):
        pass
    for path in paths:
        for globbedpath in glob.iglob(path): # useful on win32
            if os.path.isdir(globbedpath):
                for filename in os.listdir(globbedpath):
                    newpath = os.path.join(globbedpath, filename)
                    if os.path.isfile(newpath):
                        yield newpath
            elif os.path.isfile(globbedpath):
                yield globbedpath
            else:
                raise NotAFileException('{} is not a file'.format(globbedpath))


class BadImageFinder(object):

    def __init__(self, directory=None):
        self.render_pictures = test_rendering.FindPathToProgram(
            'render_pictures')
        self.test_image_decoder = test_rendering.FindPathToProgram(
            'test_image_decoder')
        assert os.path.isfile(self.render_pictures)
        assert os.path.isfile(self.test_image_decoder)
        if directory is None:
            self.saved_image_dir = tempfile.mkdtemp(prefix='skia_skp_test_')
        else:
            assert os.path.isdir(directory)
            self.saved_image_dir = directory
        self.bad_image_count = 0

    def process_files(self, skp_files):
        for path in skp_files:
            self.process_file(path)

    def process_file(self, skp_file):
        assert self.saved_image_dir is not None
        assert os.path.isfile(skp_file)
        args = [self.render_pictures, '--readPath', skp_file]
        ignores = ['^process_in', '^deserializ', '^drawing...', '^Non-defaul']
        returncode, output = execute_program(args, ignores)
        if (returncode == 0) and not output:
            return
        temp_image_dir = tempfile.mkdtemp(prefix='skia_skp_test___')
        args = [ self.render_pictures, '--readPath', skp_file,
                 '--writePath', temp_image_dir, '--writeEncodedImages']
        subprocess.call(args, stderr=open(os.devnull,'w'),
                        stdout=open(os.devnull,'w'))
        for image_name in os.listdir(temp_image_dir):
            image_path = os.path.join(temp_image_dir, image_name)
            assert(os.path.isfile(image_path))
            args = [self.test_image_decoder, image_path]
            returncode, output = execute_program(args, [])
            if (returncode == 0) and not output:
                os.remove(image_path)
                continue
            try:
                shutil.move(image_path, self.saved_image_dir)
            except (shutil.Error,):
                # If this happens, don't stop the entire process,
                # just warn the user.
                os.remove(image_path)
                sys.stderr.write('{0} is a repeat.\n'.format(image_name))
            self.bad_image_count += 1
            if returncode == 2:
                returncode = 'SkImageDecoder::DecodeFile returns false'
            elif returncode == 0:
                returncode = 'extra verbosity'
                assert output
            elif returncode == -11:
                returncode = 'segmentation violation'
            else:
                returncode = 'returncode: {}'.format(returncode)
            output = output.strip().replace('\n',' ').replace('"','\'')
            suffix = image_name[-3:]
            output_line = '"{0}","{1}","{2}","{3}","{4}"\n'.format(
                returncode, suffix, skp_file, image_name, output)
            sys.stdout.write(output_line)
            sys.stdout.flush()
        os.rmdir(temp_image_dir)
        return

def main(main_argv):
    if not main_argv or main_argv[0] in ['-h', '-?', '-help', '--help']:
        sys.stderr.write(USAGE.format(command=__file__))
        return 1
    if 'NUM_THREADS' in os.environ:
        number_of_threads = int(os.environ['NUM_THREADS'])
        if number_of_threads < 1:
            number_of_threads = 1
    else:
        number_of_threads = 1
    os.environ['skia_images_png_suppressDecoderWarnings'] = 'true'
    os.environ['skia_images_jpeg_suppressDecoderWarnings'] = 'true'

    temp_dir = tempfile.mkdtemp(prefix='skia_skp_test_')
    sys.stderr.write('Directory for bad images: {}\n'.format(temp_dir))
    sys.stdout.write('"Error","Filetype","SKP File","Image File","Output"\n')
    sys.stdout.flush()

    finders = [
        BadImageFinder(temp_dir) for index in xrange(number_of_threads)]
    arguments = [[] for index in xrange(number_of_threads)]
    for index, item in enumerate(list_files(main_argv)):
        ## split up the given targets among the worker threads
        arguments[index % number_of_threads].append(item)
    threads = [
        threading.Thread(
            target=BadImageFinder.process_files, args=(finder,argument))
        for finder, argument in zip(finders, arguments)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    number  = sum(finder.bad_image_count for finder in finders)
    sys.stderr.write('Number of bad images found: {}\n'.format(number))
    return 0

if __name__ == '__main__':
    exit(main(sys.argv[1:]))

#  LocalWords:  skp stdout csv