1840 lines
74 KiB
Python
1840 lines
74 KiB
Python
"""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("<Escape>", 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("<Configure>", self._on_pad_configure)
|
|
for seq in ("<Motion>", "<B1-Motion>", "<B2-Motion>", "<B3-Motion>"):
|
|
self.canvas.bind(seq, self._on_pad_motion)
|
|
for n in (1, 2, 3):
|
|
self.canvas.bind(f"<ButtonPress-{n}>", self._on_pad_press)
|
|
self.canvas.bind(f"<ButtonRelease-{n}>", self._on_pad_release)
|
|
self.canvas.bind("<MouseWheel>", 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("<FocusIn>", self._on_textbox_focus_in)
|
|
self.text_entry.bind("<FocusOut>", self._on_textbox_focus_out)
|
|
# Ctrl+Enter sends; plain Enter inserts a newline.
|
|
self.text_entry.bind("<Control-Return>", 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("<Return>", 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(
|
|
"<Destroy>",
|
|
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
|