| 1 | # Thread testing example. Harm van der Heijden, March 26 1999. |
| 2 | # |
| 3 | # Rule One in threading: make sure only one thread interacts with the |
| 4 | # user interface. See the wxTextCtrlQueue class for an example of how |
| 5 | # to accomplish this |
| 6 | |
| 7 | import thread |
| 8 | import time |
| 9 | from whrandom import random |
| 10 | |
| 11 | from wxPython.wx import * |
| 12 | |
| 13 | # Set this to zero to prevent entering the wxApp mainloop |
| 14 | # (for testing whether threads work at all in the absense of wxWindows) |
| 15 | use_wxpython = 1 |
| 16 | |
| 17 | # write a message to stdout every second |
| 18 | def DoThread(mesg): |
| 19 | while 1: |
| 20 | sleeptime = (random() * 3) + 0.5 |
| 21 | print "Hello from %s (%1.3f)" % (mesg, sleeptime) |
| 22 | time.sleep(sleeptime) |
| 23 | |
| 24 | # the same, but write it to a textctrl. |
| 25 | def DoTextCtrlThread(text, mesg): |
| 26 | while 1: |
| 27 | sleeptime = (random() * 3) + 0.5 |
| 28 | text.WriteText("Hello from %s (%1.3f)\n" % (mesg, sleeptime)) |
| 29 | time.sleep(sleeptime) |
| 30 | |
| 31 | # A very simple queue for textctrls. |
| 32 | # Nice demonstration of the power of OO programming too (at least I think so!) |
| 33 | # WriteText puts text in the queue, rather than writing it immediately. |
| 34 | # The main (UI) thread must call Flush to force output. (see MyFrame::OnIdle) |
| 35 | class wxTextCtrlQueue(wxTextCtrl): |
| 36 | def __init__(self, parent, id, value, pos, size, flags): |
| 37 | wxTextCtrl.__init__(self,parent, id, value, pos, size, flags) |
| 38 | self.queue = [] |
| 39 | def WriteText(self, value): |
| 40 | self.queue.append(value) |
| 41 | def Flush(self): |
| 42 | queue = self.queue |
| 43 | self.queue = [] |
| 44 | for value in queue: |
| 45 | wxTextCtrl.WriteText(self,value) |
| 46 | |
| 47 | # MyFrame and MyApp are very simple classes to test python threads in |
| 48 | # wxPython. |
| 49 | class MyFrame(wxFrame): |
| 50 | def __init__(self): |
| 51 | wxFrame.__init__(self, NULL, -1, "test threads", wxDefaultPosition, wxSize(300,200)) |
| 52 | self.text = wxTextCtrlQueue(self, -1, "thread output\n", wxDefaultPosition, wxDefaultSize, wxTE_MULTILINE) |
| 53 | menu = wxMenu() |
| 54 | menu.Append(1001, "Start thread") |
| 55 | self.cnt = 0; |
| 56 | menubar = wxMenuBar() |
| 57 | menubar.Append(menu, "Action") |
| 58 | self.SetMenuBar(menubar) |
| 59 | EVT_MENU(self, 1001, self.StartThread) |
| 60 | def StartThread(self, event): |
| 61 | self.cnt = self.cnt + 1 |
| 62 | thread.start_new_thread(DoTextCtrlThread, (self.text, "thread %d" % self.cnt)) |
| 63 | def OnIdle(self, event): |
| 64 | self.text.Flush() |
| 65 | |
| 66 | class MyApp(wxApp): |
| 67 | def OnInit(self): |
| 68 | frame = MyFrame() |
| 69 | self.SetTopWindow(frame) |
| 70 | frame.Show(TRUE) |
| 71 | return TRUE |
| 72 | |
| 73 | # Start two threads that print a message every second |
| 74 | thread.start_new_thread(DoThread, ("thread A",)) |
| 75 | thread.start_new_thread(DoThread, ("thread B",)) |
| 76 | |
| 77 | # if using wxpython, open a frame. Otherwise, just hang in while 1 |
| 78 | if use_wxpython: |
| 79 | app = MyApp(0) |
| 80 | app.MainLoop() |
| 81 | else: |
| 82 | while 1: |
| 83 | print "main loop" |
| 84 | time.sleep(4) |
| 85 | print 'done!' |