[PYTHON] Async / warte mit Kivy und tkinter
2018 war ein Schock für mich. Früher habe ich mit GUI Framekworks Kivy kivy gespielt, aber ich habe festgestellt, dass die Verwendung eines Generators hässlichen Code voller Rückruffunktionen überraschend einfach zu lesen macht. Nachdem ich verschiedene Dinge ausprobiert hatte, lernte ich die asynchrone Verarbeitung durch async / await kennen, eine Magie, die ich bis dahin nicht kannte, und ich konnte eine kleine asynchrone Verarbeitungsbibliothek erstellen. Dieser Artikel ist neu
――Der Prozess, bis ich die Wunderbarkeit des Generators und der daraus geborenen nativen Coroutine erkannte
- Der Prozess der Realisierung von Async / Warten mit Kivy und tkinter, die es verwenden
Ich möchte es formulieren.
(Um die Textmenge zu reduzieren, wird der Generator als gen und die Coroutine als coro abgekürzt.)
Die verborgene Kraft von Gen.
Gen als Gerät, das Wert schafft
Ich denke, viele Einführungsbücher führen gen als wertschöpfendes Gerät ein. (Um Verwechslungen mit "Rendite" zu vermeiden, was Rendite bedeutet, wird die Rendite als "Geben" ausgedrückt.)
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, b+a
for i in fibonacci():
print(i, end=' ')
import time; time.sleep(.1)
0 1 1 2 3 5 8 13 21 34 55 89 144 233 377 610 987 1597 2584 4181 ...
Unter dem Namen "Generator" könnte es ursprünglich dieser Zweck gewesen sein. Tatsächlich habe ich bis 2018 nur so etwas gesehen, und ich denke, es war schwierig für mich, aus diesem Konzept herauszukommen, weil ich den Wert wie oben mit "for-in" abgerufen habe.
Abrufen von Werten ohne For-In
Einige Einführungsbücher enthalten auch ein Beispiel für die Verwendung von "send ()", um einen Wert von gen abzurufen.
gen = fibonacci()
print(gen.send(None))
print(gen.send(None))
print('Eine kurze Pause')
import time;time.sleep(1)
print('Ende der Pause')
print(gen.send(None))
print(gen.send(None))
0
1
Eine kurze Pause
Ende der Pause
1
2
Ich denke, dies gibt einen Einblick in die "verborgene Kraft" von Gen, aber ich habe es zu diesem Zeitpunkt auch nicht bemerkt. Aber was ist mit dem nächsten Beispiel?
Nichts wird gen geboren
def sub_task():
print('sub:Prozess 1')
yield
print('sub:Prozess 2')
yield
print('sub:Prozess 3')
def main_task():
gen = sub_task()
try:
print('main:Prozess 1')
gen.send(None)
print('main:Prozess 2')
gen.send(None)
print('main:Prozess 3')
gen.send(None)
except StopIteration:
pass
main_task()
main:Prozess 1
sub:Prozess 1
main:Prozess 2
sub:Prozess 2
main:Prozess 3
sub:Prozess 3
Sie können sehen, dass beide Aufgaben nach und nach Fortschritte machen, wobei "gen.send ()" und "Yield" als Schaltpunkte dienen. Ist das nicht ** Parallelverarbeitung ** selbst!
Versteckte Kraft
Dies war Gens "verborgene Kraft". gen ist
- Es kann angehalten werden ... (Stoppt bei der Ausbeute und beginnt sich mit
gen.send ()zu bewegen.) --Kommunikation mit der Benutzerseite ist am Stopppunkt möglich ... (Der Wert kann durch Yield von der Gen-Seite undgen.send ()von der Benutzerseite gesendet werden.)
Es war wie eine Funktion, und aufgrund seiner stoptierbaren Funktion konnte eine ** Parallelverarbeitung durchgeführt werden, ohne sich auf "Multithreading" ** zu verlassen.
Erwartungen an gen
Dies ließ mich hoffen, dass ich unter einem hässlichen Code voller Rückruffunktionen litt. Zum Beispiel, wenn Sie mit Kivy Folgendes tun möchten
def some_task():
print('Prozess 1')
Warten Sie, bis die Taste gedrückt wird
print('Prozess 2')
Warten Sie 1 Sekunde
print('Prozess 3')
Der eigentliche Code ist
from kivy.clock import Clock
def some_task(button):
print('Prozess 1')
def callback2(button):
button.unbind(on_press=callback2)
print('Prozess 2')
Clock.schedule_once(callback3, 1)
def callback3(__):
print('Prozess 3')
button.bind(on_press=callback2)
Es wird Hässlichkeit, die nicht gesehen werden kann. Sie müssen auf etwas warten =>Sie müssen die Verarbeitung stoppen, bis etwas passiert=> Die folgende Verarbeitung muss in eine andere Funktion aufgeteilt werden. Aber ich möchte, dass Sie sich an die sub_task () erinnern, die früher herauskam.
def sub_task():
print('sub:Prozess 1')
yield
print('sub:Prozess 2')
yield
print('sub:Prozess 3')
Ebenso erscheint die Rückruffunktion nirgendwo, obwohl sie in der Mitte angehalten hat. Also begann ich mich zu fragen, ob gen verwendet werden könnte, um die Rückruffunktion in Kivy zu eliminieren.
Beseitigen Sie die Rückruffunktion (Kivy Edition)
Ich werde von nun an über die Methode nachdenken, aber ich möchte erwähnen, dass der aktuelle Hauptzweig von Kivy bereits eine vollständige asynchrone Programmierung mit [asyncio] asyncio_doc und [trio] trio_doc durchführen kann. Was wir hier also tun, ist [Neuerfindung der Räder] [Rad]. Zu dieser Zeit war dies jedoch nicht der Fall, und ich interessierte mich einfach für Gen, also entschied ich mich, etwas selbst zu tun.
Halten Sie für einen bestimmten Zeitraum an
Ich habe beschlossen, die Schaltfläche einmal beiseite zu legen und eine Funktion zu realisieren, mit der gen für diese Zeit angehalten werden kann, wenn ein numerischer Wert von gen gesendet wird. Das liegt daran, dass ich zufällig [BeeWares Video] video_beeware gesehen habe und dachte, es wäre cool, so etwas zu tun.
def some_task():
print('Prozess 1')
yield 2 #Warten Sie 2 Sekunden
print('Prozess 2')
yield 1 #Warten Sie 1 Sekunde
print('Prozess 3')
Überlegen Sie, wie das oben genannte Gen wie erwartet funktioniert. Mit dem bisherigen Wissen
- Sie müssen eine Funktion an
Clock.schedule_once ()übergeben, um den Prozess zu reservieren, den Sie nach einer bestimmten Zeit auf Kivy ausführen möchten. - Sie müssen "gen.send ()" aufrufen, um gen neu zu starten
Ich weiß das. Dann: "Warum übergeben wir die Funktion zum Neustarten von gen nicht an Clock.schedule_once ()?"
from kivy.clock import Clock
from kivy.app import App
from kivy.uix.widget import Widget
def start_gen(gen):
def step_gen(dt):
try:
Clock.schedule_once(step_gen, gen.send(None)) # C
except StopIteration:
pass
step_gen(None) # B
def some_task():
print('Prozess 1')
yield 1 # D
print('Prozess 2')
yield 2 # E
print('Prozess 3')
class SampleApp(App):
def build(self):
return Widget()
def on_start(self):
start_gen(some_task()) # A
if __name__ == '__main__':
SampleApp().run()
Es war die richtige Antwort. Dieser Code funktioniert wie folgt.
- Ein Gen wird beim Start der App erstellt und an
start_gen ()(Zeile A) übergeben. start_gen ()ruft sofortstep_gen ()auf (Zeile B)step_gen ()ruftgen.send ()auf, damit gen anfängt zu arbeiten (Zeile C)- gen stoppt beim ersten Ertragsausdruck und sendet 1 (Zeile D)
- Da das Bewertungsergebnis von "gen.send (None)" 1 ist, behält sich "step_gen ()" vor, nach 1 Sekunde erneut aufgerufen zu werden (Zeile C).
- Da nichts mehr zu tun ist, kehrt der Prozess zur Ereignisschleife von kivy zurück.
- Eine Sekunde später wird "step_gen ()" aufgerufen und "gen.send ()" aufgerufen, sodass sich gen von der Position aus bewegt, an der es das letzte Mal angehalten hat. (Linie C)
- gen stoppt an dem Punkt, an dem es den zweiten Ertragsausdruck erreicht und sendet 2 (Zeile E)
- (Unten weggelassen)
Es war schockierend, eine Weile warten zu können, ohne die Rückruffunktion zu verwenden, indem eine Funktion mit nur 7 Zeilen vorbereitet wurde (start_gen ()). Motiviert werde ich das weiter verbessern.
Verwenden Sie den an die Rückruffunktion übergebenen Wert
Die tatsächlich verstrichene Zeit wird an die Rückruffunktion übergeben, die an "Clock.schedule_once ()" übergeben wird. Da es eine große Sache ist, habe ich es so gemacht, dass die Seite 'some_task ()' es empfangen kann. Alles was Sie tun müssen, ist den Teil "gen.send (None)" von "start_gen ()" in "gen.send (dt)" zu ändern. Jetzt kann die Seite "some_task ()" die tatsächlich verstrichene Zeit wie folgt abrufen (ganzer Code).
def some_task():
print('Prozess 1')
s = yield 1
print(f"Als ich für 1 Sekunde nach einem Stopp fragte, war es tatsächlich so{s:.03f}Hielt für eine Sekunde an")
print('Prozess 2')
s = yield 2
print(f"Als ich für 2 Sekunden nach einem Stopp fragte, war es tatsächlich so{s:.03f}Hielt für eine Sekunde an")
print('Prozess 3')
Prozess 1
Als ich für 1 Sekunde nach einem Stopp fragte, war es tatsächlich 1.089 Sekunden gestoppt
Prozess 2
Als ich für 2 Sekunden nach einem Stopp fragte, war es tatsächlich 2.003 Sekunden angehalten
Prozess 3
Warten auf Veranstaltung
Als nächstes wartet auf das Ereignis, idealerweise wenn die Gen-Seite wie folgt schreibt.
def some_task(button):
print('Prozess 1')
yield event(button, 'on_press') #Warten Sie, bis die Taste gedrückt wird
print('Prozess 2')
Im Falle eines Ereignisses ist dies etwas kompliziert, da für die Verbindung der Rückruffunktion eine
def start_gen(gen):
def step_gen(*args, **kwargs):
try:
gen.send((args, kwargs, ))(step_gen)
except StopIteration:
pass
try:
gen.send(None)(step_gen)
except StopIteration:
pass
def event(ed, name):
bind_id = None
step_gen = None
def bind(step_gen_):
nonlocal bind_id, step_gen
bind_id = ed.fbind(name, callback) #Verknüpfen Sie die Rückruffunktion
assert bind_id > 0 #Überprüfen Sie, ob die Bindung erfolgreich war
step_gen = step_gen_
def callback(*args, **kwargs):
ed.unbind_uid(name, bind_id) #Lösen Sie die Rückruffunktion
step_gen(*args, **kwargs) #Lebenslauf gen
return bind
Der große Unterschied zum Zeitstopp besteht darin, dass die gesamte Verarbeitung in Bezug auf das Ereignis in event () ausgeblendet werden kann. Dank dessen hängt start_gen () überhaupt nicht von kivy ab und es ist so einfach wie das Übergeben von step_gen an den von gen gesendeten Callable.
Verallgemeinerung
Ich bin der Meinung, dass das obige Design sehr gut ist, also habe ich die mit kivy verbundene Verarbeitung aus start_gen () entfernt und sie in einer anderen Funktion versteckt, nachdem das Ereignis auf den Zeitstopp gewartet hat.
def sleep(duration):
return lambda step_gen: Clock.schedule_once(step_gen, duration)
Jetzt können Sie sleep () undevent ()mischen.
def some_task(button):
yield event(button, 'on_press') #Warten Sie, bis die Taste gedrückt wird
button.text = 'Pressed'
yield sleep(1) #Warten Sie 1 Sekunde
button.text = 'Bye'
Wie gen wieder aufgenommen wird, hängt vollständig von dem von gen gesendeten Callable ab. Wenn Sie also beispielsweise Folgendes senden
def sleep_forever():
return lambda step_gen: None
def some_task():
yield sleep_forever() #Ewig warten
Es ist auch möglich, nicht neu zu starten.
Warten Sie auf den Thread
Um seine Vielseitigkeit zu bestätigen, habe ich mich auch mit Dingen befasst, die nichts mit Kivy zu tun haben, Thread.
def thread(func, *args, **kwargs):
from threading import Thread
return_value = None
is_finished = False
def wrapper(*args, **kwargs):
nonlocal return_value, is_finished
return_value = func(*args, **kwargs)
is_finished = True
Thread(target=wrapper, args=args, kwargs=kwargs).start()
while not is_finished:
yield sleep(3)
return return_value
Es ist zu einer langweiligen Art geworden, sich umzusehen, ob es regelmäßig endet, aber jetzt kann die Gen-Seite auf das Ende warten, indem sie die übergebene Funktion in einem anderen Thread ausführt.
class SampleApp(App):
def on_start(self):
start_gen(self.some_task())
def some_task(self):
def heavy_task():
import time
for i in range(5):
time.sleep(1)
print(i)
button = self.root
button.text = 'start heavy task'
yield event(button, 'on_press') #Warten Sie, bis die Taste gedrückt wird
button.text = 'running...'
yield from thread(heavy_task) #Schwer auf einem anderen Faden_task()Und warte auf sein Ende
button.text = 'done'
Was ist Ertrag oder Ertrag von?
Hier scheint es gut zu laufen, aber einige Probleme sind offensichtlich geworden. Eine davon ist, dass Sie Ertrag und Ertrag richtig einsetzen müssen, je nachdem, worauf Sie warten. (sleep () undevent ()sind Ausbeute,thread ()ist Ausbeute von). Darüber hinaus hängt dies von der Implementierung ab, und wenn threading.Thread einen Mechanismus hatte, um das Ende des Threads mit der Rückruffunktion zu benachrichtigen, könntethread ()auch implementiert werden, damit es mit Ausbeute warten kann. Es ist nicht gut, dass die Verwendung so unterschiedlich ist, deshalb habe ich beschlossen, sie mit beiden zu vereinheitlichen.
Ich denke, die einzige Option ist "Rendite aus". Weil es einfach ist, aus dem Warten auf das, was Sie auf den Ertrag warten können, einen Ertrag zu erzielen, aber nicht immer umgekehrt. Zum Beispiel
def some_gen():
yield 1
1 ist
def some_gen():
yield from one()
def one():
yield 1
Auf diese Weise können Sie bei "Ertrag von" warten
def some_gen():
yield from another_gen()
def another_gen():
yield 1
yield 4
Another_gen () kann wahrscheinlich nicht dazu gebracht werden zu warten`` another_gen () `.
Vereinigt, um von zu ergeben
Also habe ich "sleep ()" und "event ()" umgeschrieben, um auf den Ertrag von zu warten.
def sleep(duration):
return (yield lambda step_coro: Clock.schedule_once(step_gen, duration))
def event(ed, name):
#Abkürzung
return (yield bind)
Damit muss die Benutzerseite Yield und Yield von nicht richtig verwenden.
#Immer nachgeben von
def some_task():
yield from sleep(2)
yield from event(button, 'on_press')
yield from thread(heavy_task)
Probleme bei der Verwendung der an die Rückruffunktion übergebenen Argumente
Ein weiteres Problem besteht darin, dass die Gen-Seite ursprünglich die an die Rückruffunktion übergebenen Argumente wie folgt verwendet hat.
def some_task():
s = yield 1
print(f"Als ich für 1 Sekunde nach einem Stopp fragte, war es tatsächlich so{s:.03f}Hielt für eine Sekunde an")
Was jetzt los ist, ist tatsächlich
def some_task():
args, kwargs = yield from sleep(1)
s = args[0]
print(f"Als ich für 1 Sekunde nach einem Stopp fragte, war es tatsächlich so{s:.03f}Hielt für eine Sekunde an")
Es ist schwierig, den erforderlichen Wert zu erhalten. Dies liegt daran, dass das formale Argument "def step_gen (* args, ** kwargs)" lautet, sodass "step_gen ()" jedes Argument empfangen kann. Glücklicherweise kann eine solche Verarbeitung dank der Vereinigung, aus der man hervorgehen kann, auf der Seite "sleep ()" erfolgen.
def sleep(duration):
args, kwargs = yield lambda step_coro: Clock.schedule_once(step_coro, duration)
return args[0]
Damit die Benutzerseite
def some_task():
s = yield from sleep(1)
print(f"Als ich für 1 Sekunde nach einem Stopp fragte, war es tatsächlich so{s:.03f}Hielt für eine Sekunde an")
Es wurde genug.
Einführung der Async / Warten-Syntax
Als nächstes habe ich beschlossen, das, was ich bisher gemacht habe, so zu konvertieren, dass es mit der in Python 3.5 eingeführten asynchronen / wartenden Syntax verarbeitet werden kann. Dies liegt daran, dass in verschiedenen Dokumenten geschrieben wurde, dass dies ein Ersatz für Gen als Coro ist. Für mich ist das eine Wort "Warten" kürzer und leichter zu lesen als das Zwei-Wort "Warten" aus ", aber es scheint tatsächlich mehr Vorteile zu haben. Bitte lesen Sie die Details. Siehe [Offiziell] pep492.
Konvertierungsverfahren
Erstens wurden Gen-Funktionen, die Ausbeuten ohne from enthalten, wie "sleep ()" und "event ()", "@ types.coroutine" zugewiesen. Dies liegt daran, dass ich nicht wusste, wie ich die gen-Funktion, die die return-Anweisung wie diese enthält, als asynchrone Funktion umschreiben kann. (Wenn die asynchrone Funktion einen Ertragsausdruck hat und einen Wert in der return-Anweisung zurückgibt, tritt ein Syntaxfehler auf: SyntaxError: 'return' mit Wert im asynchronen Generator)
import types
@types.coroutine
def sleep(duration):
#Abkürzung
@types.coroutine
def event(ed, name):
#Abkürzung
Andererseits könnten "thread ()" und "some_task ()" als reine asynchrone Funktionen umgeschrieben werden. Speziell
- Von "Ertrag von" bis "Warten"
- Von
defzuasync def
Ersetzt.
async def thread(func, *args, **kwargs):
#Abkürzung
while not is_finished:
await sleep(3)
#Abkürzung
class SampleApp(App):
async def some_task(self):
#Abkürzung
await event(button, 'on_press')
button.text = 'running...'
await thread(heavy_task)
button.text = 'done'
Ersetzen Sie abschließend die im Bezeichner enthaltene Zeichenfolge "gen" durch "coro", um sie zu vervollständigen.
Dies ist das Ende der Kivy-Ausgabe. Wie ich vergessen habe zu sagen, kann coro gleichzeitig mit start_coro (another_task ()) ausgeführt werden. Await another_task () wenn Sie auf den Abschluss warten möchten, start_coro (another_task ()) wenn Sie parallel laufen möchten, ohne zu warten.
Beseitigen Sie die Rückruffunktion (tkinter edition)
Dann habe ich dasselbe mit tkinter versucht, aber die Prozedur war genau die gleiche wie bei kivy (Übergabe der Funktion zum Neustart von gen / coro als Rückruffunktion), sodass alles reibungslos verlief.
Warte auf die Zeit (Schlaf)
Der Unterschied zu Kivy besteht darin, dass Kivy das einzige Objekt "kivy.clock.Clock" verwendet, während tkinter die Methode ".after ()" jedes Widgets verwendet. Sie müssen also angeben, welches Widget ".after ()" zusätzlich zu der Zeit, die Sie stoppen möchten, aufrufen soll.
@types.coroutine
def sleep(widget, duration):
yield lambda step_coro: widget.after(duration, step_coro)
Dies bedeutet, dass Sie das Widget an "thread ()" übergeben müssen, das intern "sleep ()" verwendet.
async def thread(func, *, watcher):
#Abkürzung
while not is_finished:
await sleep(watcher, 3000) #3000ms stoppen
return return_value
Warten auf Veranstaltung
Als nächstes kommt das Ereignis. Vor der Implementierung scheint es, dass "unbind ()" von tkinter einen [bug] tkinter_issue hat, daher habe ich ihn wie folgt geändert und mich dabei auf die verknüpften Informationen gestützt.
def _new_unbind(self, sequence, funcid=None):
if not funcid:
self.tk.call('bind', self._w, sequence, '')
return
func_callbacks = self.tk.call('bind', self._w, sequence, None).split('\n')
new_callbacks = [l for l in func_callbacks if l[6:6 + len(funcid)] != funcid]
self.tk.call('bind', self._w, sequence, '\n'.join(new_callbacks))
self.deletecommand(funcid)
def patch_unbind():
from tkinter import Misc
Misc.unbind = _new_unbind
Es wird durch das modifizierte "unbind ()" ersetzt, indem "patch_unbind ()" aufgerufen wird. Das Problem ist, wann man es aufruft, aber ich denke, es ist besser, es nicht ohne Erlaubnis zu tun, da tkinter selbst geändert wird. Deshalb habe ich beschlossen, dass der Benutzer mich explizit anruft. Und die Implementierung von event () ist
@types.coroutine
def event(widget, name):
bind_id = None
step_coro = None
def bind(step_coro_):
nonlocal bind_id, step_coro
bind_id = widget.bind(name, callback, '+')
step_coro = step_coro_
def callback(*args, **kwargs):
widget.unbind(name, bind_id)
step_coro(*args, **kwargs)
return (yield bind)[0][0]
Es wurde.
Wie benutzt man
#Installieren Sie asynctkinter im Voraus
# pip install git+https://github.com/gottadiveintopython/asynctkinter#egg=asynctkinter
from tkinter import Tk, Label
import asynctkinter as at
at.patch_unbind() # unbind()Fehler behoben
def heavy_task():
import time
for i in range(5):
time.sleep(1)
print('heavy task:', i)
root = Tk()
label = Label(root, text='Hello', font=('', 60))
label.pack()
async def some_task(label):
label['text'] = 'start heavy task'
event = await at.event(label, '<Button>') #Warten Sie, bis das Etikett gedrückt wurde
print(event.x, event.y)
label['text'] = 'running...'
await at.thread(heavy_task, watcher=label) #Schwer auf einem anderen Faden_task()Und warte auf sein Ende
label['text'] = 'done'
await at.sleep(label, 2000) #Warten Sie 2 Sekunden
label['text'] = 'close the window'
at.start(some_task(label))
root.mainloop()
abschließend
Die Idee, gen für die parallele Verarbeitung zu verwenden, scheint vor etwa 10 Jahren [bereits existiert] video_curious_course gewesen zu sein, und viele Leute haben es vielleicht gewusst, aber für mich war es vor ein oder zwei Jahren ein neues Wissen und es war ein Schock. Also habe ich mir diesen Artikel ausgedacht. Wenn die Bibliotheksseite die Ereignisschleife wie tkinter implementiert, verwenden Sie diese Methode. Wenn die Benutzerseite die Implementierung der Ereignisschleife wie pygame anvertraut, setzen Sie die Ereignisschleife wahrscheinlich auf "asyncio" oder Wenn Sie es als eine Aufgabe auf trio implementieren, können Sie grundsätzlich async / await für alles einführen. Auf Wiedersehen hässliche Rückruffunktion.
Link-Sammlung
- asynckivy asynckivy ... Die neueste Version von der, die ich dieses Mal gemacht habe
- asynctkinter asynctkinter ... Die neueste Version von der, die ich dieses Mal gemacht habe
- [Du willst also ein Python-Experte sein?] Video_powell ... Das Video, das mich dazu inspiriert hat, die Größe von Gen zu erkennen
- Mach dich zu einem Async für großes Wohl! ...
asyncioEin Video, in dem du dein eigenes Mock machst - Fluent Python ... Die detaillierteste japanische Literatur, die ich über Gen / Coro kenne. Die niedrige Version von Python, die ich benutze, kratzt den Ball.
Recommended Posts