Kivy ThreadWorker Example

kivy_thread_worker.py

"""
Kivy ThreadWorker example app.
"""

import kivy
kivy.require('1.10.0')

import threading
import time

from kivy.app import App
from kivy.clock import mainthread
from kivy.event import EventDispatcher
from kivy.lang import Builder
from kivy.uix.floatlayout import FloatLayout


KV = '''\
<RootLayout>:
    Button:
        id: btn
        text: 'Click me!'
        on_press: app.heavy_lifting()
'''


class RootLayout(FloatLayout):
    pass


class ThreadWorker(EventDispatcher):
    __events__ = ('on_done',)

    def __init__(self, target, *args, **kwargs):
        super(ThreadWorker, self).__init__()
        self._target = target
        self._thread = threading.Thread(target=self._run, args=args, kwargs=kwargs)
        self._thread.daemon = True

    def start(self):
        self._thread.start()

    def _run(self, *args, **kwargs):
        self._target(*args, **kwargs)
        self._notify()

    @mainthread
    def _notify(self):
        self.dispatch('on_done')

    def on_done(self, *_):
        pass


class ThreadExampleApp(App):

    def build(self):
        return RootLayout()

    def heavy_lifting(self):
        self.root.ids.btn.text = "Please wait while we do some processing..."

        # put your long-running code here, or use what you have as `target`
        def a_lot_of_hard_work_being_done_here():
            time.sleep(3)

        worker = ThreadWorker(target=a_lot_of_hard_work_being_done_here)

        def done(*_):
            print('done.')
            self.root.ids.btn.text = "All done. Thanks for your patience."

        # add some post-processing which should happen on the main thread
        worker.bind(on_done=done)
        worker.start()


if __name__ == '__main__':
    Builder.load_string(KV)
    ThreadExampleApp().run()