Refine apps, tools, and dependencies

apps/ble_rpa_tool.py

@@ -52,11 +52,10 @@ def verify_rpa(irk: str, rpa: str) -> None:
         print(color("Not Verified", "red"))
 
 
-def main():
-    cli.add_command(gen_irk)
-    cli.add_command(gen_rpa)
-    cli.add_command(verify_rpa)
-    cli()
+cli.add_command(gen_irk)
+cli.add_command(gen_rpa)
+cli.add_command(verify_rpa)
+main = cli
 
 
 # -----------------------------------------------------------------------------

apps/console.py

@@ -20,14 +20,14 @@
 # Imports
 # -----------------------------------------------------------------------------
 import asyncio
+import datetime
 import logging
 import os
 import re
 from collections import OrderedDict
+from typing import Any
 
 import click
-import humanize
-from prettytable import PrettyTable
 from prompt_toolkit import Application
 from prompt_toolkit.completion import Completer, Completion, NestedCompleter
 from prompt_toolkit.data_structures import Point
@@ -121,6 +121,55 @@ def parse_phys(phys):
     return phy_list
 
 
+def natural_time(dt: datetime.datetime) -> str:
+    now = datetime.datetime.now()
+    if dt > now:
+        delta = datetime.timedelta(seconds=0)
+    else:
+        delta = now - dt
+    seconds = int(delta.total_seconds())
+    if seconds < 1:
+        return 'now'
+    if seconds < 60:
+        return f"{seconds} second{'s' if seconds != 1 else ''} ago"
+    if seconds < 3600:
+        minutes = seconds // 60
+        return f"{minutes} minute{'s' if minutes != 1 else ''} ago"
+    if seconds < 86400:
+        hours = seconds // 3600
+        return f"{hours} hour{'s' if hours != 1 else ''} ago"
+    days = seconds // 86400
+    return f"{days} day{'s' if days != 1 else ''} ago"
+
+
+def format_table(field_names: list[str], rows: list[list[Any]]) -> str:
+    if not field_names:
+        return ''
+    widths = [len(str(h)) for h in field_names]
+    for row in rows:
+        for i, val in enumerate(row):
+            if i < len(widths):
+                widths[i] = max(widths[i], len(str(val)))
+            else:
+                widths.append(len(str(val)))
+
+    border = '+' + '+'.join('-' * (w + 2) for w in widths) + '+'
+    header = (
+        '| '
+        + ' | '.join(f'{str(h):<{widths[i]}}' for i, h in enumerate(field_names))
+        + ' |'
+    )
+    body = [
+        '| ' + ' | '.join(f'{str(v):<{widths[i]}}' for i, v in enumerate(row)) + ' |'
+        for row in rows
+    ]
+    lines = [border, header, border]
+    if body:
+        lines.extend(body)
+    lines.append(border)
+    return '\n'.join(lines)
+
+
 # -----------------------------------------------------------------------------
 # Console App
 # -----------------------------------------------------------------------------
@@ -579,7 +628,7 @@ async def rssi_monitor_loop(self):
 
     async def command(self, command):
         try:
-            (keyword, *params) = command.strip().split(' ')
+            keyword, *params = command.strip().split(' ')
             keyword = keyword.replace('-', '_').lower()
             handler = getattr(self, f'do_{keyword}', None)
             if handler:
@@ -745,7 +794,6 @@ async def do_show(self, params):
             await asyncio.sleep(1)
 
     async def do_show_local_values(self):
-        prettytable = PrettyTable()
         field_names = ["Service", "Characteristic", "Descriptor"]
 
         # if there's no connections, add a column just for value
@@ -756,6 +804,7 @@ async def do_show_local_values(self):
         for connection in self.device.connections.values():
             field_names.append(f"Connection {connection.handle}")
 
+        rows = []
         for attribute in self.device.gatt_server.attributes:
             if isinstance(attribute, Characteristic):
                 service = self.device.gatt_server.get_attribute_group(
@@ -769,7 +818,7 @@ async def do_show_local_values(self):
                 ]
                 if not values:
                     values = [attribute.read_value(None)]
-                prettytable.add_row([f"{service.uuid}", attribute.uuid, ""] + values)
+                rows.append([f"{service.uuid}", attribute.uuid, ""] + values)
 
             elif isinstance(attribute, Descriptor):
                 service = self.device.gatt_server.get_attribute_group(
@@ -791,25 +840,23 @@ async def do_show_local_values(self):
 
                 # TODO: future optimization: convert CCCD value to human readable string
 
-                prettytable.add_row(
+                rows.append(
                     [service.uuid, characteristic.uuid, attribute.type] + values
                 )
 
-        prettytable.field_names = field_names
-        self.local_values_text.text = prettytable.get_string()
+        self.local_values_text.text = format_table(field_names, rows)
         self.ui.invalidate()
 
     async def do_show_remote_values(self):
-        prettytable = PrettyTable(
-            field_names=[
-                "Connection",
-                "Service",
-                "Characteristic",
-                "Descriptor",
-                "Time",
-                "Value",
-            ]
-        )
+        field_names = [
+            "Connection",
+            "Service",
+            "Characteristic",
+            "Descriptor",
+            "Time",
+            "Value",
+        ]
+        rows = []
         for connection in self.device.connections.values():
             for handle, (time, value) in connection.gatt_client.cached_values.items():
                 row = [connection.handle]
@@ -827,10 +874,10 @@ async def do_show_remote_values(self):
                 else:
                     continue
 
-                row.extend([humanize.naturaltime(time), value])
-                prettytable.add_row(row)
+                row.extend([natural_time(time), value])
+                rows.append(row)
 
-        self.remote_values_text.text = prettytable.get_string()
+        self.remote_values_text.text = format_table(field_names, rows)
         self.ui.invalidate()
 
     async def do_get_phy(self, _):

apps/l2cap_bridge.py

@@ -355,7 +355,11 @@ def client(context, bluetooth_address, tcp_host, tcp_port):
     asyncio.run(run(context.obj['device_config'], context.obj['hci_transport'], bridge))
 
 
-# -----------------------------------------------------------------------------
-if __name__ == '__main__':
+def main() -> None:
     bumble.logging.setup_basic_logging('WARNING')
     cli(obj={})  # pylint: disable=no-value-for-parameter
+
+
+# -----------------------------------------------------------------------------
+if __name__ == '__main__':
+    main()

apps/lea_unicast/app.py

@@ -25,15 +25,16 @@
 import pathlib
 import wave
 import weakref
-from importlib import resources
 
 try:
     import lc3  # type: ignore  # pylint: disable=E0401
 except ImportError as e:
     raise ImportError("Try `python -m pip install \".[auracast]\"`.") from e
 
-import aiohttp.web
 import click
+import websockets.asyncio.server
+import websockets.exceptions
+import websockets.http11
 
 import bumble
 import bumble.logging
@@ -159,70 +160,97 @@ async def lc3_source_task(
 class UiServer:
     speaker: weakref.ReferenceType[Speaker]
     port: int
+    channel_socket: websockets.asyncio.server.ServerConnection | None
+    server: websockets.asyncio.server.Server | None
 
     def __init__(self, speaker: Speaker, port: int) -> None:
         self.speaker = weakref.ref(speaker)
         self.port = port
         self.channel_socket = None
+        self.server = None
 
     async def start_http(self) -> None:
         """Start the UI HTTP server."""
-
-        app = aiohttp.web.Application()
-        app.add_routes(
-            [
-                aiohttp.web.get('/', self.get_static),
-                aiohttp.web.get('/index.html', self.get_static),
-                aiohttp.web.get('/channel', self.get_channel),
-            ]
+        self.server = await websockets.asyncio.server.serve(
+            self.get_channel,
+            '127.0.0.1',
+            self.port,
+            process_request=self.process_request,
         )
-
-        runner = aiohttp.web.AppRunner(app)
-        await runner.setup()
-        site = aiohttp.web.TCPSite(runner, 'localhost', self.port)
+        if self.port == 0 and self.server.sockets:
+            self.port = self.server.sockets[0].getsockname()[1]
         print('UI HTTP server at ' + color(f'http://127.0.0.1:{self.port}', 'green'))
-        await site.start()
 
-    async def get_static(self, request):
+    async def close(self) -> None:
+        if self.server:
+            self.server.close()
+            await self.server.wait_closed()
+
+    async def process_request(
+        self,
+        connection: websockets.asyncio.server.ServerConnection,
+        request: websockets.asyncio.server.Request,
+    ) -> websockets.http11.Response | None:
         path = request.path
-        if path == '/':
+        if path == '/channel':
+            return None
+
+        if path in ('', '/'):
             path = '/index.html'
+
         if path.endswith('.html'):
-            content_type = 'text/html'
+            content_type = 'text/html; charset=utf-8'
         elif path.endswith('.js'):
-            content_type = 'text/javascript'
+            content_type = 'text/javascript; charset=utf-8'
         elif path.endswith('.css'):
-            content_type = 'text/css'
+            content_type = 'text/css; charset=utf-8'
         elif path.endswith('.svg'):
             content_type = 'image/svg+xml'
         else:
-            content_type = 'text/plain'
-        text = (
-            resources.files("bumble.apps.lea_unicast")
-            .joinpath(pathlib.Path(path).relative_to('/'))
-            .read_text(encoding="utf-8")
-        )
-        return aiohttp.web.Response(text=text, content_type=content_type)
+            content_type = 'text/plain; charset=utf-8'
 
-    async def get_channel(self, request):
-        ws = aiohttp.web.WebSocketResponse()
-        await ws.prepare(request)
+        try:
+            body = (pathlib.Path(__file__).parent / path.lstrip('/')).read_bytes()
+            return websockets.http11.Response(
+                status_code=200,
+                reason_phrase='OK',
+                headers=websockets.http11.Headers(
+                    [
+                        ('Content-Type', content_type),
+                        ('Content-Length', str(len(body))),
+                    ]
+                ),
+                body=body,
+            )
+        except Exception:
+            body = b'Not Found'
+            return websockets.http11.Response(
+                status_code=404,
+                reason_phrase='Not Found',
+                headers=websockets.http11.Headers(
+                    [
+                        ('Content-Type', 'text/plain; charset=utf-8'),
+                        ('Content-Length', str(len(body))),
+                    ]
+                ),
+                body=body,
+            )
 
+    async def get_channel(self, ws: websockets.asyncio.server.ServerConnection) -> None:
         # Process messages until the socket is closed.
         self.channel_socket = ws
-        async for message in ws:
-            if message.type == aiohttp.WSMsgType.TEXT:
-                logger.debug(f'<<< received message: {message.data}')
-                await self.on_message(message.data)
-            elif message.type == aiohttp.WSMsgType.ERROR:
-                logger.debug(
-                    f'channel connection closed with exception {ws.exception()}'
-                )
-
-        self.channel_socket = None
-        logger.debug('--- channel connection closed')
-
-        return ws
+        try:
+            async for message in ws:
+                if isinstance(message, str):
+                    logger.debug(f'<<< received message: {message}')
+                    await self.on_message(message)
+                else:
+                    logger.debug(f'<<< received binary message: {len(message)} bytes')
+        except websockets.exceptions.ConnectionClosed as error:
+            logger.debug(f'channel connection closed: {error}')
+        finally:
+            self.channel_socket = None
+            logger.debug('--- channel connection closed')
 
     async def on_message(self, message_str: str):
         # Parse the message as JSON
@@ -231,18 +259,21 @@ async def on_message(self, message_str: str):
         # Dispatch the message
         message_type = message['type']
         message_params = message.get('params', {})
-        handler = getattr(self, f'on_{message_type}_message')
+        handler = getattr(self, f'on_{message_type}_message', None)
         if handler:
             await handler(**message_params)
 
     async def on_hello_message(self):
+        speaker = self.speaker()
+        if not speaker:
+            return
         await self.send_message(
             'hello',
             bumble_version=bumble.__version__,
-            codec=self.speaker().codec,
-            streamState=self.speaker().stream_state.name,
+            codec=speaker.codec,
+            streamState=speaker.stream_state.name,
         )
-        if connection := self.speaker().connection:
+        if connection := speaker.connection:
             await self.send_message(
                 'connection',
                 peer_address=connection.peer_address.to_string(False),
@@ -254,14 +285,14 @@ async def send_message(self, message_type: str, **kwargs) -> None:
             return
 
         message = {'type': message_type, 'params': kwargs}
-        await self.channel_socket.send_json(message)
+        await self.channel_socket.send(json.dumps(message))
 
     async def send_audio(self, data: bytes) -> None:
         if self.channel_socket is None:
             return
 
         try:
-            await self.channel_socket.send_bytes(data)
+            await self.channel_socket.send(data)
         except Exception as error:
             logger.warning(f'exception while sending audio packet: {error}')