Merge branch 'parallel'

resources/todo

@@ -15,10 +15,6 @@ todo app wise
 todo else
 - figure out how to make mac and win versions (someone else :) <- via gui2exe
 
-todo later
-- use multiprocessing lib to take advantage of multicore/multi-CPU to compress
-    multiple files simultaneously (threads have issues in Python; see "GIL")
-
 ===========================================
 later versions:
     animate compressing.gif
@@ -37,3 +33,11 @@ later versions:
         again would currently try to recompress all 100, when only 10 would be
         worthy of trying to compress further.
 
+
+1.1.0 changes:
+- use multiprocessing for images
+- more robust file handling
+- re-adding images now results in recompressing them
+- compressing message now shows filename
+- wider array of status messages in the table
+

resources/window.ui

@@ -127,7 +127,7 @@
                <cursorShape>PointingHandCursor</cursorShape>
               </property>
               <property name="toolTip">
-               <string>Recompress selected images</string>
+               <string>Recompress all images</string>
               </property>
               <property name="text">
                <string>&amp;Recompress</string>

setup.py

@@ -3,7 +3,7 @@
 from distutils.core import setup
 
 setup(name = "trimage",
-    version = "1.0.0b3",
+    version = "1.1.0b",
     description = "Trimage image compressor - A cross-platform tool for optimizing PNG and JPG files",
     author = "Kilian Valkhof, Paul Chaplin",
     author_email = "help@trimage.org",

src/trimage/ThreadPool.py

@@ -0,0 +1,280 @@
+'''
+ThreadPool Implementation
+
+@author: Morten Holdflod Moeller - morten@holdflod.dk
+@license: LGPL v3 
+'''
+
+from __future__ import with_statement
+from threading import Thread, RLock
+from time import sleep
+from Queue import Queue, Empty
+import logging
+import sys
+
+class NullHandler(logging.Handler):
+    def emit(self, record):
+        pass
+
+h = sys.stderr
+logging.getLogger('threadpool').addHandler(h)
+logging.getLogger('threadpool.worker').addHandler(h)
+
+
+
+class ThreadPoolMixIn:
+    """Mix-in class to handle each request in a new thread from the ThreadPool."""
+
+    def __init__(self, threadpool=None):
+        if (threadpool == None):
+            threadpool = ThreadPool()
+            self.__private_threadpool = True
+        else:
+            self.__private_threadpool = False
+        
+        self.__threadpool = threadpool
+
+    def process_request_thread(self, request, client_address):
+        """Same as in BaseServer but as a thread.
+
+        In addition, exception handling is done here.
+
+        """
+        try:
+            self.finish_request(request, client_address)
+            self.close_request(request)
+        except:
+            self.handle_error(request, client_address) #IGNORE:W0702
+            self.close_request(request)
+
+    def process_request(self, request, client_address):
+        self.__threadpool.add_job(self.process_request_thread, [request, client_address])      
+
+    def shutdown(self):
+        if (self.__private_threadpool): self.__threadpool.shutdown()
+        
+
+class AddJobException(Exception):
+    '''
+    Exceptoion raised when a Job could not be added
+    to the queue
+    '''
+    def __init__(self, msg):
+        Exception.__init__(self, msg)
+    
+
+class ThreadPool:
+    '''
+    The class implementing the ThreadPool.
+    
+    Instantiate and add jobs using add_job(func, args_list) 
+    '''
+
+    class Job: #IGNORE:R0903
+        '''
+        Class encapsulating a job to be handled
+        by ThreadPool workers 
+        '''
+        def __init__(self, function, args, return_callback=None):
+            self.callable = function
+            self.arguments = args
+            self.return_callback = return_callback
+            
+        def execute(self):
+            '''
+            Called to execute the function
+            '''
+            try:
+                return_value = self.callable(*self.arguments) #IGNORE:W0142
+            except Exception, excep: #IGNORE:W0703
+                logger = logging.getLogger("threadpool.worker")
+                logger.warning("A job in the ThreadPool raised an exception: " + excep)
+                #else do nothing cause we don't know what to do...
+                return 
+                    
+            try:
+                if (self.return_callback != None):
+                    self.return_callback(return_value)
+            except Exception, _: #IGNORE:W0703 everything could go wrong... 
+                logger = logging.getLogger('threadpool')
+                logger.warning('Error while delivering return value to callback function')
+
+    class Worker(Thread):
+        '''
+        A worker thread handling jobs in the thread pool 
+        job queue
+        '''
+        
+        def __init__(self, pool):
+            Thread.__init__(self)
+            
+            if (not isinstance(pool, ThreadPool)):
+                raise TypeError("pool is not a ThreadPool instance")
+            
+            self.pool = pool
+            
+            self.alive = True
+            self.start()
+                
+        def run(self):
+            '''
+            The workers main-loop getting jobs from queue 
+            and executing them
+            '''
+            while self.alive:
+                #print self.pool.__active_worker_count,  self.pool.__worker_count
+                job = self.pool.get_job()
+                if (job != None):
+                    self.pool.worker_active()
+                    job.execute()
+                    self.pool.worker_inactive()
+                else:
+                    self.alive = False
+                
+            self.pool.punch_out()
+                       
+    def __init__(self, max_workers = 5, kill_workers_after = 3):
+        if (not isinstance(max_workers, int)):
+            raise TypeError("max_workers is not an int")
+        if (max_workers < 1):
+            raise ValueError('max_workers must be >= 1')
+        
+        if (not isinstance(kill_workers_after, int)):
+            raise TypeError("kill_workers_after is not an int")
+        
+        self.__max_workers = max_workers
+        self.__kill_workers_after = kill_workers_after
+        
+        # This Queue is assumed Thread Safe
+        self.__jobs = Queue()
+                      
+        self.__worker_count_lock = RLock() 
+        self.__worker_count = 0
+        self.__active_worker_count = 0
+        
+        self.__shutting_down = False
+        logger = logging.getLogger('threadpool')
+        logger.info('started')
+    
+    def shutdown(self,  wait_for_workers_period = 1, clean_shutdown_reties = 5):
+        if (not isinstance(clean_shutdown_reties, int)):
+            raise TypeError("clean_shutdown_reties is not an int")
+        if (not clean_shutdown_reties >= 0):
+            raise ValueError('clean_shutdown_reties must be >= 0')
+        
+        if (not isinstance(wait_for_workers_period, int)):
+            raise TypeError("wait_for_workers_period is not an int")
+        if (not wait_for_workers_period >= 0):
+            raise ValueError('wait_for_workers_period must be >= 0')
+        
+        logger = logging.getLogger("threadpool")
+        logger.info("shutting down")
+        
+        with self.__worker_count_lock:
+            self.__shutting_down = True
+            self.__max_workers = 0
+            self.__kill_workers_after = 0
+        
+        retries_left = clean_shutdown_reties
+        while (retries_left > 0):
+            
+            with self.__worker_count_lock:
+                logger.info("waiting for workers to shut down (%i), %i workers left"%(retries_left, self.__worker_count))
+                if (self.__worker_count > 0):
+                    retries_left -= 1
+                else:
+                    retries_left = 0
+                    
+                sleep(wait_for_workers_period)
+        
+        
+        with self.__worker_count_lock:
+            if (self.__worker_count > 0):
+                logger.warning("shutdown stopped waiting. Still %i active workers"%self.__worker_count)
+                clean_shutdown = False
+            else:
+                clean_shutdown = True
+            
+        logger.info("shutdown complete")
+        
+        return clean_shutdown
+    
+    def punch_out(self):
+        '''
+        Called by worker to update worker count 
+        when the worker is shutting down
+        '''
+        with self.__worker_count_lock:
+            self.__worker_count -= 1
+        
+    def __new_worker(self):
+        '''
+        Adding a new worker thread to the thread pool
+        '''
+        with self.__worker_count_lock:
+            ThreadPool.Worker(self)
+            self.__worker_count += 1
+        
+    def worker_active(self):
+        with self.__worker_count_lock:
+            self.__active_worker_count = self.__active_worker_count + 1
+            
+    def worker_inactive(self):
+        with self.__worker_count_lock:
+            self.__active_worker_count = self.__active_worker_count - 1
+        
+    def add_job(self, function, args = None, return_callback=None):
+        '''
+        Put new job into queue
+        '''
+        
+        if (not callable(function)):
+            raise TypeError("function is not a callable")
+        if (not ( args == None or isinstance(args, list))):
+            raise TypeError("args is not a list")
+        if (not (return_callback == None or callable(return_callback))):
+            raise TypeError("return_callback is not a callable")
+                
+        if (args == None):
+            args = []
+        
+        job = ThreadPool.Job(function, args, return_callback)
+        
+        with self.__worker_count_lock:
+            if (self.__shutting_down):
+                raise AddJobException("ThreadPool is shutting down")
+            
+            try:
+                start_new_worker = False
+                if (self.__worker_count < self.__max_workers):
+                    if (self.__active_worker_count == self.__worker_count):
+                        start_new_worker = True
+                        
+                self.__jobs.put(job)
+                
+                if (start_new_worker): 
+                    self.__new_worker()
+                    
+            except Exception:
+                raise AddJobException("Could not add job")
+        
+    
+    def get_job(self):
+        '''
+        Retrieve next job from queue 
+        workers die (and should) when 
+        returning None  
+        '''
+        
+        job = None
+        try:
+            if (self.__kill_workers_after < 0):
+                job = self.__jobs.get(True)
+            elif (self.__kill_workers_after == 0):
+                job = self.__jobs.get(False)
+            else:
+                job = self.__jobs.get(True, self.__kill_workers_after)
+        except Empty:
+            job = None
+        
+        return job

src/trimage/trimage.py

@@ -1,5 +1,5 @@
 #!/usr/bin/python
-
+import time
 import sys
 import errno
 from os import listdir
@@ -13,10 +13,12 @@ from hurry.filesize import *
 from imghdr import what as determinetype
 
 from Queue import Queue
+from ThreadPool import ThreadPool
+from multiprocessing import cpu_count
 
 from ui import Ui_trimage
 
-VERSION = "1.0.0b3"
+VERSION = "1.1.0b"
 
 
 class StartQT4(QMainWindow):
@@ -106,21 +108,14 @@ class StartQT4(QMainWindow):
         self.showapp = False
         dirpath = path.abspath(directory)
         imagedir = listdir(directory)
-        filelist = []
-        for image in imagedir:
-            image = path.join(dirpath, image)
-            if path.isfile(image) and self.checkname(image):
-                filelist.append(image)
+        filelist = [path.join(dirpath, image) for image in imagedir]
         self.delegator(filelist)
 
     def file_from_cmd(self, image):
         """Get the file and send it to compress_file"""
         self.showapp = False
-        image = path.abspath(image)
-        filecmdlist = []
-        if self.checkname(image):
-            filecmdlist.append(image)
-        self.delegator(filecmdlist)
+        filelist = [path.abspath(image)]
+        self.delegator(filelist)
 
     def file_drop(self, images):
         """
@@ -137,19 +132,11 @@ class StartQT4(QMainWindow):
             # this is a fix for file dialog differentiating between cases
             "Image files (*.png *.jpg *.jpeg *.PNG *.JPG *.JPEG)")
 
-        imagelist = []
-        for i, image in enumerate(images):
-            imagelist.append(unicode(image))
-
-        self.delegator(imagelist)
+        self.delegator([unicode(fullpath) for fullpath in images])
 
     def recompress_files(self):
         """Send each file in the current file list to compress_file again."""
-        newimagelist = []
-        for image in self.imagelist:
-            newimagelist.append(image[4])
-        self.imagelist = []
-        self.delegator(newimagelist)
+        self.delegator([row.image.fullpath for row in self.imagelist])
 
     """
     Compress functions
@@ -160,13 +147,22 @@ class StartQT4(QMainWindow):
         Recieve all images, check them and send them to the worker thread.
         """
         delegatorlist = []
-        for image in images:
-            if self.checkname(image):
-                delegatorlist.append((image, QIcon(image)))
-                self.imagelist.append(("Compressing...", "", "", "", image,
-                    QIcon(QPixmap(self.ui.get_image("pixmaps/compressing.gif")))))
+        for fullpath in images:
+            try: # recompress images already in the list
+                image = (i.image for i in self.imagelist
+                    if i.image.fullpath == fullpath).next()
+                if image.compressed:
+                    image.reset()
+                    image.recompression = True
+                    delegatorlist.append(image)
+            except StopIteration:
+                image = Image(fullpath)
+                if image.valid:
+                    delegatorlist.append(image)
+                    icon = QIcon(QPixmap(self.ui.get_image("pixmaps/compressing.gif")))
+                    self.imagelist.append(ImageRow(image, icon))
                 else:
-                sys.stderr.write("[error] %s not an image file\n" % image)
+                    print >> sys.stderr, u"[error] %s not a supported image file" % image.fullpath
 
         self.update_table()
         self.thread.compress_file(delegatorlist, self.showapp, self.verbose,
@@ -208,10 +204,6 @@ class StartQT4(QMainWindow):
     Helper functions
     """
 
-    def checkname(self, name):
-        """Check if the file is a jpg or png."""
-        return determinetype(name) in ["jpeg", "png"]
-
     def enable_recompress(self):
         """Enable the recompress button."""
         self.ui.recompress.setEnabled(True)
@@ -246,6 +238,7 @@ class StartQT4(QMainWindow):
                 else:
                     raise
 
+
 class TriTableModel(QAbstractTableModel):
 
     def __init__(self, parent, imagelist, header, *args):
@@ -275,7 +268,7 @@ class TriTableModel(QAbstractTableModel):
             return QVariant(data)
         elif index.column() == 0 and role == Qt.DecorationRole:
             # decorate column 0 with an icon of the image itself
-            f_icon = self.imagelist[index.row()][5]
+            f_icon = self.imagelist[index.row()][4]
             return QVariant(f_icon)
         else:
             return QVariant()
@@ -288,21 +281,120 @@ class TriTableModel(QAbstractTableModel):
         return QVariant()
 
 
+class ImageRow:
+
+    def __init__(self, image, waitingIcon=None):
+        """ Build the information visible in the table image row. """
+        self.image = image
+        d = {
+            'shortname': lambda i: self.statusStr() % i.shortname,
+            'oldfilesizestr': lambda i: size(i.oldfilesize, system=alternative)
+                if i.compressed else "",
+            'newfilesizestr': lambda i: size(i.newfilesize, system=alternative)
+                if i.compressed else "",
+            'ratiostr': lambda i:
+                "%.1f%%" % (100 - (float(i.newfilesize) / i.oldfilesize * 100))
+                if i.compressed else "",
+            'icon': lambda i: i.icon if i.compressed else waitingIcon,
+            'fullpath': lambda i: i.fullpath, #only used by cli
+        }
+        names = ['shortname', 'oldfilesizestr', 'newfilesizestr',
+                      'ratiostr', 'icon']
+        for i, n in enumerate(names):
+            d[i] = d[n]
+
+        self.d = d
+
+    def statusStr(self):
+        """ Set the status message. """
+        if self.image.failed:
+            return "ERROR: %s"
+        if self.image.compressing:
+            message = "Compressing %s..."
+            return message
+        if not self.image.compressed and self.image.recompression:
+            return "Queued for recompression..."
+        if not self.image.compressed:
+            return "Queued..."
+        return "%s"
+
+    def __getitem__(self, key):
+        return self.d[key](self.image)
+
+
+class Image:
+
+    def __init__(self, fullpath):
+        """ gather image information. """
+        self.valid = False
+        self.reset()
+        self.fullpath = fullpath
+        if path.isfile(self.fullpath):
+            self.filetype = determinetype(self.fullpath)
+            if self.filetype in ["jpeg", "png"]:
+                oldfile = QFileInfo(self.fullpath)
+                self.shortname = oldfile.fileName()
+                self.oldfilesize = oldfile.size()
+                self.icon = QIcon(self.fullpath)
+                self.valid = True
+
+    def _determinetype(self):
+        """ Determine the filetype of the file using imghdr. """
+        filetype = determinetype(self.fullpath)
+        if filetype in ["jpeg", "png"]:
+            self.filetype = filetype
+        else:
+            self.filetype = None
+        return self.filetype
+
+    def reset(self):
+        self.failed = False
+        self.compressed = False
+        self.compressing = False
+        self.recompression = False
+
+    def compress(self):
+        """ Compress the image and return it to the thread. """
+        if not self.valid:
+            raise "Tried to compress invalid image (unsupported format or not \
+            file)"
+        self.reset()
+        self.compressing = True
+        runString = {
+            "jpeg": u"jpegoptim -f --strip-all '%(file)s'",
+            "png": u"optipng -force -o7 '%(file)s'&&advpng -z4 '%(file)s'"}
+        try:
+            retcode = call(runString[self.filetype] % {"file": self.fullpath},
+                shell=True, stdout=PIPE)
+        except:
+            retcode = -1
+        if retcode == 0:
+            self.newfilesize = QFile(self.fullpath).size()
+            self.compressed = True
+        else:
+            self.failed = True
+        self.compressing = False
+        self.retcode = retcode
+        return self
+
+
 class Worker(QThread):
 
     def __init__(self, parent=None):
         QThread.__init__(self, parent)
-        self.exiting = False
-        self.toProcess=Queue()
+        self.toDisplay = Queue()
+        self.threadpool = ThreadPool(max_workers=cpu_count())
 
     def __del__(self):
-        self.exiting = True
-        self.wait()
+        self.threadpool.shutdown()
 
     def compress_file(self, images, showapp, verbose, imagelist):
         """Start the worker thread."""
         for image in images:
-            self.toProcess.put(image)
+            #FIXME:http://code.google.com/p/pythonthreadpool/issues/detail?id=5
+            time.sleep(0.05)
+            self.threadpool.add_job(image.compress, None,
+                                    return_callback=self.toDisplay.put)
         self.showapp = showapp
         self.verbose = verbose
         self.imagelist = imagelist
@@ -310,79 +402,21 @@ class Worker(QThread):
 
     def run(self):
         """Compress the given file, get data from it and call update_table."""
-        while self.showapp or not self.toProcess.empty():
-            #gather old file data
-            filename, icon = self.toProcess.get()            
-            oldfile = QFileInfo(filename)
-            name = oldfile.fileName()
-            oldfilesize = oldfile.size()
-            oldfilesizestr = size(oldfilesize, system=alternative)
-
-            filetype = determinetype(filename)
-            #decide which tool to use
-            if filetype is "jpeg":
-                runString = u"jpegoptim -f --strip-all '%(file)s'"
-            elif filetype is "png":
-                runString = (u"optipng -force -o7 '%(file)s'; advpng -z4 '%(file)s'")
-            else:
-                sys.stderr.write("[error] %s not an image file" % filename)
-
-            try:
-                retcode = call(runString % {"file": filename}, shell=True,
-                               stdout=PIPE)
-                runfile = retcode
-            except OSError as e:
-                runfile = e
-
-            if runfile == 0:
-                #gather new file data
-                newfile = QFile(filename)
-                newfilesize = newfile.size()
-                newfilesizestr = size(newfilesize, system=alternative)
-
-                #calculate ratio and make a nice string
-                ratio = 100 - (float(newfilesize) / float(oldfilesize) * 100)
-                ratiostr = "%.1f%%" % ratio
-
-                # append current image to list
-                for i, image in enumerate(self.imagelist):
-                    if image[4] == filename:
-                        self.imagelist.remove(image)
-                        self.imagelist.insert(i, (name, oldfilesizestr,
-                            newfilesizestr, ratiostr, filename, icon))
+        tp = self.threadpool
+        while self.showapp or not (tp._ThreadPool__active_worker_count == 0 and
+                                   tp._ThreadPool__jobs.empty()):
+            image = self.toDisplay.get()
 
             self.emit(SIGNAL("updateUi"))
 
-                if not self.showapp and self.verbose:
-                    # we work via the commandline
-                    print("File: " + filename + ", Old Size: "
-                        + oldfilesizestr + ", New Size: " + newfilesizestr
-                        + ", Ratio: " + ratiostr)
+            if not self.showapp and self.verbose: # we work via the commandline
+                if image.retcode == 0:
+                    ir = ImageRow(image)
+                    print("File: " + ir['fullpath'] + ", Old Size: "
+                        + ir['oldfilesizestr'] + ", New Size: "
+                        + ir['newfilesizestr'] + ", Ratio: " + ir['ratiostr'])
                 else:
-                sys.stderr.write("[error] %s" % runfile)
-
-
-class TrimageTableView(QTableView):
-    """Init the table drop event."""
-    def __init__(self, parent=None):
-        super(TrimageTableView, self).__init__(parent)
-        self.setAcceptDrops(True)
-
-    def dragEnterEvent(self, event):
-        if event.mimeData().hasFormat("text/uri-list"):
-            event.accept()
-        else:
-            event.ignore()
-
-    def dragMoveEvent(self, event):
-        event.accept()
-
-    def dropEvent(self, event):
-        files = str(event.mimeData().data("text/uri-list")).strip().split()
-        for i, file in enumerate(files):
-            files[i] = QUrl(QString(file)).toLocalFile()
-        files=[i.toUtf8().decode("utf-8") for i in files]
-        self.emit(SIGNAL("fileDropEvent"), (files))
+                    print >> sys.stderr, u"[error] %s could not be compressed" % image.fullpath
 
 if __name__ == "__main__":
     app = QApplication(sys.argv)

src/trimage/ui.py

@@ -153,7 +153,7 @@ class Ui_trimage(object):
             "Drag and drop images onto the table", None,
             QApplication.UnicodeUTF8))
         self.recompress.setToolTip(QApplication.translate("trimage",
-            "Recompress selected images", None, QApplication.UnicodeUTF8))
+            "Recompress all images", None, QApplication.UnicodeUTF8))
         self.recompress.setText(QApplication.translate("trimage",
             "&Recompress", None, QApplication.UnicodeUTF8))
         self.recompress.setShortcut(QApplication.translate("trimage",

trimage

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-#Copyright (c) 2010 Kilian Valkhof, Paul Chaplin
+#Copyright (c) 2010 Kilian Valkhof, Paul Chaplin, Tarnay Kálmán
 #
 #Permission is hereby granted, free of charge, to any person
 #obtaining a copy of this software and associated documentation

website/index.html

@@ -69,7 +69,7 @@
 
 <body>
   <div id="wrap">
-    <h1><img src="trimage-icon.png" alt=""> Trimage image compressor &ndash; 1.0.0b3 (beta)</h1>
+    <h1><img src="trimage-icon.png" alt=""> Trimage image compressor &ndash; 1.1.0b (beta)</h1>
     <span class="subtitle">A cross-platform tool for losslessly optimizing PNG and JPG files.</span>
     <p class="tri">Trimage is a cross-platform GUI and command-line interface to optimize image
     files via <a href="http://optipng.sourceforge.net/">optipng</a>,
@@ -123,7 +123,7 @@
       <ul>
         <li>Neil Wallace</li>
         <li>Jeroen Goudsmit</li>
-        <li>Kálmán Tarnay</li>
+        <li>Tarnay Kálmán</li>
       </ul>
     </div>
     <div class="block">
@@ -165,16 +165,12 @@
     </div>
 
     <h2>Planned features</h2>
-    <p>Version 1.0.0 final:</p>
+    <p>Version 1.1.0 final:</p>
     <ul>
       <li>Expand command line options</li>
       <li>Make sure a compressed file is always smaller than the original one, or don't compress</li>
       <li>General refactoring</li>
     </ul>
-    <p>Version 1.1.0</p>
-    <ul>
-      <li>Use multiprocessing instead of threading</li>
-    </ul>
     <p>Beyond that</p>
     <ul>
       <li>Deletion of rows in the table view</li>