feat(web-ui): process input demands dispatched on the bus

.github/workflows/integration_delivery.yml

@@ -2,6 +2,10 @@ name: CI/CD
 
 on:
   push:
+    branches:
+      - main
+    tags:
+      - 'v*'
   pull_request:
   workflow_dispatch:
 

CHANGELOG.md

@@ -8,6 +8,7 @@
 - chore: remove what has remained from poetry in the codebase
 - refactor(core): avoid truncating or coloring logs in log files
 - feat(web-ui): add web-ui service
+- feat(web-ui): process input demands dispatched on the bus
 
 ## Version 1.0.0
 

pyproject.toml

@@ -22,7 +22,7 @@ dependencies = [
   "platformdirs >=4.2.0",
   "dill >=0.3.8",
   "simpleaudio >=1.0.4",
-  "python-redux >=0.17.1",
+  "python-redux >=0.17.2",
   "python-debouncer >=0.1.5",
   "python-strtobool >=1.0.0",
   "python-fake >=0.1.3",

setup_scm_schemes.py

@@ -1,6 +1,6 @@
-from setuptools_scm.version import get_local_node_and_date
+from setuptools_scm.version import get_local_node_and_date # pyright: ignore
 import re
-from datetime import datetime
+from datetime import datetime, timezone
 
 
 def local_scheme(version):
@@ -11,4 +11,4 @@ def local_scheme(version):
     )
     original_local_version = get_local_node_and_date(version)
     numeric_version = original_local_version.replace('+', '').replace('.d', '')
-    return datetime.utcnow().strftime('%y%m%d') + numeric_version
+    return datetime.now(timezone.utc).strftime('%y%m%d') + numeric_version

tests/flows/results/test_wireless/wireless_flow/window-rpi-001.hash

@@ -1,2 +1,2 @@
 // window-rpi-001
-37ce32a12a80915276ecbf3176fdf9162f152f74a9f67e34e1df640f48e3f741
+80d7df0e74ea16d07c0fb984760f5e350ce2ecc0781f18978019ea833206e795

tests/flows/results/test_wireless/wireless_flow/window-rpi-002.hash

@@ -1,2 +1,2 @@
 // window-rpi-002
-37ce32a12a80915276ecbf3176fdf9162f152f74a9f67e34e1df640f48e3f741
+80d7df0e74ea16d07c0fb984760f5e350ce2ecc0781f18978019ea833206e795

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

@@ -1106,7 +1106,10 @@
     "is_pending": false,
     "status": null
   },
-  "web_ui": null,
+  "web_ui": {
+    "_type": "WebUIState",
+    "active_inputs": []
+  },
   "wifi": {
     "_type": "WiFiState",
     "connections": [

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

@@ -1176,7 +1176,10 @@
     "is_pending": false,
     "status": null
   },
-  "web_ui": null,
+  "web_ui": {
+    "_type": "WebUIState",
+    "active_inputs": []
+  },
   "wifi": {
     "_type": "WiFiState",
     "connections": [],

ubo_app/rpc/service.py

@@ -20,8 +20,7 @@
 from ubo_app.rpc.generated.ubo.v1 import Event
 from ubo_app.rpc.message_to_object import get_class, rebuild_object, reduce_group
 from ubo_app.rpc.object_to_message import build_message
-from ubo_app.store.main import store
-from ubo_app.store.operations import UboAction, UboEvent
+from ubo_app.store.main import UboAction, UboEvent, store
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator

ubo_app/services/010-voice/setup.py

@@ -77,6 +77,7 @@ async def act() -> None:
                     '.*',
                     prompt='Convert the Picovoice access key to a QR code and '
                     'scan it.',
+                    title='Picovoice Access Key',
                 )
             )[0]
             secrets.write_secret(key=PICOVOICE_ACCESS_KEY, value=access_key)

ubo_app/services/030-wifi/pages/create_wireless_connection.py

@@ -57,7 +57,7 @@ async def create_wireless_connection(self: CreateWirelessConnectionPage) -> None
         try:
             _, match = await qrcode_input(
                 BARCODE_PATTERN,
-                prompt='Scan WiFi QR-Code With Front Camera',
+                prompt='Enter WiFi connection',
                 extra_information=NotificationExtraInformation(
                     text='Go to your phone settings, choose QR code and hold it in '
                     'front of the camera to scan it.',

ubo_app/services/040-camera/reducer.py

@@ -11,20 +11,21 @@
     ReducerResult,
 )
 
+from ubo_app.store.operations import (
+    InputDemandAction,
+    InputProvideAction,
+    InputProvideEvent,
+)
 from ubo_app.store.services.camera import (
-    CameraAction,
-    CameraBarcodeEvent,
     CameraEvent,
     CameraReportBarcodeAction,
-    CameraStartViewfinderAction,
     CameraStartViewfinderEvent,
     CameraState,
     CameraStopViewfinderEvent,
-    InputDescription,
 )
 from ubo_app.store.services.keypad import Key, KeypadKeyPressAction
 
-Action = InitAction | CameraAction
+Action = InitAction | InputDemandAction
 
 
 def pop_queue(state: CameraState) -> CameraState:
@@ -41,28 +42,45 @@ def pop_queue(state: CameraState) -> CameraState:
 def reducer(
     state: CameraState | None,
     action: Action,
-) -> ReducerResult[CameraState, Action, CameraEvent]:
+) -> ReducerResult[CameraState, Action, CameraEvent | InputProvideEvent]:
     if state is None:
         if isinstance(action, InitAction):
             return CameraState(is_viewfinder_active=False, queue=[])
         raise InitializationActionError(action)
 
-    if isinstance(action, CameraStartViewfinderAction):
+    if isinstance(action, InputDemandAction):
         if state.is_viewfinder_active:
             return replace(
                 state,
                 queue=[
                     *state.queue,
-                    InputDescription(id=action.id, pattern=action.pattern),
+                    action.description,
                 ],
             )
         return CompleteReducerResult(
             state=replace(
                 state,
                 is_viewfinder_active=True,
-                current=InputDescription(id=action.id, pattern=action.pattern),
+                current=action.description,
             ),
-            events=[CameraStartViewfinderEvent(pattern=action.pattern)],
+            events=[CameraStartViewfinderEvent(pattern=action.description.pattern)],
+        )
+
+    if isinstance(action, InputProvideAction):
+        if state.current and state.current.id == action.id:
+            return CompleteReducerResult(
+                state=pop_queue(state),
+                events=[
+                    CameraStopViewfinderEvent(id=state.current.id),
+                ],
+            )
+        return replace(
+            state,
+            queue=[
+                description
+                for description in state.queue
+                if description.id != action.id
+            ],
         )
 
     if isinstance(action, CameraReportBarcodeAction) and state.current:
@@ -73,10 +91,10 @@ def reducer(
                     return CompleteReducerResult(
                         state=pop_queue(state),
                         events=[
-                            CameraBarcodeEvent(
+                            InputProvideEvent(
                                 id=state.current.id,
-                                code=code,
-                                group_dict=match.groupdict(),
+                                value=code,
+                                data=match.groupdict(),
                             ),
                             CameraStopViewfinderEvent(id=None),
                         ],
@@ -85,10 +103,10 @@ def reducer(
                 return CompleteReducerResult(
                     state=pop_queue(state),
                     events=[
-                        CameraBarcodeEvent(
+                        InputProvideEvent(
                             id=state.current.id,
-                            code=code,
-                            group_dict=None,
+                            value=code,
+                            data=None,
                         ),
                         CameraStopViewfinderEvent(id=None),
                     ],

ubo_app/services/040-camera/setup.py

@@ -12,7 +12,6 @@
 import png
 from debouncer import DebounceOptions, debounce
 from kivy.clock import Clock, mainthread
-from pyzbar.pyzbar import decode
 from typing_extensions import override
 from ubo_gui.page import PageWidget
 
@@ -115,6 +114,8 @@ def feed_viewfinder(picamera2: Picamera2 | None) -> None:
         data = None
 
     if data is not None:
+        from pyzbar.pyzbar import decode
+
         barcodes = decode(data)
         if len(barcodes) > 0:
             create_task(

ubo_app/services/050-vscode/commands.py

@@ -102,7 +102,7 @@ async def check_status() -> None:
                     ),
                 ),
             )
-    logger.info(
+    logger.debug(
         'Checked VSCode Tunnel Status',
         extra={
             'status': status_data,

ubo_app/services/080-docker/setup.py

@@ -5,7 +5,7 @@
 import asyncio
 import contextlib
 import functools
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 import docker
 import docker.errors
@@ -210,8 +210,10 @@ async def act() -> None:
         try:
             credentials = (
                 await qrcode_input(
-                    r'^[^|]*\|[^|]*\|[^|]*$|^[^|]*|[^|]*$',
+                    r'^(?P<Service>[^|]*)\|(?P<Username>[^|]*)\|(?P<Password>[^|]*)$|'
+                    r'(?P<Username_>^[^|]*)|(?P<Password_>[^|]*)$',
                     prompt='Format: [i]SERVICE|USERNAME|PASSWORD[/i]',
+                    title='Enter Docker Credentials',
                     extra_information=NotificationExtraInformation(
                         text="""To generate your QR code for login, format your \
 details by separating your service, username, and password with the pipe symbol. For \
@@ -231,15 +233,15 @@ async def act() -> None:
 default.""",
                     ),
                 )
-            )[0]
-            if credentials.count('|') == 1:
-                username, password = credentials.split('|')
-                registry = 'docker.io'
-            else:
-                registry, username, password = credentials.split('|')
-            registry = registry.strip()
-            username = username.strip()
-            password = password.strip()
+            )[1]
+            if not credentials:
+                return
+            username = credentials.get('Username', credentials.get('Username_', ''))
+            password = credentials.get('Password', credentials.get('Password_', ''))
+            registry = credentials.get('Service', 'docker.io')
+            username = cast(str, username).strip()
+            password = cast(str, password).strip()
+            registry = cast(str, registry).strip()
             docker_client = docker.from_env()
             docker_client.login(
                 username=username,

ubo_app/services/090-web-ui/reducer.py

@@ -1,15 +1,45 @@
 # ruff: noqa: D100, D101, D102, D103, D104, D107
 from __future__ import annotations
 
+from dataclasses import replace
 from typing import TYPE_CHECKING
 
+from redux import InitAction, InitializationActionError
+
+from ubo_app.store.operations import (
+    InputAction,
+    InputDemandAction,
+    InputProvideAction,
+)
+from ubo_app.store.services.web_ui import WebUIState
+
 if TYPE_CHECKING:
-    from redux import BaseAction, ReducerResult
+    from redux import ReducerResult
 
 
 def reducer(
-    state: None,
-    action: BaseAction,
-) -> ReducerResult[None, None, None]:
-    _ = action
+    state: WebUIState | None,
+    action: InputAction,
+) -> WebUIState | ReducerResult[WebUIState, None, None]:
+    if state is None:
+        if isinstance(action, InitAction):
+            return WebUIState(active_inputs=[])
+        raise InitializationActionError(action)
+
+    if isinstance(action, InputDemandAction):
+        return replace(
+            state,
+            active_inputs=[*state.active_inputs, action.description],
+        )
+
+    if isinstance(action, InputProvideAction):
+        return replace(
+            state,
+            active_inputs=[
+                description
+                for description in state.active_inputs
+                if description.id != action.id
+            ],
+        )
+
     return state

ubo_app/services/090-web-ui/setup.py

@@ -1,24 +1,53 @@
 """Implementation of the web-ui service."""
 
 import asyncio
+import re
 from pathlib import Path
 
-from quart import Quart
+from quart import Quart, render_template, request
 from redux import FinishEvent
 
 from ubo_app.constants import WEB_UI_DEBUG_MODE, WEB_UI_LISTEN_HOST, WEB_UI_LISTEN_PORT
 from ubo_app.store.main import store
+from ubo_app.store.operations import InputDescription, InputProvideEvent
 
 
 async def init_service() -> None:
     """Initialize the web-ui service."""
-    app = Quart('ubo-app')
-    app.debug = False
+    app = Quart(
+        'ubo-app',
+        template_folder=(Path(__file__).parent / 'templates').absolute().as_posix(),
+    )
+    app.debug = WEB_UI_DEBUG_MODE
     shutdown_event: asyncio.Event = asyncio.Event()
 
-    @app.get('/')
-    async def hello_world() -> str:
-        return (Path(__file__).parent / 'static' / 'index.html').read_text()
+    @store.view(lambda state: state.web_ui.active_inputs)
+    def inputs(inputs: list[InputDescription]) -> list[InputDescription]:
+        return inputs
+
+    @app.route('/', methods=['GET', 'POST'])
+    async def inputs_form() -> str:
+        if request.method == 'POST':
+            data = dict(await request.form)
+            id = data.pop('id')
+            value = data.pop('value', '')
+            store.dispatch(
+                InputProvideEvent(
+                    id=id,
+                    value=value,
+                    data=data,
+                ),
+            )
+            await asyncio.sleep(0.2)
+        return await render_template('index.jinja2', inputs=inputs(), re=re)
+
+    if WEB_UI_DEBUG_MODE:
+
+        @app.errorhandler(Exception)
+        async def handle_error(_: Exception) -> str:
+            import traceback
+
+            return f'<pre>{traceback.format_exc()}</pre>'
 
     store.subscribe_event(FinishEvent, shutdown_event.set)