Merge pull request #5 from ubopod/clear-pause-resume

feat: add `clear_at_exit` setting and `pause` and `resume` methods

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## Version 0.5.6
+
+- fix: ignore data hash and render a single frame after resume
+
+## Version 0.5.3
+
+- feat: add `clear_at_exit` setting and `pause` and `resume` methods
+
 ## Version 0.5.2
 
 - chore: simplify `pyproject.toml` and the setup instructions

README.md

@@ -39,6 +39,8 @@ poetry --group dev headless-kivy-pi
        display_class=ST7789,
        double_buffering=True,
        synchronous_clock=True,
+       automatic_fps=True,
+       clear_at_exit=True,
    )
    ```
 
@@ -91,7 +93,15 @@ Is set to `True`, it will let Kivy generate the next frame while sending the las
 
 #### `synchronous_clock`
 
-If set to True, Kivy will wait for the LCD before rendering next frames. This will cause Headless to skip frames if they are rendered before the LCD has finished displaying the previous frames. If set to False, frames will be rendered asynchronously, letting Kivy render frames regardless of display being able to catch up or not at the expense of possible frame skipping.
+If set to `True`, Kivy will wait for the LCD before rendering next frames. This will cause Headless to skip frames if they are rendered before the LCD has finished displaying the previous frames. If set to False, frames will be rendered asynchronously, letting Kivy render frames regardless of display being able to catch up or not at the expense of possible frame skipping.
+
+#### `automatic_fps`
+
+If set to `True`, it will monitor the hash of the screen data, if this hash changes, it will increase the fps to the maximum and if the hash doesn't change for a while, it will drop the fps to the minimum.
+
+#### `clear_at_exit`
+
+If set to `True`, it will clear the screen before exiting.
 
 ## ⚒️ Contribution
 

headless_kivy_pi/__init__.py

@@ -13,13 +13,14 @@
 """
 from __future__ import annotations
 
+import atexit
 import os
 import time
 from distutils.util import strtobool
 from pathlib import Path
 from queue import Queue
 from threading import Semaphore, Thread
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 import kivy
 import numpy as np
@@ -95,37 +96,44 @@
     )
     == 1
 )
+CLEAR_AT_EXIT = (
+    strtobool(os.environ.get('HEADLESS_KIVY_PI_CLEAR_AT_EXIT', 'False')) == 1
+)
 
 
 class SetupHeadlessConfig(TypedDict):
     """Arguments of `setup_headless` function."""
 
-    min_fps: NotRequired[int]
     """Minimum frames per second for when the Kivy application is idle."""
-    max_fps: NotRequired[int]
+    min_fps: NotRequired[int]
     """Maximum frames per second for the Kivy application."""
-    width: NotRequired[int]
+    max_fps: NotRequired[int]
     """The width of the display in pixels."""
-    height: NotRequired[int]
+    width: NotRequired[int]
     """The height of the display in pixels."""
-    baudrate: NotRequired[int]
+    height: NotRequired[int]
     """The baud rate for the display connection."""
-    debug_mode: NotRequired[bool]
+    baudrate: NotRequired[int]
     """If set to True, the application will consume computational resources to log
     additional debug information."""
-    display_class: NotRequired[type[DisplaySPI]]
+    debug_mode: NotRequired[bool]
     """The display class to use (default is ST7789)."""
-    double_buffering: NotRequired[bool]
+    display_class: NotRequired[type[DisplaySPI]]
     """Is set to `True`, it will let Kivy generate the next frame while sending the
     last frame to the display."""
-    synchronous_clock: NotRequired[bool]
+    double_buffering: NotRequired[bool]
     """If set to `True`, Kivy will wait for the LCD before rendering next frames. This
     will cause Headless to skip frames if they are rendered before the LCD has finished
     displaying the previous frames. If set to False, frames will be rendered
     asynchronously, letting Kivy render frames regardless of display being able to catch
     up or not at the expense of possible frame skipping."""
+    synchronous_clock: NotRequired[bool]
+    """If set to `True`, it will monitor the hash of the screen data, if this hash
+    changes, it will increase the fps to the maximum and if the hash doesn't change for
+    a while, it will drop the fps to the minimum."""
     automatic_fps: NotRequired[bool]
-    """If set to `True` fps will adjust automatically."""
+    """If set to `True`, it will clear the screen before exiting."""
+    clear_at_eixt: NotRequired[bool]
 
 
 def setup_headless(config: SetupHeadlessConfig | None = None) -> None:
@@ -154,6 +162,7 @@ def setup_headless(config: SetupHeadlessConfig | None = None) -> None:
         'automatic_fps',
         AUTOMATIC_FPS_CONTROL,
     )
+    clear_at_exit = config.get('clear_at_exit', CLEAR_AT_EXIT)
 
     if HeadlessWidget.debug_mode:
         add_stdout_handler()
@@ -179,6 +188,8 @@ def setup_headless(config: SetupHeadlessConfig | None = None) -> None:
     HeadlessWidget.width = width
     HeadlessWidget.height = height
 
+    HeadlessWidget.is_paused = False
+
     Config.set('kivy', 'kivy_clock', 'default')
     Config.set('graphics', 'fbo', 'force-hardware')
     Config.set('graphics', 'fullscreen', '0')
@@ -196,7 +207,7 @@ def setup_headless(config: SetupHeadlessConfig | None = None) -> None:
         cs_pin = digitalio.DigitalInOut(board.CE0)
         dc_pin = digitalio.DigitalInOut(board.D25)
         reset_pin = digitalio.DigitalInOut(board.D24)
-        HeadlessWidget._display = display_class(
+        display = display_class(
             spi,
             height=height,
             width=width,
@@ -208,6 +219,17 @@ def setup_headless(config: SetupHeadlessConfig | None = None) -> None:
             rst=reset_pin,
             baudrate=baudrate,
         )
+        HeadlessWidget._display = display
+        if clear_at_exit:
+            atexit.register(
+                lambda: display._block(
+                    0,
+                    0,
+                    width - 1,
+                    height - 1,
+                    bytes(width * height * 2),
+                ),
+            )
     else:
         from kivy.core.window import Window
         from screeninfo import get_monitors
@@ -222,11 +244,13 @@ def setup_headless(config: SetupHeadlessConfig | None = None) -> None:
 class HeadlessWidget(Widget):
     """Headless Kivy widget class rendering on SPI connected display."""
 
+    should_ignore_hash: ClassVar = False
     texture = ObjectProperty(None, allownone=True)
     _display: DisplaySPI
 
     min_fps: int
     max_fps: int
+    is_paused: bool
     width: int
     height: int
     debug_mode: bool
@@ -340,11 +364,23 @@ def release_frame(self: HeadlessWidget) -> None:
 
         def release_task() -> None:
             time.sleep(1 / self.fps)
-            type(self).fps_control_queue.release()
+            HeadlessWidget.fps_control_queue.release()
 
         self.latest_release_thread = Thread(target=release_task)
         self.latest_release_thread.start()
 
+    @classmethod
+    def pause(cls: type[HeadlessWidget]) -> None:
+        """Pause writing to the display."""
+        HeadlessWidget.is_paused = True
+
+    @classmethod
+    def resume(cls: type[HeadlessWidget]) -> None:
+        """Resume writing to the display."""
+        cls.is_paused = False
+        cls.should_ignore_hash = True
+        cls.reset_fps_control_queue()
+
     @classmethod
     def reset_fps_control_queue(cls: type[HeadlessWidget]) -> None:
         """Dequeue `fps_control_queue` forcfully to render the next frame.
@@ -369,7 +405,7 @@ def activate_high_fps_mode(cls: type[HeadlessWidget]) -> None:
         cls.reset_fps_control_queue()
 
     @classmethod
-    def _activate_low_fps_mode(cls: type[HeadlessWidget], *_: Any) -> None:
+    def _activate_low_fps_mode(cls: type[HeadlessWidget], *_: object) -> None:
         logger.info('Activating low fps mode, dropping FPS to `min_fps`')
         cls.fps = cls.min_fps
 
@@ -404,7 +440,7 @@ def transfer_to_display(
             last_render_thread.join()
 
         # Only render when running on a Raspberry Pi
-        if IS_RPI:
+        if IS_RPI and not HeadlessWidget.is_paused:
             HeadlessWidget._display._block(
                 0,
                 0,
@@ -416,7 +452,7 @@ def transfer_to_display(
     def render_on_display(self: HeadlessWidget, *_: Any) -> None:  # noqa: ANN401
         """Render the widget on display connected to the SPI controller."""
         # Block if it is rendering more FPS than expected
-        type(self).fps_control_queue.acquire()
+        HeadlessWidget.fps_control_queue.acquire()
         self.release_frame()
 
         # Log the number of skipped and rendered frames in the last second
@@ -454,7 +490,7 @@ def render_on_display(self: HeadlessWidget, *_: Any) -> None:  # noqa: ANN401
             -1,
         )
         data_hash = hash(data.data.tobytes())
-        if data_hash == self.last_hash:
+        if data_hash == self.last_hash and not HeadlessWidget.should_ignore_hash:
             # Only drop FPS when the screen has not changed for at least one second
             if (
                 self.automatic_fps_control
@@ -467,6 +503,8 @@ def render_on_display(self: HeadlessWidget, *_: Any) -> None:  # noqa: ANN401
             # Considering the content has not changed, this frame can safely be ignored
             return
 
+        HeadlessWidget.should_ignore_hash = False
+
         self.last_change = time.time()
         self.last_hash = data_hash
         if self.automatic_fps_control and self.fps != self.max_fps:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "headless-kivy-pi"
-version = "0.5.2"
+version = "0.5.6"
 description = "Headless renderer for Kivy framework on Raspberry Pi"
 authors = ["Sassan Haradji <[email protected]>"]
 license = "Apache-2.0"