main.py

main.py

import functools
import time

import trio
from kivy.app import App
from kivy.clock import Clock
from kivy.lang import Builder
from kivy.properties import NumericProperty


KV = '''
BoxLayout:
    orientation: 'vertical'

    Label:
        text: 'Weeeeee'
        font_size: sp(24)

        canvas.before:
            PushMatrix:
            Rotate:
                origin: self.center
                angle: app.time * 180

        canvas.after:
            PopMatrix:

    Button:
        text: 'Press me!'
        on_press: app.start_freeze_task(1)
'''


class AsyncTest(App):
    time = NumericProperty()

    async def async_run(self, async_lib=None):
        super_run = super().async_run

        async with trio.open_nursery() as nursery:
            self.nursery = nursery

            async def _run():
                await super_run(async_lib)
                nursery.cancel_scope.cancel()

            nursery.start_soon(_run)

    def build(self):
        Clock.schedule_interval(self.update_time, 0)
        self.freeze_cancel_scope = None
        self.freeze_id = 0
        return Builder.load_string(KV)

    def update_time(self, dt):
        if dt >= .1:
            print(f'We just froze for {dt:.3f}s')

        self.time += dt

    def start_freeze_task(self, duration):
        self.nursery.start_soon(
            functools.partial(
                self.freeze,
                duration,
                freeze_id=self.freeze_id,
            ),
        )
        self.freeze_id += 1

    async def freeze(self, duration, freeze_id):
        if self.freeze_cancel_scope:
            self.freeze_cancel_scope.cancel()

        self.freeze_cancel_scope = trio.CancelScope()

        with self.freeze_cancel_scope:
            print(
                f'Starting expensive task with id {freeze_id} '
                f'lasting {duration:.3f}s...'
            )
            await trio.sleep(duration)
            print(
                f'Expensive task with id {freeze_id} '
                f'lasting {duration:.3f}s done'
            )


trio.run(AsyncTest().async_run, 'trio')