-
-
Save akloster/a5446416cf9a99f208797e5fa289e53c to your computer and use it in GitHub Desktop.
Asyncio test
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import os | |
os.environ['KIVY_EVENTLOOP'] = 'asyncio' | |
from kivy.lang.builder import Builder | |
from kivy.app import App | |
from asyncio import sleep, ensure_future, CancelledError | |
kv = ''' | |
BoxLayout: | |
orientation: 'vertical' | |
Button: | |
id: btn | |
text: 'Press me' | |
BoxLayout: | |
Label: | |
id: label | |
text: 'Button is "{}"'.format(btn.state) | |
Label: | |
id: test | |
text: "" | |
''' | |
class MyApp(App): | |
def build(self): | |
return Builder.load_string(kv) | |
def on_start(self): | |
self.main_task = ensure_future(self.main()) | |
def on_stop(self): | |
self.main_task.cancel() | |
async def main(self): | |
i = 0 | |
try: | |
while 1: | |
await sleep(1) | |
i+=1 | |
self.root.ids.test.text = str(i) | |
except CancelledError: | |
print("stopping app") | |
await sleep(0.2) | |
print("stopped") | |
if __name__ == '__main__': | |
loop = asyncio.get_event_loop() | |
app = MyApp() | |
app_task = ensure_future(app.async_run()) | |
loop.run_until_complete(app_task) | |
print("async_run has finished") | |
loop.run_until_complete(app.main_task) | |
loop.close() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thank you for this example. Any idea how one could trigger the background task from a budget widget or other bind event? I can only find examples similar to this one that start the async task at the same time that the whole async app starts.