Improve the support for keyboard events

docs/api.md

@@ -71,6 +71,7 @@ other scheme (`http` or `https`), within about half a minute.
 | Methods | Requires |
 |---|---|
 | `paste_text` | all |
+| `key_down`, `key_up`, `key_tap` | all, see below |
 | `mouse_move_abs`, `mouse_move_rel`, `mouse_down`, `mouse_up`, `mouse_click`, `mouse_scroll` | all |
 | `get_hid_mode`, `set_hid_mode` | non-Pro 2.2.5 / Pro |
 | `reset_hid` | non-Pro / Pro 1.1.6 |
@@ -81,6 +82,37 @@ other scheme (`http` or `https`), within about half a minute.
 
 `paste_text` accepts only the characters in `nanokvm.client.PASTE_CHAR_MAP`.
 
+The key methods send USB HID key codes, such as `KeyCode.F12` or any `int`,
+with modifiers such as `KeyModifier.LEFT_CTRL`:
+
+```python
+from nanokvm.models import KeyCode, KeyModifier
+
+await client.key_tap(
+    KeyCode.DELETE, modifiers=KeyModifier.LEFT_CTRL | KeyModifier.LEFT_ALT
+)
+```
+
+- `key_down` holds up to six keys and replaces the keys held before. `key_up`
+  releases them all. `key_tap` presses and releases, and still releases when
+  it is cancelled.
+- Before non-Pro 2.3.2 and Pro 1.2.6, the device holds one key at a time and
+  applies modifiers only together with a key. Other reports raise
+  `NanoKVMNotSupportedError`.
+- The keyboard and mouse share one WebSocket. While it is open, the client
+  sends a heartbeat every 10 seconds, as the web UI does, so that non-Pro 2.5.0
+  and newer do not close it when idle.
+
+To catch a short boot-menu window, tap a key repeatedly:
+
+```python
+loop = asyncio.get_running_loop()
+end = loop.time() + 8
+while loop.time() < end:
+    await client.key_tap(KeyCode.F12)
+    await asyncio.sleep(0.4)
+```
+
 ### Video and streaming
 
 | Methods | Requires |

nanokvm/components/hid.py

@@ -2,9 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
+import math
+
 from aiohttp import hdrs
 
 from ..compatibility import require_application_version, require_hardware
+from ..exceptions import NanoKVMNotSupportedError
 from ..models.common import (
     AddShortcutReq,
     DeleteShortcutReq,
@@ -15,6 +19,8 @@
     GetShortcutsRsp,
     HidMode,
     HWFamily,
+    KeyCode,
+    KeyModifier,
     MouseJigglerMode,
     PasteReq,
     SetHidModeReq,
@@ -30,6 +36,9 @@
     "[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"
 )
 
+_KEYBOARD_EVENT = 1
+_KEYBOARD_REPORT_KEYS = 6
+
 
 class HidController(Controller):
     """Implement HID operations behind the public client facade."""
@@ -148,3 +157,63 @@ async def paste_text(self, text: str) -> None:
             "/hid/paste",
             data=PasteReq(content=text),
         )
+
+    async def key_down(
+        self, *keys: KeyCode | int, modifiers: KeyModifier | int = 0
+    ) -> None:
+        """Hold keys and modifiers down, replacing those held before.
+
+        Args:
+            keys: Up to six USB HID key codes, such as ``KeyCode.F12``.
+            modifiers: Modifier keys to hold, such as ``KeyModifier.LEFT_CTRL``.
+
+        Before non-Pro 2.3.2 and Pro 1.2.6, the device holds one key at a time
+        and applies modifiers only together with a key.
+        """
+        await self._send_keyboard_report(int(modifiers), [int(key) for key in keys])
+
+    async def key_up(self) -> None:
+        """Release all keys and modifiers."""
+        await self._send_keyboard_report(0, [])
+
+    async def key_tap(
+        self,
+        *keys: KeyCode | int,
+        modifiers: KeyModifier | int = 0,
+        hold: float = 0.05,
+    ) -> None:
+        """Press keys and modifiers, and release them after ``hold`` seconds."""
+        if not math.isfinite(hold) or hold < 0:
+            raise ValueError("hold must be a finite, non-negative number of seconds")
+
+        await self.key_down(*keys, modifiers=modifiers)
+        try:
+            await asyncio.sleep(hold)
+        finally:
+            # Release even when cancelled, so no key stays held on the host.
+            await self.key_up()
+
+    async def _send_keyboard_report(self, modifiers: int, keys: list[int]) -> None:
+        """Send a keyboard report in the wire format of the connected device."""
+        if len(keys) > _KEYBOARD_REPORT_KEYS:
+            raise ValueError("A keyboard report holds at most six keys")
+        if not all(0 <= value <= 0xFF for value in (modifiers, *keys)):
+            raise ValueError("Key codes and modifiers must be between 0 and 255")
+
+        session = self._session
+        if await session.uses_binary_hid_protocol():
+            # Boot keyboard report: modifiers, a reserved byte and six key slots.
+            report = bytes((modifiers, 0, *keys)).ljust(8, bytes(1))
+            await session.send_ws(
+                lambda ws: ws.send_bytes(bytes((_KEYBOARD_EVENT,)) + report)
+            )
+            return
+
+        if len(keys) > 1 or (modifiers and not any(keys)):
+            raise NanoKVMNotSupportedError(
+                "Holding several keys, or modifiers without a key, requires the "
+                "binary keyboard protocol"
+            )
+        # The legacy server combines the four modifier fields into one byte.
+        message = [_KEYBOARD_EVENT, keys[0] if keys else 0, modifiers, 0, 0, 0]
+        await session.send_ws(lambda ws: ws.send_json(message))

nanokvm/components/mouse.py

@@ -3,33 +3,18 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Callable
 import logging
 import math
 
-import aiohttp
-
 from ..exceptions import NanoKVMNotSupportedError
 from ..models.common import MouseButton
 from .session import Controller
 
 _LOGGER = logging.getLogger(__name__)
 
-_BINARY_MOUSE_MIN_NON_PRO_VERSION = "2.3.2"
-_BINARY_MOUSE_MIN_PRO_VERSION = "1.2.6"
-
 
 class MouseController(Controller):
-    """Own the mouse protocol and its lazily-created WebSocket transport."""
-
-    async def _uses_binary_mouse_protocol(self) -> bool:
-        """Select the mouse wire format supported by the connected device."""
-        # Without a token the WebSocket reports the authentication error, so
-        # the current protocol stays the default when detection is impossible.
-        return await self._session.application_version_at_least(
-            non_pro=_BINARY_MOUSE_MIN_NON_PRO_VERSION,
-            pro=_BINARY_MOUSE_MIN_PRO_VERSION,
-        )
+    """Own the mouse protocol on the session's shared WebSocket."""
 
     @staticmethod
     def _require_finite(*values: float) -> None:
@@ -79,21 +64,9 @@ def _report_for_current_mode(self, *, wheel: int = 0) -> bytes:
             return self._absolute_report(wheel)
         return self._relative_report(wheel=wheel)
 
-    async def _send_ws(
-        self,
-        send: Callable[[aiohttp.ClientWebSocketResponse], Awaitable[None]],
-    ) -> None:
-        """Send one mouse message and invalidate the connection on failure."""
-        ws = await self._session.get_ws()
-        try:
-            await send(ws)
-        except (aiohttp.ClientConnectionError, ConnectionError, RuntimeError):
-            await self._session.invalidate_ws(ws)
-            raise
-
     async def _send_mouse_report(self, report: bytes) -> None:
         """Send a binary NanoKVM mouse event and HID report."""
-        await self._send_ws(lambda ws: ws.send_bytes(bytes((2,)) + report))
+        await self._session.send_ws(lambda ws: ws.send_bytes(bytes((2,)) + report))
 
     async def _send_legacy_mouse_event(
         self, event_type: int, button_state: int, x: float, y: float
@@ -114,7 +87,7 @@ async def _send_legacy_mouse_event(
 
         message = [2, event_type, button_state, x_value, y_value]
         _LOGGER.debug("Sending legacy mouse event: %s", message)
-        await self._send_ws(lambda ws: ws.send_json(message))
+        await self._session.send_ws(lambda ws: ws.send_json(message))
 
     async def mouse_move_abs(self, x: float, y: float) -> None:
         """Move mouse to absolute position.
@@ -125,7 +98,7 @@ async def mouse_move_abs(self, x: float, y: float) -> None:
         """
         self._require_finite(x, y)
         session = self._session
-        if not await self._uses_binary_mouse_protocol():
+        if not await self._session.uses_binary_hid_protocol():
             await self._send_legacy_mouse_event(2, 0, x, y)
             return
 
@@ -142,7 +115,7 @@ async def mouse_move_rel(self, dx: float, dy: float) -> None:
         """
         self._require_finite(dx, dy)
         session = self._session
-        if not await self._uses_binary_mouse_protocol():
+        if not await self._session.uses_binary_hid_protocol():
             await self._send_legacy_mouse_event(3, 0, dx, dy)
             return
 
@@ -161,7 +134,7 @@ async def mouse_down(self, button: MouseButton = MouseButton.LEFT) -> None:
         """
 
         session = self._session
-        if not await self._uses_binary_mouse_protocol():
+        if not await self._session.uses_binary_hid_protocol():
             if button in (MouseButton.BACK, MouseButton.FORWARD):
                 raise NanoKVMNotSupportedError(
                     "Back and Forward mouse buttons require the binary mouse protocol"
@@ -192,7 +165,7 @@ async def mouse_up(self) -> None:
         The report releases all currently held buttons.
         """
         session = self._session
-        if not await self._uses_binary_mouse_protocol():
+        if not await self._session.uses_binary_hid_protocol():
             await self._send_legacy_mouse_event(0, 0, 0.0, 0.0)
             return
 
@@ -239,7 +212,7 @@ async def mouse_scroll(self, dx: float, dy: float) -> None:
                 negative scrolls down.
         """
         self._require_finite(dx, dy)
-        if not await self._uses_binary_mouse_protocol():
+        if not await self._session.uses_binary_hid_protocol():
             await self._send_legacy_mouse_event(4, 0, dx, dy)
             return
 

nanokvm/components/session.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Awaitable, Callable
 import contextlib
 from http.cookies import Morsel, SimpleCookie
 import json
@@ -14,6 +14,7 @@
 from typing import Any, Literal, NoReturn, TypeVar, overload
 
 from aiohttp import (
+    ClientConnectionError,
     ClientResponse,
     ClientResponseError,
     ClientSession,
@@ -59,6 +60,13 @@
 _SESSION_COOKIE_NAME = "nano-kvm-token"
 _NOT_STARTED = "NanoKVMClient is not started; use 'async with NanoKVMClient(...)'"
 _CURRENT_PASSWORD_MIN_NON_PRO_VERSION = (2, 5, 1)
+_BINARY_HID_MIN_NON_PRO_VERSION = "2.3.2"
+_BINARY_HID_MIN_PRO_VERSION = "1.2.6"
+_WS_SEND_ERRORS = (ClientConnectionError, ConnectionError, RuntimeError)
+# Non-Pro 2.5.0 and newer close a WebSocket that sends nothing for 90 seconds.
+# The web UI sends this heartbeat every 10 seconds.
+_WS_HEARTBEAT = bytes((0,))
+_WS_HEARTBEAT_INTERVAL = 10.0
 
 
 class SessionController:
@@ -85,6 +93,7 @@ def __init__(
         self._request_timeout = request_timeout
         self._image_transfer_lock = asyncio.Lock()
         self._ws: ClientWebSocketResponse | None = None
+        self._ws_heartbeat: asyncio.Task[None] | None = None
         self._ws_session: ClientSession | None = None
         self._ws_lock = asyncio.Lock()
         self._mouse_buttons = 0
@@ -444,8 +453,7 @@ async def clear_local_session(
             self.forget_versions()
             self._mouse_buttons = 0
             self.clear_session_cookies()
-            ws = self._ws
-            self._ws = None
+            ws = self._detach_ws()
         # Detach state atomically, then close only the old transport. Network I/O
         # must not block a new login or close its replacement WebSocket.
         if ws is not None and not ws.closed:
@@ -643,39 +651,82 @@ async def get_hardware(self) -> GetHardwareRsp:
     async def close_ws(self) -> None:
         """Close and forget the current WebSocket connection."""
         async with self._ws_lock:
-            ws = self._ws
-            self._ws = None
+            ws = self._detach_ws()
             if ws is not None and not ws.closed:
                 await ws.close()
 
     async def invalidate_ws(self, ws: ClientWebSocketResponse) -> None:
         """Forget a failed WebSocket without closing a replacement connection."""
         async with self._ws_lock:
             if self._ws is ws:
-                self._ws = None
+                self._detach_ws()
             if not ws.closed:
                 await ws.close()
 
+    def _detach_ws(self) -> ClientWebSocketResponse | None:
+        """Forget the shared WebSocket and stop its heartbeat; hold _ws_lock."""
+        ws = self._ws
+        self._ws = None
+        if self._ws_heartbeat is not None:
+            self._ws_heartbeat.cancel()
+            self._ws_heartbeat = None
+        return ws
+
+    async def _send_heartbeats(self, ws: ClientWebSocketResponse) -> None:
+        """Keep the shared WebSocket open while no input is sent."""
+        while True:
+            await asyncio.sleep(_WS_HEARTBEAT_INTERVAL)
+            try:
+                await ws.send_bytes(_WS_HEARTBEAT)
+            except _WS_SEND_ERRORS:
+                # The next input finds the broken connection and replaces it.
+                return
+
     async def get_ws(self) -> ClientWebSocketResponse:
-        """Get or create WebSocket connection for mouse events."""
+        """Get or create the WebSocket shared by keyboard and mouse events."""
         generation = self._session_generation
         try:
             async with self._ws_lock:
                 if self._ws is not None and not self._ws.closed:
                     return self._ws
 
-                if self._ws is not None:
-                    await self._ws.close()
-                    self._ws = None
+                stale_ws = self._detach_ws()
+                if stale_ws is not None:
+                    await stale_ws.close()
 
                 generation = self._session_generation
-                self._ws = await self._connect_ws("ws")
-                return self._ws
+                ws = await self._connect_ws("ws")
+                self._ws = ws
+                self._ws_heartbeat = asyncio.create_task(self._send_heartbeats(ws))
+                return ws
         except ClientResponseError as err:
             # _clear_local_session acquires _ws_lock, so do this after the
             # lock above has been released to avoid a self-deadlock.
             await self._raise_ws_error(err, "ws", generation)
 
+    async def send_ws(
+        self, send: Callable[[ClientWebSocketResponse], Awaitable[None]]
+    ) -> None:
+        """Send one input message on the shared WebSocket.
+
+        A failed send forgets the connection, so the next one reconnects.
+        """
+        ws = await self.get_ws()
+        try:
+            await send(ws)
+        except _WS_SEND_ERRORS:
+            await self.invalidate_ws(ws)
+            raise
+
+    async def uses_binary_hid_protocol(self) -> bool:
+        """Select the keyboard and mouse wire format of the connected device."""
+        # Without a token the WebSocket reports the authentication error, so
+        # the current protocol stays the default when detection is impossible.
+        return await self.application_version_at_least(
+            non_pro=_BINARY_HID_MIN_NON_PRO_VERSION,
+            pro=_BINARY_HID_MIN_PRO_VERSION,
+        )
+
     async def open_ws(self, path: str, **kwargs: Any) -> ClientWebSocketResponse:
         """Open a new, unshared WebSocket on ``path``, relative to the API base.