pb-service: aio-rfcomm 1.1, and say when macOS is asking for Bluetooth

packages/pb-cli/src/pb_cli/commands/hub.py

@@ -14,16 +14,19 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import AsyncIterator, Mapping
+import sys
+from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Annotated, Final
 
+import anyio
 import cyclopts
 from rich.console import Console, RenderableType
 
 import pb_client
 from pb_cli import errors, keys, picker, ui
 from pb_cli.app import HUB_COMMANDS, Verbose, app, configure_logging
-from pb_cli.errors import EXIT_OK
+from pb_cli.errors import EXIT_OK, PERMISSION_NAME
 from pb_core.jsonrpc import ErrorCode
 
 logger = logging.getLogger(__name__)
@@ -93,6 +96,63 @@ def _connect_hints(transport: str | None) -> Mapping[int, str]:
     return {ErrorCode.HUB_NOT_FOUND: hint}
 
 
+def _asked(platform: str = sys.platform) -> str:
+    """Say that the system is asking whether the service may use Bluetooth.
+
+    Args:
+        platform: What to say it for. Only macOS asks, today.
+
+    Returns:
+        The sentence, with markup.
+    """
+    who = "macOS" if platform == "darwin" else "The system"
+    return (
+        f"{who} is asking whether [bold]{PERMISSION_NAME}[/] may use "
+        "Bluetooth. Answer the dialog to carry on."
+    )
+
+
+@asynccontextmanager
+async def _telling_when_asked(
+    service: pb_client.Service, say: Callable[[bool], None]
+) -> AsyncGenerator[None, None]:
+    """Say when the system starts or stops asking for Bluetooth, while hubs are sought.
+
+    Open it before the scan or connection it is for, so that nothing it reports
+    is missed. The search itself carries on once the user answers, so this only
+    has to tell them to; what changes is whether they look for the dialog now
+    or are told to answer it once the search has given up.
+
+    Args:
+        service: Where the search runs.
+        say: Called with True when the system starts asking, and with False if
+            it stops before the search ends.
+
+    Yields:
+        Nothing. The block is the search.
+    """
+
+    async def follow(
+        changes: AsyncIterator[tuple[pb_client.UnusableAdapter, ...]],
+    ) -> None:
+        asked = False
+
+        async for unusable in changes:
+            now = any(adapter.awaiting_permission for adapter in unusable)
+
+            if now != asked:
+                asked = now
+                say(asked)
+
+    async with service.watching_adapters() as changes, anyio.create_task_group() as tg:
+        tg.start_soon(follow, changes)
+
+        try:
+            yield
+        finally:
+            tg.cancel_scope.cancel()
+
+
 def _looks_like_id(value: str) -> bool:
     """Whether a string is a hub id rather than a hub name."""
     return value.startswith(_ID_PREFIXES)
@@ -134,32 +194,55 @@ async def _scan(seconds: float, name: str | None, transport: str | None) -> int:
     seen: dict[str, pb_client.DeviceInfo] = {}
     caption = f"Scanning for {seconds:.0f}s..."
 
-    async with (
-        pb_client.connect() as service,
-        service.scanning(seconds=seconds, name=name, transport=transport) as devices,
-    ):
+    def scanning(
+        service: pb_client.Service,
+    ) -> AbstractAsyncContextManager[AsyncIterator[pb_client.DeviceInfo]]:
+        return service.scanning(seconds=seconds, name=name, transport=transport)
+
+    async with pb_client.connect() as service:
         if out.is_terminal:
             # Redrawn in place as advertisements arrive, so a hub appears the
             # moment it is heard and its signal moves as it is carried about.
             with ui.live(ui.scan_table((), caption), out) as display:
-                async for device in devices:
-                    seen[device.id] = device
+
+                def say(asked: bool) -> None:
+                    # In the caption, not printed: this display is on stdout,
+                    # and a line on stderr would land in the middle of it.
+                    nonlocal caption
+                    caption = _asked() if asked else f"Scanning for {seconds:.0f}s..."
                     display.update(ui.scan_table(seen.values(), caption), refresh=True)
 
+                async with (
+                    _telling_when_asked(service, say),
+                    scanning(service) as devices,
+                ):
+                    async for device in devices:
+                        seen[device.id] = device
+                        display.update(
+                            ui.scan_table(seen.values(), caption), refresh=True
+                        )
+
                 display.update(ui.scan_table(seen.values(), f"Found {len(seen)}."))
         else:
-            # Piped: one line per hub, printed the moment it is first heard, so
-            # that "pb scan | head -1" answers as soon as there is an answer.
-            async for device in devices:
-                if device.id not in seen:
-                    line = (
-                        f"{device.name}\t{device.id}"
-                        if ui.SHOW_HUB_IDS
-                        else device.name
-                    )
-                    ui.plain(out, line)
 
-                seen[device.id] = device
+            def tell(asked: bool) -> None:
+                if asked:
+                    err.print(_asked())
+
+            async with _telling_when_asked(service, tell), scanning(service) as devices:
+                # Piped: one line per hub, printed the moment it is first
+                # heard, so that "pb scan | head -1" answers as soon as there
+                # is an answer.
+                async for device in devices:
+                    if device.id not in seen:
+                        line = (
+                            f"{device.name}\t{device.id}"
+                            if ui.SHOW_HUB_IDS
+                            else device.name
+                        )
+                        ui.plain(out, line)
+
+                    seen[device.id] = device
 
     if not seen:
         if transport == "rfcomm":
@@ -225,11 +308,20 @@ async def connect_cmd(
 async def _connect(hub: str | None, scan: bool, transport: str | None) -> int:
     err = ui.console(err=True)
 
+    def tell(asked: bool) -> None:
+        # Printed above the spinner or the chooser, both of which are drawn
+        # on this console and so make room for it.
+        if asked:
+            err.print(_asked())
+
     async with pb_client.connect() as service:
         if scan:
             # No time limit: the hub may still be in the next room. The scan
             # ends when the block does, whether the user chose or gave up.
-            async with service.scanning(seconds=None, transport=transport) as devices:
+            async with (
+                _telling_when_asked(service, tell),
+                service.scanning(seconds=None, transport=transport) as devices,
+            ):
                 device = await picker.choose_hub(err, devices, ui.chooser_table)
 
             if device is None:
@@ -243,11 +335,12 @@ async def _connect(hub: str | None, scan: bool, transport: str | None) -> int:
                 # chosen the route, since each row is one.
         else:
             with ui.spinner(err, "Connecting..."):
-                info = await service.connect_hub(
-                    hub_id=hub if hub and _looks_like_id(hub) else None,
-                    name=hub if hub and not _looks_like_id(hub) else None,
-                    transport=transport,
-                )
+                async with _telling_when_asked(service, tell):
+                    info = await service.connect_hub(
+                        hub_id=hub if hub and _looks_like_id(hub) else None,
+                        name=hub if hub and not _looks_like_id(hub) else None,
+                        transport=transport,
+                    )
 
     # The protocol version belongs to pb status: it explains why a command was
     # refused, but on a successful connection it is one more number to read

packages/pb-cli/src/pb_cli/errors.py

@@ -52,7 +52,9 @@
     "EXIT_INTERRUPTED",
     "EXIT_NOTHING_CONNECTED",
     "EXIT_OK",
+    "PERMISSION_NAME",
     "Problem",
+    "bluetooth_advice",
     "explain",
     "find",
     "flatten",
@@ -154,6 +156,70 @@ def serial_port_advice(platform: str = sys.platform) -> str:
     )
 
 
+PERMISSION_NAME = "Pybricks RFCOMM Service"
+"""What macOS calls this program when it asks about Bluetooth.
+
+pb-service's ``PERMISSION_NAME``, which it passes to aio-rfcomm's
+``prompt_under_name()``, and so the one in the dialog and under Privacy &
+Security. Repeated rather than imported, so that explaining an error does not
+load the service; a test holds the two together.
+"""
+
+_BLUETOOTH = (
+    "Switch Bluetooth on, or plug in an adapter, and check that this "
+    "computer lets pb use Bluetooth."
+)
+"""What to suggest when the service could not say why Bluetooth is unusable."""
+
+
+def bluetooth_advice(availability: str | None, platform: str = sys.platform) -> str:
+    """Say what to do about Bluetooth that cannot be used.
+
+    Args:
+        availability: Why, as the service names it -- one word per member of
+            aio-rfcomm's ``Availability`` -- or ``None`` if it did not say.
+        platform: What to advise for. An argument for the same reason as in
+            :func:`serial_port_advice`.
+
+    Returns:
+        What to suggest.
+    """
+    mac = platform == "darwin"
+
+    match availability:
+        case "switched_off":
+            return "Switch Bluetooth on, then try again."
+        case "no_adapter":
+            return "Plug in a Bluetooth adapter, then try again."
+        case "awaiting_permission":
+            # Only once the service has given up waiting: until then the CLI
+            # says this as it happens, and carries on when it is answered.
+            what = "the dialog" if mac else "the system's question"
+            return (
+                f"Answer {what} asking whether [bold]{PERMISSION_NAME}[/] may "
+                "use Bluetooth, then try again."
+            )
+        case "not_allowed" if mac:
+            # Conditional, because a profile or parental controls refuse it
+            # in the same words, and those are not the user's to undo.
+            return (
+                f"If you refused it for [bold]{PERMISSION_NAME}[/], allow it "
+                "in System Settings under Privacy & Security > Bluetooth, "
+                "then try again."
+            )
+        case "not_allowed":
+            return "Check that this computer lets pb use Bluetooth."
+        case "service_unavailable":
+            return (
+                "The system's Bluetooth service usually restarts by itself. "
+                "Wait a moment, then try again."
+            )
+        case "unsupported":
+            return "If the hub has a USB port, connect it with a cable instead."
+        case _:
+            return _BLUETOOTH
+
+
 _RULES: Mapping[int, _Rule] = {
     # Hubs. The service's own wording is specific -- "no connected hub named
     # 'Spare Hub'" -- so it is usually kept as the headline and only the
@@ -202,12 +268,9 @@ def serial_port_advice(platform: str = sys.platform) -> str:
     ),
     # Transport and lifecycle.
     # The service repeats what the system said, which is what tells a radio
-    # that is switched off from one this program may not use.
-    ErrorCode.BLUETOOTH_UNAVAILABLE: _Rule(
-        "Switch Bluetooth on, or plug in an adapter, and check that this "
-        "computer lets pb use Bluetooth.",
-        exit_code=EXIT_ERROR,
-    ),
+    # that is switched off from one this program may not use, and names it in
+    # a word, which is what chooses the suggestion (see bluetooth_advice).
+    ErrorCode.BLUETOOTH_UNAVAILABLE: _Rule(_BLUETOOTH, exit_code=EXIT_ERROR),
     ErrorCode.SCAN_TIMEOUT: _Rule(
         "Switch the hub on and press its button so that its light flashes."
     ),
@@ -373,6 +436,24 @@ def _slot_hint(exc: JsonRpcError, default: str | None) -> str | None:
     return _ONE_SLOT if data.get("slots") == 1 else default
 
 
+def _bluetooth_hint(exc: JsonRpcError) -> str:
+    """Suggest what to do about Bluetooth, by the reason the service gave.
+
+    Args:
+        exc: The error the service sent.
+
+    Returns:
+        The suggestion to print.
+    """
+    payload: object = exc.data
+    if not isinstance(payload, Mapping):
+        return bluetooth_advice(None)
+
+    data = cast("Mapping[str, object]", payload)
+    availability = data.get("availability")
+    return bluetooth_advice(availability if isinstance(availability, str) else None)
+
+
 def _describe_rpc(exc: JsonRpcError, hints: Mapping[int, str] | None) -> Problem | None:
     """Present an error the service reported, or ``None`` if the code is unknown."""
     rule = _RULES.get(exc.code)
@@ -391,6 +472,9 @@ def _describe_rpc(exc: JsonRpcError, hints: Mapping[int, str] | None) -> Problem
     if exc.code == ErrorCode.PROGRAM_SLOT_NOT_SELECTED:
         hint = _slot_hint(exc, hint)
 
+    if exc.code == ErrorCode.BLUETOOTH_UNAVAILABLE:
+        hint = _bluetooth_hint(exc)
+
     if exc.code == ErrorCode.UNSUPPORTED_PROTOCOL_VERSION:
         message, hint = _version_mismatch(exc, message, hint)
 

packages/pb-cli/tests/test_errors.py

@@ -11,6 +11,7 @@
 
 import asyncio
 import io
+import sys
 from pathlib import Path
 
 import anyio
@@ -439,6 +440,57 @@ def test_unusable_bluetooth_keeps_the_systems_words() -> None:
     assert "Low Energy" not in problems[0].hint
 
 
+@pytest.mark.parametrize(
+    ("availability", "expected"),
+    [
+        ("switched_off", "Switch Bluetooth on"),
+        ("no_adapter", "Plug in a Bluetooth adapter"),
+        # The two only macOS reports, worded for it there.
+        (
+            "not_allowed",
+            "Privacy & Security" if sys.platform == "darwin" else "lets pb use",
+        ),
+        ("awaiting_permission", "Answer the"),
+        ("service_unavailable", "Wait a moment"),
+        ("unsupported", "USB"),
+        (None, "Switch Bluetooth on, or plug in an adapter"),
+        ("something_newer", "Switch Bluetooth on, or plug in an adapter"),
+    ],
+)
+def test_unusable_bluetooth_gets_the_advice_for_its_reason(
+    availability: str | None, expected: str
+) -> None:
+    # One suggestion covering every reason sent somebody whose permission
+    # dialog was open off to switch on a radio that was already on.
+    exc = JsonRpcError(
+        ErrorCode.BLUETOOTH_UNAVAILABLE,
+        "Bluetooth cannot be used: whatever the system said",
+        {"availability": availability} if availability else None,
+    )
+    problems = errors.explain(exc)
+
+    assert problems is not None
+    assert problems[0].hint is not None
+    assert expected in problems[0].hint
+    assert "whatever the system said" in problems[0].message
+
+
+def test_the_name_is_the_one_the_service_asks_under() -> None:
+    from pb_service.backend import PERMISSION_NAME
+
+    assert errors.PERMISSION_NAME == PERMISSION_NAME
+
+
+def test_the_permission_advice_names_what_macos_shows() -> None:
+    # The dialog and System Settings name the service, not pb or Python.
+    for availability in ("awaiting_permission", "not_allowed"):
+        advice = errors.bluetooth_advice(availability, "darwin")
+        assert errors.PERMISSION_NAME in advice
+
+    # Elsewhere there is no System Settings to send anybody to.
+    assert "System Settings" not in errors.bluetooth_advice("not_allowed", "linux")
+
+
 @pytest.mark.parametrize(
     ("platform", "expected"),
     [