6 Using Asynchronous programming inside a Kivy application
timothy eichler edited this page 2016-12-18 21:07:02 +01:00

Summary

  • author: timeyyy
  • kivy: >= Unknown

Using an async style of programming is much easier to reason about than using threads. If you are using good architecture it is possible to wrap all function calls from the gui to your main application in a decorator and no longer have to worry about any blocking issues. Fanning out work to multiple processes or threads is also possible.

The async landscape for kivy is a bit lacking. The best option has been explained here. Compare Working with threads in kivy. While the complexity of the thread looks manageable, Sooner or later you will be bitten with buggy non deterministic behaviour and or deadlocks.

Warning: When using the async model with kivy, certain errors will crash kivy without a traceback. This is rather annoying and means you have to step through the code in a debugger (the exception will be raised in this case). It is unknown if this is an issue with kivy or the async implementation.

Usage

from kivy.app import App
from kivy.uix.button import Button
from kivy.uix.label import Label
from kivy.uix.gridlayout import GridLayout
import time

from async_gui.engine import Task, MultiProcessTask
from async_gui.toolkits.kivy import KivyEngine

from cpu_work import is_prime, PRIMES


engine = KivyEngine()

class MyApp(App):
    def build(self):
        grid = GridLayout(cols=4)

        grid.add_widget(Button(text='Check Primes',
                               on_press=self.cpu_bound))
        self.status = Label(text='Status')
        grid.add_widget(self.status)
        return grid

    @engine.async
    def cpu_bound(self, *_):
        t = time.time()
        self.status.text = "calculating..."
        prime_flags = yield MultiProcessTask(
            [Task(is_prime, n) for n in PRIMES]
        )
        print(time.time() - t)
        text = '\n'.join("%s: %s" % (n, prime)
                         for n, prime in zip(PRIMES, prime_flags))
        self.status.text = text

if __name__ == '__main__':
    MyApp().run()