feat(core): allow waiting for N4W1 emulator "connection"

"Hold the tag" layout should wait for tap event.

Data read/write should be done later (using a different "progress" layout).

[no changelog]

Author: romanz

core/src/apps/debug/n4w1_mock.py

@@ -2,10 +2,11 @@
 
 from trezor import log, loop
 from trezor.messages import DebugLinkN4W1Read, DebugLinkN4W1Response, DebugLinkN4W1Write
+from trezor.ui import Layout
 
 if TYPE_CHECKING:
     from buffer_types import AnyBytes
-    from typing import Any
+    from typing import Any, Awaitable, Iterator
 
     from trezor.wire.context import Context
     from typing_extensions import Self
@@ -28,6 +29,11 @@ def __exit__(self, exc_type: Any, exc_val: Any, tb: Any) -> None:
         log.debug(__name__, "N4W1 exchange done")
         self.tx.put(None)
 
+    async def connect(self) -> None:
+        """Wait for N4W1 connection notification."""
+        res = await self.rx
+        assert res.value is None
+
     async def read(self, key: str) -> AnyBytes | None:
         """Read a specific entry from N4W1."""
         log.debug(__name__, "N4W1 read: %s", key)
@@ -50,9 +56,51 @@ async def write(self, key: str, value: AnyBytes | None) -> AnyBytes | None:
 
     async def handle(self, ctx: Context) -> None:
         """Called from `apps.debug.dispatch_DebugLinkConnected()`."""
+        self.rx.put(DebugLinkN4W1Response(value=None))  # notify `self.connect()`
         while (req := await self.tx) is not None:
             res = await ctx.call(req, DebugLinkN4W1Response)
             self.rx.put(res)
 
+    def confirm_connect(
+        self, *, title: str, description: str, button: str, br_name: str | None
+    ) -> Awaitable[None]:
+        """Show a layout waiting for N4W1 connection, allowing cancellation."""
+
+        from trezor import TR
+        from trezor.ui.layouts.menu import Menu, confirm_with_menu
+        from trezorui_api import show_info
+
+        self_ctx: N4W1Context = self
+
+        class _Connect(Layout):
+
+            def create_tasks(self) -> Iterator[loop.Task[None]]:
+                from trezor.ui import Shutdown
+                from trezorui_api import CONFIRMED
+
+                async def _task() -> None:
+                    await self_ctx.connect()  # blocks until N4W1 is connected.
+                    try:
+                        # emitting a message raises Shutdown exception
+                        self._emit_message(CONFIRMED)
+                    except Shutdown:
+                        pass
+
+                yield from super().create_tasks()
+                yield _task()
+
+        main = show_info(
+            title=title,
+            description=description,
+            button=(button, False),
+            external_menu=True,
+        )
+        return confirm_with_menu(
+            main,
+            Menu.root(cancel=TR.buttons__cancel),
+            br_name=br_name,
+            layout_type=_Connect,
+        )
+
 
 ctx = N4W1Context()