"""Keyboard β€” multi-device live keystroke streamer (BLE or ESP-NOW hub). A standalone Toplevel window. Lets the user stream their host keyboard to many M5Stack ATOMS3 / ATOMS3 Lite devices simultaneously, with per- device enable/disable, recording, and replay. Transport is chosen per session (the Keyboard button pops a mode picker): * "ble" β€” one direct BLE link per device (MultiBleKeyboardManager). Simple, no USB hub required, but Windows only holds ~3-4 reliable concurrent links, so it soft-caps at 4 devices. * "hub" β€” the window borrows the app's USB serial link and switches the plugged-in device into ESP-NOW HUB mode. The hub broadcasts the stream to every node over ESP-NOW (no Bluetooth, no WiFi AP), sidestepping the BLE ceiling and soft-capping at 12 devices. On close the hub reverts to a normal node and the serial port is handed back to the app. Both managers expose the same public surface (see mesh_manager's module docstring), so streaming / recording / replay / profiles are identical regardless of the transport the user picked. UX overview (left panel, sectioned): CONNECTION: [ Discover ] [ Stream ] MACRO: [ ● Record ] [ Macros | πŸ“ ] Record captures keys + mouse (trackpad) + Ctrl+Alt+Del; Stop & Save stores it into a folder in the macro library (bt_macros). The split Macros button runs the loaded macro (toggles to Stop while running); the πŸ“ opens the library (Quick Run / Load / Delete + Loop, and folder Create / Rename / Delete). PROFILES: [ πŸ’Ύ Save Profile ] [ πŸ“‚ Load Profile ] DEVICES: one row per BLE slot β€” [enable] label status ev bytes [Remove] Right panel: 16:9 virtual trackpad (absolute mouse to all devices) + a multi-line "type to all devices" box. Security: Each slot is its own BLELiveKeystrokeClient. Each client looks up the matching per-MAC key in ble_keystore and AES-GCM encrypts every frame under that key β€” same security model as the single-device macro recorder. Adding a second slot doesn't share a key between devices; each pair stays end-to-end encrypted with its own. """ from __future__ import annotations import time import tkinter as tk from tkinter import ttk, messagebox, simpledialog from typing import Optional import bt_macros import bt_profiles from mesh_link import MeshLink from mesh_manager import MeshKeyboardManager, MAX_SLOTS as HUB_MAX_SLOTS from ble_multi import MultiBleKeyboardManager, MAX_SLOTS as BLE_MAX_SLOTS from widgets.macro_library_picker import MacroLibraryPicker # Per-device JOIN timeout while loading a profile, before the # Skip/Retry dialog pops up. The mesh JOINs in ~tens of ms, so a node # that hasn't acked within a few seconds is genuinely out of range. LOAD_TIMEOUT_S = 8.0 try: from utils.win_keyboard_hook import ( WinKeyboardHook, is_supported as _winhook_supported, ) except ImportError: WinKeyboardHook = None def _winhook_supported() -> bool: return False # Tk mouse button number -> absolute-mouse bitmask (bit0 left, bit1 right, # bit2 middle). Tk uses 1=left, 2=middle, 3=right. _TK_BTN_TO_MASK = {1: 0x01, 2: 0x04, 3: 0x02} # Modifier / special HID usage codes used by the Ctrl+Alt+Del and type-text # helpers. HID_LCTRL = 0xE0 HID_LSHIFT = 0xE1 HID_LALT = 0xE2 HID_DELETE = 0x4C def _build_char_map() -> dict: """ASCII char -> (HID usage code, needs_shift) for a US keyboard.""" m: dict = {} for c in range(ord('a'), ord('z') + 1): code = 0x04 + (c - ord('a')) m[chr(c)] = (code, False) m[chr(c).upper()] = (code, True) for i, ch in enumerate("1234567890"): m[ch] = (0x1E + i, False) for i, ch in enumerate("!@#$%^&*()"): m[ch] = (0x1E + i, True) m.update({ ' ': (0x2C, False), '\n': (0x28, False), '\r': (0x28, False), '\t': (0x2B, False), '-': (0x2D, False), '=': (0x2E, False), '[': (0x2F, False), ']': (0x30, False), '\\': (0x31, False), ';': (0x33, False), "'": (0x34, False), '`': (0x35, False), ',': (0x36, False), '.': (0x37, False), '/': (0x38, False), '_': (0x2D, True), '+': (0x2E, True), '{': (0x2F, True), '}': (0x30, True), '|': (0x31, True), ':': (0x33, True), '"': (0x34, True), '~': (0x35, True), '<': (0x36, True), '>': (0x37, True), '?': (0x38, True), }) return m _CHAR_TO_HID = _build_char_map() class _NullManager: """No-op stand-in used when the mesh hub couldn't be brought up (no device on USB). Keeps every send/query call site working β€” they just reach zero devices β€” so the window stays interactive and the user can fix the connection and reopen.""" def slots(self): return [] def stats(self): return [] def take_beacons(self): return [] def slot_count(self): return 0 def over_soft_limit(self): return False def set_callbacks(self, **_kw): pass def set_hub(self, _mac): pass def add_device(self, *a, **k): return None def get_slot(self, _addr): return None def remove_device(self, *a, **k): return False def set_enabled(self, *a, **k): pass def set_label(self, *a, **k): pass def identify(self, *a, **k): pass def send_event(self, *a, **k): return 0 def send_event_with_t(self, *a, **k): return 0 def send_mouse(self, *a, **k): return 0 def reset_session_clocks(self): pass def shutdown(self): pass def choose_keyboard_mode(parent) -> Optional[str]: """Modal transport picker shown when the Keyboard button is clicked. Presents the two streaming modes with their soft device caps and returns "ble", "hub", or None if the user cancels.""" dlg = tk.Toplevel(parent) dlg.title("Keyboard β€” choose a mode") dlg.configure(bg="#2D2D3D") dlg.transient(parent.winfo_toplevel()) dlg.resizable(False, False) dlg.grab_set() choice = {"mode": None} def _pick(mode): choice["mode"] = mode dlg.destroy() tk.Label(dlg, text="How do you want to reach your devices?", bg="#2D2D3D", fg="white", font=("Segoe UI", 12, "bold")).pack(padx=22, pady=(16, 4)) tk.Label(dlg, text="Pick one transport for this session. Reopen this window " "to switch modes.", bg="#2D2D3D", fg="#94A3B8", font=("Segoe UI", 9), wraplength=440, justify="left").pack(padx=22, pady=(0, 10)) def _card(title, desc, cap, accent, active, btn_text, mode): card = tk.Frame(dlg, bg="#1E1E2E", highlightbackground=accent, highlightthickness=1) card.pack(fill="x", padx=22, pady=6) tk.Label(card, text=title, bg="#1E1E2E", fg=accent, font=("Segoe UI", 11, "bold")).pack(anchor="w", padx=14, pady=(10, 2)) tk.Label(card, text=desc, bg="#1E1E2E", fg="#D1D5DB", font=("Segoe UI", 9), wraplength=380, justify="left").pack(anchor="w", padx=14) tk.Label(card, text=f"Recommended maximum connections: {cap}", bg="#1E1E2E", fg="#F59E0B", font=("Segoe UI", 9, "bold")).pack( anchor="w", padx=14, pady=(6, 2)) tk.Button(card, text=btn_text, bg=accent, fg="white", activebackground=active, font=("Segoe UI", 9, "bold"), relief="flat", padx=16, pady=4, command=lambda: _pick(mode)).pack(anchor="e", padx=14, pady=(0, 12)) _card("πŸ”΅ Bluetooth (BLE)", "Your PC connects to each device directly over its own Bluetooth " "Low Energy link. No hub device β€” the PC drives every device itself.", BLE_MAX_SLOTS, "#3B82F6", "#2563EB", "Use Bluetooth", "ble") _card("πŸ“‘ ESP-NOW Hub", "The M5Stack plugged into your PC becomes a hub and broadcasts your " "keystrokes to every other device over Wi-Fi (ESP-NOW). No pairing.", HUB_MAX_SLOTS, "#7C3AED", "#6D28D9", "Use Hub", "hub") tk.Button(dlg, text="Cancel", bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=16, command=dlg.destroy).pack(pady=(2, 14)) # Center over the parent window. dlg.update_idletasks() try: top = parent.winfo_toplevel() x = top.winfo_rootx() + (top.winfo_width() - dlg.winfo_width()) // 2 y = top.winfo_rooty() + (top.winfo_height() - dlg.winfo_height()) // 3 dlg.geometry(f"+{max(0, x)}+{max(0, y)}") except Exception: pass parent.wait_window(dlg) return choice["mode"] class BtKeyboardWindow(tk.Toplevel): """Multi-device live keystroke streaming dialog (BLE or ESP-NOW hub).""" ACTION_DOWN = 0 ACTION_UP = 1 def __init__(self, parent, serial_manager, mode="hub", pause_var_sync=None, resume_var_sync=None, pause_port_watcher=None, resume_port_watcher=None): super().__init__(parent) self.parent = parent self.serial_manager = serial_manager # Transport for this session: "ble" (direct per-device BLE) or # "hub" (USB device switched into an ESP-NOW broadcast hub). self._mode = "ble" if mode == "ble" else "hub" self._soft_cap = BLE_MAX_SLOTS if self._mode == "ble" else HUB_MAX_SLOTS self._pause_var_sync = pause_var_sync self._resume_var_sync = resume_var_sync self._pause_port_watcher = pause_port_watcher self._resume_port_watcher = resume_port_watcher self._var_sync_paused = False # Bring up the chosen transport and wire the manager to it. On # failure (no device plugged in / handshake error / no BLE adapter) # the window still opens but in a clearly-disabled state. self._link: MeshLink | None = None self.manager = None self._hub_error: str | None = None if self._mode == "hub": self._start_hub() else: self._start_ble() if self.manager is None: self.manager = _NullManager() else: self.manager.set_callbacks( on_status_change=lambda addr, status: self.after(0, self._refresh_slots), ) # Capture state self._streaming = False self._recording = False self._record_t0_ns: Optional[int] = None # Recorded macro buffer β€” tagged events mixing keys + mouse + CAD: # ["k", t_ms, action, hid] | ["m", t_ms, buttons, x, y, wheel] # (x/y absolute 0..32767, so playback is desync-proof.) self._record_buffer: list = [] # Dedupes auto-repeat (hooks fire repeatedly on a held key). self._held_codes: set[int] = set() # Macro library: the macro Loaded into the Macros button, and the # after()-driven playback state machine (Quick Run / loaded run). self._loaded_macro = None # {"folder","name","events","loop"} | None self._macro_running = False self._macro_events: list = [] self._macro_loop = False self._macro_i = 0 self._macro_start = 0.0 self._macro_held_keys: set[int] = set() self._macro_held_buttons = 0 self._macro_last_xy = (0.5, 0.5) # Virtual-trackpad state self._mouse_buttons = 0 # current abs-mouse button bitmask self._pad_box = None # (x0, y0, w, h) of the 16:9 region self._last_move_send = 0.0 # monotonic ts of last move frame sent self._fullscreen = True # Low-level Windows hook (None on other platforms or if install # fails). Installed on demand when streaming OR recording starts. self._win_hook: Optional[WinKeyboardHook] = None # True while the global keyboard hook is temporarily paused because # the user is typing in the multi-line text box. We re-arm it when # focus leaves the box (if streaming/recording is still active). self._hook_paused_for_text = False # True while capture is paused because a modal dialog that needs # keyboard input (e.g. the macro Save dialog, the library picker's # folder prompts) is open. Re-armed when the dialog closes. self._capture_paused_for_dialog = False # Profile-load state machine (after()-driven so the UI stays live). self._load_queue = None # list of {"address","label"} or None self._load_idx = 0 self._load_deadline = 0.0 self._build_ui() self._refresh_slots() # Poll stats / status periodically so the UI shows live counts. self._stats_job = self.after(500, self._tick_stats) self.protocol("WM_DELETE_WINDOW", self._on_close) # Surface a transport-setup failure once the window is up. if self._hub_error: title = ("Mesh hub not ready" if self._mode == "hub" else "Bluetooth not ready") self.after(150, lambda: messagebox.showwarning( title, self._hub_error, parent=self)) # ---- UI construction ---- def _build_ui(self): mode_name = "Bluetooth" if self._mode == "ble" else "ESP-NOW hub" self.title(f"Keyboard β€” multi-device streamer ({mode_name})") self.configure(bg="#2D2D3D") self.transient(self.parent.winfo_toplevel()) # Open full-screen; Exit Fullscreen button (and Escape) restore it. self._fullscreen = True try: self.attributes("-fullscreen", True) except tk.TclError: self.state("zoomed") self.bind("", lambda _e: self._exit_fullscreen()) # ---- Top bar (full width) ---- top = tk.Frame(self, bg="#1E1E2E") top.pack(fill="x", side="top") tk.Label(top, text=f"Keyboard β€” {mode_name}", bg="#1E1E2E", fg="white", font=("Segoe UI", 13, "bold")).pack( side="left", padx=14, pady=8) self.capture_var = tk.StringVar(value="Capture: OFF") tk.Label(top, textvariable=self.capture_var, bg="#1E1E2E", fg="#94A3B8", font=("Segoe UI", 10, "bold")).pack( side="left", padx=12) self.fs_btn = tk.Button(top, text="Exit Fullscreen", bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=12, command=self._toggle_fullscreen) self.fs_btn.pack(side="right", padx=(6, 14), pady=6) tk.Button(top, text="Ctrl+Alt+Del", bg="#B91C1C", fg="white", activebackground="#991B1B", font=("Segoe UI", 9, "bold"), relief="flat", padx=12, command=self._send_cad).pack(side="right", padx=6, pady=6) tk.Button(top, text="Close", bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=12, command=self._on_close).pack(side="right", padx=6, pady=6) # ---- Main split: left 1/3 devices, right 2/3 trackpad + text ---- main = tk.Frame(self, bg="#2D2D3D") main.pack(fill="both", expand=True) main.columnconfigure(0, weight=1, uniform="cols") main.columnconfigure(1, weight=2, uniform="cols") main.rowconfigure(0, weight=1) left = tk.Frame(main, bg="#2D2D3D") left.grid(row=0, column=0, sticky="nsew", padx=(10, 6), pady=10) right = tk.Frame(main, bg="#2D2D3D") right.grid(row=0, column=1, sticky="nsew", padx=(6, 10), pady=10) self._build_left(left) self._build_right(right) def _section(self, left, title): """A titled, visually-grouped section; returns the inner frame to pack buttons into.""" tk.Label(left, text=title, bg="#2D2D3D", fg="#9CA3AF", font=("Segoe UI", 9, "bold")).pack(anchor="w", pady=(8, 2)) outer = tk.Frame(left, bg="#1E1E2E") outer.pack(fill="x") inner = tk.Frame(outer, bg="#1E1E2E") inner.pack(fill="x", padx=6, pady=6) return inner def _build_left(self, left): tk.Label(left, text=("Leave each M5Stack idle on its macro selector, then " "Discover β€” they connect automatically."), bg="#2D2D3D", fg="#94A3B8", font=("Segoe UI", 8), wraplength=340, justify="left").pack(anchor="w", pady=(0, 2)) # --- Section: Connection --- conn = self._section(left, "CONNECTION") self.discover_btn = tk.Button(conn, text="Discover", bg="#0EA5E9", fg="white", activebackground="#0284C7", font=("Segoe UI", 9, "bold"), relief="flat", padx=10, pady=3, command=self._discover) self.discover_btn.pack(side="left") self.stream_btn = tk.Button(conn, text="β–Ά Stream", bg="#22C55E", fg="white", activebackground="#16A34A", font=("Segoe UI", 9, "bold"), relief="flat", padx=10, pady=3, command=self._toggle_stream) self.stream_btn.pack(side="left", padx=(6, 0)) # --- Section: Macro --- mac = self._section(left, "MACRO") self.rec_btn = tk.Button(mac, text="● Record", bg="#DC2626", fg="white", activebackground="#B91C1C", font=("Segoe UI", 9, "bold"), relief="flat", padx=10, pady=3, command=self._toggle_record) self.rec_btn.pack(side="left") # Split "Macros | πŸ“" button: left runs the loaded macro (toggles to # Stop while running); folder icon opens the library. split = tk.Frame(mac, bg="#1E1E2E") split.pack(side="left", padx=(8, 0)) self.macros_run_btn = tk.Button(split, text="Macros", bg="#A78BFA", fg="#1E1E2E", activebackground="#8B5CF6", font=("Segoe UI", 9, "bold"), relief="flat", padx=10, pady=3, state="disabled", command=self._on_macros_run) self.macros_run_btn.pack(side="left") self.macros_folder_btn = tk.Button(split, text="πŸ“", bg="#7C3AED", fg="white", activebackground="#6D28D9", font=("Segoe UI", 10, "bold"), relief="flat", padx=8, pady=3, command=self._open_macro_library) self.macros_folder_btn.pack(side="left", padx=(1, 0)) # --- Section: Profiles --- prof = self._section(left, "PROFILES") tk.Button(prof, text="πŸ’Ύ Save Profile", bg="#0F766E", fg="white", activebackground="#0D5C56", font=("Segoe UI", 9, "bold"), relief="flat", padx=10, pady=3, command=self._save_profile).pack(side="left") tk.Button(prof, text="πŸ“‚ Load Profile", bg="#1D4ED8", fg="white", activebackground="#1E40AF", font=("Segoe UI", 9, "bold"), relief="flat", padx=10, pady=3, command=self._load_profile).pack(side="left", padx=(6, 0)) # --- Device list (header + body) --- tk.Label(left, text="DEVICES", bg="#2D2D3D", fg="#9CA3AF", font=("Segoe UI", 9, "bold")).pack(anchor="w", pady=(8, 2)) # Brief, non-blocking warning shown when more than the current # mode's soft cap of devices is connected (the count is uncapped). self._warn_var = tk.StringVar(value="") tk.Label(left, textvariable=self._warn_var, bg="#2D2D3D", fg="#F59E0B", font=("Segoe UI", 8, "bold"), wraplength=340, justify="left").pack(anchor="w") list_frame = tk.Frame(left, bg="#2D2D3D") list_frame.pack(fill="both", expand=True) header = tk.Frame(list_frame, bg="#1E1E2E") header.pack(fill="x") for text, w in [("On", 4), ("Address / Label", 24), ("Status", 13), ("Lag", 7), ("Events", 9), ("Bytes", 9), ("", 8)]: tk.Label(header, text=text, bg="#1E1E2E", fg="#9CA3AF", font=("Segoe UI", 9, "bold"), width=w, anchor="w").pack(side="left", padx=4, pady=4) body = tk.Frame(list_frame, bg="#1E1E2E") body.pack(fill="both", expand=True) self.slots_frame = body self.recording_status_var = tk.StringVar( value="No recording in buffer.") tk.Label(left, textvariable=self.recording_status_var, bg="#2D2D3D", fg="#9CA3AF", font=("Segoe UI", 9)).pack(anchor="w", pady=(6, 0)) def _build_right(self, right): right.rowconfigure(0, weight=1) right.columnconfigure(0, weight=1) pad_wrap = tk.Frame(right, bg="#2D2D3D") pad_wrap.grid(row=0, column=0, sticky="nsew") tk.Label(pad_wrap, text=("Virtual trackpad β€” controls every connected device " "(absolute position, no desync)"), bg="#2D2D3D", fg="#94A3B8", font=("Segoe UI", 9)).pack(anchor="w", pady=(0, 4)) self.canvas = tk.Canvas(pad_wrap, bg="#11131A", highlightthickness=1, highlightbackground="#3B3B52", cursor="tcross") self.canvas.pack(fill="both", expand=True) self.canvas.bind("", self._on_pad_configure) for seq in ("", "", "", ""): self.canvas.bind(seq, self._on_pad_motion) for n in (1, 2, 3): self.canvas.bind(f"", self._on_pad_press) self.canvas.bind(f"", self._on_pad_release) self.canvas.bind("", self._on_pad_wheel) # ---- Type-to-all row ---- # Multi-line text box capped at 5 visible lines with a scrollbar # (infinite lines via scroll). Focusing it pauses the global key # capture so the user can type into the box without it being # streamed; leaving it re-arms capture if streaming/recording. text_row = tk.Frame(right, bg="#2D2D3D") text_row.grid(row=1, column=0, sticky="ew", pady=(8, 0)) text_row.columnconfigure(0, weight=1) tk.Label(text_row, text="Type to all devices (Ctrl+Enter or Send):", bg="#2D2D3D", fg="white", font=("Segoe UI", 9)).grid( row=0, column=0, columnspan=2, sticky="w", pady=(0, 2)) self.text_entry = tk.Text(text_row, height=5, wrap="word", bg="#1E1E2E", fg="white", insertbackground="white", font=("Segoe UI", 10), relief="flat", highlightthickness=1, highlightbackground="#3B3B52", highlightcolor="#3B82F6") self.text_entry.grid(row=1, column=0, sticky="ew") text_sb = tk.Scrollbar(text_row, orient="vertical", command=self.text_entry.yview) text_sb.grid(row=1, column=1, sticky="ns") self.text_entry.config(yscrollcommand=text_sb.set) self.text_entry.bind("", self._on_textbox_focus_in) self.text_entry.bind("", self._on_textbox_focus_out) # Ctrl+Enter sends; plain Enter inserts a newline. self.text_entry.bind("", self._on_textbox_send_key) tk.Button(text_row, text="Send", bg="#22C55E", fg="white", activebackground="#16A34A", font=("Segoe UI", 9, "bold"), relief="flat", padx=16, command=self._send_text).grid( row=1, column=2, sticky="ns", padx=(8, 0)) # ---- Fullscreen control ---- def _toggle_fullscreen(self): self._set_fullscreen(not self._fullscreen) def _exit_fullscreen(self): if self._fullscreen: self._set_fullscreen(False) def _set_fullscreen(self, on: bool): self._fullscreen = bool(on) try: self.attributes("-fullscreen", self._fullscreen) except tk.TclError: try: self.state("zoomed" if self._fullscreen else "normal") except tk.TclError: pass if not self._fullscreen: self.geometry("1100x720") try: self.fs_btn.config( text="Exit Fullscreen" if self._fullscreen else "Fullscreen") except tk.TclError: pass # ---- Ctrl+Alt+Del ---- def _emit_key_action(self, action, hid): """Send a key to devices AND record it if recording. Used by the Ctrl+Alt+Del button (whose chord the OS would otherwise swallow).""" if self._recording and self._record_t0_ns is not None: t_ms = (time.monotonic_ns() - self._record_t0_ns) // 1_000_000 self._record_buffer.append(["k", int(t_ms), int(action), int(hid)]) self.manager.send_event(action, hid) def _send_cad(self): """Send Ctrl+Alt+Del to every connected device (chord then release).""" for h in (HID_LCTRL, HID_LALT, HID_DELETE): self._emit_key_action(self.ACTION_DOWN, h) self.after(60, self._release_cad) def _release_cad(self): for h in (HID_DELETE, HID_LALT, HID_LCTRL): self._emit_key_action(self.ACTION_UP, h) # ---- Virtual trackpad ---- def _on_pad_configure(self, _e=None): self._recompute_pad_box() self._draw_pad() def _recompute_pad_box(self): try: cw = self.canvas.winfo_width() ch = self.canvas.winfo_height() except tk.TclError: return if cw < 8 or ch < 8: self._pad_box = None return # Largest 16:9 box centered in the canvas. if cw / ch > 16 / 9: h = ch w = int(h * 16 / 9) else: w = cw h = int(w * 9 / 16) self._pad_box = ((cw - w) // 2, (ch - h) // 2, w, h) def _draw_pad(self, cursor=None): c = self.canvas c.delete("all") if not self._pad_box: return x0, y0, w, h = self._pad_box c.create_rectangle(x0, y0, x0 + w, y0 + h, outline="#3B82F6", width=2, fill="#0B0D14") c.create_text(x0 + w // 2, y0 + 16, text="16:9 β€” move / click / scroll here", fill="#475569", font=("Segoe UI", 9)) if cursor is not None: cxp, cyp = cursor c.create_line(cxp - 9, cyp, cxp + 9, cyp, fill="#22C55E") c.create_line(cxp, cyp - 9, cxp, cyp + 9, fill="#22C55E") def _pad_norm(self, ev): if not self._pad_box: return None x0, y0, w, h = self._pad_box xn = (ev.x - x0) / w yn = (ev.y - y0) / h if xn < 0 or xn > 1 or yn < 0 or yn > 1: return None return xn, yn def _clamp_norm(self, ev): if not self._pad_box: return None x0, y0, w, h = self._pad_box return (min(1.0, max(0.0, (ev.x - x0) / w)), min(1.0, max(0.0, (ev.y - y0) / h))) def _maybe_record_mouse(self, buttons, n, wheel): """Append a mouse event to the record buffer (absolute 0..32767).""" if n is None or not (self._recording and self._record_t0_ns is not None): return t_ms = (time.monotonic_ns() - self._record_t0_ns) // 1_000_000 x = int(round(max(0.0, min(1.0, n[0])) * 32767)) y = int(round(max(0.0, min(1.0, n[1])) * 32767)) self._record_buffer.append( ["m", int(t_ms), int(buttons) & 0x7, x, y, int(wheel)]) def _on_pad_motion(self, ev): n = self._pad_norm(ev) if n is None: return now = time.monotonic() # Rate-limit moves (~33 Hz). Absolute positioning means a skipped # move is harmless β€” the next one re-pins every cursor. if now - self._last_move_send < 0.03: return self._last_move_send = now self.manager.send_mouse(self._mouse_buttons, n[0], n[1], 0) self._maybe_record_mouse(self._mouse_buttons, n, 0) self._draw_pad((ev.x, ev.y)) def _on_pad_press(self, ev): self.canvas.focus_set() mask = _TK_BTN_TO_MASK.get(ev.num, 0) if not mask: return self._mouse_buttons |= mask n = self._pad_norm(ev) or self._clamp_norm(ev) if n is not None: self.manager.send_mouse(self._mouse_buttons, n[0], n[1], 0) self._maybe_record_mouse(self._mouse_buttons, n, 0) def _on_pad_release(self, ev): mask = _TK_BTN_TO_MASK.get(ev.num, 0) if not mask: return self._mouse_buttons &= ~mask n = self._pad_norm(ev) or self._clamp_norm(ev) if n is not None: self.manager.send_mouse(self._mouse_buttons, n[0], n[1], 0) self._maybe_record_mouse(self._mouse_buttons, n, 0) def _on_pad_wheel(self, ev): n = self._pad_norm(ev) or self._clamp_norm(ev) if n is None: return ticks = int(ev.delta / 120) if ev.delta else 0 if ticks == 0: ticks = 1 if ev.delta > 0 else -1 self.manager.send_mouse(self._mouse_buttons, n[0], n[1], ticks) self._maybe_record_mouse(self._mouse_buttons, n, ticks) # ---- Type text to all devices ---- def _on_textbox_send_key(self, _e=None): self._send_text() return "break" # don't also insert a newline def _on_textbox_focus_in(self, _e=None): # Pause global key capture so typing goes into the box, not the # devices. Remember to re-arm on focus-out. if self._win_hook is not None: self._uninstall_hook() self._hook_paused_for_text = True def _on_textbox_focus_out(self, _e=None): if self._hook_paused_for_text: self._hook_paused_for_text = False if self._streaming or self._recording: self._install_hook() def _send_text(self): txt = self.text_entry.get("1.0", "end-1c") if not txt: return events = [] for ch in txt: m = _CHAR_TO_HID.get(ch) if m is None: continue hid, shift = m if shift: events.append((self.ACTION_DOWN, HID_LSHIFT)) events.append((self.ACTION_DOWN, hid)) events.append((self.ACTION_UP, hid)) if shift: events.append((self.ACTION_UP, HID_LSHIFT)) if not events: return # Leave the text in the box after sending so the user can resend or # edit it without retyping. self._pump_text_events(events, 0) def _pump_text_events(self, events, i): if i >= len(events) or not self._dialog_alive(): return action, hid = events[i] self.manager.send_event(action, hid) # Pace so target apps don't coalesce a fast burst into dropped keys. self.after(6, lambda: self._pump_text_events(events, i + 1)) # ---- Slot rendering ---- def _refresh_slots(self): # Wipe and re-render. Cheap (small N), avoids stale callbacks. for child in self.slots_frame.winfo_children(): child.destroy() slots = self.manager.slots() # Soft-limit warning (non-blocking) β€” the count is uncapped; this # just flags when the current mode's soft cap is exceeded. if hasattr(self, "_warn_var"): n = len(slots) if n > self._soft_cap: if self._mode == "hub": self._warn_var.set( f"⚠ {n} devices (soft cap {self._soft_cap}) β€” watch the " f"Lag column; if it climbs, try a clearer Wi-Fi channel " f"in Settings.") else: self._warn_var.set( f"⚠ {n} devices (soft cap {self._soft_cap}) β€” BLE " f"bandwidth and latency degrade past this; expect lag.") else: self._warn_var.set("") if not slots: tk.Label(self.slots_frame, text="No devices connected. Click Discover to add one.", bg="#1E1E2E", fg="#6B7280", font=("Segoe UI", 9, "italic"), pady=20).pack(fill="x") return stats_by_addr = {s["address"]: s for s in self.manager.stats()} for slot in slots: row = tk.Frame(self.slots_frame, bg="#1E1E2E") row.pack(fill="x", pady=1) stats = stats_by_addr.get(slot.address, {}) # Enable toggle enable_var = tk.BooleanVar(value=slot.enabled) def make_toggle(addr=slot.address, var=enable_var): def cb(): self.manager.set_enabled(addr, var.get()) return cb tk.Checkbutton(row, variable=enable_var, bg="#1E1E2E", activebackground="#1E1E2E", selectcolor="#1E1E2E", fg="#22C55E", command=make_toggle(), width=2).pack(side="left", padx=4) # Label / address label_text = slot.display_label() tk.Label(row, text=label_text, bg="#1E1E2E", fg="white", font=("Consolas", 9), width=26, anchor="w").pack(side="left", padx=4) # Status status = stats.get("status", slot.status) status_color = { "connected": "#22C55E", "connecting": "#F59E0B", "lagging": "#F59E0B", "scanning": "#F59E0B", "disconnected": "#EF4444", "error": "#EF4444", }.get(status, "#94A3B8") tk.Label(row, text=status, bg="#1E1E2E", fg=status_color, font=("Segoe UI", 9, "bold"), width=13, anchor="w").pack(side="left", padx=4) # ACK lag (events the device is behind the head of the stream). # Only the ESP-NOW hub reports this; BLE mode shows a dash. lag = stats.get("ack_lag") lag_text = "β€”" if lag is None else str(lag) over = lag is not None and lag > 32 tk.Label(row, text=lag_text, bg="#1E1E2E", fg=("#EF4444" if over else "#D1D5DB"), font=("Consolas", 9), width=7, anchor="w").pack(side="left", padx=4) # Counters tk.Label(row, text=str(stats.get("events_sent", 0)), bg="#1E1E2E", fg="#D1D5DB", font=("Consolas", 9), width=9, anchor="w").pack(side="left", padx=4) tk.Label(row, text=str(stats.get("bytes_sent", 0)), bg="#1E1E2E", fg="#D1D5DB", font=("Consolas", 9), width=9, anchor="w").pack(side="left", padx=4) tk.Button(row, text="Remove", bg="#4A4A6A", fg="white", font=("Segoe UI", 8), relief="flat", padx=8, command=lambda a=slot.address: self._remove_slot(a) ).pack(side="left", padx=4) def _tick_stats(self): if not self._dialog_alive(): return try: self._refresh_slots() except Exception: pass self._stats_job = self.after(750, self._tick_stats) # ---- Discover ---- def _discover(self): """Discover idle, in-range devices and let the user add one. The hub polls the ESP-NOW channel; BLE mode scans with Bleak. Both funnel into the shared _discover_done picker.""" if self._mode == "hub": self._discover_mesh() else: self._discover_ble() def _discover_mesh(self): """Ask the hub to poll the mesh for idle nodes, collect the beacons they send back for ~3 s, then present anything not already a slot. No Bluetooth involved β€” discovery rides the same ESP-NOW channel the keystroke stream uses.""" if isinstance(self.manager, _NullManager) or self._link is None: messagebox.showwarning( "Mesh hub not ready", self._hub_error or "The mesh hub isn't connected.", parent=self) return # Drain any stale beacons, turn on discovery polling, collect. self.manager.take_beacons() self._link.send_json({"cmd": "mesh_poll", "on": True}) self.discover_btn.config(state="disabled", text="Scanning...") self.after(3000, self._discover_collect) def _discover_ble(self): """One-shot Bleak scan for idle live-mode devices (those advertising LIVE_SERVICE_UUID). Runs on a background thread so Tk stays live.""" if isinstance(self.manager, _NullManager): messagebox.showwarning( "Bluetooth not ready", self._hub_error or "BLE streaming isn't available.", parent=self) return # Pause var-sync during the scan so its scanner doesn't fight ours # for the BLE adapter (idempotent β€” already paused on open). self._maybe_pause_var_sync() import threading self.discover_btn.config(state="disabled", text="Scanning...") threading.Thread(target=self._discover_worker, daemon=True).start() def _discover_worker(self): import asyncio try: from bleak import BleakScanner from ble_server import LIVE_SERVICE_UUID except ImportError as exc: self.after(0, lambda e=exc: self._discover_done([], str(e))) return async def go(): seen = {} def cb(d, adv): if LIVE_SERVICE_UUID in (adv.service_uuids or []): seen[d.address.upper()] = { "address": d.address, "name": d.name, } scanner = BleakScanner(detection_callback=cb) await scanner.start() await asyncio.sleep(4.0) await scanner.stop() return list(seen.values()) try: results = asyncio.run(go()) except Exception as exc: self.after(0, lambda e=exc: self._discover_done([], repr(e))) return self.after(0, lambda: self._discover_done(results, None)) def _discover_collect(self): if not self._dialog_alive(): return if self._link is not None: self._link.send_json({"cmd": "mesh_poll", "on": False}) results = self.manager.take_beacons() if self.manager else [] self._discover_done(results, None) def _discover_done(self, results, err): self.discover_btn.config(state="normal", text="Discover") if err: messagebox.showerror("Discover failed", err, parent=self) return # Filter out devices we're already tracking. existing = {s.address.upper() for s in self.manager.slots()} candidates = [r for r in results if r["address"].upper() not in existing] if not candidates: messagebox.showinfo( "Nothing new found", "No additional M5Stack devices found in range.\n\n" "Make sure each device is powered on and idle on its macro " "selector (not running a routine or mid-USB-upload).", parent=self) return # Quick picker: a small Toplevel listing addresses; pick adds it. self._open_picker(candidates) def _open_picker(self, candidates): picker = tk.Toplevel(self) picker.title("Pick a device to add") picker.configure(bg="#2D2D3D") picker.transient(self) picker.grab_set() picker.geometry("420x320") tk.Label(picker, text="Select a device to add to the stream.", bg="#2D2D3D", fg="white", font=("Segoe UI", 10)).pack(padx=14, pady=(12, 6)) listbox = tk.Listbox(picker, bg="#1E1E2E", fg="white", selectbackground="#3B82F6", font=("Consolas", 10), activestyle="none", relief="flat") listbox.pack(fill="both", expand=True, padx=14, pady=(0, 8)) for c in candidates: label = c["address"] extra = c.get("name") or c.get("board") if extra: label += f" ({extra})" listbox.insert("end", label) def add_selected(): sel = listbox.curselection() if not sel: return idx = sel[0] c = candidates[idx] # Add the device FIRST so its BLE client starts connecting, then # light up its screen with the Bluetooth logo so the user can # see which physical M5Stack they're labeling. The label prompt # is modal but the BLE worker keeps running in the background, so # the identify request is delivered as soon as the link is up. slot = self.manager.add_device(c["address"], label="", board=c.get("board", "")) if slot is None: messagebox.showerror( "Add failed", "Could not add this device (already tracked, or it's the " "hub).", parent=picker) return picker.destroy() self._refresh_slots() self.manager.identify(c["address"], True) try: label = simpledialog.askstring( "Label", "The selected M5Stack is showing a Bluetooth logo on its " f"screen.\n\nOptional friendly label for {c['address']}:", parent=self) finally: # Always clear the logo, even if the user cancels the prompt. self.manager.identify(c["address"], False) if label: self.manager.set_label(c["address"], label) self._refresh_slots() btn_row = tk.Frame(picker, bg="#2D2D3D") btn_row.pack(fill="x", padx=14, pady=(0, 12)) tk.Button(btn_row, text="Cancel", bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=14, command=picker.destroy).pack(side="right") tk.Button(btn_row, text="Add", bg="#22C55E", fg="white", font=("Segoe UI", 9, "bold"), relief="flat", padx=14, command=add_selected).pack(side="right", padx=(0, 6)) def _remove_slot(self, address: str): self.manager.remove_device(address) self._refresh_slots() # ---- Device profiles (save / load) ---- def _save_profile(self): slots = self.manager.slots() if not slots: messagebox.showinfo( "Nothing to save", "Add at least one device (Discover) before saving a profile.", parent=self) return name = simpledialog.askstring( "Save profile", "Profile name:", parent=self) if name is None: return name = name.strip() if not name: return if name in bt_profiles.list_profiles() and not messagebox.askyesno( "Overwrite?", f"A profile named '{name}' already exists. Overwrite it?", parent=self): return devices = [(s.address, s.display_label()) for s in slots] try: bt_profiles.save_profile(name, devices) except OSError as exc: messagebox.showerror("Save failed", str(exc), parent=self) return messagebox.showinfo( "Profile saved", f"Saved {len(devices)} device(s) to '{name}'.", parent=self) def _load_profile(self): if self._load_queue is not None: messagebox.showinfo("Busy", "A profile is still loading.", parent=self) return names = bt_profiles.list_profiles() if not names: messagebox.showinfo( "No profiles", "No saved profiles yet. Add devices and click Save Profile " "first.", parent=self) return self._open_profile_picker(names) def _open_profile_picker(self, names): picker = tk.Toplevel(self) picker.title("Load device profile") picker.configure(bg="#2D2D3D") picker.transient(self) picker.grab_set() picker.geometry("420x320") tk.Label(picker, text="Pick a profile to connect to:", bg="#2D2D3D", fg="white", font=("Segoe UI", 10)).pack(padx=14, pady=(12, 6)) listbox = tk.Listbox(picker, bg="#1E1E2E", fg="white", selectbackground="#3B82F6", font=("Consolas", 10), activestyle="none", relief="flat") listbox.pack(fill="both", expand=True, padx=14, pady=(0, 8)) def repopulate(sel_names): listbox.delete(0, "end") for nm in sel_names: devs = bt_profiles.load_profile(nm) listbox.insert("end", f"{nm} ({len(devs)} device(s))") current = list(names) repopulate(current) def do_load(): sel = listbox.curselection() if not sel: return nm = current[sel[0]] devs = bt_profiles.load_profile(nm) picker.destroy() if not devs: messagebox.showinfo("Empty profile", f"Profile '{nm}' has no devices.", parent=self) return self._begin_load(devs) def do_delete(): sel = listbox.curselection() if not sel: return nm = current[sel[0]] if not messagebox.askyesno("Delete profile?", f"Delete profile '{nm}'?", parent=picker): return bt_profiles.delete_profile(nm) current.clear() current.extend(bt_profiles.list_profiles()) repopulate(current) btn_row = tk.Frame(picker, bg="#2D2D3D") btn_row.pack(fill="x", padx=14, pady=(0, 12)) tk.Button(btn_row, text="Cancel", bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=14, command=picker.destroy).pack(side="right") tk.Button(btn_row, text="Load", bg="#22C55E", fg="white", font=("Segoe UI", 9, "bold"), relief="flat", padx=14, command=do_load).pack(side="right", padx=(0, 6)) tk.Button(btn_row, text="Delete", bg="#7F1D1D", fg="white", font=("Segoe UI", 9), relief="flat", padx=12, command=do_delete).pack(side="left") # ---- Profile load state machine (one device at a time) ---- def _mesh_addr(self, addr: str) -> str: """Map a stored profile address to the mesh (STA) MAC. New profiles store the STA MAC directly. Legacy profiles saved under the BLE transport stored the Bleak BT MAC, which on the ESP32-S3 default eFuse layout is the STA MAC + 2. We prefer an exact key-store match, then try the BT->STA (-2) arithmetic, else fall back to the stored value verbatim. In BLE mode there is no STA-MAC remap β€” the manager connects by the Bleak BT MAC, so the stored value is used as-is.""" if self._mode != "hub": return addr.upper() import ble_keystore a = addr.upper() known = {m.upper() for m in ble_keystore.all_known_macs()} if a in known: return a try: parts = [int(p, 16) for p in a.split(":")] val = int.from_bytes(bytes(parts), "big") - 2 cand = ":".join(f"{b:02X}" for b in val.to_bytes(6, "big")) if cand in known: return cand except (ValueError, OverflowError): pass return a def _begin_load(self, devices): # Pause var-sync so its scanner doesn't fight ours during the # sequential connects. self._maybe_pause_var_sync() # Migrate any legacy BLE-MAC profile entries to mesh STA MACs. migrated = [] for d in devices: migrated.append({"address": self._mesh_addr(d["address"]), "label": d.get("label", "")}) self._load_queue = list(migrated) self._load_idx = 0 self.recording_status_var.set( f"Loading profile β€” 0/{len(self._load_queue)} connected...") self._load_next() def _load_next(self): if not self._dialog_alive(): self._load_queue = None return q = self._load_queue if q is None: return if self._load_idx >= len(q): self._load_queue = None self._refresh_slots() self.recording_status_var.set( f"Profile loaded ({len(q)} device(s)).") return dev = q[self._load_idx] addr, label = dev["address"], dev["label"] slot = self.manager.get_slot(addr) if slot is None: slot = self.manager.add_device(addr, label=label) if slot is None: # Uncapped now β€” None means it's already tracked; just move on. self._load_idx += 1 self.after(50, self._load_next) return else: # Already tracked β€” just (re)apply the saved label. self.manager.set_label(addr, label) self._refresh_slots() self._load_deadline = time.monotonic() + LOAD_TIMEOUT_S self.after(200, self._load_poll) def _load_poll(self): if not self._dialog_alive(): self._load_queue = None return q = self._load_queue if q is None: return dev = q[self._load_idx] addr, label = dev["address"], dev["label"] slot = self.manager.get_slot(addr) connected = bool(slot and slot.status == "connected") if connected: # Push the saved name to the device so its screen shows it too. self.manager.set_label(addr, label) self._refresh_slots() self._load_idx += 1 self.recording_status_var.set( f"Loading profile β€” {self._load_idx}/{len(q)} connected...") self.after(150, self._load_next) return if time.monotonic() >= self._load_deadline: choice = self._ask_retry_skip(label or addr, addr) if choice == "retry": self._load_deadline = time.monotonic() + LOAD_TIMEOUT_S self.after(200, self._load_poll) elif choice == "skip": # Leave the slot in place β€” its worker keeps retrying in the # background β€” and move on to the next device. self._load_idx += 1 self.after(50, self._load_next) else: # cancel the whole load self._load_queue = None self._refresh_slots() self.recording_status_var.set("Profile load cancelled.") return self.after(200, self._load_poll) def _ask_retry_skip(self, label: str, address: str) -> str: """Modal shown when a device can't be reached during a load. Returns 'retry', 'skip', or 'cancel'.""" dlg = tk.Toplevel(self) dlg.title("Device not found") dlg.configure(bg="#2D2D3D") dlg.transient(self) dlg.grab_set() dlg.geometry("440x230") tk.Label(dlg, text="Couldn't connect to:", bg="#2D2D3D", fg="#F59E0B", font=("Segoe UI", 10, "bold")).pack( padx=16, pady=(16, 2), anchor="w") tk.Label(dlg, text=label, bg="#2D2D3D", fg="white", font=("Segoe UI", 11, "bold")).pack(padx=16, anchor="w") tk.Label(dlg, text=address, bg="#2D2D3D", fg="#94A3B8", font=("Consolas", 9)).pack(padx=16, anchor="w") tk.Label(dlg, text=("Make sure it's powered on and idle on its macro " "selector (not running a routine or mid-USB-upload), " "then Retry β€” or Skip it for now."), bg="#2D2D3D", fg="#CBD5E1", font=("Segoe UI", 9), wraplength=400, justify="left").pack(padx=16, pady=(8, 12), anchor="w") result = {"v": "skip"} def choose(v): result["v"] = v try: dlg.destroy() except tk.TclError: pass btns = tk.Frame(dlg, bg="#2D2D3D") btns.pack(fill="x", padx=16, pady=(0, 14)) tk.Button(btns, text="Retry", bg="#22C55E", fg="white", activebackground="#16A34A", font=("Segoe UI", 9, "bold"), relief="flat", padx=16, command=lambda: choose("retry") ).pack(side="left") tk.Button(btns, text="Skip", bg="#475569", fg="white", font=("Segoe UI", 9, "bold"), relief="flat", padx=16, command=lambda: choose("skip")).pack(side="left", padx=(8, 0)) tk.Button(btns, text="Cancel load", bg="#7F1D1D", fg="white", font=("Segoe UI", 9), relief="flat", padx=12, command=lambda: choose("cancel")).pack(side="right") dlg.protocol("WM_DELETE_WINDOW", lambda: choose("skip")) self.wait_window(dlg) return result["v"] # ---- Capture state ---- def _toggle_stream(self): if self._streaming and not self._recording: # Streaming-only -> off self._streaming = False if not self._recording: self._uninstall_hook() self.stream_btn.config(text="β–Ά Stream", bg="#22C55E") self.capture_var.set("Capture: OFF") else: self._streaming = True if self._install_hook(): self.stream_btn.config(text="β–  Stop streaming", bg="#0EA5E9") self._update_capture_label() else: self._streaming = False messagebox.showerror( "Hook install failed", "Couldn't install the system keyboard hook. Streaming " "needs it to capture the Windows key and suppress " "local OS shortcuts while typing.", parent=self) def _toggle_record(self): if self._recording: self._stop_recording() else: self._start_recording() def _start_recording(self): if self._recording: return self._record_buffer = [] self._record_t0_ns = time.monotonic_ns() self._recording = True # Recording implies streaming (so the recorded actions also reach # the devices live as you perform them). if not self._streaming: self._streaming = True self.stream_btn.config(text="β–  Stop streaming", bg="#0EA5E9") if not self._install_hook(): self._recording = False self._streaming = False self.stream_btn.config(text="β–Ά Stream", bg="#22C55E") messagebox.showerror( "Hook install failed", "Couldn't install the system keyboard hook.", parent=self) return self.rec_btn.config(text="β–  Stop & Save", bg="#B91C1C") self.recording_status_var.set("● Recording (keys + mouse + CAD)...") self._update_capture_label() def _stop_recording(self): if not self._recording: return self._recording = False # Release anything still held so a replay doesn't leave it latched. if self._record_t0_ns is not None: t_ms = (time.monotonic_ns() - self._record_t0_ns) // 1_000_000 for code in list(self._held_codes): self._record_buffer.append(["k", int(t_ms), self.ACTION_UP, code]) if self._streaming: self.manager.send_event(self.ACTION_UP, code) self._held_codes.clear() if self._mouse_buttons: # Release held mouse buttons at the last known position. self._record_buffer.append( ["m", int(t_ms), 0, 0, 0, 0]) self._mouse_buttons = 0 self.rec_btn.config(text="● Record", bg="#DC2626") self._update_capture_label() events = self._record_buffer n = len(events) dur_ms = events[-1][1] if events else 0 if not events: self.recording_status_var.set("Recording stopped β€” nothing captured.") return self.recording_status_var.set( f"Recording stopped β€” {n} events, {dur_ms/1000.0:.2f}s. Saving…") self._open_save_macro_dialog(events, dur_ms) # ---- Save dialog (custom; name + folder) ---- def _open_save_macro_dialog(self, events, duration_ms): # Pause global key capture so the user can actually type the name β€” # otherwise the still-installed hook swallows every keystroke. self._pause_capture_for_dialog() dlg = tk.Toplevel(self) dlg.title("Save macro") dlg.configure(bg="#2D2D3D") dlg.transient(self) dlg.grab_set() dlg.geometry("420x220") tk.Label(dlg, text="Save recorded macro", bg="#2D2D3D", fg="white", font=("Segoe UI", 11, "bold")).pack(anchor="w", padx=16, pady=(14, 8)) body = tk.Frame(dlg, bg="#2D2D3D") body.pack(fill="x", padx=16) body.columnconfigure(1, weight=1) tk.Label(body, text="Name:", bg="#2D2D3D", fg="white", font=("Segoe UI", 9)).grid(row=0, column=0, sticky="w", pady=4) name_var = tk.StringVar() name_entry = tk.Entry(body, textvariable=name_var, bg="#1E1E2E", fg="white", insertbackground="white", font=("Segoe UI", 10), relief="flat") name_entry.grid(row=0, column=1, sticky="ew", padx=(8, 0), pady=4) name_entry.focus_set() tk.Label(body, text="Folder:", bg="#2D2D3D", fg="white", font=("Segoe UI", 9)).grid(row=1, column=0, sticky="w", pady=4) folder_var = tk.StringVar(value=bt_macros.DEFAULT_FOLDER) folder_box = ttk.Combobox(body, textvariable=folder_var, values=bt_macros.list_folders(), font=("Segoe UI", 9)) # editable: type a new name folder_box.grid(row=1, column=1, sticky="ew", padx=(8, 0), pady=4) tk.Label(body, text="(type a new name to create a folder)", bg="#2D2D3D", fg="#94A3B8", font=("Segoe UI", 8)).grid( row=2, column=1, sticky="w", padx=(8, 0)) def finish(): try: dlg.destroy() except tk.TclError: pass self._resume_capture_after_dialog() def do_save(): name = name_var.get().strip() folder = folder_var.get().strip() or bt_macros.DEFAULT_FOLDER if not name: messagebox.showinfo("Name required", "Enter a macro name.", parent=dlg) return if bt_macros.macro_exists(folder, name) and not messagebox.askyesno( "Overwrite?", f"'{name}' already exists in '{folder}'. Overwrite?", parent=dlg): return bt_macros.save_macro(folder, name, events, duration_ms) self.recording_status_var.set( f"Saved macro '{name}' to '{folder}'.") finish() btns = tk.Frame(dlg, bg="#2D2D3D") btns.pack(fill="x", padx=16, pady=(14, 14)) tk.Button(btns, text="Cancel", bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=14, command=finish).pack(side="right") tk.Button(btns, text="Save", bg="#22C55E", fg="white", font=("Segoe UI", 9, "bold"), relief="flat", padx=16, command=do_save).pack(side="right", padx=(0, 6)) dlg.protocol("WM_DELETE_WINDOW", finish) name_entry.bind("", lambda _e: do_save()) # ---- Macro library button + playback ---- def _open_macro_library(self): # Pause capture while the picker is open so its folder name prompts # (and Delete confirmations) receive keyboard input normally. self._pause_capture_for_dialog() picker = MacroLibraryPicker(self, select_mode=False, on_quick_run=self._quick_run_macro, on_load=self._load_macro_into_button) picker.bind( "", lambda e, p=picker: (self._resume_capture_after_dialog() if e.widget is p else None)) def _quick_run_macro(self, folder, name, events, loop): self._run_macro(events, loop) self.recording_status_var.set( f"Running '{name}'{' (loop)' if loop else ''}…") def _load_macro_into_button(self, folder, name, events, loop): self._loaded_macro = {"folder": folder, "name": name, "events": events, "loop": bool(loop)} try: self.macros_run_btn.config(state="normal") except tk.TclError: pass self.recording_status_var.set( f"Loaded '{name}'{' (loop)' if loop else ''} into Macros button.") def _on_macros_run(self): if self._macro_running: self._stop_macro() return if not self._loaded_macro: return self._run_macro(self._loaded_macro["events"], self._loaded_macro["loop"]) def _set_macros_running(self, running: bool): try: if running: self.macros_run_btn.config(text="β–  Stop", bg="#DC2626", fg="white", state="normal") else: self.macros_run_btn.config(text="Macros", bg="#A78BFA", fg="#1E1E2E", state="normal" if self._loaded_macro else "disabled") except tk.TclError: pass def _run_macro(self, events, loop): if self._macro_running or not events: return self._macro_events = sorted(events, key=lambda e: e[1]) self._macro_loop = bool(loop) self._macro_running = True self._macro_i = 0 self._macro_start = time.monotonic() self._macro_held_keys.clear() self._macro_held_buttons = 0 self.manager.reset_session_clocks() self._set_macros_running(True) self._macro_pump() def _macro_pump(self): if not self._macro_running or not self._dialog_alive(): return elapsed_ms = (time.monotonic() - self._macro_start) * 1000.0 evs = self._macro_events while self._macro_i < len(evs): ev = evs[self._macro_i] if ev[1] > elapsed_ms: break self._emit_macro_event(ev) self._macro_i += 1 if self._macro_i >= len(evs): if self._macro_loop and self._macro_running: # Release held state, then restart from the top. self._release_macro_held() self._macro_i = 0 self._macro_start = time.monotonic() self.manager.reset_session_clocks() self.after(10, self._macro_pump) else: self._stop_macro() return self.after(5, self._macro_pump) def _emit_macro_event(self, ev): tag = ev[0] if tag == "k": _, _t, action, hid = ev self.manager.send_event(action, hid) if action == self.ACTION_DOWN: self._macro_held_keys.add(hid) else: self._macro_held_keys.discard(hid) elif tag == "m": _, _t, buttons, x, y, wheel = ev xn, yn = x / 32767.0, y / 32767.0 self.manager.send_mouse(buttons, xn, yn, wheel) self._macro_held_buttons = buttons self._macro_last_xy = (xn, yn) def _release_macro_held(self): for hid in list(self._macro_held_keys): self.manager.send_event(self.ACTION_UP, hid) self._macro_held_keys.clear() if self._macro_held_buttons: xn, yn = self._macro_last_xy self.manager.send_mouse(0, xn, yn, 0) self._macro_held_buttons = 0 def _stop_macro(self): self._macro_running = False self._release_macro_held() self._set_macros_running(False) self.recording_status_var.set("Macro stopped.") def _update_capture_label(self): flags = [] if self._streaming: flags.append("STREAM") if self._recording: flags.append("REC") if flags: self.capture_var.set("Capture: " + "+".join(flags)) else: self.capture_var.set("Capture: OFF") # ---- Win hook integration ---- def _install_hook(self) -> bool: if self._win_hook is not None: return True if WinKeyboardHook is None or not _winhook_supported(): # Non-Windows host: there's no platform-native equivalent # yet. Streaming requires the hook to suppress local OS # behavior (e.g., Win key opening Start menu) and to see # keys regardless of which Tk widget has focus. return False try: # No on_escape handler: in the streamer, Escape is just another # key the user wants forwarded to the M5Stack (HID 0x29), not a # local "stop" gesture. Stopping is done via the Stream / Stop # buttons (or the mouse). Leaving on_escape unset lets Escape # flow through _vk_to_hid β†’ get streamed and suppressed locally # like any other key. self._win_hook = WinKeyboardHook( on_event=self._on_hook_event, ) return self._win_hook.start(suppress_local=True) except Exception as exc: print(f"[bt_kbd] hook install failed: {exc}") self._win_hook = None return False def _uninstall_hook(self): if self._win_hook is not None: try: self._win_hook.stop() except Exception: pass self._win_hook = None self._held_codes.clear() def _pause_capture_for_dialog(self): """Uninstall the global keyboard hook while a modal dialog that needs typed input is open. Without this, the hook keeps capturing (and suppressing) keys so the dialog's entries receive nothing.""" if self._win_hook is not None: self._uninstall_hook() self._capture_paused_for_dialog = True def _resume_capture_after_dialog(self): if self._capture_paused_for_dialog: self._capture_paused_for_dialog = False if self._streaming or self._recording: self._install_hook() def _on_hook_event(self, action: int, hid: int): # Hook fires on its own thread β€” marshal to Tk for the # bookkeeping + send. self.after(0, lambda a=action, h=hid: self._dispatch(a, h)) def _dispatch(self, action: int, hid: int): # Dedupe auto-repeat (hook fires WM_KEYDOWN repeatedly on a # held key) and ensure we never UP a key we didn't see DOWN. if action == self.ACTION_DOWN: if hid in self._held_codes: return self._held_codes.add(hid) else: if hid not in self._held_codes: return self._held_codes.discard(hid) if self._recording and self._record_t0_ns is not None: t_ms = (time.monotonic_ns() - self._record_t0_ns) // 1_000_000 self._record_buffer.append(["k", int(t_ms), action, hid]) if self._streaming: self.manager.send_event(action, hid) # ---- Transport lifecycle ---- def _start_ble(self): """BLE mode: no serial/hub takeover. Build the per-device BLE fan-out manager and pause var-sync so its scanner doesn't fight ours for the Bluetooth adapter. The USB port watcher keeps running (we don't touch the serial port in this mode).""" self._maybe_pause_var_sync() try: self.manager = MultiBleKeyboardManager(max_slots=BLE_MAX_SLOTS) except Exception as exc: self._hub_error = f"Couldn't start Bluetooth streaming: {exc}" self.manager = None def _start_hub(self): """Borrow the app's USB serial link and switch the plugged-in device into ESP-NOW hub mode, then build the mesh manager on top of it. Pauses the port watcher so it can't reconnect/poll on the port we're taking over.""" sm = self.serial_manager if self._pause_port_watcher is not None: try: self._pause_port_watcher() except Exception: pass # Pause var-sync too β€” it shares the radio with the mesh on nodes, # and its scanner shouldn't run while we drive the fleet. self._maybe_pause_var_sync() if sm is None: self._hub_error = "No serial manager available." return if not sm.connected: try: sm.scan_and_connect() except Exception: pass if not sm.connected or sm.ser is None: self._hub_error = ( "No M5Stack found on USB. Plug one in over USB to act as the " "mesh hub, then reopen this window.") return # Clean JSON round-trip to enter hub mode BEFORE the MeshLink # reader thread takes over the port. rsp = sm.send_command({"cmd": "espnow_hub", "on": True}) if not rsp or rsp.get("rsp") != "hub": self._hub_error = ( "The connected device didn't enter hub mode. Re-flash it " "with the current firmware and try again.") return hub_mac = rsp.get("sta_mac", "") # Hand the raw serial handle to the MeshLink (exclusive owner now). self._link = MeshLink(sm.ser) self.manager = MeshKeyboardManager(self._link, max_slots=HUB_MAX_SLOTS) self.manager.set_hub(hub_mac) def _stop_transport(self): """Tear the active manager down. In hub mode also stop the mesh link and return the USB device to normal node mode, handing the serial port back to the app. In BLE mode there's no serial to restore β€” just stop the per-device BLE workers.""" if self.manager is not None: try: self.manager.shutdown() except Exception: pass self.manager = None if self._mode != "hub": return if self._link is not None: try: self._link.stop() except Exception: pass self._link = None # Now that the reader thread is stopped, it's safe to use the # serial manager again to revert hub mode. Restore a normal read # timeout first (MeshLink had shortened it for snappy shutdown). sm = self.serial_manager if sm is not None and sm.connected and sm.ser is not None: try: sm.ser.timeout = 2 except Exception: pass try: sm.send_command({"cmd": "espnow_hub", "on": False}) except Exception: pass if self._resume_port_watcher is not None: try: self._resume_port_watcher() except Exception: pass # ---- Var-sync coordination ---- def _maybe_pause_var_sync(self): if not self._var_sync_paused and self._pause_var_sync is not None: try: self._pause_var_sync() self._var_sync_paused = True except Exception: pass def _maybe_resume_var_sync(self): if self._var_sync_paused and self._resume_var_sync is not None: try: self._resume_var_sync() except Exception: pass self._var_sync_paused = False # ---- Close ---- def _on_close(self): # Abort any in-progress profile load + macro playback so their # after() chains stop. self._load_queue = None self._macro_running = False try: if self._stats_job is not None: self.after_cancel(self._stats_job) except Exception: pass self._uninstall_hook() self._stop_transport() self._maybe_resume_var_sync() self.destroy() # ---- Utilities ---- def _dialog_alive(self) -> bool: try: return bool(self.winfo_exists()) except tk.TclError: return False