asyncio_events.py

''' Shows a couple of ideas around coroutines for event handling.
'''
import asyncio
from asyncio import sleep, ensure_future, Future
import os

os.environ['KIVY_EVENTLOOP'] = 'async'
'''async needs to be set so that asyncio will be used for the event loop. '''

from kivy.app import App
from kivy.lang.builder import Builder

kv = '''
BoxLayout:
    orientation: 'vertical'
    BoxLayout:
        Button:
            id: btn1
            group: 'a'
            text: 'Test'
    Label:
        id: label
        status: 'Reading'
        text: 'Beach status is "{}"'.format(self.status)
'''


class EventListener(object):
    def __init__(self, dispatcher, event_name, form=0):
        super().__init__()
        self.dispatcher = dispatcher
        self.event_name = event_name
        self.form = form
        dispatcher.fbind(event_name, self.trigger)
        self._fut = None


    def trigger(self,*args, **kwargs):
        if self._fut is not None: 
            if self.form==2:
                self._fut.set_result((args, kwargs))
            elif self.form==1:
                self._fut.set_result(args)
            else:
                self._fut.set_result(args[0])
            self._fut = None


    def __await__(self):
        if self._fut is None:
            self._fut = Future()
        result = yield from self._fut
        return result

    async def __aiter__(self):
        while 1:
            event = await self
            yield event
def listen_to(dispatcher, event_name):
    return EventListener(dispatcher, event_name)

class AsyncApp(App):
    def build(self):
        return Builder.load_string(kv)
    def on_start(self, *args, **kwargs):
        loop = asyncio.get_event_loop()
        self.main_task = ensure_future(self.main(), loop=loop)
    def on_stop(self, *args, **kwargs):
        self.main_task.cancel()

    async def main(self):
        await sleep(1)
        btn1 = self.root.ids.btn1
        button_press = listen_to(btn1, "on_press")
        await button_press
        print("First button press")
        try:
            async for event in button_press:
                print("A button was pressed:", event)
        except asyncio.CancelledError as e:
            print("Ending")


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(AsyncApp().async_run())
    loop.close()