Introduce app.timer

nicegui/app/app.py

@@ -30,6 +30,7 @@ class State(Enum):
 
 
 class App(FastAPI):
+    from ..timer import Timer as timer  # pylint: disable=import-outside-toplevel
 
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs, docs_url=None, redoc_url=None, openapi_url=None)

nicegui/elements/timer.py

@@ -1,103 +1,18 @@
-import asyncio
-import time
 from contextlib import nullcontext
-from typing import Any, Awaitable, Callable, Optional
+from typing import ContextManager
 
-from .. import background_tasks, core
-from ..awaitable_response import AwaitableResponse
-from ..binding import BindableProperty
 from ..client import Client
 from ..element import Element
 from ..logging import log
+from ..timer import Timer as BaseTimer
 
 
-class Timer(Element, component='timer.js'):
-    active = BindableProperty()
-    interval = BindableProperty()
+class Timer(BaseTimer, Element, component='timer.js'):
 
-    def __init__(self,
-                 interval: float,
-                 callback: Callable[..., Any], *,
-                 active: bool = True,
-                 once: bool = False,
-                 ) -> None:
-        """Timer
+    def _get_context(self) -> ContextManager:
+        return self.parent_slot or nullcontext()
 
-        One major drive behind the creation of NiceGUI was the necessity to have a simple approach to update the interface in regular intervals,
-        for example to show a graph with incoming measurements.
-        A timer will execute a callback repeatedly with a given interval.
-
-        :param interval: the interval in which the timer is called (can be changed during runtime)
-        :param callback: function or coroutine to execute when interval elapses
-        :param active: whether the callback should be executed or not (can be changed during runtime)
-        :param once: whether the callback is only executed once after a delay specified by `interval` (default: `False`)
-        """
-        super().__init__()
-        self.interval = interval
-        self.callback: Optional[Callable[..., Any]] = callback
-        self.active = active
-        self._is_canceled: bool = False
-
-        coroutine = self._run_once if once else self._run_in_loop
-        if core.app.is_started:
-            background_tasks.create(coroutine(), name=str(callback))
-        else:
-            core.app.on_startup(coroutine)
-
-    def activate(self) -> None:
-        """Activate the timer."""
-        assert not self._is_canceled, 'Cannot activate a canceled timer'
-        self.active = True
-
-    def deactivate(self) -> None:
-        """Deactivate the timer."""
-        self.active = False
-
-    def cancel(self) -> None:
-        """Cancel the timer."""
-        self._is_canceled = True
-
-    async def _run_once(self) -> None:
-        try:
-            if not await self._connected():
-                return
-            with self.parent_slot or nullcontext():
-                await asyncio.sleep(self.interval)
-                if self.active and not self._should_stop():
-                    await self._invoke_callback()
-        finally:
-            self._cleanup()
-
-    async def _run_in_loop(self) -> None:
-        try:
-            if not await self._connected():
-                return
-            with self.parent_slot or nullcontext():
-                while not self._should_stop():
-                    try:
-                        start = time.time()
-                        if self.active:
-                            await self._invoke_callback()
-                        dt = time.time() - start
-                        await asyncio.sleep(self.interval - dt)
-                    except asyncio.CancelledError:
-                        break
-                    except Exception as e:
-                        core.app.handle_exception(e)
-                        await asyncio.sleep(self.interval)
-        finally:
-            self._cleanup()
-
-    async def _invoke_callback(self) -> None:
-        try:
-            assert self.callback is not None
-            result = self.callback()
-            if isinstance(result, Awaitable) and not isinstance(result, AwaitableResponse):
-                await result
-        except Exception as e:
-            core.app.handle_exception(e)
-
-    async def _connected(self, timeout: float = 60.0) -> bool:
+    async def _can_start(self) -> bool:
         """Wait for the client connection before the timer callback can be allowed to manipulate the state.
 
         See https://github.com/zauberzeug/nicegui/issues/206 for details.
@@ -107,24 +22,23 @@ async def _connected(self, timeout: float = 60.0) -> bool:
             return True
 
         # ignore served pages which do not reconnect to backend (e.g. monitoring requests, scrapers etc.)
+        TIMEOUT = 60.0
         try:
-            await self.client.connected(timeout=timeout)
+            await self.client.connected(timeout=TIMEOUT)
             return True
         except TimeoutError:
-            log.error(f'Timer cancelled because client is not connected after {timeout} seconds')
+            log.error(f'Timer cancelled because client is not connected after {TIMEOUT} seconds')
             return False
 
     def _should_stop(self) -> bool:
         return (
             self.is_deleted or
             self.client.id not in Client.instances or
-            self._is_canceled or
-            core.app.is_stopping or
-            core.app.is_stopped
+            super()._should_stop()
         )
 
     def _cleanup(self) -> None:
-        self.callback = None
+        super()._cleanup()
         if not self._deleted:
             assert self.parent_slot
             self.parent_slot.parent.remove(self)

nicegui/timer.py

@@ -0,0 +1,116 @@
+import asyncio
+import time
+from contextlib import nullcontext
+from typing import Any, Awaitable, Callable, ContextManager, Optional
+
+from . import background_tasks, core
+from .awaitable_response import AwaitableResponse
+from .binding import BindableProperty
+
+
+class Timer:
+    active = BindableProperty()
+    interval = BindableProperty()
+
+    def __init__(self,
+                 interval: float,
+                 callback: Callable[..., Any], *,
+                 active: bool = True,
+                 once: bool = False,
+                 immediate: bool = True,
+                 ) -> None:
+        """Timer
+
+        One major drive behind the creation of NiceGUI was the necessity to have a simple approach to update the interface in regular intervals,
+        for example to show a graph with incoming measurements.
+        A timer will execute a callback repeatedly with a given interval.
+
+        :param interval: the interval in which the timer is called (can be changed during runtime)
+        :param callback: function or coroutine to execute when interval elapses
+        :param active: whether the callback should be executed or not (can be changed during runtime)
+        :param once: whether the callback is only executed once after a delay specified by `interval` (default: `False`)
+        :param immediate: whether the callback should be executed immediately (default: `True`, ignored if `once` is `True`)
+        """
+        super().__init__()
+        self.interval = interval
+        self.callback: Optional[Callable[..., Any]] = callback
+        self.active = active
+        self._is_canceled = False
+        self._immediate = immediate
+
+        coroutine = self._run_once if once else self._run_in_loop
+        if core.app.is_started:
+            background_tasks.create(coroutine(), name=str(callback))
+        else:
+            core.app.on_startup(coroutine)
+
+    def _get_context(self) -> ContextManager:
+        return nullcontext()
+
+    def activate(self) -> None:
+        """Activate the timer."""
+        assert not self._is_canceled, 'Cannot activate a canceled timer'
+        self.active = True
+
+    def deactivate(self) -> None:
+        """Deactivate the timer."""
+        self.active = False
+
+    def cancel(self) -> None:
+        """Cancel the timer."""
+        self._is_canceled = True
+
+    async def _run_once(self) -> None:
+        try:
+            if not await self._can_start():
+                return
+            with self._get_context():
+                await asyncio.sleep(self.interval)
+                if self.active and not self._should_stop():
+                    await self._invoke_callback()
+        finally:
+            self._cleanup()
+
+    async def _run_in_loop(self) -> None:
+        try:
+            if not self._immediate:
+                await asyncio.sleep(self.interval)
+            if not await self._can_start():
+                return
+            with self._get_context():
+                while not self._should_stop():
+                    try:
+                        start = time.time()
+                        if self.active:
+                            await self._invoke_callback()
+                        dt = time.time() - start
+                        await asyncio.sleep(self.interval - dt)
+                    except asyncio.CancelledError:
+                        break
+                    except Exception as e:
+                        core.app.handle_exception(e)
+                        await asyncio.sleep(self.interval)
+        finally:
+            self._cleanup()
+
+    async def _invoke_callback(self) -> None:
+        try:
+            assert self.callback is not None
+            result = self.callback()
+            if isinstance(result, Awaitable) and not isinstance(result, AwaitableResponse):
+                await result
+        except Exception as e:
+            core.app.handle_exception(e)
+
+    async def _can_start(self) -> bool:
+        return True
+
+    def _should_stop(self) -> bool:
+        return (
+            self._is_canceled or
+            core.app.is_stopping or
+            core.app.is_stopped
+        )
+
+    def _cleanup(self) -> None:
+        self.callback = None

tests/test_timer.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from nicegui import ui
+from nicegui import app, ui
 from nicegui.testing import Screen, User
 
 
@@ -145,3 +145,23 @@ def count():
     await asyncio.sleep(0.1)
     gc.collect()
     assert count() == 1, 'only current timer object is in memory'
+
+
+def test_app_timer(screen: Screen):
+    counter = Counter()
+    timer = app.timer(0.1, counter.increment)
+    ui.button('Activate', on_click=timer.activate)
+    ui.button('Deactivate', on_click=timer.deactivate)
+
+    screen.open('/')
+    screen.wait(0.5)
+    assert counter.value > 0, 'timer is running after starting the server'
+
+    screen.click('Deactivate')
+    value = counter.value
+    screen.wait(0.5)
+    assert counter.value == value, 'timer is not running anymore after deactivating it'
+
+    screen.click('Activate')
+    screen.wait(0.5)
+    assert counter.value > value, 'timer is running again after activating it'

website/documentation/content/timer_documentation.py

@@ -32,4 +32,32 @@ def handle_click():
     ui.button('Notify after 1 second', on_click=handle_click)
 
 
+@doc.demo("Don't start immediately", '''
+    By default, the timer will start immediately.
+    You can change this behavior by setting the `immediate` parameter to `False`.
+    This will delay the first execution of the callback by the given interval.
+''')
+def start_immediately_demo():
+    from datetime import datetime
+
+    label = ui.label()
+    ui.timer(1.0, lambda: label.set_text(f'{datetime.now():%X}'), immediate=False)
+
+
+@doc.demo('Global app timer', '''
+    While `ui.timer` is kind of a UI element that runs in the context of the current page,
+    you can also use the global `app.timer` for UI-independent timers.
+''')
+def app_timer_demo():
+    from nicegui import app
+
+    counter = {'value': 0}
+    app.timer(1.0, lambda: counter.update(value=counter['value'] + 1))
+
+    # @ui.page('/')
+    def page():
+        ui.label().bind_text_from(counter, 'value', lambda value: f'Count: {value}')
+    page()  # HIDE
+
+
 doc.reference(ui.timer)