fix(notification): avoid passing color components bigger than 255

feat(ssh): show an error message if anything goes wrong during creating temporary account instead of crashing

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## Version 0.12.5
+
+- fix(notification): avoid passing color components bigger than 255
+- feat(ssh): show an error message if anything goes wrong during creating temporary
+  account instead of crashing
+- fix(keypad): return from `key_press_cb` if no input is changed
+
 ## Version 0.12.4
 
 - build(packer): disable userconfig service

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ubo-app"
-version = "0.12.4"
+version = "0.12.5"
 description = "Ubo main app, running on device initialization. A platform for running other apps."
 authors = ["Sassan Haradji <[email protected]>"]
 license = "Apache-2.0"
@@ -51,7 +51,7 @@ pytest-asyncio = "^0.23.5.post1"
 pytest-cov = "^4.1.0"
 pytest-timeout = "^2.3.1"
 pytest-xdist = "^3.5.0"
-ruff = "^0.3.5"
+ruff = "^0.3.7"
 tenacity = "^8.2.3"
 toml = "^0.10.2"
 pytest-mock = "^3.14.0"

ubo_app/services/000-sound/audio_manager.py

@@ -134,16 +134,16 @@ def play(self: AudioManager, filename: str) -> None:
         try:
             with wave.open(filename, 'rb') as wf:
                 self.is_playing = True
-                stream = self.stream = self.pyaudio.open(
+                self.stream = self.pyaudio.open(
                     format=self.pyaudio.get_format_from_width(wf.getsampwidth()),
                     channels=wf.getnchannels(),
                     rate=wf.getframerate(),
                     output=True,
                     output_device_index=self.find_respeaker_index(),
                 )
                 data = wf.readframes(CHUNK_SIZE)
-                while data and not self.should_stop and stream.is_active():
-                    stream.write(data)
+                while data and not self.should_stop and self.stream.is_active():
+                    self.stream.write(data)
                     data = wf.readframes(CHUNK_SIZE)
         except Exception:
             create_task(self.restart_pulse_audio())

ubo_app/services/010-notifications/reducer.py

@@ -68,9 +68,9 @@ def reducer(
             actions=[
                 RgbRingBlinkAction(
                     color=(
-                        round(kivy_color[0] * 256),
-                        round(kivy_color[1] * 256),
-                        round(kivy_color[2] * 256),
+                        round(kivy_color[0] * 255),
+                        round(kivy_color[1] * 255),
+                        round(kivy_color[2] * 255),
                     ),
                     repetitions={
                         Importance.LOW: 1,

ubo_app/services/020-keypad/setup.py

@@ -235,24 +235,26 @@ def key_press_cb(self: Keypad, _: object) -> None:
         self.event_queue.append(event)
         self.logger.info(self.event_queue)
         self.logger.debug('Current Inputs', extra={'inputs': f'{self.inputs:016b}'})
-        previos_event = self.event_queue.pop(0)
+        previous_event = self.event_queue.pop(0)
         self.logger.debug(
             'Previous Inputs',
-            extra={'inputs': f'{previos_event.inputs:016b}'},
+            extra={'inputs': f'{previous_event.inputs:016b}'},
         )
         # XOR the last recorded input values with the current input values
         # to see which bits have changed. Technically there can only be one
         # bit change in every callback
-        change_mask = previos_event.inputs ^ self.inputs
+        change_mask = previous_event.inputs ^ self.inputs
         self.logger.debug('Change', extra={'change_mask': f'{change_mask:016b}'})
+        if change_mask == 0:
+            return
         # use the change mask to see if the button was the change was
         # falling (1->0) indicating a pressed action
         # or risign (0->1) indicating a release action
         self.index = (int)(math.log2(change_mask))
         self.logger.info('button index', extra={'button_index': self.index})
         # Check for rising edge or falling edge action (press or release)
         self.button_label = self.buttons.get_label(self.index)
-        if (previos_event.inputs & change_mask) == 0:
+        if (previous_event.inputs & change_mask) == 0:
             self.logger.info(
                 'Button pressed',
                 extra={'button': str(self.index), 'label': self.button_label},

ubo_app/services/050-ssh/setup.py

@@ -5,6 +5,7 @@
 import asyncio
 from typing import TYPE_CHECKING
 
+from ubo_gui.constants import DANGER_COLOR, SUCCESS_COLOR
 from ubo_gui.menu.types import (
     ActionItem,
     ApplicationItem,
@@ -83,6 +84,22 @@ async def act() -> None:
             'service ssh create_temporary_ssh_account',
             has_output=True,
         )
+        if not result:
+            dispatch(
+                NotificationsAddAction(
+                    notification=Notification(
+                        title='Failed to create temporary SSH account',
+                        content='An error occurred while creating the temporary SSH '
+                        'account.',
+                        importance=Importance.MEDIUM,
+                        icon='󰣀',
+                        display_type=NotificationDisplayType.STICKY,
+                        color=DANGER_COLOR,
+                    ),
+                ),
+            )
+
+            return
         username, password = result.split(':')
         dispatch(
             NotificationsAddAction(
@@ -95,6 +112,7 @@ async def act() -> None:
                     importance=Importance.MEDIUM,
                     icon='󰣀',
                     display_type=NotificationDisplayType.STICKY,
+                    color=SUCCESS_COLOR,
                 ),
             ),
         )

ubo_app/utils/qrcode.py

@@ -93,9 +93,9 @@ async def handle_barcode_event(event: CameraBarcodeEvent) -> None:
             dispatch(
                 RgbRingBlinkAction(
                     color=(
-                        round(kivy_color[0] * 256),
-                        round(kivy_color[1] * 256),
-                        round(kivy_color[2] * 256),
+                        round(kivy_color[0] * 255),
+                        round(kivy_color[1] * 255),
+                        round(kivy_color[2] * 255),
                     ),
                     repetitions=1,
                     wait=200,

ubo_app/utils/server.py

@@ -3,43 +3,42 @@
 from __future__ import annotations
 
 import asyncio
-import threading
 from typing import Literal, overload
 
 from ubo_app.constants import SERVER_SOCKET_PATH
 from ubo_app.logging import logger
 from ubo_app.utils import IS_RPI
 
-thread_lock = threading.Lock()
-
 
 @overload
 async def send_command(command: str) -> None: ...
 
 
 @overload
-async def send_command(command: str, *, has_output: Literal[True]) -> str: ...
+async def send_command(command: str, *, has_output: Literal[True]) -> str | None: ...
 
 
 async def send_command(command: str, *, has_output: bool = False) -> str | None:
     """Send a command to the system manager socket."""
     if not IS_RPI:
         return None
-    reader, writer = await asyncio.open_unix_connection(SERVER_SOCKET_PATH)
 
-    logger.debug('Sending command:', extra={'command': command})
+    try:
+        reader, writer = await asyncio.open_unix_connection(SERVER_SOCKET_PATH)
+
+        logger.debug('Sending command:', extra={'command': command})
 
-    response = None
-    with thread_lock:
+        response = None
         writer.write(command.encode() + b'\0')
         if has_output:
             datagram = (await reader.readuntil(b'\0'))[:-1]
             if datagram:
                 response = datagram.decode('utf-8')
                 logger.debug('Server response:', extra={'response': response})
         writer.close()
-        await writer.wait_closed()
 
-    logger.debug('Received response:', extra={'response': response})
-
-    return response
+    except Exception:
+        logger.exception('Failed to send command to the server')
+        return None
+    else:
+        return response