feat(display): add display service and put display content in the bus…

… via `DisplayRenderEvent`

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## Upcoming
+
+- feat(display): add display service and put display content in the bus via `DisplayRenderEvent`
+
 ## Version 0.16.1
 
 - feat(lightdm): set wayland as the default session for lightdm after installing raspberrypi-ui-mods

pyproject.toml

@@ -161,11 +161,15 @@ profile = "black"
 exclude = ["typings"]
 
 [[tool.pyright.executionEnvironments]]
-root = "ubo_app/services/000-keypad"
+root = "ubo_app/services/000-audio"
 extraPaths = ["."]
 
 [[tool.pyright.executionEnvironments]]
-root = "ubo_app/services/000-audio"
+root = "ubo_app/services/000-display"
+extraPaths = ["."]
+
+[[tool.pyright.executionEnvironments]]
+root = "ubo_app/services/000-keypad"
 extraPaths = ["."]
 
 [[tool.pyright.executionEnvironments]]

scripts/deploy.sh

@@ -11,7 +11,6 @@ function cleanup() {
 trap cleanup ERR
 trap cleanup EXIT
 
-LATEST_VERSION=$(basename $(ls -rt dist/*.whl | tail -n 1))
 deps=${deps:-"False"}
 bootstrap=${bootstrap:-"False"}
 kill=${kill:-"False"}
@@ -21,6 +20,7 @@ env=${env:-"False"}
 perl -i -pe 's/^(packages = \[.*)$/\1\nexclude = ["ubo_app\/services\/*-voice\/models\/*"]/' pyproject.toml
 poetry build
 cleanup
+LATEST_VERSION=$(basename $(ls -rt dist/*.whl | tail -n 1))
 
 function run_on_pod() {
   if [ $# -lt 1 ]; then

tests/flows/test_wireless.py

@@ -61,7 +61,7 @@ def store_snapshot_selector(state: RootState) -> WiFiState:
 
     app = MenuApp()
     app_context.set_app(app)
-    load_services(['camera', 'wifi', 'notifications'])
+    load_services(['camera', 'display', 'notifications', 'wifi'])
 
     @wait_for(wait=wait_fixed(1), run_async=True)
     def check_icon(expected_icon: str) -> None:

tests/integration/results/test_services/all_services_register/store-desktop-000.jsonc

@@ -7,16 +7,20 @@
     "capture_volume": 0.5,
     "is_capture_mute": false,
     "is_playback_mute": false,
-    "playback_volume": 0.5
+    "playback_volume": 0.15
   },
   "camera": {
     "_type": "CameraState",
     "current": null,
     "is_viewfinder_active": false,
     "queue": []
   },
+  "display": {
+    "_type": "DisplayState",
+    "is_paused": false
+  },
   "docker": {
-    "_id": "b5d32b1666194cb1d71037d1b83e90ec",
+    "_id": "a0116be5ab0c1681c8f8e3d0d3290a4c",
     "_type": "DockerState",
     "home_assistant": {
       "_type": "ImageState",
@@ -577,7 +581,7 @@
                                       {
                                         "_type": "DispatchItem",
                                         "action": "<function:<lambda>>",
-                                        "background_color": "#FF3F51",
+                                        "background_color": "#FFC107",
                                         "color": [
                                           1,
                                           1,

tests/integration/results/test_services/all_services_register/store-rpi-000.jsonc

@@ -7,16 +7,20 @@
     "capture_volume": 0.5,
     "is_capture_mute": true,
     "is_playback_mute": false,
-    "playback_volume": 0.5
+    "playback_volume": 0.15
   },
   "camera": {
     "_type": "CameraState",
     "current": null,
     "is_viewfinder_active": false,
     "queue": []
   },
+  "display": {
+    "_type": "DisplayState",
+    "is_paused": false
+  },
   "docker": {
-    "_id": "b5d32b1666194cb1d71037d1b83e90ec",
+    "_id": "a0116be5ab0c1681c8f8e3d0d3290a4c",
     "_type": "DockerState",
     "home_assistant": {
       "_type": "ImageState",
@@ -577,7 +581,7 @@
                                       {
                                         "_type": "DispatchItem",
                                         "action": "<function:<lambda>>",
-                                        "background_color": "#FF3F51",
+                                        "background_color": "#FFC107",
                                         "color": [
                                           1,
                                           1,
@@ -642,7 +646,7 @@
                                       {
                                         "_type": "DispatchItem",
                                         "action": "<function:<lambda>>",
-                                        "background_color": "#FF3F51",
+                                        "background_color": "#FFC107",
                                         "color": [
                                           1,
                                           1,

tests/integration/results/test_services/all_services_register/window-desktop-000.hash

@@ -1,2 +1,2 @@
 // window-desktop-000
-12cac7ceb6198b823530c73a9965c46a793f38eccb63aa77d51490cf8c9b72e5
+3620d71f464832b8c78da353873df66e3f2daf54a324150709b4e8190b6dbafb

tests/integration/results/test_services/all_services_register/window-rpi-000.hash

@@ -1,2 +1,2 @@
 // window-rpi-000
-b64fc25d16a3142afc786cd59c2700fc0b0abbb85ed8ea209e3faaff64c73713
+006d4f47428eb62a54e3111ed4ffae84fbf14a4baff55a93a213475f7457a13b

tests/integration/test_services.py

@@ -13,6 +13,7 @@
 ALL_SERVICES_IDS = [
     'audio',
     'camera',
+    'display',
     'docker',
     'ethernet',
     'ip',

ubo_app/display.py

@@ -2,53 +2,45 @@
 
 from __future__ import annotations
 
-import atexit
 from typing import TYPE_CHECKING, cast
 
 import numpy as np
 from adafruit_rgb_display.st7789 import ST7789
+from fake import Fake
+
+from ubo_app.store.main import store
+from ubo_app.store.services.display import DisplayRenderEvent
+from ubo_app.utils import IS_RPI
 
 if TYPE_CHECKING:
     from threading import Thread
 
     from numpy._typing import NDArray
 
 
-from fake import Fake
-
-from ubo_app.constants import BYTES_PER_PIXEL
-from ubo_app.utils import IS_RPI
-
 if IS_RPI:
     import board
     import digitalio
 
+    from ubo_app.constants import HEIGHT, WIDTH
+
     cs_pin = digitalio.DigitalInOut(board.CE0)
     dc_pin = digitalio.DigitalInOut(board.D25)
     reset_pin = digitalio.DigitalInOut(board.D24)
     spi = board.SPI()
-
-
-def setup_display() -> ST7789:
-    """Set the display for the Raspberry Pi."""
-    if IS_RPI:
-        from ubo_app.constants import HEIGHT, WIDTH
-
-        display = ST7789(
-            spi,
-            height=HEIGHT,
-            width=WIDTH,
-            y_offset=80,
-            x_offset=0,
-            cs=cs_pin,
-            dc=dc_pin,
-            rst=reset_pin,
-            baudrate=60000000,
-        )
-    else:
-        display = cast(ST7789, Fake())
-
-    return display
+    display = ST7789(
+        spi,
+        height=HEIGHT,
+        width=WIDTH,
+        y_offset=80,
+        x_offset=0,
+        cs=cs_pin,
+        dc=dc_pin,
+        rst=reset_pin,
+        baudrate=60000000,
+    )
+else:
+    display = cast(ST7789, Fake())
 
 
 def render_on_display(
@@ -59,89 +51,46 @@ def render_on_display(
     last_render_thread: Thread,
 ) -> None:
     """Transfer data to the display via SPI controller."""
-    if IS_RPI and state.is_running:
-        from ubo_app.logging import logger
-
-        logger.verbose('Rendering frame', extra={'data_hash': data_hash})
-
-        data_ = data.astype(np.uint16)
-        color = (
-            ((data_[:, :, 0] & 0xF8) << 8)
-            | ((data_[:, :, 1] & 0xFC) << 3)
-            | (data_[:, :, 2] >> 3)
-        )
-        data_bytes = bytes(
-            np.dstack(((color >> 8) & 0xFF, color & 0xFF)).flatten().tolist(),
-        )
-
-        # Wait for the last render thread to finish
-        if last_render_thread:
-            last_render_thread.join()
-
-        # Only render when running on a Raspberry Pi
-        state.block(rectangle, data_bytes)
-
-
-class _DisplayState:
-    """The state of the display."""
-
-    is_running = True
-    display = setup_display()
-
-    def __init__(self: _DisplayState, splash_screen: bytes | None = None) -> None:
-        if IS_RPI:
-            from RPi import GPIO  # pyright: ignore [reportMissingModuleSource]
-
-            GPIO.setmode(GPIO.BCM)
-            GPIO.setup(26, GPIO.OUT)
-            GPIO.output(26, GPIO.HIGH)
-
-        from ubo_app.constants import HEIGHT, WIDTH
-
-        self.block(
-            (0, 0, WIDTH - 1, HEIGHT - 1),
-            bytes(WIDTH * HEIGHT * BYTES_PER_PIXEL)
-            if splash_screen is None
-            else splash_screen,
-        )
-
-        atexit.register(self.turn_off)
-
-    def pause(self: _DisplayState) -> None:
-        """Pause the display."""
-        self.is_running = False
-
-    def resume(self: _DisplayState) -> None:
-        """Resume the display."""
-        self.is_running = True
-
-    def turn_off(self: _DisplayState) -> None:
-        """Destroy the display."""
-        from ubo_app.constants import HEIGHT, WIDTH
-
-        self.block(
-            (0, 0, WIDTH - 1, HEIGHT - 1),
-            np.zeros((WIDTH, HEIGHT, BYTES_PER_PIXEL), dtype=np.uint8).tobytes(),
-        )
-
-        if IS_RPI:
-            from RPi import GPIO  # pyright: ignore [reportMissingModuleSource]
+    data_ = data.astype(np.uint16)
+    color = (
+        ((data_[:, :, 0] & 0xF8) << 8)
+        | ((data_[:, :, 1] & 0xFC) << 3)
+        | (data_[:, :, 2] >> 3)
+    )
+    data_bytes = bytes(
+        np.dstack(((color >> 8) & 0xFF, color & 0xFF)).flatten().tolist(),
+    )
+    if last_render_thread:
+        last_render_thread.join()
+    render_block(rectangle, data_bytes)
+    store.dispatch(
+        DisplayRenderEvent(
+            data=data.tobytes(),
+            data_hash=data_hash,
+            rectangle=rectangle,
+        ),
+    )
+
+
+@store.view(lambda state: state.display.is_paused)
+def render_block(
+    is_paused: bool,  # noqa: FBT001
+    rectangle: tuple[int, int, int, int],
+    data_bytes: bytes,
+    *,
+    bypass_pause: bool = False,
+) -> None:
+    """Block the display."""
+    if not is_paused or bypass_pause:
+        display._block(*rectangle, data_bytes)  # noqa: SLF001
 
-            GPIO.setmode(GPIO.BCM)
-            GPIO.setup(26, GPIO.OUT)
-            GPIO.output(26, GPIO.LOW)
-            GPIO.cleanup(26)
 
-    def block(
-        self: _DisplayState,
-        rectangle: tuple[int, int, int, int],
-        data_bytes: bytes,
-        *,
-        bypass_pause: bool = False,
-    ) -> None:
-        """Block the display."""
-        if self.is_running or bypass_pause:
-            self.display._block(*rectangle, data_bytes)  # noqa: SLF001
+def turn_off() -> None:
+    """Turn off the display."""
+    display._block = lambda *args, **kwargs: (args, kwargs)  # noqa: SLF001
+    render_blank()
 
 
-state = _DisplayState()
+def render_blank() -> None:
+    """Render a blank screen."""
+    display.fill(0)

ubo_app/services/000-display/reducer.py

@@ -0,0 +1,36 @@
+# ruff: noqa: D100, D101, D102, D103, D104, D107, N999
+from __future__ import annotations
+
+from dataclasses import replace
+
+from redux import (
+    InitAction,
+    InitializationActionError,
+)
+
+from ubo_app.store.services.display import (
+    DisplayAction,
+    DisplayPauseAction,
+    DisplayResumeAction,
+    DisplayState,
+)
+
+Action = InitAction | DisplayAction
+
+
+def reducer(
+    state: DisplayState | None,
+    action: Action,
+) -> DisplayState:
+    if state is None:
+        if isinstance(action, InitAction):
+            return DisplayState()
+        raise InitializationActionError(action)
+
+    if isinstance(action, DisplayPauseAction):
+        return replace(state, is_paused=True)
+
+    if isinstance(action, DisplayResumeAction):
+        return replace(state, is_paused=False)
+
+    return state