add 'event_freq'

examples/painter4.py

@@ -3,7 +3,7 @@
 =======
 
 * can handle mutiple touches simultaneously
-* uses 'touch_up_event' instead of 'rest_of_touch_events'
+* uses 'event_freq' and 'move_on_when' instead of 'rest_of_touch_events'
 '''
 
 from functools import cached_property, partial
@@ -43,12 +43,15 @@ async def draw_rect(self, touch):
             Color(*get_random_color())
             line = Line(width=2)
 
-        async with ak.move_on_when(ak.event(Window, 'on_touch_up', filter=lambda w, t: t is touch)):
-            touch_move_event = partial(
-                ak.event, self, 'on_touch_move', stop_dispatching=True,
-                filter=lambda w, t: t is touch)
+        def touch_filter(w, t, touch=touch):
+            return t is touch
+
+        async with (
+            ak.move_on_when(ak.event(Window, 'on_touch_up', filter=touch_filter)),
+            ak.event_freq(self, 'on_touch_move', filter=touch_filter, stop_dispatching=True) as on_touch_move,
+        ):
             while True:
-                __, touch = await touch_move_event()
+                await on_touch_move()
                 x, y = self_to_local(*touch.pos)
                 min_x, max_x = (x, ox) if x < ox else (ox, x)
                 min_y, max_y = (y, oy) if y < oy else (oy, y)

sphinx/notes-ja.rst

@@ -71,6 +71,7 @@ async操作が禁じられている場所
 * :func:`asynckivy.interpolate`
 * :func:`asynckivy.interpolate_seq`
 * ``asynckivy.anim_with_xxx``
+* :any:`asynckivy.event_freq`
 
 .. code-block::
 

sphinx/notes.rst

@@ -112,6 +112,7 @@ Here is a list of them:
 - :func:`asynckivy.interpolate`
 - :func:`asynckivy.interpolate_seq`
 - ``asynckivy.anim_with_xxx``
+- :any:`asynckivy.event_freq`
 
 .. code-block::
 

src/asynckivy/__init__.py

@@ -9,6 +9,7 @@
     'anim_with_ratio',
     'create_texture_from_text',
     'event',
+    'event_freq',
     'fade_transition',
     'interpolate',
     'interpolate_seq',
@@ -30,7 +31,7 @@
 from asyncgui import *
 from ._exceptions import MotionEventAlreadyEndedError
 from ._sleep import sleep, sleep_free, repeat_sleeping, move_on_after
-from ._event import event
+from ._event import event, event_freq
 from ._anim_with_xxx import anim_with_dt, anim_with_et, anim_with_ratio, anim_with_dt_et, anim_with_dt_et_ratio
 from ._anim_attrs import anim_attrs, anim_attrs_abbr
 from ._interpolate import interpolate, interpolate_seq, fade_transition

src/asynckivy/_event.py

@@ -1,4 +1,4 @@
-__all__ = ('event', )
+__all__ = ('event', 'event_freq', )
 
 import typing as T
 import types
@@ -25,7 +25,6 @@ def event(event_dispatcher, event_name, *, filter=None, stop_dispatching=False)
         # Wait for 'widget.x' to change.
         __, x = await ak.event(widget, 'x')
 
-
     The ``filter`` parameter:
 
     .. code-block::
@@ -43,15 +42,71 @@ def event(event_dispatcher, event_name, *, filter=None, stop_dispatching=False)
       See :ref:`kivys-event-system` for details.
     '''
     task = (yield _current_task)[0][0]
-    bind_id = event_dispatcher.fbind(event_name, partial(_callback, filter, task, stop_dispatching))
+    bind_id = event_dispatcher.fbind(event_name, partial(_callback, filter, task._step, stop_dispatching))
     assert bind_id  # check if binding succeeded
     try:
         return (yield _sleep_forever)[0]
     finally:
         event_dispatcher.unbind_uid(event_name, bind_id)
 
 
-def _callback(filter, task, stop_dispatching, *args, **kwargs):
+def _callback(filter, task_step, stop_dispatching, *args, **kwargs):
     if (filter is None) or filter(*args, **kwargs):
-        task._step(*args)
+        task_step(*args)
         return stop_dispatching
+
+
+class event_freq:
+    '''
+    When handling a frequently occurring event, such as ``on_touch_move``, the following code might cause performance
+    issues:
+
+    .. code-block::
+
+        __, touch = await event(widget, 'on_touch_down')
+        while True:
+            await event(widget, 'on_touch_move', filter=lambda w, t: t is touch)
+            ...
+
+    If that happens, try the following code instead. It might resolve the issue:
+
+    .. code-block::
+
+        __, touch = await event(widget, 'on_touch_down')
+        async with event_freq(widget, 'on_touch_move', filter=lambda w, t: t is touch) as on_touch_move:
+            while True:
+                await on_touch_move()
+                ...
+
+    The trade-off is that within the context manager, you can't perform any async operations except the
+    ``await on_touch_move()``.
+
+    .. code-block::
+
+        async with event_freq(...) as xxx:
+            await xxx()  # OK
+            await something_else()  # Don't
+
+    .. versionadded:: 0.7.1
+    '''
+    __slots__ = ('_disp', '_name', '_filter', '_stop', '_bind_id', )
+
+    def __init__(self, event_dispatcher, event_name, *, filter=None, stop_dispatching=False):
+        self._disp = event_dispatcher
+        self._name = event_name
+        self._filter = filter
+        self._stop = stop_dispatching
+
+    @types.coroutine
+    def __aenter__(self):
+        task = (yield _current_task)[0][0]
+        self._bind_id = self._disp.fbind(self._name, partial(_callback, self._filter, task._step, self._stop))
+        return self._wait_one
+
+    async def __aexit__(self, *args):
+        self._disp.unbind_uid(self._name, self._bind_id)
+
+    @staticmethod
+    @types.coroutine
+    def _wait_one():
+        return (yield _sleep_forever)[0]

tests/test_event_freq.py

@@ -0,0 +1,109 @@
+import pytest
+
+
+@pytest.fixture(scope='module')
+def ed_cls():
+    from kivy.event import EventDispatcher
+    class ConcreteEventDispatcher(EventDispatcher):
+        __events__ = ('on_test', 'on_test2', )
+        def on_test(self, *args, **kwargs):
+            pass
+        def on_test2(self, *args, **kwargs):
+            pass
+    return ConcreteEventDispatcher
+
+
+@pytest.fixture()
+def ed(ed_cls):
+    return ed_cls()
+
+
+def test_properly_cleanuped(ed):
+    import asynckivy as ak
+    async def async_fn():
+        async with ak.event_freq(ed, 'on_test') as on_test:
+            await on_test()
+            await on_test()
+        await ak.sleep_forever()
+
+    task = ak.start(async_fn())
+    ed.dispatch('on_test')
+    assert not task.finished
+    ed.dispatch('on_test')
+    assert not task.finished
+    ed.dispatch('on_test')
+    assert not task.finished
+    task._step()
+    assert task.finished
+
+
+def test_event_parameters(ed):
+    import asynckivy as ak
+
+    async def async_fn():
+        async with ak.event_freq(ed, 'on_test') as on_test:
+            assert (ed, 1, 2, ) == await on_test()
+            assert (ed, 3, 4, ) == await on_test()  # kwarg is ignored
+
+    task = ak.start(async_fn())
+    assert not task.finished
+    ed.dispatch('on_test', 1, 2)
+    assert not task.finished
+    ed.dispatch('on_test', 3, 4, kwarg='A')
+    assert task.finished
+
+
+def test_filter(ed):
+    import asynckivy as ak
+
+    async def async_fn():
+        async with ak.event_freq(ed, 'on_test', filter=lambda *args: args == (ed, 3, 4, )) as on_test:
+            await on_test()
+
+    task = ak.start(async_fn())
+    assert not task.finished
+    ed.dispatch('on_test', 1, 2)
+    assert not task.finished
+    ed.dispatch('on_test', 3, 4)
+    assert task.finished
+
+
+def test_stop_dispatching(ed):
+    import asynckivy as ak
+
+    called = []
+
+    async def async_fn():
+        ed.bind(on_test=lambda *args: called.append(1))
+        async with ak.event_freq(ed, 'on_test', stop_dispatching=True) as on_test:
+            await on_test()
+
+    task = ak.start(async_fn())
+    assert not called
+    ed.dispatch('on_test')
+    assert not called
+    assert task.finished
+    ed.dispatch('on_test')
+    assert called
+
+
+def test_cancel(ed):
+    import asynckivy as ak
+
+    async def async_fn(ed):
+        def filter_func(*args):
+            nonlocal called; called = True
+            return True
+        async with ak.event_freq(ed, 'on_test', filter=filter_func) as on_test:
+            await on_test()
+
+    called = False
+    task = ak.start(async_fn(ed))
+    assert not task.finished
+    assert not called
+    task.close()
+    assert not task.finished
+    assert not called
+    ed.dispatch('on_test')
+    assert not task.finished
+    assert not called