Das deutsche Python-Forum

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: iso-8859-1 -*-
"""
Threading mit Event - Beispiel
"""

import threading
import time
import Tkinter
import sys


class Motorsteuerung(threading.Thread):
    """
    Simulierte Schrittmotorsteuerung
    """
   
    def __init__(self, motormeldungen = None):
        """
        schalter = threading.Event
        motormeldungen = Rückgabe als Tkinter.StringVar, damit eine
                         Meldung in einem Label angezeigt werden kann.
        """
       
        threading.Thread.__init__(self)
       
        self.schalter = threading.Event()
        self.motormeldungen = motormeldungen
        self.canceled = False


    def run(self):
        """
        Der Motor wird initialisiert.
        Enthaelt die Steuerschleife.
        """
       
        i = 0
       
        while True:
            # Hier wird darauf gewartet, dass der Schalter
            # eingeschaltet wird.
            self.schalter.wait()
            if self.canceled:
                break
           
            # Statt einer Schrittmotor-Aansteuerung wird hier einfach
            # ein Text an die Variable "motormeldungen" uebergeben.
            # Statt mit einer Tkinter.StringVar könnte man natürlich
            # auch mit print etwas anzeigen lassen.
            i += 1
            if self.motormeldungen:
                self.motormeldungen.set("Motor laeuft (%s)" % i)
            else:
                print "Motor laeuft (%s)" % i
            time.sleep(0.5)
   
   
    def motor_start(self):
        """
        Schaltet den Motor ein
        """
        self.schalter.set()


    def motor_stopp(self):
        """
        Stoppt den Motor aus
        """
        self.schalter.clear()
   

    def motor_aus(self):
        """
        Schaltet den Motor aus
        """
        self.canceled = True
        self.schalter.set()
   

class MeinFenster(Tkinter.Tk):
    """
    GUI
    """
   
    def _center(self, *args):
        """
        Zentriert das Fenster
        """
        xpos = (self.winfo_screenwidth() - self.winfo_width()) / 2
        ypos = ((self.winfo_screenheight() - self.winfo_height()) / 2) / 100 * 90
        self.wm_geometry("+%d+%d" % (xpos,ypos))

   
    def motor_aus(self):
        """
        Schaltet die Schleife aus und schliesst das Fenster
        """
        self.motor.motor_aus()
        self.destroy()
       

    def __init__(self):
        """
        Anzeigen und initialisieren
        """
       
        # Init
        Tkinter.Tk.__init__(self)

        self.motormeldungen = Tkinter.StringVar()

        # Motor initialisieren
        self.motor = Motorsteuerung(self.motormeldungen)
        self.motor.start()

        # Fenster und Buttons
        self.config(bd = 10)
        frame = Tkinter.Frame(self)
        frame.pack()
        Tkinter.Button(
            frame,
            text = "Start",
            command = self.motor.motor_start,
            bd = 10, padx = 10, pady = 10
        ).grid(row = 0, column = 0)
        Tkinter.Button(
            frame,
            text = "Stopp",
            command = self.motor.motor_stopp,
            bd = 10, padx = 10, pady = 10
        ).grid(row = 0, column = 1)
       
        # Label
        lab = Tkinter.Label(
            self,
            text = "",
            textvariable = self.motormeldungen,
            bd = 10, padx = 10, pady = 10
        ).pack()
       
        # Aus
        Tkinter.Button(
            self,
            text = "Motor AUS",
            command = self.motor_aus,
            bd = 4
        ).pack()
       
        # Warten bis das Fenster angezeigt wird und dann zentrieren
        self.wait_visibility()
        self._center()
       

if __name__ == "__main__":
    fenster = MeinFenster()
    Tkinter.mainloop()
    sys.exit(0)

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: iso-8859-1 -*-
"""
******************************************************************************
* Zwei Threads; Ein Thread zählt ständig nach oben, bis dieser durch ein
* Stopp-Signal (Event) aufgefordert wird, sich zu beenden. Der zweite Thread
* ist sozusagen der Timer, der von 10 bis 0 nach unten
* zählt und bei erreichen von 0 dem ersten Thread das Signal zum Beenden gibt.
* Der erste Thread erhält das Signal zum Beenden und schaltet den Hauptschalter
* aus, der sich darum gekümmert hat, dass die Hauptfunktion stehen bleibt.
******************************************************************************
"""

import time
import threading


class NachObenZaehler(threading.Thread):
    """
    Diese Klasse enthält einen Zähler, der ständig nach oben zählt und
    darauf wartet, bis er ausgeschaltet wird.
    """
   
    def __init__(self, ausschalter, hauptschalter):
        """
        Bindet den Ausschalter an eine Instanzvariable und initialisiert
        den Thread. Der Hauptschalter wird auch an eine Instanzvariable
        gebunden. Dieser hält die Hauptfunktion so lange am Leben, bis
        dieser Thread beendet wird.
        """
       
        threading.Thread.__init__(self)
       
        self.ausschalter = ausschalter
        self.hauptschalter = hauptschalter


    def run(self):
        """
        'run' ist die Methode, die aufgerufen wird, wenn 'NachObenZaehler.start()'
        ausgeführt wird.
        """
       
        i = 0
        while True:
            i += 1
           
            # Prüfen ob der Ausschalter betätigt wurde
            if self.ausschalter.isSet():
                # Der Ausschalter wurde betätigt --> Hauptschalter umlegen
                print (
                    "Erkannt, dass der Ausschalter umgelegt wurde --> "
                    "Hauptschalter umlegen"
                )
                self.hauptschalter.set()
                break
            else:
                print "NachObenZaehler: %s" % i
                time.sleep(1)


class NachUntenZaehler(threading.Thread):
    """
    Diese Klasse enthält einen Zähler, der von 10 nach 0 zählt und bei
    erreichen von 0 einen übergebenen Schalter (=threading.Event)
    einschaltet.
    """
   
    def __init__(self, ausschalter):
        """
        Bindet den Ausschalter an eine Instanzvariable und initialisiert
        den Thread
        """
       
        threading.Thread.__init__(self)

        self.ausschalter = ausschalter


    def run(self):
        """
        'run' ist die Methode, die aufgerufen wird, wenn 'NachUntenZaehler.start()'
        ausgeführt wird.
        """

        # Nach unten Zählen
        for i in range(10, 0, -1):
            print "    NachUntenZaehler: %s" % i
            time.sleep(1)
       
        # Schalter umlegen :-)
        self.ausschalter.set()
        print "    Schalter umgelegt"


def main():
    """
    Hauptprozedur: Erstellt die zwei Klasseninstanzen und startet
    danach die Threads.
    """
   
    # Hauptschalter erstellen. Dieser kümmert sich darum, dass diese
    # Funktion erst dann beendet wird, wenn der NachObenZaehler-Thread auch
    # wirklich fertig ist.
    hauptschalter = threading.Event()
   
    # Ausschalter erstellen. Dieser wird an die beiden Threads übergeben
    ausschalter = threading.Event()
   
    # Klasseninstanzen erstellen und die Schalter übergeben.
    # Die Klasseninstanz 'count_down' wird den Schalter nach Ablauf der
    # Zeit "einschalten" und damit der anderen Klasse signalisieren, dass
    # sie auslaufen soll.
    count_up = NachObenZaehler(ausschalter, hauptschalter)
    count_down = NachUntenZaehler(ausschalter)
   
    # Starten
    count_up.start()
    count_down.start()
   
    # Hier wird gewartet bis der Hauptschalter umgelegt wurde.
    hauptschalter.wait()
   
    # Meldung von der Hauptprozedur
    print "Hauptprozedur ist fertig"
   

if __name__ == "__main__":
    main()

Code: Alles auswählen

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

import threading
import Queue
import time


class WartenderThread(threading.Thread):
    """
    Dieser Thread wartet auf die Anweisung, verschiedenste Jobs abzuarbeiten.
    """
   
    def __init__(self, jobqueue, ausschalter):
        """
        Übernimmt die Queue, die später die Jobanweisungen übergibt.
        """
       
        threading.Thread.__init__(self)
       
        self.jobqueue = jobqueue
        self.ausschalter = ausschalter
   
   
    def hallo_welt_1(self):
        """
        Diese Funktion wird aufgerufen, wenn an die
        Jobqueue "hallo_welt_1" übergeben wurde.
        """
       
        print "    Ausgefuehrt: Hallo Welt 1"


    def hallo_welt_2(self):
        """
        Diese Funktion wird aufgerufen, wenn an die
        Jobqueue "hallo_welt_2" übergeben wurde.
        """
       
        print "    Ausgefuehrt: Hallo Welt 2"


    def run(self):
        """
        'run' ist die Methode, die aufgerufen wird, wenn 'WartenderThread.start()'
        ausgeführt wird.
        """
       
        # In einer Schleife wird darauf **gewartet** bis ein neuer Job
        # an die Jobqueue übergeben wird.
        while True:
           
            # Hier wird gewartet
            anweisung = self.jobqueue.get()
           
            # Diesen Thread kurz schlafen schicken, damit wird auch die
            # Unabhängigkeit des Threades demonstriert.
            time.sleep(1)

            print "Anweisung: %s" % anweisung
           
            # Wenn Quit übergeben wurde, dann darf die Schleife
            # nicht weitergeführt werden.
            if anweisung == "quit":
                print "    Ausschalter setzen"
                self.ausschalter.set()
                break

            # Prüfen ob die Methode mit dem Namen %(anweisung)s
            # existiert. Wenn Ja, ausführen.
            if hasattr(self, anweisung):
                getattr(self, anweisung)()


class AnweisenderThread(threading.Thread):
    """
    Dieser Thread gibt Anweisungen an den anderen Thread weiter.
    """
   
    def __init__(self, jobqueue):
        """
        Übernimmt die Queue, mit der die Jobanweisungen an den anderen
        Thread übergeben werden.
        """
       
        threading.Thread.__init__(self)

        self.jobqueue = jobqueue


    def run(self):
        """
        'run' ist die Methode, die aufgerufen wird, wenn 'AnweisenderThread.start()'
        ausgeführt wird.
        """
       
        # Ein paar Anweisungen in eine Liste stecken...
        # "hallo_welt_3" ist nur ein Platzhalter, ohne zugeh. Funktion
        # im wartenden Thread.
        anweisungen = ["hallo_welt_1", "hallo_welt_2", "hallo_welt_3"]
       
        # Die ersten Anweisungen in die Queue stellen.
        for anweisung in anweisungen:
            self.jobqueue.put(anweisung)
       
        # Durch diese Wartezeit, sieht man ziemlich gut, dass der wartende
        # Thread wirklich auf die nächste Anweisung aus der Queue wartet.
        time.sleep(10)
       
        # Die nächsten zwei Anweisungen in die Queue stellen.
        for anweisung in anweisungen:
            self.jobqueue.put(anweisung)
       
        # Die Anweisung zum Beenden schicken.
        self.jobqueue.put("quit")
        return


def main():
    """
    Startet die Threads und führt den Test durch.
    """
   
    # Jobqueue erstellen (Queue)
    q = Queue.Queue()
   
    # Ausschalter erstellen (Event)
    ausschalter = threading.Event()
   
    # Threads initialisieren und dabei die Jobqueue und den
    # Ausschalter übergeben
    wt = WartenderThread(q, ausschalter)
    at = AnweisenderThread(q)
   
    # Threads starten
    wt.start() # Dieser wartet so lange bis Jobanweisungen geschickt werden.
    at.start() # Dieser schickt Jobanweisungen über die Jobqueue.
   
    # Warten bis der Ausschalter betätigt wurde. Sonst würde das
    # Programm sofort beendet werden.
    ausschalter.wait()
   
   
if __name__ == "__main__":
    main()

Code: Alles auswählen

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

import threading
import Queue
import time
import types


class WartenderThread(threading.Thread):
    """
    Dieser Thread wartet auf die Anweisung, verschiedenste Jobs abzuarbeiten.
    """
   
    def __init__(self, jobqueue, ausschalter):
        """
        Übernimmt die Queue, die später die Jobanweisungen übergibt.
        """
       
        threading.Thread.__init__(self)
       
        self.jobqueue = jobqueue
        self.ausschalter = ausschalter
   
   
    def hallo_welt_1(self):
        """
        Diese Funktion wird aufgerufen, wenn an die
        Jobqueue "hallo_welt_1" übergeben wurde.
        """
       
        print "    Ausgefuehrt: Hallo Welt 1"


    def hallo_welt_2(self, **dargs):
        """
        Diese Funktion wird aufgerufen, wenn an die
        Jobqueue "hallo_welt_2" übergeben wurde.
        """
       
        print "    Ausgefuehrt: Hallo Welt 2"
        if dargs:
            for key, value in dargs.items():
                print "        %s: %s" % (key, value)


    def run(self):
        """
        'run' ist die Methode, die aufgerufen wird, wenn 'WartenderThread.start()'
        ausgeführt wird.
        """
       
        # In einer Schleife wird darauf **gewartet** bis ein neuer Job
        # an die Jobqueue übergeben wird.
        while True:
           
            # Hier wird gewartet
            anweisung = self.jobqueue.get()
           
            # Diesen Thread kurz schlafen schicken, damit wird auch die
            # Unabhängigkeit des Threades demonstriert.
            time.sleep(1)

            print "Anweisung: %s" % str(anweisung)
           
            # Wenn die Anweisung ein Tupel oder eine Liste ist, dann
            # wird diese(r) in die Anweisung und die Argumente zerlegt.
            if isinstance(anweisung, (types.ListType, types.TupleType)):
                anweisung, dargs = anweisung
            else:
                dargs = {}

            # Wenn Quit übergeben wurde, dann darf die Schleife
            # nicht weitergeführt werden.
            if anweisung == "quit":
                print "    Ausschalter setzen"
                self.ausschalter.set()
                break

            # Prüfen ob die Methode mit dem Namen %(anweisung)s
            # existiert. Wenn Ja, ausführen.
            if hasattr(self, anweisung):
                getattr(self, anweisung)(**dargs)


class AnweisenderThread(threading.Thread):
    """
    Dieser Thread gibt Anweisungen an den anderen Thread weiter.
    """
   
    def __init__(self, jobqueue):
        """
        Übernimmt die Queue, mit der die Jobanweisungen an den anderen
        Thread übergeben werden.
        """
       
        threading.Thread.__init__(self)

        self.jobqueue = jobqueue


    def run(self):
        """
        'run' ist die Methode, die aufgerufen wird, wenn 'AnweisenderThread.start()'
        ausgeführt wird.
        """
       
        # Ein paar Anweisungen in einen Tupel stecken...
        # "hallo_welt_3" ist nur ein Platzhalter, ohne zugeh. Funktion
        # im wartenden Thread.
        anweisungen = ("hallo_welt_1", "hallo_welt_2", "hallo_welt_3")
       
        # Die ersten Anweisungen in die Queue stellen.
        for anweisung in anweisungen:
            self.jobqueue.put(anweisung)
       
        # Durch diese Wartezeit, sieht man ziemlich gut, dass der wartende
        # Thread wirklich auf die nächste Anweisung aus der Queue wartet.
        time.sleep(10)
       
        # An die Anweisung "hallo_welt_2" werden auch die Parameter
        # übergeben.
        anweisungen = (
            "hallo_welt_1",
            ("hallo_welt_2", {"vorname": "Gerold", "nachname": "Penz"}),
        )

        # Die nächsten zwei Anweisungen in die Queue stellen.
        for anweisung in anweisungen:
            self.jobqueue.put(anweisung)
       
        # Die Anweisung zum Beenden schicken.
        self.jobqueue.put("quit")
        return


def main():
    """
    Startet die Threads und führt den Test durch.
    """
   
    # Jobqueue erstellen (Queue)
    q = Queue.Queue()
   
    # Ausschalter erstellen (Event)
    ausschalter = threading.Event()
   
    # Threads initialisieren und dabei die Jobqueue und den
    # Ausschalter übergeben
    wt = WartenderThread(q, ausschalter)
    at = AnweisenderThread(q)
   
    # Threads starten
    wt.start() # Dieser wartet so lange bis Jobanweisungen geschickt werden.
    at.start() # Dieser schickt Jobanweisungen über die Jobqueue.
   
    # Warten bis der Ausschalter betätigt wurde. Sonst würde das
    # Programm sofort beendet werden.
    ausschalter.wait()
   
   
if __name__ == "__main__":
    main()