"""Multi-device BLE keyboard streaming manager. Maintains up to N independent BLELiveKeystrokeClient instances, each pinned to a distinct M5Stack by BLE MAC. Provides a single fan-out API (:meth:`send_event`) so the caller (the BT Keyboard window) can stream one stream of keystrokes to many devices simultaneously, with per-device enable/disable so the user can selectively mute targets without disconnecting them. Architecture: - One BLELiveKeystrokeClient per slot. Each runs its own asyncio worker thread and BLE link, so a stall on one device cannot block another. - Slots are added explicitly via :meth:`add_device` after a one- shot scan discovers an in-range live-mode device. Removal is explicit too. - send_event(action, hid) walks all slots and forwards the event to every enabled, connected client. Each client encrypts under that device's own per-MAC key (see ble_keystore + ble_frame); nothing about multi-device streaming changes the security model. - send_event_with_t(action, hid, t_ms) is the replay path — caller supplies the original recorded timestamp so the device-side cadence-preserving scheduler reproduces the typing rhythm. Threading: - Public API is Tk-thread safe (each call delegates to per-client asyncio.call_soon_threadsafe). - on_status / on_error callbacks fire on the worker thread for the slot that changed; the UI marshals back to Tk. """ from __future__ import annotations import asyncio import threading import time from dataclasses import dataclass, field from typing import Callable, Optional from ble_live import ( BLELiveKeystrokeClient, ST_CONNECTED, ST_CONNECTING, ST_SCANNING, ST_DISCONNECTED, ST_ERROR, ) # Soft warning threshold (NOT a hard cap). The manager accepts an unlimited # number of devices; the UI shows a brief, non-blocking warning once more # than this many are connected at once. BLE mode caps lower than the ESP-NOW # hub (see mesh_manager.MAX_SLOTS) because Windows only holds ~3-4 reliable # concurrent BLE links before latency/loss degrades. MAX_SLOTS = 4 @dataclass class DeviceSlot: address: str # Bleak MAC ("AA:BB:CC:DD:EE:FF") label: str = "" # Friendly name (defaults to MAC) enabled: bool = True # User toggle: stream to this device or not client: BLELiveKeystrokeClient | None = None status: str = ST_DISCONNECTED last_status_change: float = field(default_factory=time.monotonic) added_at: float = field(default_factory=time.monotonic) def display_label(self) -> str: return self.label or self.address class MultiBleKeyboardManager: """Owns N BLELiveKeystrokeClient slots and fans events out across every enabled, currently-connected slot.""" def __init__(self, max_slots: int = MAX_SLOTS): self._max_slots = max_slots self._slots: dict[str, DeviceSlot] = {} self._lock = threading.Lock() # External callbacks self._on_status_change: Optional[Callable[[str, str], None]] = None self._on_stats_change: Optional[Callable[[], None]] = None # ---- Public API ---- def set_callbacks(self, *, on_status_change=None, on_stats_change=None): """on_status_change(address, status) — fires on every slot status change. on_stats_change() — fires periodically when sent counters change.""" self._on_status_change = on_status_change self._on_stats_change = on_stats_change def slots(self) -> list[DeviceSlot]: """Return a stable-ordered snapshot of current slots.""" with self._lock: return sorted(self._slots.values(), key=lambda s: s.added_at) def slot_count(self) -> int: with self._lock: return len(self._slots) def is_full(self) -> bool: # Uncapped — there is no hard device limit anymore. return False def over_soft_limit(self) -> bool: """True when more devices than the soft threshold are tracked.""" return self.slot_count() > self._max_slots def add_device(self, address: str, label: str = "", board: str = "") -> DeviceSlot | None: """Add a slot for ``address`` and start its BLE worker. Returns the new slot, or None only if the address is already tracked (the device count is uncapped). ``board`` is accepted for call-surface parity with the mesh manager and ignored here (BLE discovery doesn't report board type).""" addr = address.upper() with self._lock: if addr in self._slots: return None slot = DeviceSlot(address=addr, label=label or addr) slot.client = BLELiveKeystrokeClient(target_address=addr) self._slots[addr] = slot self._start_slot(slot) return slot def remove_device(self, address: str) -> bool: """Disconnect and forget ``address``. Returns True if it was tracked.""" addr = address.upper() with self._lock: slot = self._slots.pop(addr, None) if slot is None: return False if slot.client is not None: try: slot.client.stop(timeout=2.0) except Exception: pass return True def set_enabled(self, address: str, enabled: bool) -> None: """Toggle whether send_event reaches this slot.""" addr = address.upper() with self._lock: slot = self._slots.get(addr) if slot is not None: slot.enabled = bool(enabled) def get_slot(self, address: str) -> DeviceSlot | None: with self._lock: return self._slots.get(address.upper()) def set_label(self, address: str, label: str) -> None: """Update a slot's friendly label (falls back to the MAC if empty) and push it to the device so it shows on its screen too.""" addr = address.upper() with self._lock: slot = self._slots.get(addr) if slot is not None: slot.label = label or slot.address client = slot.client else: client = None if client is not None: try: client.set_device_label(label or "") except Exception: pass def identify(self, address: str, on: bool = True) -> None: """Ask one device to show (on) / hide (off) its Bluetooth identify logo, so the user can see which physical M5Stack a slot maps to. No-op if the slot/client isn't present.""" slot = self.get_slot(address) if slot is not None and slot.client is not None: try: slot.client.set_identify(on) except Exception: pass # ---- Streaming ---- def send_event(self, action: int, hid_code: int) -> int: """Fan-out one event to every enabled, connected slot. Returns the count of slots the event was delivered to. Timestamp is captured per-slot (each BLELiveKeystrokeClient anchors its own session clock on first send), so each slot sees consistent t=0...delta_t cadence even if they were added at different times. """ n = 0 for slot in self.slots(): if not slot.enabled or slot.client is None: continue if slot.client.is_connected(): if slot.client.send_event(action, hid_code): n += 1 return n def send_event_with_t(self, action: int, hid_code: int, t_ms: int) -> int: """Replay path: fan-out an event with its ORIGINAL timestamp instead of "now". Used by the replay loop to send recorded events in their original cadence.""" n = 0 for slot in self.slots(): if not slot.enabled or slot.client is None: continue if slot.client.is_connected(): if slot.client.send_event_with_t(action, hid_code, t_ms): n += 1 return n def send_mouse(self, buttons: int, x: float, y: float, wheel: int = 0) -> int: """Fan-out one absolute pointer update to every enabled, connected slot. ``x``/``y`` are normalized [0,1] screen coordinates so every device's cursor lands at the same relative position with no drift. Returns the count of slots it reached.""" n = 0 for slot in self.slots(): if not slot.enabled or slot.client is None: continue if slot.client.is_connected(): if slot.client.send_mouse(buttons, x, y, wheel): n += 1 return n def reset_session_clocks(self) -> None: """Reset every slot's session anchor. Called at the start of a replay so each slot's t=0 corresponds to the first replay event.""" for slot in self.slots(): if slot.client is not None: try: slot.client.reset_session_clock() except Exception: pass # ---- Stats ---- def stats(self) -> list[dict]: """Snapshot of per-slot stats for the UI.""" out = [] for slot in self.slots(): client = slot.client out.append({ "address": slot.address, "label": slot.display_label(), "enabled": slot.enabled, "status": slot.status, "events_sent": getattr(client, "events_sent", 0) if client else 0, "bytes_sent": getattr(client, "bytes_sent", 0) if client else 0, }) return out # ---- Shutdown ---- def shutdown(self) -> None: """Stop every BLE worker. Called when the BT Keyboard window closes or the app exits.""" with self._lock: slots = list(self._slots.values()) self._slots.clear() for slot in slots: if slot.client is not None: try: slot.client.stop(timeout=2.0) except Exception: pass # ---- Internal ---- def _start_slot(self, slot: DeviceSlot) -> None: client = slot.client if client is None: return def on_status(status: str, _addr=slot.address): with self._lock: s = self._slots.get(_addr) if s is None: return s.status = status s.last_status_change = time.monotonic() if self._on_status_change: try: self._on_status_change(_addr, status) except Exception: pass def on_error(err_code, ref_seq, label, _addr=slot.address): # Surface as a status update for now; the UI can decide to # render it differently if it tracks the most recent error. if self._on_status_change: try: self._on_status_change(_addr, f"error:{label}") except Exception: pass client.start(on_status=on_status, on_error=on_error)