Das deutsche Python-Forum

Code: Alles auswählen

ergebnis = test.run()

Code: Alles auswählen

class MyWorker(threading.Thread):
   
    def __init__(self, finished_function = None):
        threading.Thread.__init__(self)
        self.canceled = threading.Event()
        self.finished_function = finished_function
   
   
    def run(self):
        for i in xrange(10):
            if self.canceled.isSet():
                print "Gestoppt..."
                break
            print "Ich arbeite..."
            time.sleep(1)
       
        if self.finished_function:
            self.finished_function()
   
   
    def stop(self):
        self.canceled.set()

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: iso-8859-15 -*-

import time


def my_worker(start = 0, stop = 10, step = 1, status_func = None, finished_func = None):
    """
    Arbeitende Funktion.
    
    :param start: Gibt die Zahl an, ab der gezählt werden soll.
    :param stop: Gibt die Zahl an, bis zu der gezählt werden soll.
    :param step: Schrittwert
    :param status_func: Wenn angegeben, dann wird eine Funktion erwartet, an die
        man als Parameter den Zwischenstand übergeben kann.
        Z.B. so: ``status_func("Neuer Zwischenstand")``
    :param finished_func: Wird ausgeführt, wenn der Worker mit seiner Arbeit fertig ist.
        Es wird eine ausführbare Funktion oder None erwartet.
    """
    
    if not callable(status_func):
        status_func = None
    
    for i in range(start, stop, step):
        if status_func:
            status_func("Zwischenstand: %i" % i)
        time.sleep(0.6)
    
    if status_func:
        status_func("Fertig")
    
    if callable(finished_func):
        finished_func()


def main():
    # Testen
    
    def status_func(message):
        print message
    
    def finished_func():
        print "END"
    
    my_worker(2, 8, status_func = status_func, finished_func = finished_func)


if __name__ == "__main__":
    main()

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: iso-8859-15 -*-

import wx
from thread import start_new_thread
import example_lib


wx.SetDefaultPyEncoding("iso-8859-15")


class MyFrame(wx.Frame):
    
    def __init__(
        self, parent = None, title = "Example", size = wx.Size(550, 420)
    ):
        wx.Frame.__init__(self, parent, -1, title, size = size)
        
        panel = wx.Panel(self)
        
        vbox_main = wx.BoxSizer(wx.VERTICAL)
        panel.SetSizer(vbox_main)
        
        vbox = wx.BoxSizer(wx.VERTICAL)
        vbox_main.Add(vbox, 1, wx.EXPAND | wx.ALL, 5)
        
        txt_status = wx.TextCtrl(
            panel, style = wx.TE_MULTILINE | wx.TE_AUTO_SCROLL | wx.TE_READONLY
        )
        vbox.Add(txt_status, 1, wx.EXPAND | wx.ALL)
        self.txt_status = txt_status
        
        btn_start = wx.Button(panel, label = "Start")
        vbox.Add(btn_start, 0, wx.ALIGN_CENTER_HORIZONTAL | wx.ALL, 5)
        btn_start.Bind(wx.EVT_BUTTON, self.start_worker)
        self.btn_start = btn_start
    
    
    def show_status(self, message):
        """
        Zeigt den neuen Status in der Textbox an.
        
        :param message: Dieser Text wird in der Textbox angehängt.
        """
        
        message = message.rstrip() + "\n"
        
        # Hier findet eine Trennung zwischen den Threads statt
        wx.CallAfter(self.txt_status.AppendText, message)
    
    
    def worker_finished(self):
        """
        Wird ausgeführt, wenn der Worker fertig ist. Damit könnte man den
        Start-Button wieder aktivieren, falls man diesen vorher deaktiviert
        hat.
        """
        
        # Hier findet eine Trennung zwischen den Threads statt
        wx.CallAfter(self.btn_start.Enable)
    
    
    def start_worker(self, event):
        """
        Startet den Worker aus dem Modul "example_lib".
        """
        
        # self.btn_start.Disable()
        
        self.txt_status.SetValue("")
        start_new_thread(
            example_lib.my_worker, (), 
            dict(
                start = 10, stop = 29, step = 2, status_func = self.show_status,
                finished_func = self.worker_finished
            )
        )


def main():
    """Testing"""
    app = wx.PySimpleApp()
    f = MyFrame()
    f.Center()
    f.Show()
    app.MainLoop()


if __name__ == "__main__":
    main()

Code: Alles auswählen

        start_new_thread(
            example_lib.my_worker, (),
            dict(
                start = 10, stop = 29, step = 2, status_func = self.show_status,
                finished_func = self.worker_finished
            )
        )

Code: Alles auswählen

class MeineKlasse(object):
    def __init__(self,...):
        ...
    def my_worker(self):
        ...

meine_klasseninstanz = MeineKlasse(start = 1, ...)

start_new_thread(meine_klasseninstanz.my_worker, ())