893 lines
36 KiB
Python
893 lines
36 KiB
Python
"""Persistent low-latency BLE channel for streaming live keystrokes to the
|
||
M5Stack during macro recording.
|
||
|
||
The variable-sync client in ble_server.py is exchange-driven: connect,
|
||
swap one round of frames, disconnect. Live recording needs the opposite —
|
||
a connection that stays up for the entire recording session, with the
|
||
host emitting many small write-without-response frames as the user types.
|
||
|
||
Architecture:
|
||
- Owned by widgets/macro_recorder.py.
|
||
- The M5Stack auto-advertises the live service whenever it is idle
|
||
(no routine running). This client scans for that advertisement,
|
||
connects, subscribes to LIVE_KEYS_NOTIFY, exchanges a START
|
||
handshake, then accepts send_event() calls from the Tk thread.
|
||
- send_event() is fire-and-forget: it queues onto the asyncio loop;
|
||
writes use BLE WRITE_NO_RESPONSE so they cost one L2CAP frame and
|
||
no ACK round-trip.
|
||
- Disconnect detection bubbles up via on_status("disconnected").
|
||
|
||
Security model:
|
||
BLE link-layer access is intentionally open — no pairing, no PIN.
|
||
That's deliberate so the host machine (running Python, possibly a
|
||
different physical computer than the one the M5Stack is plugged
|
||
into) can connect at any time without out-of-band setup.
|
||
|
||
Confidentiality and integrity come from the application layer:
|
||
every frame in both directions is AES-256-GCM, with the device-tag
|
||
string ("M5Stack|<MAC>") bound as AAD. The 32-byte key lives on the
|
||
device's LittleFS partition and on the host's `<repo>/config/.blekeyfile`
|
||
(see ble_keystore.py). The key was originally pulled from the device
|
||
over USB during initial profile setup; once provisioned, neither
|
||
side ever transmits it.
|
||
|
||
Consequence: any BLE-range device can OPEN a connection to the
|
||
M5Stack, but every frame it sends fails GCM auth (no key) and is
|
||
dropped silently in onLiveWriteReceived. Replay-protection (per-MAC
|
||
session_id + monotonic seq) blocks captured frames being replayed
|
||
even by a key-holder, scoped to the current device boot session.
|
||
|
||
This is the same security model the existing variable-sync channel
|
||
uses; live recording inherits it unchanged.
|
||
|
||
Frame protocol (each direction, after AES-GCM unwrap of ble_frame.py):
|
||
|
||
byte 0: msg_type
|
||
0x01 = START (host -> device, no body)
|
||
0x02 = KEYS (host -> device, body = event_count + events)
|
||
0x03 = STOP (host -> device, no body)
|
||
0x10 = ACK (device -> host)
|
||
0x11 = ERROR (device -> host)
|
||
0x12 = HELLO (device -> host, first frame after subscribe)
|
||
bytes 1..8: session_id (uint64 little-endian)
|
||
bytes 9..16:seq (uint64 little-endian)
|
||
byte 17: body per msg_type
|
||
|
||
KEYS body:
|
||
byte 17: event_count (1..16)
|
||
bytes 18..: event_count × { uint8 action, uint8 hid_code, uint32 t_ms_le }
|
||
action: 0=DOWN, 1=UP
|
||
t_ms_le: host-clock ms since the first event of this
|
||
live session (uint32 little-endian). Device
|
||
uses this to preserve typing cadence on
|
||
emission (see live_keystroke.h).
|
||
|
||
ACK body:
|
||
byte 17..24: ref_seq (uint64 LE) — seq of the frame being acknowledged
|
||
|
||
ERROR body:
|
||
byte 17: err_code (1=BUFFER_FULL, 2=NOT_LIVE_MODE, 3=HID_FAILURE,
|
||
4=BAD_MSG)
|
||
bytes 18..25: ref_seq (uint64 LE)
|
||
|
||
Replay protection: same (session_id, seq) shape as the variable-sync
|
||
channel, validated via ble_replay.ReplayState.
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
import struct
|
||
import threading
|
||
import time
|
||
|
||
import ble_keystore
|
||
import ble_replay
|
||
import ble_debug_log as _bled
|
||
from ble_frame import build_frame, parse_frame_auto, DEVICE_TAG_PREFIX
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
def _dbg(kind: str, **fields) -> None:
|
||
"""Mirror live-mode lifecycle into the shared .ble_debug.log so the
|
||
file survives across runs and is grep-able alongside the var-sync
|
||
events. Each event is tagged so it's easy to filter from var-sync."""
|
||
try:
|
||
_bled.event("live_" + kind, **fields)
|
||
except Exception:
|
||
pass
|
||
|
||
# ---- Message type bytes ----
|
||
MSG_START = 0x01
|
||
MSG_KEYS = 0x02
|
||
MSG_STOP = 0x03
|
||
MSG_IDENTIFY = 0x04 # host -> device: show/hide the Bluetooth identify logo
|
||
MSG_MOUSE = 0x05 # host -> device: absolute pointer {buttons, x_u16, y_u16, wheel_i8}
|
||
MSG_LABEL = 0x06 # host -> device: friendly label (UTF-8 body)
|
||
MSG_ACK = 0x10
|
||
MSG_ERROR = 0x11
|
||
MSG_HELLO = 0x12
|
||
|
||
ACTION_DOWN = 0
|
||
ACTION_UP = 1
|
||
|
||
ERR_LABELS = {
|
||
1: "BUFFER_FULL",
|
||
2: "NOT_LIVE_MODE",
|
||
3: "HID_FAILURE",
|
||
4: "BAD_MSG",
|
||
}
|
||
|
||
# Status strings published via on_status callback.
|
||
ST_IDLE = "idle"
|
||
ST_SCANNING = "scanning"
|
||
ST_CONNECTING = "connecting"
|
||
ST_CONNECTED = "connected" # subscribed and START acked
|
||
ST_DISCONNECTED = "disconnected"
|
||
ST_ERROR = "error"
|
||
|
||
# How long to scan for the device before giving up one cycle.
|
||
SCAN_TIMEOUT_S = 4
|
||
# How long to wait for the device's hello frame after subscribing.
|
||
HELLO_TIMEOUT_S = 8
|
||
# How long to wait for the START ack before giving up.
|
||
START_ACK_TIMEOUT_S = 5
|
||
# Worker queue capacity. Keystrokes that overflow are dropped on the host
|
||
# side (with a warning) rather than waiting and growing latency.
|
||
SEND_QUEUE_MAX = 512
|
||
|
||
|
||
def _pack_header(msg_type: int, session_id: int, seq: int) -> bytes:
|
||
return struct.pack("<BQQ", msg_type, session_id & 0xFFFFFFFFFFFFFFFF,
|
||
seq & 0xFFFFFFFFFFFFFFFF)
|
||
|
||
|
||
def _parse_header(plain: bytes):
|
||
if len(plain) < 17:
|
||
return None
|
||
msg_type, sid, seq = struct.unpack("<BQQ", plain[:17])
|
||
return msg_type, sid, seq, plain[17:]
|
||
|
||
|
||
class BLELiveKeystrokeClient:
|
||
"""Persistent BLE link for live keystroke streaming.
|
||
|
||
Lifecycle:
|
||
c = BLELiveKeystrokeClient()
|
||
c.start(on_status=cb, on_error=cb) # scan + connect + START
|
||
c.send_event(action, hid_code) # called from Tk thread
|
||
c.stop() # STOP + disconnect
|
||
|
||
All callbacks run on the asyncio worker thread — bounce through
|
||
Tk's `after(0, ...)` to touch widgets.
|
||
"""
|
||
|
||
def __init__(self, target_address: str | None = None):
|
||
"""
|
||
target_address — optional BLE MAC (Bleak's d.address, e.g.
|
||
"AA:BB:CC:DD:EE:FF"). When set, the scanner only accepts a
|
||
match on this address. Used by the multi-device manager
|
||
(ble_multi.py) to pin each client to a distinct device so
|
||
N>1 clients don't fight for the same advertisement.
|
||
None = match the first live-mode device found (existing
|
||
single-device behavior).
|
||
"""
|
||
self._target_address = target_address.upper() if target_address else None
|
||
self._running = False
|
||
self._thread: threading.Thread | None = None
|
||
self._loop: asyncio.AbstractEventLoop | None = None
|
||
self._client = None
|
||
self._key: bytes | None = None
|
||
self._device_tag: str | None = None
|
||
self._replay = ble_replay.ReplayState()
|
||
# Counts consecutive frames that failed AES-GCM auth — surfaced
|
||
# to the UI as "key mismatch" after a few in a row so the user
|
||
# gets a clear error instead of silent failure.
|
||
self._auth_fail_count = 0
|
||
# Host clock anchor (time.monotonic_ns of the first send_event of
|
||
# the current live session). Per-event timestamps are computed
|
||
# relative to this so the device can preserve typing cadence.
|
||
self._session_t0_ns: int | None = None
|
||
# Per-session running totals exposed to the UI.
|
||
self.events_sent = 0
|
||
self.bytes_sent = 0
|
||
# Whether the device should be showing its Bluetooth "identify"
|
||
# logo. Latched here so it survives reconnects and is (re)sent the
|
||
# moment a session reaches the connected state.
|
||
self._identify_on = False
|
||
# Friendly label to display on the device; latched so it's re-sent
|
||
# on every (re)connect (a device reboot loses it from RAM).
|
||
self._device_label = ""
|
||
# Last mouse button mask we sent — used to decide which mouse frames
|
||
# warrant a reliable (response=True) write vs a fire-and-forget one.
|
||
self._last_mouse_buttons = 0
|
||
# asyncio.Queue of pending outgoing payloads (List of (action, hid))
|
||
# The worker batches whatever is ready when it wakes up so a burst
|
||
# of keystrokes pays one BLE radio cycle instead of N.
|
||
self._send_q: asyncio.Queue | None = None
|
||
# Set when the worker observes a clean disconnect.
|
||
self._disconnected_event: asyncio.Event | None = None
|
||
# Set when the device acks our most recent START.
|
||
self._start_acked: asyncio.Event | None = None
|
||
self._on_status = None
|
||
self._on_error = None
|
||
# Used to tag errors emitted from the worker with the seq they reference
|
||
# so the UI can correlate (mostly diagnostic).
|
||
self._last_status = ST_IDLE
|
||
|
||
# ---- Public API (Tk thread) ----
|
||
|
||
def start(self, on_status=None, on_error=None) -> None:
|
||
"""Begin scanning and connecting. Non-blocking.
|
||
|
||
on_status(status_str) — fires on every state transition.
|
||
on_error(err_code:int, ref_seq:int|None, label:str)
|
||
"""
|
||
if self._running:
|
||
return
|
||
self._on_status = on_status
|
||
self._on_error = on_error
|
||
self._running = True
|
||
self._thread = threading.Thread(target=self._thread_main, daemon=True,
|
||
name="BLELiveKeystrokeClient")
|
||
self._thread.start()
|
||
|
||
def stop(self, timeout: float = 4.0) -> None:
|
||
"""Send STOP if connected, tear down the worker thread."""
|
||
if not self._running:
|
||
return
|
||
self._running = False
|
||
loop = self._loop
|
||
if loop and loop.is_running():
|
||
# Wake the worker so it observes _running=False and exits cleanly.
|
||
loop.call_soon_threadsafe(self._wake_for_shutdown)
|
||
if self._thread and self._thread.is_alive():
|
||
self._thread.join(timeout=timeout)
|
||
self._thread = None
|
||
self._loop = None
|
||
self._client = None
|
||
|
||
def send_event(self, action: int, hid_code: int) -> bool:
|
||
"""Queue one (action, hid_code) event for transmission.
|
||
|
||
Timestamps are captured HERE (Tk thread, on the actual keypress)
|
||
so BLE/asyncio jitter doesn't pollute the recorded cadence. The
|
||
timestamp is host-monotonic ms relative to the first event of
|
||
the session.
|
||
|
||
Returns True on enqueue, False if the channel is not open or
|
||
the queue is full. Safe to call from the Tk thread.
|
||
"""
|
||
if self._last_status != ST_CONNECTED:
|
||
return False
|
||
now_ns = time.monotonic_ns()
|
||
if self._session_t0_ns is None:
|
||
self._session_t0_ns = now_ns
|
||
t_ms = (now_ns - self._session_t0_ns) // 1_000_000
|
||
if t_ms > 0xFFFFFFFF:
|
||
t_ms = 0xFFFFFFFF
|
||
return self.send_event_with_t(action, hid_code, int(t_ms))
|
||
|
||
def send_event_with_t(self, action: int, hid_code: int,
|
||
t_ms: int) -> bool:
|
||
"""Like send_event but the caller supplies the timestamp. Used
|
||
by the Replay path so each event carries its ORIGINAL
|
||
recorded timestamp instead of "now"."""
|
||
loop = self._loop
|
||
q = self._send_q
|
||
if loop is None or q is None or not self._running:
|
||
return False
|
||
if self._last_status != ST_CONNECTED:
|
||
return False
|
||
try:
|
||
loop.call_soon_threadsafe(self._enqueue_send_threadsafe,
|
||
int(action) & 0xFF,
|
||
int(hid_code) & 0xFF,
|
||
int(t_ms) & 0xFFFFFFFF)
|
||
return True
|
||
except RuntimeError:
|
||
return False
|
||
|
||
def reset_session_clock(self) -> None:
|
||
"""Forget the current session anchor so the next send_event
|
||
becomes t=0. Called when a new recording session begins."""
|
||
self._session_t0_ns = None
|
||
|
||
def set_identify(self, on: bool) -> None:
|
||
"""Ask the device to show (on=True) or hide (on=False) its
|
||
Bluetooth identify logo. Safe to call from the Tk thread.
|
||
|
||
The desire is latched in self._identify_on so it's (re)applied on
|
||
every (re)connect; if we're already connected we also push the
|
||
change immediately. If we're not connected yet, the session sends
|
||
the current flag the moment it comes up."""
|
||
self._identify_on = bool(on)
|
||
loop = self._loop
|
||
q = self._send_q
|
||
if loop is None or q is None or not self._running:
|
||
return
|
||
if self._last_status != ST_CONNECTED:
|
||
return # will be applied on connect by _session
|
||
try:
|
||
loop.call_soon_threadsafe(self._enqueue_identify_threadsafe,
|
||
bool(on))
|
||
except RuntimeError:
|
||
pass
|
||
|
||
def _enqueue_identify_threadsafe(self, on: bool) -> None:
|
||
if self._send_q is None:
|
||
return
|
||
try:
|
||
self._send_q.put_nowait(("identify", bool(on)))
|
||
except asyncio.QueueFull:
|
||
pass
|
||
|
||
def send_mouse(self, buttons: int, x: float, y: float,
|
||
wheel: int = 0) -> bool:
|
||
"""Send one absolute pointer update. ``x``/``y`` are normalized
|
||
screen coordinates in [0, 1] (so the remote cursor never desyncs —
|
||
every report fully specifies the position). ``buttons`` is a bitmask
|
||
(bit0 left, bit1 right, bit2 middle). ``wheel`` is a relative tick.
|
||
Safe to call from the Tk thread."""
|
||
loop = self._loop
|
||
q = self._send_q
|
||
if loop is None or q is None or not self._running:
|
||
return False
|
||
if self._last_status != ST_CONNECTED:
|
||
return False
|
||
xi = int(max(0.0, min(1.0, x)) * 32767)
|
||
yi = int(max(0.0, min(1.0, y)) * 32767)
|
||
try:
|
||
loop.call_soon_threadsafe(self._enqueue_mouse_threadsafe,
|
||
int(buttons) & 0xFF, xi, yi, int(wheel))
|
||
return True
|
||
except RuntimeError:
|
||
return False
|
||
|
||
def _enqueue_mouse_threadsafe(self, buttons: int, x: int, y: int,
|
||
wheel: int) -> None:
|
||
if self._send_q is None:
|
||
return
|
||
try:
|
||
self._send_q.put_nowait(("mouse", buttons, x, y, wheel))
|
||
except asyncio.QueueFull:
|
||
pass
|
||
|
||
def set_device_label(self, label: str) -> None:
|
||
"""Push a friendly label to show on the device's screen. Latched so
|
||
it survives reconnects (re-sent by _session on connect)."""
|
||
self._device_label = label or ""
|
||
loop = self._loop
|
||
q = self._send_q
|
||
if loop is None or q is None or not self._running:
|
||
return
|
||
if self._last_status != ST_CONNECTED:
|
||
return
|
||
try:
|
||
loop.call_soon_threadsafe(self._enqueue_label_threadsafe,
|
||
self._device_label)
|
||
except RuntimeError:
|
||
pass
|
||
|
||
def _enqueue_label_threadsafe(self, label: str) -> None:
|
||
if self._send_q is None:
|
||
return
|
||
try:
|
||
self._send_q.put_nowait(("label", label))
|
||
except asyncio.QueueFull:
|
||
pass
|
||
|
||
def is_connected(self) -> bool:
|
||
return self._last_status == ST_CONNECTED
|
||
|
||
def status(self) -> str:
|
||
return self._last_status
|
||
|
||
# ---- Worker thread / asyncio glue ----
|
||
|
||
def _wake_for_shutdown(self) -> None:
|
||
# Drop a sentinel on the queue so the writer task unblocks.
|
||
if self._send_q is not None:
|
||
try:
|
||
self._send_q.put_nowait(None)
|
||
except asyncio.QueueFull:
|
||
pass
|
||
|
||
def _enqueue_send_threadsafe(self, action: int, hid_code: int,
|
||
t_ms: int) -> None:
|
||
if self._send_q is None:
|
||
return
|
||
try:
|
||
self._send_q.put_nowait((action, hid_code, t_ms))
|
||
except asyncio.QueueFull:
|
||
print(f"[live] send queue full — dropping event "
|
||
f"(action={action} hid=0x{hid_code:02X})")
|
||
_dbg("send_queue_full", action=action, hid=hid_code)
|
||
|
||
def _thread_main(self) -> None:
|
||
try:
|
||
from bleak import BleakClient, BleakScanner # noqa
|
||
except ImportError:
|
||
self._set_status(ST_ERROR)
|
||
log.error("bleak not installed — pip install bleak")
|
||
return
|
||
self._BleakClient = BleakClient
|
||
self._BleakScanner = BleakScanner
|
||
|
||
self._loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(self._loop)
|
||
try:
|
||
self._loop.run_until_complete(self._run())
|
||
except RuntimeError:
|
||
pass
|
||
except Exception as exc:
|
||
log.warning("live worker error: %s", exc, exc_info=True)
|
||
finally:
|
||
try:
|
||
self._loop.close()
|
||
except Exception:
|
||
pass
|
||
|
||
async def _run(self) -> None:
|
||
"""Outer worker loop. Repeats scan → connect → session until the
|
||
caller calls stop(). The M5Stack advertises the live service
|
||
automatically whenever it is idle (no routine running, no USB
|
||
upload in progress), so the device is normally already
|
||
discoverable. The loop also recovers mid-session drops
|
||
(interference / range) by re-scanning and reconnecting; the device
|
||
keeps listening across drops so recording resumes seamlessly."""
|
||
# We scan for LIVE_SERVICE_UUID specifically — the var-sync
|
||
# client uses SERVICE_UUID, and the device advertises one or
|
||
# the other depending on whether it's in live mode. This keeps
|
||
# the two clients from racing for the same BLE connection.
|
||
from ble_server import LIVE_SERVICE_UUID
|
||
|
||
self._send_q = asyncio.Queue(maxsize=SEND_QUEUE_MAX)
|
||
self._start_acked = asyncio.Event()
|
||
|
||
# Per-MAC keystore can supply the right key once we read the
|
||
# device tag from the first frame. We pre-load the legacy
|
||
# single-key file as a "any unknown device" fallback so
|
||
# existing single-device setups continue working.
|
||
self._key = ble_keystore.load_key()
|
||
# Note: no `if self._key is None: return` here — multi-device
|
||
# users may have only per-MAC keys, no legacy file. We resolve
|
||
# the actual key per-frame based on the tag in the frame.
|
||
|
||
print("[live] worker started, scanning for live-mode device...")
|
||
_dbg("worker_start", scan_uuid=LIVE_SERVICE_UUID)
|
||
backoff_s = 0.5
|
||
while self._running:
|
||
try:
|
||
await self._one_cycle(LIVE_SERVICE_UUID)
|
||
except Exception as exc:
|
||
print(f"[live] cycle error: {exc!r}")
|
||
_dbg("cycle_error", error=repr(exc))
|
||
self._set_status(ST_DISCONNECTED)
|
||
if not self._running:
|
||
break
|
||
await asyncio.sleep(backoff_s)
|
||
backoff_s = min(backoff_s * 1.5, 3.0)
|
||
# Reset transient signalling for the next cycle.
|
||
self._start_acked = asyncio.Event()
|
||
print("[live] worker exiting")
|
||
_dbg("worker_exit")
|
||
|
||
async def _one_cycle(self, service_uuid: str) -> None:
|
||
"""One scan-connect-session pass. Returns whether or not it
|
||
succeeded — caller decides whether to retry."""
|
||
self._set_status(ST_SCANNING)
|
||
_dbg("scan_start", timeout_s=SCAN_TIMEOUT_S, uuid=service_uuid)
|
||
|
||
# Collect ALL device sightings (with their advertised service
|
||
# UUIDs) during the scan window, so the log shows what was
|
||
# actually nearby. The filter we return controls connection.
|
||
all_sightings: dict = {}
|
||
|
||
def _filter(d, adv):
|
||
try:
|
||
addr = getattr(d, "address", None)
|
||
if addr and addr not in all_sightings:
|
||
all_sightings[addr] = {
|
||
"name": getattr(d, "name", None),
|
||
"uuids": list(adv.service_uuids or []),
|
||
}
|
||
except Exception:
|
||
pass
|
||
if service_uuid not in (adv.service_uuids or []):
|
||
return False
|
||
# If we were pinned to a specific BLE MAC (multi-device
|
||
# manager spawns one client per known device), only match
|
||
# that exact address. This lets multiple BLELiveKeystrokeClients
|
||
# coexist without racing for the first advertisement.
|
||
if self._target_address is not None:
|
||
addr = getattr(d, "address", "")
|
||
if not addr or addr.upper() != self._target_address:
|
||
return False
|
||
return True
|
||
|
||
try:
|
||
device = await self._BleakScanner.find_device_by_filter(
|
||
_filter, timeout=SCAN_TIMEOUT_S,
|
||
)
|
||
except Exception as exc:
|
||
print(f"[live] scan failed: {exc!r}")
|
||
_dbg("scan_error", error=repr(exc))
|
||
self._set_status(ST_DISCONNECTED)
|
||
return
|
||
if not device:
|
||
self._set_status(ST_SCANNING)
|
||
_dbg("scan_idle", sighted=all_sightings)
|
||
return
|
||
if not self._running:
|
||
return
|
||
|
||
print(f"[live] found device {getattr(device, 'name', '?')} "
|
||
f"({getattr(device, 'address', '?')}), connecting...")
|
||
_dbg("scan_found",
|
||
name=getattr(device, "name", None),
|
||
address=getattr(device, "address", None))
|
||
self._set_status(ST_CONNECTING)
|
||
# Fresh per-cycle disconnect event so a previous cycle's set()
|
||
# doesn't immediately tear down this one.
|
||
self._disconnected_event = asyncio.Event()
|
||
|
||
def on_disc(_client):
|
||
try:
|
||
if self._loop and self._loop.is_running():
|
||
self._loop.call_soon_threadsafe(
|
||
self._disconnected_event.set)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
async with self._BleakClient(device, timeout=15.0,
|
||
disconnected_callback=on_disc) as client:
|
||
print("[live] BLE connected, starting session")
|
||
_dbg("connected", address=getattr(device, "address", None))
|
||
self._client = client
|
||
await self._session(client)
|
||
except asyncio.TimeoutError:
|
||
print("[live] connect timed out")
|
||
_dbg("connect_timeout")
|
||
except Exception as exc:
|
||
print(f"[live] connect/session error: {exc!r}")
|
||
_dbg("connect_error", error=repr(exc))
|
||
finally:
|
||
self._client = None
|
||
_dbg("session_end")
|
||
if self._last_status == ST_CONNECTED:
|
||
self._set_status(ST_DISCONNECTED)
|
||
|
||
async def _session(self, client) -> None:
|
||
from ble_server import LIVE_KEYS_WRITE_UUID, LIVE_KEYS_NOTIFY_UUID
|
||
|
||
hello_evt: asyncio.Future = self._loop.create_future()
|
||
|
||
def handle_notify(_char, data: bytearray):
|
||
raw = bytes(data)
|
||
result = parse_frame_auto(raw)
|
||
if result is None:
|
||
# Fully malformed (no tag, bad length). Drop silently.
|
||
_dbg("malformed_frame", raw_len=len(raw))
|
||
return
|
||
tag, plain = result
|
||
if plain is None:
|
||
# Tag was valid but no key worked. Either we don't have
|
||
# a key for this device (it was reflashed / never had a
|
||
# profile uploaded over USB), or the stored key is
|
||
# stale. Surface KEY_MISMATCH after a few in a row so
|
||
# the user gets a clear remediation.
|
||
self._auth_fail_count += 1
|
||
_dbg("auth_fail", count=self._auth_fail_count,
|
||
raw_len=len(raw), tag=tag)
|
||
if self._auth_fail_count == 3 and self._on_error:
|
||
try:
|
||
self._on_error(0, None, "KEY_MISMATCH")
|
||
except Exception:
|
||
pass
|
||
return
|
||
# First good frame nails down the per-device key for
|
||
# outgoing writes too.
|
||
self._auth_fail_count = 0
|
||
mac = tag[len(DEVICE_TAG_PREFIX):]
|
||
resolved = ble_keystore.load_key_for_mac_or_default(mac)
|
||
if resolved is not None:
|
||
self._key = resolved
|
||
hdr = _parse_header(plain)
|
||
if hdr is None:
|
||
return
|
||
msg_type, sid, seq, body = hdr
|
||
# Validate replay
|
||
ok, _reason = self._replay.accept_received(mac, sid, seq)
|
||
if not ok:
|
||
print(f"[live] replay reject seq={seq} sid={sid}")
|
||
return
|
||
if msg_type == MSG_HELLO:
|
||
self._device_tag = tag
|
||
if not hello_evt.done():
|
||
hello_evt.set_result(True)
|
||
return
|
||
if msg_type == MSG_ACK:
|
||
if len(body) >= 8:
|
||
ref_seq, = struct.unpack("<Q", body[:8])
|
||
# The START ack is the only one we explicitly wait on.
|
||
if not self._start_acked.is_set():
|
||
self._start_acked.set()
|
||
return
|
||
if msg_type == MSG_ERROR:
|
||
if len(body) >= 9:
|
||
err = body[0]
|
||
ref_seq, = struct.unpack("<Q", body[1:9])
|
||
else:
|
||
err = body[0] if body else 0
|
||
ref_seq = None
|
||
label = ERR_LABELS.get(err, f"ERR_{err}")
|
||
log.warning("live: device error %s ref_seq=%s", label, ref_seq)
|
||
if self._on_error:
|
||
try:
|
||
self._on_error(err, ref_seq, label)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
await client.start_notify(LIVE_KEYS_NOTIFY_UUID, handle_notify)
|
||
print("[live] subscribed to LIVE_KEYS_NOTIFY")
|
||
_dbg("subscribed")
|
||
except Exception as exc:
|
||
print(f"[live] start_notify failed: {exc!r}")
|
||
_dbg("start_notify_failed", error=repr(exc))
|
||
self._set_status(ST_ERROR)
|
||
return
|
||
|
||
# ---- Wait for hello ----
|
||
try:
|
||
await asyncio.wait_for(hello_evt, timeout=HELLO_TIMEOUT_S)
|
||
print("[live] received hello")
|
||
_dbg("hello_ok")
|
||
except asyncio.TimeoutError:
|
||
print(f"[live] hello timeout after {HELLO_TIMEOUT_S}s")
|
||
_dbg("hello_timeout", timeout_s=HELLO_TIMEOUT_S)
|
||
self._set_status(ST_ERROR)
|
||
return
|
||
if not self._running:
|
||
return
|
||
|
||
# ---- Send START ----
|
||
if not await self._send_control(client, LIVE_KEYS_WRITE_UUID, MSG_START):
|
||
print("[live] START write failed")
|
||
_dbg("start_write_failed")
|
||
self._set_status(ST_ERROR)
|
||
return
|
||
_dbg("start_sent")
|
||
try:
|
||
await asyncio.wait_for(self._start_acked.wait(),
|
||
timeout=START_ACK_TIMEOUT_S)
|
||
print("[live] START ack received")
|
||
_dbg("start_ack")
|
||
except asyncio.TimeoutError:
|
||
print(f"[live] START ack timeout after {START_ACK_TIMEOUT_S}s")
|
||
_dbg("start_ack_timeout", timeout_s=START_ACK_TIMEOUT_S)
|
||
self._set_status(ST_ERROR)
|
||
return
|
||
|
||
self._set_status(ST_CONNECTED)
|
||
print("[live] session ready — recording can begin")
|
||
_dbg("session_ready")
|
||
|
||
# Re-apply any pending identify request now that we're connected
|
||
# (e.g. set_identify(True) was called while we were still scanning).
|
||
if self._identify_on:
|
||
self._enqueue_identify_threadsafe(True)
|
||
# Re-send the device label so it survives reconnects / reboots.
|
||
if self._device_label:
|
||
self._enqueue_label_threadsafe(self._device_label)
|
||
|
||
# ---- Stream loop ----
|
||
try:
|
||
await self._stream_loop(client, LIVE_KEYS_WRITE_UUID)
|
||
finally:
|
||
# ---- Send STOP best-effort ----
|
||
try:
|
||
await asyncio.wait_for(
|
||
self._send_control(client, LIVE_KEYS_WRITE_UUID, MSG_STOP),
|
||
timeout=1.5)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
await client.stop_notify(LIVE_KEYS_NOTIFY_UUID)
|
||
except Exception:
|
||
pass
|
||
|
||
async def _stream_loop(self, client, write_uuid: str) -> None:
|
||
"""Drain self._send_q to BLE writes. One BLE write per drain cycle —
|
||
we coalesce whatever is queued at wake-up to amortize radio time.
|
||
|
||
Uses Write WITH response (response=True) for delivery guarantees.
|
||
Write Without Response can silently drop frames under congestion
|
||
— confirmed by dropped-keystroke reports during live recording.
|
||
The added ~15 ms per write is invisible behind the 100 ms
|
||
device-side replay buffer.
|
||
"""
|
||
# Each event is now 6 bytes (action,hid,t_ms_u32). With the 17-byte
|
||
# plain header + 1-byte count, a 16-event frame is 17+1+16*6 = 114
|
||
# bytes plaintext, well under the negotiated MTU even on Windows.
|
||
MAX_BATCH = 16
|
||
q = self._send_q
|
||
disc = self._disconnected_event
|
||
|
||
while self._running and not disc.is_set():
|
||
try:
|
||
first = await q.get()
|
||
except asyncio.CancelledError:
|
||
break
|
||
if first is None:
|
||
# Shutdown sentinel
|
||
break
|
||
# Control item (e.g. identify) is tagged with a str first
|
||
# element; keystroke items are (int action, int hid, int t).
|
||
if isinstance(first[0], str):
|
||
await self._handle_control_item(client, write_uuid, first)
|
||
continue
|
||
batch = [first]
|
||
pending_identify = None
|
||
while len(batch) < MAX_BATCH:
|
||
try:
|
||
nxt = q.get_nowait()
|
||
except asyncio.QueueEmpty:
|
||
break
|
||
if nxt is None:
|
||
self._running = False
|
||
break
|
||
if isinstance(nxt[0], str):
|
||
# A control item slipped in mid-batch — flush the
|
||
# keystrokes first, then handle it after the send.
|
||
pending_identify = nxt
|
||
break
|
||
batch.append(nxt)
|
||
|
||
# Build KEYS frame (binary, per protocol in module docstring)
|
||
seq = self._replay.next_send_seq()
|
||
sid = self._replay.host_session_id()
|
||
body = bytes([len(batch)]) + b"".join(
|
||
struct.pack("<BBI", a & 0xFF, h & 0xFF, t & 0xFFFFFFFF)
|
||
for (a, h, t) in batch
|
||
)
|
||
plain = _pack_header(MSG_KEYS, sid, seq) + body
|
||
try:
|
||
frame = build_frame(self._key, self._device_tag, plain)
|
||
except Exception as exc:
|
||
print(f"[live] build_frame failed: {exc!r}")
|
||
continue
|
||
try:
|
||
await client.write_gatt_char(write_uuid, frame, response=True)
|
||
self.events_sent += len(batch)
|
||
self.bytes_sent += len(frame)
|
||
except Exception as exc:
|
||
print(f"[live] write failed: {exc!r}")
|
||
_dbg("write_failed", error=repr(exc))
|
||
# Treat as disconnect — the on_disc callback will also fire,
|
||
# but bail out promptly either way.
|
||
break
|
||
|
||
if pending_identify is not None:
|
||
await self._handle_control_item(client, write_uuid,
|
||
pending_identify)
|
||
|
||
async def _handle_control_item(self, client, write_uuid: str,
|
||
item: tuple) -> None:
|
||
"""Dispatch a tagged control item pulled from the send queue."""
|
||
kind = item[0]
|
||
if kind == "identify":
|
||
await self._send_identify(client, write_uuid, bool(item[1]))
|
||
elif kind == "mouse":
|
||
await self._send_mouse(client, write_uuid,
|
||
item[1], item[2], item[3], item[4])
|
||
elif kind == "label":
|
||
await self._send_label(client, write_uuid, item[1])
|
||
|
||
async def _send_mouse(self, client, write_uuid: str,
|
||
buttons: int, x: int, y: int, wheel: int) -> bool:
|
||
"""Send an absolute pointer report. Pure-move frames go out
|
||
fire-and-forget (write-without-response) since absolute positions
|
||
are self-correcting; button changes and scrolls use a reliable
|
||
write so a click / wheel tick is never lost."""
|
||
seq = self._replay.next_send_seq()
|
||
sid = self._replay.host_session_id()
|
||
w = max(-127, min(127, int(wheel)))
|
||
body = struct.pack("<BHHb", buttons & 0xFF, x & 0xFFFF, y & 0xFFFF, w)
|
||
plain = _pack_header(MSG_MOUSE, sid, seq) + body
|
||
try:
|
||
frame = build_frame(self._key,
|
||
self._device_tag or self._scan_tag(), plain)
|
||
except Exception:
|
||
return False
|
||
reliable = (buttons != self._last_mouse_buttons) or (w != 0)
|
||
self._last_mouse_buttons = buttons
|
||
try:
|
||
await client.write_gatt_char(write_uuid, frame, response=reliable)
|
||
self.bytes_sent += len(frame)
|
||
return True
|
||
except Exception as exc:
|
||
_dbg("mouse_write_failed", error=repr(exc))
|
||
return False
|
||
|
||
async def _send_label(self, client, write_uuid: str, label: str) -> bool:
|
||
seq = self._replay.next_send_seq()
|
||
sid = self._replay.host_session_id()
|
||
body = (label or "").encode("utf-8")[:38]
|
||
plain = _pack_header(MSG_LABEL, sid, seq) + body
|
||
try:
|
||
frame = build_frame(self._key,
|
||
self._device_tag or self._scan_tag(), plain)
|
||
except Exception as exc:
|
||
log.warning("live: build_frame(label) failed: %s", exc)
|
||
return False
|
||
try:
|
||
await client.write_gatt_char(write_uuid, frame, response=True)
|
||
_dbg("label_sent", label=label)
|
||
return True
|
||
except Exception as exc:
|
||
_dbg("label_write_failed", error=repr(exc))
|
||
return False
|
||
|
||
async def _send_identify(self, client, write_uuid: str,
|
||
on: bool) -> bool:
|
||
"""Tell the device to show (on) or hide (off) its Bluetooth
|
||
identify logo. Same AES-GCM envelope as every other frame."""
|
||
seq = self._replay.next_send_seq()
|
||
sid = self._replay.host_session_id()
|
||
plain = (_pack_header(MSG_IDENTIFY, sid, seq)
|
||
+ bytes([1 if on else 0]))
|
||
try:
|
||
frame = build_frame(self._key,
|
||
self._device_tag or self._scan_tag(), plain)
|
||
except Exception as exc:
|
||
log.warning("live: build_frame(identify) failed: %s", exc)
|
||
return False
|
||
try:
|
||
await client.write_gatt_char(write_uuid, frame, response=True)
|
||
_dbg("identify_sent", on=on)
|
||
return True
|
||
except Exception as exc:
|
||
_dbg("identify_write_failed", error=repr(exc))
|
||
return False
|
||
|
||
async def _send_control(self, client, write_uuid: str,
|
||
msg_type: int) -> bool:
|
||
"""Send a control frame (START/STOP) with response=True so the
|
||
device's receive callback runs before we move on. Returns True on
|
||
wire-level success.
|
||
"""
|
||
seq = self._replay.next_send_seq()
|
||
sid = self._replay.host_session_id()
|
||
plain = _pack_header(msg_type, sid, seq)
|
||
try:
|
||
frame = build_frame(self._key, self._device_tag or self._scan_tag(),
|
||
plain)
|
||
except Exception as exc:
|
||
log.warning("live: build_frame(control) failed: %s", exc)
|
||
return False
|
||
try:
|
||
await client.write_gatt_char(write_uuid, frame, response=True)
|
||
return True
|
||
except Exception as exc:
|
||
log.warning("live: control write failed: %s", exc)
|
||
return False
|
||
|
||
def _scan_tag(self) -> str:
|
||
# Should never be reached — hello sets _device_tag before control
|
||
# frames go out. Fallback to a sentinel so build_frame doesn't blow
|
||
# up on None during error paths.
|
||
return DEVICE_TAG_PREFIX + "00:00:00:00:00:00"
|
||
|
||
def _set_status(self, status: str) -> None:
|
||
if status == self._last_status:
|
||
return
|
||
self._last_status = status
|
||
if self._on_status:
|
||
try:
|
||
self._on_status(status)
|
||
except Exception as exc:
|
||
log.warning("live: on_status callback error: %s", exc)
|