In diesem Beispiel möchte ich aufzeigen, wie man Events von einem arbeitenden Thread-Objekt aus als Benachrichtigung zum Aktualisieren der Anzeige an ein oder mehrere Frames weitergeben kann.
Erklärungen zum Beispiel:
Die "MyWorkingThread"-Klasse soll den arbeitenden Thread simulieren. Diese sendet jede Sekunde ein Event an alle angemeldeten Event-Empfänger-Objekte.
Jedes Mal, wenn die "MyWorkingThread"-Klasseninstanz das Event "EVT_NEW_STATUS" auslöst, dann wird die Eventhandler-Methode des Frames ausgeführt. Diese Methode aktualisiert die Anzeige im Frame.
Code: Alles auswählen
#!/usr/bin/env python
# -*- coding: iso-8859-15 -*-
import wx
import threading
import wx.lib.newevent
import time
wx.SetDefaultPyEncoding("iso-8859-15")
# Event-Klasse und Eventbinder-Funktionen erstellen
(NewStatusEvent, EVT_NEW_STATUS) = wx.lib.newevent.NewEvent()
class MyWorkingThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.cancel = threading.Event()
# Liste mit Eventempfänger-Objekten.
# An jedes Objekt dieser Liste wird das Event geschickt.
self.event_recipients = []
def add_event_recipient(self, recipient):
"""
Fügt der Liste der Eventempfänger einen neuen Empfänger hinzu.
"""
self.event_recipients.append(recipient)
def run(self):
"""
Working Thread
Dient nur zum Simulieren von ARBEIT! ;-)
"""
while True:
# Warten
time.sleep(1)
# Abbrechen, falls das Signal dazu gegeben wurde
if self.cancel.isSet():
break
# damit man sieht, dass etwas passiert...
print "Neuer Schleifendurchlauf im Thread..."
# Event-Instanz erstellen
evt = NewStatusEvent()
# Aktuelle Zeit an die Event-Instanz binden
evt.current_time = time.asctime()
# Das Event an alle Empfänger übergeben und auslösen.
for recipient in self.event_recipients:
if recipient:
wx.CallAfter(recipient.ProcessEvent, evt)
def stop(self):
self.cancel.set()
class MyFrame(wx.Frame):
def __init__(
self, parent = None, id = -1, title = "Example", size = wx.Size(300, 200)
):
wx.Frame.__init__(self, parent, id, title, size = size)
panel = wx.Panel(self)
vbox_main = wx.BoxSizer(wx.VERTICAL)
panel.SetSizer(vbox_main)
# Label zum Anzeigen des Status bzw. hier im Beispiel der Uhrzeit
self.lab_status = wx.StaticText(panel)
vbox_main.Add(self.lab_status, 0, wx.ALL, 10)
# Das Event, das von der **MyWorkingThread-Instanz** gesendet wird,
# an eine Methode (=Eventhandler) binden
self.Bind(EVT_NEW_STATUS, self.update_status)
def update_status(self, event):
"""
Dieser Eventhandler bekommt ein Event übergeben. An dieses Event
wurde die aktuelle Zeit als Attribut ``current_time`` angehängt.
Diese wird jetzt im StaticText-Feld angezeigt.
"""
print "Neue Zeit in %s anzeigen..." % self.GetTitle()
self.lab_status.SetLabel(event.current_time)
class MyApp(wx.PySimpleApp):
def OnInit(self):
# Frame1 anzeigen
myframe1 = MyFrame(title = "Fenster 1")
myframe1.Show()
# Frame2 anzeigen
myframe2 = MyFrame(title = "Fenster 2")
myframe2.Show()
# Arbeitenden Thread erstellen
myworkingthread = MyWorkingThread()
# Frame-Objekte als Empfänger für die Events übergeben
myworkingthread.add_event_recipient(myframe1)
myworkingthread.add_event_recipient(myframe2)
# Arbeitenden Thread starten
myworkingthread.start()
# ab jetzt sendet myworkingthread jede Sekunde ein Event an
# alle Empfänger-Objekte.
# Warten, bis alle Fenster geschlossen wurden
self.MainLoop()
# Das Signal zum Stoppen des arbeitenden Threads geben und warten bis
# der Thread auch wirklich gestoppt wurde.
myworkingthread.stop()
myworkingthread.join()
return True
def main():
app = MyApp()
if __name__ == "__main__":
main()
mfg
Gerold
Edit: Ich hatte die Threadtrennung mit ``wx.CallAfter`` nicht eingebaut. (Zeile 62)