Qiita記事 async/await構文へ適応

main.py

import types
from kivy.clock import Clock


def start_coro(coro):
    def step_coro(*args, **kwargs):
        try:
            coro.send((args, kwargs, ))(step_coro)
        except StopIteration:
            pass
    try:
        coro.send(None)(step_coro)
    except StopIteration:
        pass


async 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:
        await sleep(3)
    return return_value


@types.coroutine
def sleep(duration):
    args, kwargs = yield lambda step_coro: Clock.schedule_once(step_coro, duration)
    return args[0]


@types.coroutine
def event(ed, name):
    bind_id = None
    step_coro = None
    def bind(step_coro_):
        nonlocal bind_id, step_coro
        bind_id = ed.fbind(name, callback)
        assert bind_id > 0  # bindingに成功したか確認
        step_coro = step_coro_
    def callback(*args, **kwargs):
        ed.unbind_uid(name, bind_id)
        step_coro(*args, **kwargs)
    return (yield bind)


from kivy.app import App
from kivy.lang import Builder
from kivy.factory import Factory


KV_CODE = '''
Button:
'''

class SampleApp(App):
    def build(self):
        return Builder.load_string(KV_CODE)
    def on_start(self):
        start_coro(self.some_task())
    async 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'
        await event(button, 'on_press')
        button.text = 'running...'
        await thread(heavy_task)
        button.text = 'done'


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