feat(core): show the progress of the update using the new progress …

…property of the `Notification` object

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Version 0.15.5
 
 - feat(notifications): add `progress` and `progress_weight` properties to `Notification` object and show the progress on the header of the app
+- feat(core): show the progress of the update using the new `progress` property of the `Notification` object
 
 ## Version 0.15.4
 

pyproject.toml

@@ -17,7 +17,7 @@ priority = "primary"
 [tool.poetry.dependencies]
 python = "^3.11"
 psutil = "^6.0.0"
-ubo-gui = "^0.12.0"
+ubo-gui = "^0.12.1"
 headless-kivy = [
   { version = "^0.9.4", markers = "extra=='default'", extras = [
     "default",
@@ -105,7 +105,7 @@ target-version = 'py311'
 
 [tool.ruff.lint]
 select = ["ALL"]
-ignore = ["INP001", "PLR0911", "D203", "D213"]
+ignore = ["INP001", "PLR0911", "D203", "D213", "PLC0415"]
 fixable = ["ALL"]
 unfixable = []
 logger-objects = ['ubo_app.logging.logger']

ubo_app/logging.py

@@ -20,7 +20,7 @@ def __init__(self: UboLogger, name: str, level: int = logging.NOTSET) -> None:
 
         logging.addLevelName(VERBOSE, 'VERBOSE')
 
-    def verbose(  # noqa: PLR0913
+    def verbose(
         self: UboLogger,
         msg: object,
         *args: tuple[object],

ubo_app/services/010-notifications/reducer.py

@@ -84,19 +84,25 @@ def reducer(
                 else None,
             ),
             actions=[
-                RgbRingBlinkAction(
-                    color=(
-                        round(kivy_color[0] * 255),
-                        round(kivy_color[1] * 255),
-                        round(kivy_color[2] * 255),
-                    ),
-                    repetitions={
-                        Importance.LOW: 1,
-                        Importance.MEDIUM: 2,
-                        Importance.HIGH: 3,
-                        Importance.CRITICAL: 4,
-                    }[action.notification.importance],
-                    wait=400,
+                *(
+                    [
+                        RgbRingBlinkAction(
+                            color=(
+                                round(kivy_color[0] * 255),
+                                round(kivy_color[1] * 255),
+                                round(kivy_color[2] * 255),
+                            ),
+                            repetitions={
+                                Importance.LOW: 1,
+                                Importance.MEDIUM: 2,
+                                Importance.HIGH: 3,
+                                Importance.CRITICAL: 4,
+                            }[action.notification.importance],
+                            wait=400,
+                        ),
+                    ]
+                    if action.notification.blink
+                    else []
                 ),
                 *(
                     [SoundPlayChimeAction(name=action.notification.chime)]

ubo_app/setup.py

@@ -8,11 +8,11 @@
 import dotenv
 import numpy as np
 
-from ubo_app.constants import INSTALLATION_PATH
-
 
 def setup_hostname() -> None:
     """Set the hostname to 'ubo'."""
+    from ubo_app.constants import INSTALLATION_PATH
+
     available_letters = list(
         set(string.ascii_lowercase + string.digits + '-') - set('I1lO'),
     )
@@ -26,7 +26,7 @@ def setup_hostname() -> None:
     id = id_path.read_text().strip()
 
     # Set hostname of the system
-    Path('/etc/hostname').write_text(id)
+    Path('/etc/hostname').write_text(id, encoding='utf-8')
 
 
 def setup() -> None:
@@ -97,7 +97,7 @@ def fake_create_subprocess_exec(*args: str, **kwargs: Any) -> object:  # noqa: A
                 command = command.as_posix()
             if any(i in command for i in ('reboot', 'poweroff')):
                 return Fake()
-            if command in ('curl', 'tar') or command.endswith('/code'):
+            if command in {'curl', 'tar'} or command.endswith('/code'):
                 return original_asyncio_create_subprocess_exec(*args, **kwargs)
             return Fake(
                 _Fake__await_value=Fake(

ubo_app/store/services/notifications.py

@@ -89,7 +89,7 @@ class Notification(Immutable):
     content: str
     extra_information: str | None = None
     importance: Importance = Importance.LOW
-    chime: Chime | None = Chime.DONE
+    chime: Chime | None = None
     timestamp: datetime = field(default_factory=lambda: datetime.now(tz=UTC))
     is_read: bool = False
     sender: str | None = None
@@ -102,6 +102,7 @@ class Notification(Immutable):
     dismissable: bool = True
     dismiss_on_close: bool = False
     on_close: Callable[[], Any] | None = None
+    blink: bool = True
     progress: float | None = None
     progress_weight: float = 1
 

ubo_app/store/update_manager/utils.py

@@ -11,7 +11,7 @@
 
 import aiohttp
 import requests
-from ubo_gui.constants import DANGER_COLOR, SUCCESS_COLOR
+from ubo_gui.constants import DANGER_COLOR, INFO_COLOR, SUCCESS_COLOR
 from ubo_gui.menu.types import ActionItem, Item
 
 from ubo_app.constants import INSTALLATION_PATH
@@ -59,7 +59,7 @@ async def check_version() -> None:
                 try:
                     eeprom_json_data = Path(
                         '/proc/device-tree/hat/custom_0',
-                    ).read_text()
+                    ).read_text(encoding='utf-8')
                     eeprom_data = json.loads(eeprom_json_data)
                     serial_number = eeprom_data['serial_number']
                 except Exception:
@@ -86,12 +86,33 @@ async def check_version() -> None:
 async def update() -> None:
     """Update the Ubo app."""
     logger.info('Updating Ubo app...')
+    dispatch(
+        NotificationsAddAction(
+            notification=Notification(
+                id='ubo:update_manager',
+                title='Updating...',
+                content='Fetching the latest version of Ubo app...',
+                display_type=NotificationDisplayType.BACKGROUND,
+                color=INFO_COLOR,
+                icon='󰇚',
+                blink=False,
+                progress=0,
+            ),
+        ),
+    )
 
     async def download_files() -> None:
         target_path = Path(f'{INSTALLATION_PATH}/_update/')
         shutil.rmtree(target_path, ignore_errors=True)
         target_path.mkdir(parents=True, exist_ok=True)
 
+        packages_count_path = f'{INSTALLATION_PATH}/.packages-count'
+
+        try:
+            packages_count = int(Path(packages_count_path).read_text(encoding='utf-8'))
+        except FileNotFoundError:
+            packages_count = 55
+
         process = await asyncio.create_subprocess_exec(
             '/usr/bin/env',
             'pip',
@@ -101,10 +122,65 @@ async def download_files() -> None:
             'ubo-app[default]',
             'setuptools',
             'wheel',
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
         )
+        if process.stdout is None:
+            logger.exception('Failed to update (pip has no stdout)')
+            dispatch(
+                NotificationsAddAction(
+                    notification=Notification(
+                        id='ubo:update_manager',
+                        title='Failed to update',
+                        content='Failed to download packages',
+                        display_type=NotificationDisplayType.FLASH,
+                        color=DANGER_COLOR,
+                        icon='󰜺',
+                        chime=Chime.FAILURE,
+                    ),
+                ),
+                UpdateManagerSetStatusAction(status=UpdateStatus.CHECKING),
+            )
+            return
+        counter = 0
+        while True:
+            line = (await process.stdout.readline()).decode()
+            if not line:
+                break
+            if line.startswith('Collecting'):
+                counter += 1
+                dispatch(
+                    NotificationsAddAction(
+                        notification=Notification(
+                            id='ubo:update_manager',
+                            title='Updating...',
+                            content=f'Downloading {line.partition(" ")[2].strip()}',
+                            display_type=NotificationDisplayType.BACKGROUND,
+                            color=INFO_COLOR,
+                            icon='󰇚',
+                            blink=False,
+                            progress=min(counter / (packages_count * 2), 1),
+                        ),
+                    ),
+                )
         await process.wait()
+
+        # Update the packages count estimate for the next update
+        Path(packages_count_path).write_text(str(counter), encoding='utf-8')
+
+        dispatch(
+            NotificationsAddAction(
+                notification=Notification(
+                    id='ubo:update_manager',
+                    title='Updating...',
+                    content='All packages downloaded successfully, rebooting...',
+                    display_type=NotificationDisplayType.STICKY,
+                    color=INFO_COLOR,
+                    icon='󰇚',
+                    progress=1,
+                ),
+            ),
+        )
         if process.returncode != 0:
             msg = 'Failed to download packages'
             raise RuntimeError(msg)

ubo_app/system/install.sh

@@ -115,9 +115,10 @@ if [ "$UPDATE" = true ]; then
   fi
 else
   if [ "$ALPHA" = true ]; then
-    "$INSTALLATION_PATH/env/bin/python" -m pip install --pre "$SOURCE"[default]
+    "$INSTALLATION_PATH/env/bin/python" -m pip install --pre "$SOURCE"[default] | grep -c '^Collecting ' > $INSTALLATION_PATH/.packages-count
   else
-    "$INSTALLATION_PATH/env/bin/python" -m pip install "$SOURCE"[default]
+    # Count number of Collecting instances
+    "$INSTALLATION_PATH/env/bin/python" -m pip install "$SOURCE"[default] | grep -c '^Collecting ' > $INSTALLATION_PATH/.packages-count
   fi
 fi