Wir zeigen anhand Benjamins schönem Beispielprogramm, wie Funktionen als eigenständige Threads ausgeführt werden können.
Es verwendet das simulierte RPi.GPIO.add_event_callback(…) von RevPiModIO – Die Funktion ist bekannt vom RaspberryPi.
In „BennisRun“ wird durch setzen des Inputs eine Funktion gestartet, die nacheinander alle Outputs auf True setzt. Sie verwendet die sleep() Funktion und braucht daher „viel“ Zeit. Wenn der Input zurückgesetzt wird, startet eine andere Funktion, die alle Outputs nacheinander auf False setzt.
Normalerweise werden alle, durch Events ausgelösten, Funktionen komplett abgearbeitet, bevor der mainloop() weiter läuft und weitere Events ausgelöst werden können. Dies ist gewollt und wird realisiert, indem die Funktionen vom selben Thread ausgeführt werden, den auch der mainloop() verwendet (Die Prozessbildaktualisierung läuft in einem anderen Thread und wird dadurch natürlich NICHT angehalten! Es sei denn, beim Eintritt in den mainloop() wird der parameter freeze=True übergeben).
Für „BennisRun“ bedeutet dies, dass erst die eine Funktion alle Outputs auf True setzen muss und erst dann die andere Funktion aufgerufen werden kann, die alle Outputs wieder auf False setzt.
ES SEI DENN: Wir verwenden bei reg_event() den Parameter „as_thread=True“! Dies ähnelt dem Verhalten von add_event_callback(…).
Alte RevPiModIO Syntax
self.revpi.devices["devname"].reg_event("Inputname", self.eventfunktion, edge=revpimodio.RISING, as_thread=True)
Neue RevPiModIO2 Syntax
self.revpi.io.Inputname.reg_event(self.eventfunktion, edge=revpimodio2.RISING, as_thread=True)
Dann werden die Funktionen als einzelne Threads gestartet und können parallel laufen – Wenn im Code nicht überprüft, sogar die selbe Funktion mehrfach (Die Funktion muss als Übergabeparameter das Keyword „thread“ enthalten).
Das sieht dann so aus:
Und das ist der kleine Quelltext:
Wir haben aber noch weitaus mehr Möglichkeiten, die in unserem übergebenen „thread“ Objekt, vom Typen RevPiCallback, stecken! In diesem extremeren Beispiel verwenden wir nur eine einzige Event-Funktion In der „runner“ Funktion fragen wir über Über Zum Warten verwenden wir nun kein Und genau dieses Mit einem globalen threading.Lock() Objekt #!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# (c) Benjamin
"""Beispielprogramm fuer mainloop() und as_thread."""
import revpimodio2
import time
# Diese Werte müssen an die eigene piCtory Konfiguration angepasst werden!
# Name des Inputs: I_1
outputformat = "O_{}"
class RevPiModIOTest():
"""Programmklasse."""
def __init__(self):
"""Instanziiert unser Programm."""
self.out = [outputformat.format(x) for x in range(1, 15)]
# RevPiModIO2 Instanziieren und "Programm beenden"-Signal verwalten
self.revpi = revpimodio2.RevPiModIO(autorefresh=True)
self.revpi.handlesignalend(self.exitfunktion)
# Events Registrieren, die als THREAD ausgeführt werden
self.revpi.io.I_1.reg_event(
self.eventfunktion1, edge=revpimodio2.RISING, as_thread=True
)
self.revpi.io.I_1.reg_event(
self.eventfunktion2, edge=revpimodio2.FALLING, as_thread=True
)
def exitfunktion(self):
"""Diese Funktion wird beim Beenden des Programms ausgefuert."""
self.revpi.core.A1 = revpimodio2.OFF
def eventfunktion1(self, thread):
"""Lichter der Reihe nach einschalten."""
for i in range(len(self.out)):
self.revpi.io[self.out[i]].value = True
time.sleep(0.1)
def eventfunktion2(self, thread):
"""Lichter der Reihe nach ausschalten."""
for i in range(len(self.out)):
self.revpi.io[self.out[i]].value = False
time.sleep(0.1)
def start(self):
"""Startet den mainloop()."""
self.revpi.core.A1 = revpimodio2.GREEN
print("Gehe in den mainloop()")
self.revpi.mainloop()
print("Verlasse mainloop()")
if __name__ == "__main__":
root = RevPiModIOTest()
root.start()
#!/usr/bin/python3
# (c) Benjamin
# -*- coding: utf-8 -*-
import revpimodio
import signal
import time
# Diese Werte müssen an die eigene piCtory Konfiguration angepasst werden!
outputformat = "O_{}"
iodevname = "dio02"
eventinput = "I_1"
class RevPiModIOTest():
def __init__(self):
self.out = [outputformat.format(x) for x in range(1, 15)]
self.revpi = revpimodio.RevPiModIO(auto_refresh=True)
# Events Registrieren, die als THREAD ausgeführt werden
self.revpi.devices[iodevname].reg_event(
eventinput, self.eventfunktion1,
edge=revpimodio.RISING, as_thread=True
)
self.revpi.devices[iodevname].reg_event(
eventinput, self.eventfunktion2,
edge=revpimodio.FALLING, as_thread=True
)
signal.signal(signal.SIGINT, self._sigexit)
signal.signal(signal.SIGTERM, self._sigexit)
def _sigexit(self, signum, frame):
self.revpi.devices.exit()
print("Verlasse mainloop()")
def eventfunktion1(self, thread):
for i in range(len(self.out)):
self.revpi.devices[iodevname][self.out[i]].value = 1
time.sleep(0.1)
def eventfunktion2(self, thread):
for i in range(len(self.out)):
self.revpi.devices[iodevname][self.out[i]].value = 0
time.sleep(0.1)
def start(self):
self.revpi.devices.core.A1 = revpimodio.GREEN
print("Gehe in den mainloop()")
self.revpi.devices.mainloop()
self.revpi.devices.core.A1 = revpimodio.OFF
self.revpi.devices.writeprocimg()
if __name__ == "__main__":
root = RevPiModIOTest()
root.start()
Die vollen Möglichkeiten!
def runner(self, thread):. Zwei Inputs „I_1“, „I_2“ für die Lauflichtrichtung und „I_3“ als Abbruchbedingung um frühzeitig das Verlassen der for-Schleifen zu ermöglichen.thread.ioname, den Namen des Inputs ab, der die Funktion aktuell aufgerufen hat. Damit bestimmen wir die Laufrichtung.thread.iovalue erhalten wir den zum Auslösezeitpunkt anstehenden Wert des Inputs. Mit dessen Zuweisung an die Outputs schalten wir diese an oder aus.time.sleep(0.1) mehr, da wir dort einfach feststecken würden und warten MÜSSEN. Unser „thread“ Objekt stellt uns das Event thread.exit zur Verfügung, mit dessen .wait(sekunden) Funktion wir warten. Dieses Warten kann durch .set() jederzeit abgebrochen werden..set() ruft unsere Funktion def killer(self, ioname=None, iovalue=None): für jeden aktiven Thread auf, welche durch „I_3“ als normales Event gestartet wird!self.threadlock = Lock() verhindern wir den gleichzeitigen Zugriff auf die Liste mit den derzeit aktiven Threads self.lst_threads = [] (beachtet die Umfangreiche Quellcodedoku)#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# (c) Benjamin, Sven Sager
"""Beispielprogramm fuer mainloop() und as_thread."""
import revpimodio2
from threading import Lock
# Diese Werte müssen an die eigene piCtory Konfiguration angepasst werden!
# Namen der Inputs: I_1, I_2, I_3
outputformat = "O_{}"
class RevPiModIOTest():
"""Programmklasse."""
def __init__(self):
"""Instanziiert unser Programm."""
self.out = [outputformat.format(x) for x in range(1, 15)]
# RevPiModIO2 Instanziieren und "Programm beenden"-Signal verwalten
self.revpi = revpimodio2.RevPiModIO(autorefresh=True)
self.revpi.handlesignalend(self.exitfunktion)
# Threadliste führen
self.lst_threads = []
self.threadlock = Lock()
# Events Registrieren, die als THREAD ausgeführt werden
self.revpi.io.I_1.reg_event(self.runner, as_thread=True)
self.revpi.io.I_2.reg_event(self.runner, as_thread=True)
# Event um alle aktiven Threads zu "killen" registrieren
self.revpi.io.I_3.reg_event(self.killer, edge=revpimodio2.RISING)
def exitfunktion(self):
"""Diese Funktion wird beim Beenden des Programms ausgefuert."""
print("Beende alle laufenden Threads")
self.killer()
self.revpi.core.A1 = revpimodio2.OFF
def killer(self, ioname=None, iovalue=None):
""" Alle laufenden Threads beenden.
Lockobjekt holen, welches verhindert, dass die abgebrochenen Threads
sich selbst aus der Liste loeschen und diese beim iterieren veraendern
Die abgebrochenen Threads benoetigen auch dieses Lock-Objekt und
warten auf dessen Freigabe
"""
with self.threadlock:
# Durch alle aktiven Threads gehen und sie beenden
for thread in self.lst_threads:
# Die thread.stop() Funktion von RevPiCallback aufrufen,
# welche das exit() Event setzt. Dieses Event verwenden
# wir als Abbruchbedingung.
thread.stop()
def runner(self, thread):
"""Lauflicht starten."""
# Diesen Thread in aktive Liste aufnehmen
with self.threadlock:
self.lst_threads.append(thread)
# In thread.ioname steht der Name des Inputs, der das Event auslöste
if thread.ioname == "I_1":
for i in range(len(self.out)):
# In thread.iovalue steht der Wert zum Auslösezeitpunkt
self.revpi.io[self.out[i]].value = thread.iovalue
# Über das Event thread.exit(sekunden) kann gewartet werden.
# Wird das Event zum Abbrechen auf .set() gesetzt, erhält man
# True und bricht die Schleife ab. Wenn die Zeit abgelaufen ist
# ohne ein .set() zum Abbrechen erhält man ein False
if thread.exit.wait(0.1):
break
# Bei anderen Schleifenkonstruktionen kann das exit-Event
# auch über thread.exit.is_set() abgefragt werden, welches
# True zurückgibt, wenn für den Thread die .stop() Funktion
# aufgerufen wurde.
#
# while not thread.exit.is_set():
# Bleibt in der Whileschleife, bis an anderer Stelle
# die .stop() Funktion des threads aufgerufen wird.
else:
# Alles wie oben, nur von rechts nach links
for i in range(len(self.out) - 1, -1, -1):
self.revpi.io[self.out[i]].value = thread.iovalue
if thread.exit.wait(0.1):
break
with self.threadlock:
# Diesen Thread nach Fertigstellung aus aktive Liste entfernen
self.lst_threads.remove(thread)
def start(self):
"""Startet den mainloop()."""
self.revpi.core.A1 = revpimodio2.GREEN
print("Gehe in den mainloop()")
self.revpi.mainloop()
print("Verlasse mainloop()")
if __name__ == "__main__":
root = RevPiModIOTest()
root.start()
#!/usr/bin/python3
# (c) Benjamin, Sven Sager
# -*- coding: utf-8 -*-
import revpimodio
import signal
from threading import Lock
# Diese Werte müssen an die eigene piCtory Konfiguration angepasst werden!
outputformat = "O_{}"
iodevname = "dio02"
eventlinksrechts = "I_1"
eventrechtslinks = "I_2"
eventabbruch = "I_3"
class RevPiModIOTest():
def __init__(self):
self.out = [outputformat.format(x) for x in range(1, 15)]
self.revpi = revpimodio.RevPiModIO(auto_refresh=True)
# Threadliste führen
self.lst_threads = []
self.threadlock = Lock()
# Events Registrieren, die als THREAD ausgeführt werden
self.revpi.devices[iodevname].reg_event(
eventlinksrechts, self.runner, as_thread=True
)
self.revpi.devices[iodevname].reg_event(
eventrechtslinks, self.runner, as_thread=True
)
# Event um alle aktiven Threads zu "killen" registrieren
self.revpi.devices[iodevname].reg_event(
eventabbruch, self.killer, edge=revpimodio.RISING
)
signal.signal(signal.SIGINT, self._sigexit)
signal.signal(signal.SIGTERM, self._sigexit)
def _sigexit(self, signum, frame):
print("Verlasse mainloop()")
self.revpi.devices.exit()
print("Beende alle laufenden Threads")
self.killer()
def killer(self, ioname=None, iovalue=None):
# Lockobjekt holen, welches verhindert, dass die abgebrochenen Threads
# sich selbst aus der Liste löschen und diese beim iterieren verändern
# Die abgebrochenen Threads benötigen auch dieses Lock-Objekt und
# warten auf dessen Freigabe
with self.threadlock:
# Durch alle aktiven Threads gehen und sie beenden
for thread in self.lst_threads:
# Die thread.stop() Funktion von RevPiCallback aufrufen,
# welche das exit() Event setzt. Dieses Event verwenden
# wir als Abbruchbedingung.
thread.stop()
def runner(self, thread):
# Diesen Thread in aktive Liste aufnehmen
with self.threadlock:
self.lst_threads.append(thread)
# In thread.ioname steht der Name des Inputs, der das Event auslöste
if thread.ioname == eventlinksrechts:
for i in range(len(self.out)):
# In thread.iovalue steht der Wert zum Auslösezeitpunkt
self.revpi.devices[iodevname][self.out[i]].value = thread.iovalue
# Über das Event thread.exit(sekunden) kann gewartet werden.
# Wird das Event zum Abbrechen auf .set() gesetzt, erhält man
# True und bricht die Schleife ab. Wenn die Zeit abgelaufen ist
# ohne ein .set() zum Abbrechen erhält man ein False
if thread.exit.wait(0.1):
break
# Bei anderen Schleifenkonstruktionen kann das exit-Event
# auch über thread.exit.is_set() abgefragt werden, welches
# True zurückgibt, wenn für den Thread die .stop() Funktion
# aufgerufen wurde.
#
# while not thread.exit.is_set():
# Bleibt in der Whileschleife, bis an anderer Stelle
# die .stop() Funktion des threads aufgerufen wird.
# Alles wie oben, nur von rechts nach links durch Abfrage auf anderen Input
if thread.ioname == eventrechtslinks:
for i in range(len(self.out) - 1, -1, -1):
self.revpi.devices[iodevname][self.out[i]].value = thread.iovalue
if thread.exit.wait(0.1):
break
with self.threadlock:
# Diesen Thread nach Fertigstellung aus aktive Liste entfernen
self.lst_threads.remove(thread)
def start(self):
self.revpi.devices.core.A1 = revpimodio.GREEN
print("Gehe in den mainloop()")
self.revpi.devices.mainloop()
self.revpi.devices.core.A1 = revpimodio.OFF
self.revpi.devices.writeprocimg()
if __name__ == "__main__":
root = RevPiModIOTest()
root.start()