feat(core): improve update notification for phase-2 of the update pro…

…cess and add a spinner on top-left

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Version 0.15.9
 
 - build(packer): set `autologin-user` to `ubo` in `/etc/lightdm/lightdm.conf`
+- feat(core): improve update notification for phase-2 of the update process and add a spinner on top-left
 
 ## Version 0.15.8
 

ubo_app/menu_app/menu_central.py

@@ -8,7 +8,6 @@
 from debouncer import DebounceOptions, debounce
 from kivy.clock import mainthread
 from ubo_gui.app import UboApp
-from ubo_gui.constants import INFO_COLOR
 from ubo_gui.menu.menu_widget import MenuWidget
 from ubo_gui.menu.stack_item import StackItem, StackMenuItem
 
@@ -25,18 +24,8 @@
 )
 from ubo_app.store.main import autorun, dispatch, subscribe_event
 from ubo_app.store.services.keypad import Key, KeypadKeyPressEvent
-from ubo_app.store.services.notifications import (
-    Importance,
-    Notification,
-    NotificationDisplayType,
-    NotificationsAddAction,
-    NotificationsClearAction,
-    NotificationsDisplayEvent,
-)
-from ubo_app.store.update_manager import (
-    UPDATE_MANAGER_NOTIFICATION_ID,
-    UpdateManagerSetUpdateServiceStatusAction,
-)
+from ubo_app.store.services.notifications import NotificationsDisplayEvent
+from ubo_app.store.update_manager import UpdateManagerSetUpdateServiceStatusAction
 from ubo_app.utils.async_ import create_task
 
 from .home_page import HomePage
@@ -96,35 +85,6 @@ def _(menu: Menu | None) -> None:
                 return
             self.menu_widget.set_root_menu(menu)
 
-        background_update_notification = Notification(
-            id=UPDATE_MANAGER_NOTIFICATION_ID,
-            title='Update in progress',
-            content="""\
-Please keep the device powered on.
-This may take around 20 minutes to complete.""",
-            importance=Importance.LOW,
-            icon='󰚰',
-            display_type=NotificationDisplayType.STICKY,
-            dismissable=False,
-            dismiss_on_close=False,
-            color=INFO_COLOR,
-        )
-
-        @autorun(lambda state: state.update_manager.is_update_service_active)
-        def _(is_active: bool) -> None:  # noqa: FBT001
-            if is_active:
-                dispatch(
-                    NotificationsAddAction(
-                        notification=background_update_notification,
-                    ),
-                )
-            else:
-                dispatch(
-                    NotificationsClearAction(
-                        notification=background_update_notification,
-                    ),
-                )
-
     def build(self: UboApp) -> Widget | None:
         root = super().build()
         self.menu_widget.padding_top = root.ids.header_layout.height

ubo_app/services/010-notifications/reducer.py

@@ -20,6 +20,7 @@
     NotificationsAddAction,
     NotificationsClearAction,
     NotificationsClearAllAction,
+    NotificationsClearByIdAction,
     NotificationsClearEvent,
     NotificationsDisplayEvent,
     NotificationsState,
@@ -113,20 +114,48 @@ def reducer(
             events=events,
         )
     if isinstance(action, NotificationsClearAction):
+        new_notifications = [
+            notification
+            for notification in state.notifications
+            if notification is not action.notification
+        ]
         return CompleteReducerResult(
             state=replace(
                 state,
-                notifications=[
-                    notification
-                    for notification in state.notifications
-                    if notification is not action.notification
-                ],
-                unread_count=state.unread_count - 1
-                if action.notification in state.notifications
-                else state.unread_count,
+                notifications=new_notifications,
+                unread_count=sum(
+                    1 for notification in new_notifications if not notification.is_read
+                ),
             ),
             events=[NotificationsClearEvent(notification=action.notification)],
         )
+    if isinstance(action, NotificationsClearByIdAction):
+        to_be_removed = next(
+            (
+                notification
+                for notification in state.notifications
+                if notification.id == action.id
+            ),
+            None,
+        )
+        if to_be_removed is None:
+            return state
+
+        new_notifications = [
+            notification
+            for notification in state.notifications
+            if notification.id != action.id
+        ]
+        return CompleteReducerResult(
+            state=replace(
+                state,
+                notifications=new_notifications,
+                unread_count=sum(
+                    1 for notification in new_notifications if not notification.is_read
+                ),
+            ),
+            events=[NotificationsClearEvent(notification=to_be_removed)],
+        )
     if isinstance(action, NotificationsClearAllAction):
         return replace(state, notifications=[], unread_count=0)
     return state

ubo_app/store/services/notifications.py

@@ -123,6 +123,10 @@ class NotificationsClearAction(NotificationsAction):
     notification: Notification
 
 
+class NotificationsClearByIdAction(NotificationsAction):
+    id: str
+
+
 class NotificationsClearAllAction(NotificationsAction): ...
 
 

ubo_app/store/update_manager/utils.py

@@ -6,7 +6,9 @@
 import importlib.metadata
 import shutil
 import subprocess
+import time
 from pathlib import Path
+from typing import TypedDict
 
 import aiohttp
 import requests
@@ -24,11 +26,13 @@
 from ubo_app.store.main import autorun, dispatch
 from ubo_app.store.services.notifications import (
     Chime,
+    Importance,
     Notification,
     NotificationActionItem,
     NotificationDisplayType,
     NotificationExtraInformation,
     NotificationsAddAction,
+    NotificationsClearByIdAction,
 )
 from ubo_app.store.update_manager import (
     UPDATE_MANAGER_NOTIFICATION_ID,
@@ -330,3 +334,50 @@ def about_menu_items(state: UpdateManagerState) -> list[Item]:
             ),
         ]
     return []
+
+
+class _UpdateManagerServiceState(TypedDict):
+    is_running: bool
+    is_presented: bool
+    progress: int
+
+
+@autorun(
+    lambda state: _UpdateManagerServiceState(
+        is_running=state.update_manager.is_update_service_active,
+        is_presented=any(
+            notification.id == UPDATE_MANAGER_NOTIFICATION_ID
+            for notification in state.notifications.notifications
+        ),
+        progress=int(time.time() / 2),
+    ),
+)
+def _(state: _UpdateManagerServiceState) -> None:
+    if state['is_running']:
+        dispatch(
+            NotificationsAddAction(
+                notification=Notification(
+                    id=UPDATE_MANAGER_NOTIFICATION_ID,
+                    title='Update in progress',
+                    content="""\
+                Please keep the device powered on.
+                This may take around 20 minutes to complete.""",
+                    importance=Importance.LOW,
+                    icon='󰚰',
+                    display_type=NotificationDisplayType.BACKGROUND
+                    if state['is_presented']
+                    else NotificationDisplayType.STICKY,
+                    dismissable=False,
+                    dismiss_on_close=False,
+                    color=INFO_COLOR,
+                    progress=(state['progress'] % 4 + 1) / 4,
+                    blink=not state['is_presented'],
+                ),
+            ),
+        )
+    else:
+        dispatch(
+            NotificationsClearByIdAction(
+                id=UPDATE_MANAGER_NOTIFICATION_ID,
+            ),
+        )