wxWidgets/wxPython/wx/lib/delayedresult.py

"""
This module supports the thread-safe, asynchronous transmission of data 
('delayed results') from a worker (non-GUI) thread to the main thread. Ie you don't 
need to mutex lock any data, the worker thread doesn't wait (or even check) 
for the result to be received, and the main thread doesn't wait for the 
worker thread to send the result. Instead, the consumer will be called 
automatically by the wx app when the worker thread result is available. 

In most cases you just need to use startWorker() with the correct parameters
(your worker function and your 'consumer' in the simplest of cases). The 
only requirement on consumer is that it must accept a DelayedResult instance 
as first arg. 

In the following example, this will call consumer(delayedResult) with the 
return value from workerFn::

    from delayedresult import startWorker
    startWorker(consumer, workerFn)

More advanced uses: 

- The other parameters to startWorker()
- Derive from Producer to override _extraInfo (e.g. to provide traceback info)
- Create your own worker-function-thread wrapper instead of using Producer
- Create your own Handler-like wrapper to pre- or post-process the result 
  (see PreProcessChain)
- Derive from Sender to use your own way of making result hop over the 
  "thread boundary" (from non-main thread to main thread), e.g. using Queue

Thanks to Josiah Carlson for critical feedback/ideas that helped me 
improve this module. 

:Copyright: (c) 2006 by Oliver Schoenborn
:License: wxWidgets license
:Version: 1.0

"""

__author__  = 'Oliver Schoenborn at utoronto dot ca'
__version__ = '1.0'

__all__ = ('Sender', 'SenderNoWx', 'SenderWxEvent', 'SenderCallAfter', 
    'Handler', 'DelayedResult', 'Producer', 'startWorker', 'PreProcessChain')


import wx
import threading


class Struct:
    """
    An object that has attributes built from the dictionary given in 
    constructor. So ss=Struct(a=1, b='b') will satisfy assert ss.a == 1
    and assert ss.b == 'b'.
    """
    
    def __init__(self, **kwargs):
        self.__dict__.update( kwargs )


class Handler:
    """
    Bind some of the arguments and keyword arguments of a callable ('listener'). 
    Then when the Handler instance is called (e.g. handler(result, **kwargs))
    the result is passed as first argument to callable, the kwargs is 
    combined with those given at construction, and the args are those
    given at construction. Its return value is returned.
    """
    def __init__(self, listener, *args, **kwargs ):
        """Bind args and kwargs to listener. """
        self.__listener = listener
        self.__args = args
        self.__kwargs = kwargs
        
    def __call__(self, result, **moreKwargs):
        """Listener is assumed to take result as first arg, then *args, 
        then the combination of moreKwargs and the kwargs given at construction."""
        if moreKwargs:
            moreKwargs.update(self.__kwargs)
        else:
            moreKwargs = self.__kwargs
        return self.__listener(result, *self.__args, **moreKwargs)

        
class Sender:
    """
    Base class for various kinds of senders. A sender sends a result
    produced by a worker funtion to a result handler (listener). Note
    that each sender can be given a "job id". This can be anything
    (number, string, id, and object, etc) and is not used, it is 
    simply added as attribute whenever a DelayedResult is created. 
    This allows you to know, if desired, what result corresponds to 
    which sender. Note that uniqueness is not necessary. 
    
    Derive from this class if none of the existing derived classes
    are adequate, and override _sendImpl(). 
    """
    
    def __init__(self, jobID=None):
        """The optional jobID can be anything that you want to use to 
        track which sender particular results come from. """
        self.__jobID = jobID

    def getJobID(self):
        """Return the jobID given at construction"""
        return self.__jobID
    
    def sendResult(self, result):
        """This will send the result to handler, using whatever 
        technique the derived class uses. """
        delayedResult = DelayedResult(result, jobID=self.__jobID)
        self._sendImpl(delayedResult)

    def sendException(self, exception, extraInfo = None):
        """Use this when the worker function raised an exception.
        The *exception* is the instance of Exception caught. The extraInfo
        could be anything you want (e.g. locals or traceback etc), 
        it will be added to the exception as attribute 'extraInfo'. The
        exception will be raised when DelayedResult.get() is called."""
        assert exception is not None
        delayedResult = DelayedResult(extraInfo, 
            exception=exception, jobID=self.__jobID)
        self._sendImpl(delayedResult)

    def _sendImpl(self, delayedResult):
        msg = '_sendImpl() must be implemented in %s' % self.__class__
        raise NotImplementedError(msg)
        
    
class SenderNoWx( Sender ):
    """
    Sender that works without wx. The results are sent directly, ie
    the consumer will get them "in the worker thread". So it should 
    only be used for testing. 
    """
    def __init__(self, consumer, jobID=None, args=(), kwargs={}):
        """The consumer can be any callable of the form 
        callable(result, *args, **kwargs)"""
        Sender.__init__(self, jobID)
        if args or kwargs:
            self.__consumer = Handler(consumer, *args, **kwargs)
        else: 
            self.__consumer = consumer
            
    def _sendImpl(self, delayedResult):
        self.__consumer(delayedResult)
        

class SenderWxEvent( Sender ):
    """
    This sender sends the delayed result produced in the worker thread
    to an event handler in the main thread, via a wx event of class 
    *eventClass*. The result is an attribute of the event (default: 
    "delayedResult". 
    """
    def __init__(self, handler, eventClass, resultAttr="delayedResult", 
        jobID=None, **kwargs):
        """The handler must derive from wx.EvtHandler. The event class 
        is typically the first item in the pair returned by 
        wx.lib.newevent.NewEvent(). You can use the *resultAttr* 
        to change the attribute name of the generated event's 
        delayed result. """
        Sender.__init__(self, jobID)
        if not isinstance(handler, wx.EvtHandler):
            msg = 'SenderWxEvent(handler=%s, ...) not allowed,' % type(handler)
            msg = '%s handler must derive from wx.EvtHandler' % msg
            raise ValueError(msg)
        self.__consumer = Struct(handler=handler, eventClass=eventClass, 
                                 resultAttr=resultAttr, kwargs=kwargs)
        
    def _sendImpl(self, delayedResult):
        """Must not modify the consumer (that was created at construction) 
        since might be shared by several senders, each sending from 
        separate threads."""
        consumer = self.__consumer
        kwargs = consumer.kwargs.copy()
        kwargs[ consumer.resultAttr ] = delayedResult
        event = consumer.eventClass(** kwargs)
        wx.PostEvent(consumer.handler, event)


class SenderCallAfter( Sender ):
    """
    This sender sends the delayed result produced in the worker thread
    to a callable in the main thread, via wx.CallAfter. 
    """
    def __init__(self, listener, jobID=None, args=(), kwargs={}):
        Sender.__init__(self, jobID)
        if args or kwargs:
            self.__consumer = Handler(listener, *args, **kwargs)
        else:
            self.__consumer = listener
            
    def _sendImpl(self, delayedResult):
        wx.CallAfter(self.__consumer, delayedResult)
        

class DelayedResult:
    """
    Represent the actual delayed result coming from the non-main thread. 
    An instance of this is given to the result handler. This result is 
    either a (reference to a) the value sent, or an exception. 
    If the latter, the exception is raised when the get() method gets
    called. 
    """
    
    def __init__(self, result, jobID=None, exception = None):
        """You should never have to call this yourself. A DelayedResult 
        is created by a concrete Sender for you."""
        self.__result = result
        self.__exception = exception
        self.__jobID = jobID

    def getJobID(self):
        """Return the jobID given when Sender initialized, 
        or None if none given. """
        return self.__jobID 
    
    def get(self):
        """Get the result. If an exception was sent instead of a result, 
        (via Sender's sendExcept()), that **exception is raised**.
        Otherwise the result is simply returned. """
        if self.__exception: # exception was raised!
            self.__exception.extraInfo = self.__result
            raise self.__exception
        
        return self.__result


class AbortedException(Exception):
    """Raise this in your worker function so that the sender knows 
    not to send a result to handler."""
    pass
    

class Producer(threading.Thread):
    """
    Represent the worker thread that produces delayed results. 
    It causes the given function to run in a separate thread, 
    and a sender to be used to send the return value of the function.
    As with any threading.Thread, instantiate and call start().
    Note that if the workerFn raises AbortedException, the result is not 
    sent and the thread terminates gracefully.
    """
    
    def __init__(self, sender, workerFn, args=(), kwargs={}, 
                 name=None, group=None, daemon=False, 
                 sendReturn=True, senderArg=None):
        """The sender will send the return value of 
        workerFn(*args, **kwargs) to the main thread. The name and group 
        are same as threading.Thread constructor parameters. Daemon causes 
        setDaemon() to be called. If sendReturn is False, then the return 
        value of workerFn() will not be sent. If senderArg is given, it 
        must be the name of the keyword arg to use to pass the sender into 
        the workerFn, so the function can send (typically many) results."""
        if senderArg:
            kwargs[senderArg] = sender
        def wrapper():
            try: 
                result = workerFn(*args, **kwargs)
            except AbortedException:
                pass
            except Exception, exc:
                extraInfo = self._extraInfo(exc)
                sender.sendException(exc, extraInfo)
            else:
                if sendReturn:
                    sender.sendResult(result)
            
        threading.Thread.__init__(self, name=name, group=group, target=wrapper)
        if daemon:
            self.setDaemon(daemon)
    
    def _extraInfo(self, exception):
        """This method could be overridden in a derived class to provide 
        extra information when an exception is being sent instead of a 
        result. """
        return None


class AbortEvent:
    """
    Convenience class that represents a kind of threading.Event that
    raises AbortedException when called (see the __call__ method, everything
    else is just to make it look like threading.Event).
    """
    
    def __init__(self):
        self.__ev = threading.Event()

    def __call__(self, timeout=None):
        """See if event has been set (wait at most timeout if given).  If so, 
        raise AbortedException. Otherwise return None. Allows you to do
        'while not event():' which will always succeed unless the event 
        has been set (then AbortedException will cause while to exit)."""
        if timeout:
            self.__ev.wait(timeout)
        if self.__ev.isSet():
            raise AbortedException()
        return None
    
    def __getattr__(self, name):
        """This allows us to be a kind of threading.Event."""
        if name in ('set','clear','wait','isSet'):
            return getattr(self.__ev, name)


def startWorker(
    consumer, workerFn, 
    cargs=(), ckwargs={}, 
    wargs=(), wkwargs={},
    jobID=None, group=None, daemon=False, 
    sendReturn=True, senderArg=None):
    """
    Convenience function to send data produced by workerFn(*wargs, **wkwargs) 
    running in separate thread, to a consumer(*cargs, **ckwargs) running in
    the main thread. This function merely creates a SenderCallAfter (or a
    SenderWxEvent, if consumer derives from wx.EvtHandler), and a Producer,
    and returns immediately after starting the Producer thread. The jobID
    is used for the Sender and as name for the Producer thread. Returns the 
    thread created, in case caller needs join/etc.
    """
    
    if isinstance(consumer, wx.EvtHandler):
        eventClass = cargs[0]
        sender = SenderWxEvent(consumer, eventClass, jobID=jobID, **ckwargs)
    else:
        sender = SenderCallAfter(consumer, jobID, args=cargs, kwargs=ckwargs)
        
    thread = Producer(
        sender, workerFn, args=wargs, kwargs=wkwargs, 
        name=jobID, group=group, daemon=daemon, 
        senderArg=senderArg, sendReturn=sendReturn)
        
    thread.start() 
    return thread


class PreProcessChain:
    """
    Represent a 'delayed result pre-processing chain', a kind of Handler. 
    Useful when lower-level objects need to apply a sequence of transformations 
    to the delayed result before handing it over to a final handler. 
    This allows the starter of the worker function to not know 
    anything about the lower-level objects. 
    """
    def __init__(self, handler, *args, **kwargs):
        """Wrap handler(result, *args, **kwargs) so that the result 
        it receives has been transformed by us. """
        if handler is None:# assume rhs is a chain
            self.__chain = args[0]
        else:
            if args or kwargs:
                handler = Handler(handler, *args, **kwargs)
            self.__chain = [handler]

    def addSub(self, callable, *args, **kwargs):
        """Add a sub-callable, ie a callable(result, *args, **kwargs)
        that returns a transformed result to the previously added
        sub-callable (or the handler given at construction, if this is 
        the first call to addSub). """
        self.__chain.append( Handler(callable, *args, **kwargs) )
        
    def clone(self):
        """Clone the chain. Shallow only. Useful when several threads 
        must be started but have different sub-callables. """
        return PreProcessChain(None, self.__chain[:] )
    
    def cloneAddSub(self, callable, *args, **kwargs):
        """Convenience method that first clones self, then calls addSub() 
        on that clone with given arguments. """
        cc = self.clone()
        cc.addSub(callable, *args, **kwargs)
        
    def count(self):
        """How many pre-processors in the chain"""
        return len(self.__chain)
    
    class Traverser:
        """
        Traverses the chain of pre-processors it is given, transforming
        the original delayedResult along the way. The return value of each 
        callable added via addSub() is given to the previous addSub() callable,
        until the handler is reached. 
        """
        def __init__(self, delayedResult, chain):
            self.__dr = delayedResult
            self.__chain = chain
            
        def get(self):
            """This makes handler think we are a delayedResult."""
            if not self.__chain:
                return self.__dr.get()
            
            handler = self.__chain[0]
            del self.__chain[0]
            return handler(self)
        
        def getJobID(self):
            """Return the job id for the delayedResult we transform."""
            return self.__dr.getJobID()


    def __call__(self, delayedResult):
        """This makes us a Handler. We just call handler(Traverser). The
        handler will think it is getting a delayed result, but in fact 
        will be getting an instance of Traverser, which will take care
        of properly applying the chain of transformations to delayedResult."""
        chainTrav = self.Traverser(delayedResult, self.__chain[1:])
        handler = self.__chain[0]
        handler( chainTrav )