[Twisted-Python] modal wxPython, Twisted, xml-rpc; example code

[Andrew Dalke]

--Apple-Mail-10--997149463
Content-Transfer-Encoding: 7bit
Content-Type: text/plain;
	charset=US-ASCII;
	format=flowed

I've attached some example code which shows how to use
a modal wxPython dialog and Twisted to produce a responsive
GUI which uses XML-RPC to query a remote XML-RPC service to
get a US state name given it's position in a sorted list.

I've also contributed this as an ActiveState recipe.

					Andrew
					dalke@dalkescientific.com

--Apple-Mail-10--997149463
Content-Disposition: attachment;
	filename=wx_modal_twisted.py
Content-Transfer-Encoding: 7bit
Content-Type: application/octet-stream;
	x-unix-mode=0755;
	name="wx_modal_twisted.py"

# In the public domain
# Author: Andrew Dalke <dalke@dalkescientific.com
#
# This is an example of a wxPython-based modal progress bar
# which uses Twisted to do XML-RPC requests from the state name
# demo server.  It's easily extensible to other tasks.

from __future__ import generators
from wxPython.wx import *

from twisted.internet import reactor
from twisted.web.xmlrpc import Proxy

# returned from the main modal
COMPLETED, CANCELED, ERROR = range(3)

# returned from the (sub)modal when there's an XML-RPC error
RETRY, SKIP, STOP = range(20, 23)

# The progress modal does N "tasks".  Each task has a 
# "start" method, which returns a deferred.  The deferred
# is chained to call "good" with the result or "bad"
# with the failure.  The "bad" method may return one of
# RETRY, SKIP, or STOP to tell the modal how to recover.
class Task:
    def start(self):
        """returns a defered"""
        raise NotImplementedError
    def good(self, result):
        pass
    def bad(self, fail):
        pass

# To show examples of error modes
TEST_ERRORS = 1

# A Task to get the state name corresponding to the given number.
class StateTask(Task):
    def __init__(self, i):
        self.i = i

    def start(self):
        proxy = Proxy("http://beatty.userland.com/RPC2")
        if TEST_ERRORS:
            i = self.i
            if i == 3:
                proxy = Proxy("http://illegal-host_name/")
            elif i == 6:
                proxy = Proxy("http://beatty.userland.com/")
            elif i == 8:
                proxy = Proxy("http://beatty.userland.com/testing_xmlrpc_error_case")
            
        return proxy.callRemote('examples.getStateName', self.i)

    def good(self, result):
        print "state", self.i, "is", result

    def bad(self, fail):
        # pop up a submodal
        status = wxMessageBox("Cannot get name for state %d.  Try again?\n"
                              "\n"
                              "The problem is: %s" % (self.i, fail.getErrorMessage()),
                              "Connection problem",
                              wxCANCEL | wxYES_NO | wxICON_QUESTION)
        if status == wxYES:
            return RETRY
        elif status == wxNO:
            return SKIP
        elif status == wxCANCEL:
            return STOP
        else:
            raise AssertionError(status)

# The progress dialog must be passed a "task list" object which
# implements len() (needed to know how many steps to show) and
# does forward iteration.
class StateTaskList:
    def __init__(self, min=0, max=50):
        self.min = min
        self.max = max
    def __iter__(self):
        for i in range(self.min, self.max):
            yield StateTask(i)
            
    def __len__(self):
        return self.max - self.min
        

class Progress(wxDialog):
    def __init__(self, parent, ID, title, tasks,
                 pos=wxDefaultPosition, size=wxDefaultSize,
                 style=wxDEFAULT_DIALOG_STYLE):
        wxDialog.__init__(self, parent, ID, title, pos, size, style)

        n = len(tasks)
        self.task_iter = iter(tasks)

        sizer = wxBoxSizer(wxVERTICAL)

        self.gauge = wxGauge(self, -1, 100, size = (300, -1))
        sizer.Add(self.gauge, 0, wxALIGN_CENTER|wxALL, 5)

        box = wxBoxSizer(wxHORIZONTAL)
        spacer = wxStaticText(self, -1, "")
        box.Add(spacer, 1, wxALIGN_CENTRE|wxALL|wxGROW)
        btn = wxButton(self, wxID_CANCEL, " Cancel ")
        box.Add(btn, 0, wxALIGN_CENTRE|wxALL, 5)
        spacer = wxStaticText(self, -1, "")
        box.Add(spacer, 1, wxALIGN_CENTRE|wxALL|wxGROW)

        sizer.AddSizer(box, 0, wxALIGN_CENTER_VERTICAL|wxALL|wxGROW, 5)

        EVT_BUTTON(self, wxID_CANCEL, self.OnCancel)

        self.SetSizer(sizer)
        self.SetAutoLayout(true)
        sizer.Fit(self)

        self.i = 0
        self.Start(n)

        self._canceled = 0
        self.Feed()

    def OnCancel(self, event):
        self._canceled = 1
        self.EndModal(CANCELED)

    def Feed(self):
        # Get the next task and start it up
        try:
            task = self.task_iter.next()
        except StopIteration:
            self.End()
            return
        self.StartTask(task)

    def StartTask(self, task):
        defered = task.start()
        def do_good(result):
            self.Good(task, result)
        def do_bad(fail):
            self.Bad(task, fail)
        defered.addCallbacks(do_good, do_bad)

    def Good(self, task, result):
        if self._canceled:
            return
        task.good(result)
        self.Update(1)
        self.Feed()

    def Bad(self, task, fail):
        if self._canceled:
            return
        try_again = task.bad(fail)
        if try_again == RETRY:
            self.StartTask(task)
        elif try_again == SKIP:
            self.Update(1)
            self.Feed()
        elif try_again == STOP:
            self._canceled = 1
            self.EndModal(ERROR)
        else:
            raise AssertionError(try_again)

    def Start(self, count):
        self.gauge.SetRange(count)
        self.gauge.SetValue(0)

    def Update(self, incr):
        # Increment the counter.
        self.i += incr
        self.gauge.SetValue(self.i)
        
    def End(self):
        self.gauge.SetValue(self.gauge.GetRange())
        self.Close()
        self.EndModal(COMPLETED)

# Thanks to Uwe C. Schroeder and his "Using wxPython with Twisted
# Python" recipe at aspn.ActiveState.com

class MyApp(wxApp):
    def OnInit(self):
        # Twisted Reactor code
        reactor.startRunning()
        EVT_TIMER(self, 999999, self.OnTimer)
        self.timer = wxTimer(self, 999999)
        self.timer.Start(150, False)

        return true

    def OnTimer(self, event):
        reactor.runUntilCurrent()
        reactor.doIteration(0)

    def __del__(self):
        self.timer.Stop()
        reactor.stop()
        wxApp.__del__(self)

def main():
    app = MyApp(0)
    win = Progress(None, -1, "Processing ...", StateTaskList(1, 10))

    status = win.ShowModal()
    if status == COMPLETED:
        print "All done"
    elif status == CANCELED:
        print "Okay, I stopped."
    elif status == ERROR:
        print "What happened?"
    else:
        raise AssertionError(status)
    
if __name__ == "__main__":
    main()

--Apple-Mail-10--997149463--