299 lines
11 KiB
Python
299 lines
11 KiB
Python
"""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)
|