737 lines
30 KiB
Python
737 lines
30 KiB
Python
"""Live macro recorder dialog.
|
|
|
|
User-facing surface of the live-BLE recording feature.
|
|
|
|
Lifecycle:
|
|
- User opens the dialog. Idle state — no BLE link yet, Record button
|
|
is disabled.
|
|
- User clicks **Open BLE**. We spawn a BLELiveKeystrokeClient that
|
|
scans for an idle M5Stack. The device advertises the live service
|
|
automatically while sitting on its macro selector — there is no
|
|
on-device gesture to perform. The worker retries until it finds one,
|
|
then sends an encrypted START which puts the device into recording.
|
|
- Once status flips to ST_CONNECTED, the Record button enables.
|
|
- User clicks **● Record**: timestamp begins, KeyPress / KeyRelease
|
|
events get captured locally AND streamed over BLE to the M5Stack
|
|
which emits them as USB HID to the target machine.
|
|
- User clicks **■ Stop**: capture pauses but the BLE link stays up so
|
|
the user can immediately ● Record again.
|
|
- If the BLE link drops mid-session: the worker auto-retries. The
|
|
dialog shows the changing status; key capture pauses while the
|
|
link is down (we don't want to silently drop events into a black
|
|
hole, and the recorded timeline naturally pauses since host clock
|
|
isn't being read). When the worker reconnects, capture resumes.
|
|
- User clicks **Close BLE**: stops the worker, resumes the var-sync
|
|
client. Record disables again.
|
|
- User clicks Save or Cancel: same as Close BLE plus dialog closes.
|
|
|
|
Events can be hand-edited via double-click (opens MacroEventEditorDialog)
|
|
or right-click for insert before/after / delete. On Save we warn if
|
|
events aren't monotonic in time and offer to sort them.
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
|
|
from utils.constants import TKKEYSYM_TO_HID, hid_code_label
|
|
from widgets.macro_event_editor import MacroEventEditorDialog
|
|
|
|
try:
|
|
from utils.win_keyboard_hook import WinKeyboardHook, is_supported as _winhook_supported
|
|
except ImportError:
|
|
WinKeyboardHook = None
|
|
def _winhook_supported() -> bool:
|
|
return False
|
|
|
|
|
|
class MacroRecorderDialog:
|
|
"""Modal dialog: live-record + edit the ``data['events']`` payload."""
|
|
|
|
ACTION_DOWN = 0
|
|
ACTION_UP = 1
|
|
|
|
def __init__(self, parent, data, on_save, *,
|
|
ble_live_client_factory=None,
|
|
pause_var_sync=None, resume_var_sync=None):
|
|
"""
|
|
ble_live_client_factory — zero-arg callable returning a fresh
|
|
BLELiveKeystrokeClient. If None, the live Record path is
|
|
disabled and only the offline editor is available.
|
|
pause_var_sync / resume_var_sync — callables the dialog invokes
|
|
around its live session so the var-sync BLE client doesn't
|
|
fight for the same advertisement. Both optional.
|
|
"""
|
|
self.parent = parent
|
|
self.data = data
|
|
self.on_save = on_save
|
|
self._ble_factory = ble_live_client_factory
|
|
self._pause_var_sync = pause_var_sync
|
|
self._resume_var_sync = resume_var_sync
|
|
|
|
self._recording = False
|
|
self._record_start_ms: int = 0
|
|
# Dedupes Tk's auto-repeat KeyPress storms while a key is held
|
|
self._held_keys: set = set()
|
|
# Working copy; committed back to ``data`` only on Save
|
|
self._events: list = [list(e) for e in (data.get("events") or [])]
|
|
|
|
# Live-BLE state
|
|
self._ble_client = None
|
|
self._ble_status = "disconnected"
|
|
self._var_sync_paused = False
|
|
# Disconnect modal stub (some earlier flows referenced it; kept
|
|
# so reconnect UI can be added back without churn).
|
|
self._disconnect_modal = None
|
|
|
|
# Low-level Windows keyboard hook (None on non-Windows or when
|
|
# the import failed). Captures the Windows key and other system
|
|
# shortcuts Tkinter can't see, and suppresses them locally so
|
|
# they only fire on the target machine over BLE.
|
|
self._win_hook = None
|
|
# Dedupes auto-repeat from the low-level hook the same way
|
|
# _held_keys does for Tk. Keys here are HID codes, not Tk keysyms.
|
|
self._hook_held_codes: set = set()
|
|
|
|
self._build_dialog()
|
|
self._refresh_event_list()
|
|
self._refresh_button_states()
|
|
|
|
# Auto-start the BLE scanner. The worker retries until it finds
|
|
# an M5Stack in Live Mode, so the user can take their time
|
|
# putting the device into Live Mode after opening this dialog.
|
|
if self._ble_factory is not None:
|
|
try:
|
|
self.dlg.after(50, self._open_ble)
|
|
except tk.TclError:
|
|
pass
|
|
|
|
# ---- UI construction ----
|
|
|
|
def _build_dialog(self):
|
|
self.dlg = tk.Toplevel(self.parent)
|
|
self.dlg.title("Configure Macro")
|
|
self.dlg.geometry("620x600")
|
|
self.dlg.configure(bg="#2D2D3D")
|
|
self.dlg.transient(self.parent.winfo_toplevel())
|
|
self.dlg.grab_set()
|
|
|
|
tk.Label(self.dlg, text="Macro recording (live over BLE)",
|
|
bg="#2D2D3D", fg="white", font=("Segoe UI", 12, "bold")).pack(
|
|
anchor="w", padx=14, pady=(12, 2))
|
|
|
|
# Persistent reminder banner — drawn in a high-contrast color so
|
|
# the user knows the one external prerequisite. The device connects
|
|
# automatically; it just has to be idle (not running a routine).
|
|
reminder_frame = tk.Frame(self.dlg, bg="#1E40AF", bd=0)
|
|
reminder_frame.pack(fill="x", padx=14, pady=(2, 8))
|
|
tk.Label(reminder_frame,
|
|
text="The M5Stack connects automatically — just leave it idle on its "
|
|
"macro selector (not running a routine).",
|
|
bg="#1E40AF", fg="white",
|
|
font=("Segoe UI", 9, "bold"), padx=10, pady=6,
|
|
anchor="w", justify="left").pack(fill="x")
|
|
|
|
name_row = tk.Frame(self.dlg, bg="#2D2D3D")
|
|
name_row.pack(fill="x", padx=14, pady=(0, 8))
|
|
tk.Label(name_row, text="Name:",
|
|
bg="#2D2D3D", fg="white", font=("Segoe UI", 9)).pack(side="left")
|
|
self.name_var = tk.StringVar(value=self.data.get("name", ""))
|
|
tk.Entry(name_row, textvariable=self.name_var,
|
|
bg="#1E1E2E", fg="white", insertbackground="white",
|
|
font=("Segoe UI", 9), relief="flat").pack(
|
|
side="left", fill="x", expand=True, padx=(8, 0))
|
|
|
|
self.status_var = tk.StringVar(value=self._idle_status())
|
|
self.status_label = tk.Label(self.dlg, textvariable=self.status_var,
|
|
bg="#2D2D3D", fg="#F59E0B",
|
|
font=("Segoe UI", 9, "bold"))
|
|
self.status_label.pack(anchor="w", padx=14, pady=(0, 6))
|
|
|
|
# ---- BLE control row ----
|
|
ble_row = tk.Frame(self.dlg, bg="#2D2D3D")
|
|
ble_row.pack(fill="x", padx=14, pady=(0, 6))
|
|
self.open_btn = tk.Button(ble_row, text="Open BLE",
|
|
bg="#0EA5E9", fg="white",
|
|
activebackground="#0284C7",
|
|
font=("Segoe UI", 9, "bold"),
|
|
relief="flat", padx=12,
|
|
command=self._open_ble)
|
|
self.open_btn.pack(side="left")
|
|
self.close_btn = tk.Button(ble_row, text="Close BLE",
|
|
bg="#475569", fg="white",
|
|
font=("Segoe UI", 9), relief="flat",
|
|
padx=12, command=self._close_ble)
|
|
self.close_btn.pack(side="left", padx=(6, 0))
|
|
|
|
self.ble_status_var = tk.StringVar(value="BLE: disconnected")
|
|
tk.Label(ble_row, textvariable=self.ble_status_var,
|
|
bg="#2D2D3D", fg="#94A3B8", font=("Segoe UI", 9)).pack(
|
|
side="left", padx=(12, 0))
|
|
|
|
# ---- Record / housekeeping row ----
|
|
btn_row = tk.Frame(self.dlg, bg="#2D2D3D")
|
|
btn_row.pack(fill="x", padx=14, pady=(0, 8))
|
|
|
|
self.rec_btn = tk.Button(btn_row, text="● Record",
|
|
bg="#DC2626", fg="white",
|
|
activebackground="#B91C1C",
|
|
font=("Segoe UI", 10, "bold"), relief="flat",
|
|
padx=14, command=self._toggle_record)
|
|
self.rec_btn.pack(side="left")
|
|
|
|
tk.Button(btn_row, text="Clear all", bg="#7F1D1D", fg="white",
|
|
activebackground="#991B1B",
|
|
font=("Segoe UI", 9), relief="flat", padx=12,
|
|
command=self._clear_all).pack(side="left", padx=(8, 0))
|
|
|
|
tk.Button(btn_row, text="Sort by time", bg="#4A4A6A", fg="white",
|
|
font=("Segoe UI", 9), relief="flat", padx=12,
|
|
command=self._sort_events).pack(side="left", padx=(8, 0))
|
|
|
|
list_frame = tk.Frame(self.dlg, bg="#2D2D3D")
|
|
list_frame.pack(fill="both", expand=True, padx=14, pady=(4, 8))
|
|
|
|
style = ttk.Style()
|
|
try:
|
|
style.configure("Macro.Treeview",
|
|
background="#1E1E2E", foreground="white",
|
|
fieldbackground="#1E1E2E", rowheight=22,
|
|
font=("Consolas", 9))
|
|
style.configure("Macro.Treeview.Heading",
|
|
background="#2D2D3D", foreground="#CCCCCC",
|
|
font=("Segoe UI", 9, "bold"))
|
|
style.map("Macro.Treeview", background=[("selected", "#3B82F6")])
|
|
except tk.TclError:
|
|
pass
|
|
|
|
columns = ("t", "action", "key")
|
|
self.tree = ttk.Treeview(list_frame, columns=columns, show="headings",
|
|
style="Macro.Treeview", selectmode="extended")
|
|
self.tree.heading("t", text="Time (ms)")
|
|
self.tree.heading("action", text="Action")
|
|
self.tree.heading("key", text="Key")
|
|
self.tree.column("t", width=100, anchor="e")
|
|
self.tree.column("action", width=80, anchor="center")
|
|
self.tree.column("key", width=240, anchor="w")
|
|
|
|
vsb = ttk.Scrollbar(list_frame, orient="vertical", command=self.tree.yview)
|
|
self.tree.configure(yscrollcommand=vsb.set)
|
|
self.tree.grid(row=0, column=0, sticky="nsew")
|
|
vsb.grid(row=0, column=1, sticky="ns")
|
|
list_frame.grid_rowconfigure(0, weight=1)
|
|
list_frame.grid_columnconfigure(0, weight=1)
|
|
|
|
self.tree.bind("<Double-Button-1>", self._on_tree_double_click)
|
|
self.tree.bind("<Button-3>", self._on_tree_right_click)
|
|
|
|
edit_row = tk.Frame(self.dlg, bg="#2D2D3D")
|
|
edit_row.pack(fill="x", padx=14, pady=(0, 8))
|
|
tk.Button(edit_row, text="Edit", bg="#4A4A6A", fg="white",
|
|
font=("Segoe UI", 9), relief="flat", padx=10,
|
|
command=self._edit_selected).pack(side="left")
|
|
tk.Button(edit_row, text="Insert before", bg="#4A4A6A", fg="white",
|
|
font=("Segoe UI", 9), relief="flat", padx=10,
|
|
command=lambda: self._insert_relative(before=True)).pack(
|
|
side="left", padx=(6, 0))
|
|
tk.Button(edit_row, text="Insert after", bg="#4A4A6A", fg="white",
|
|
font=("Segoe UI", 9), relief="flat", padx=10,
|
|
command=lambda: self._insert_relative(before=False)).pack(
|
|
side="left", padx=(6, 0))
|
|
tk.Button(edit_row, text="Delete selected", bg="#4A4A6A", fg="white",
|
|
font=("Segoe UI", 9), relief="flat", padx=10,
|
|
command=self._delete_selected).pack(side="left", padx=(6, 0))
|
|
tk.Label(edit_row, text="(double-click a row to edit)",
|
|
bg="#2D2D3D", fg="#888888", font=("Segoe UI", 8)).pack(
|
|
side="left", padx=(10, 0))
|
|
|
|
footer = tk.Frame(self.dlg, bg="#2D2D3D")
|
|
footer.pack(fill="x", side="bottom", padx=14, pady=12)
|
|
tk.Button(footer, text="Cancel", bg="#4A4A6A", fg="white",
|
|
font=("Segoe UI", 9), relief="flat", padx=14,
|
|
command=self._cancel).pack(side="right")
|
|
tk.Button(footer, text="Save", bg="#22C55E", fg="white",
|
|
activebackground="#16A34A",
|
|
font=("Segoe UI", 9, "bold"), relief="flat", padx=14,
|
|
command=self._save).pack(side="right", padx=(0, 6))
|
|
|
|
self.dlg.focus_set()
|
|
self.dlg.protocol("WM_DELETE_WINDOW", self._cancel)
|
|
|
|
def _idle_status(self) -> str:
|
|
n = len(self._events)
|
|
if n == 0:
|
|
return "No events recorded."
|
|
last_t = self._events[-1][0]
|
|
secs = last_t / 1000.0
|
|
return f"{n} events, {secs:.2f}s total."
|
|
|
|
# ---- BLE lifecycle ----
|
|
|
|
def _open_ble(self):
|
|
if self._ble_client is not None:
|
|
return
|
|
if self._ble_factory is None:
|
|
messagebox.showerror(
|
|
"Live record unavailable",
|
|
"BLE client factory not configured.",
|
|
parent=self.dlg)
|
|
return
|
|
|
|
# Note: we deliberately do NOT pause the var-sync client here.
|
|
# The device advertises a different SERVICE_UUID when in live
|
|
# mode, so the two scanners filter to disjoint device sets and
|
|
# don't race. Pausing var-sync via stop() would interrupt any
|
|
# in-flight WinRT BleakClient cleanup and leave the host BLE
|
|
# subsystem in a state that breaks the next connect attempt.
|
|
self._ble_client = self._ble_factory()
|
|
self._ble_client.start(
|
|
on_status=self._on_ble_status,
|
|
on_error=self._on_ble_error,
|
|
)
|
|
self._set_status("BLE: scanning for idle M5Stack...", "#F59E0B")
|
|
self._refresh_button_states()
|
|
|
|
def _close_ble(self):
|
|
# Stops capture too — pressing Close BLE during a recording is
|
|
# equivalent to pressing Stop first.
|
|
if self._recording:
|
|
self._stop_recording()
|
|
if self._ble_client is not None:
|
|
try:
|
|
self._ble_client.stop(timeout=2.0)
|
|
except Exception:
|
|
pass
|
|
self._ble_client = None
|
|
self._ble_status = "disconnected"
|
|
self.ble_status_var.set("BLE: disconnected")
|
|
self._set_status(self._idle_status(), "#F59E0B")
|
|
self._refresh_button_states()
|
|
|
|
def _on_ble_status(self, status: str):
|
|
self.dlg.after(0, lambda s=status: self._apply_ble_status(s))
|
|
|
|
def _on_ble_error(self, err_code: int, ref_seq, label: str):
|
|
# KEY_MISMATCH is host-side: ble_live.py emits it when a few
|
|
# consecutive notify frames fail AES-GCM auth. Surface it with
|
|
# an actionable remediation instead of an opaque label.
|
|
if label == "KEY_MISMATCH":
|
|
msg = ("BLE key mismatch — the host and the M5Stack have "
|
|
"different keys. Upload the profile over USB to "
|
|
"re-sync, then try again.")
|
|
self.dlg.after(0, lambda m=msg: self._set_status(m, "#EF4444"))
|
|
# Also a modal so the user can't miss it. Done once per
|
|
# session: the auth-fail counter only fires the callback
|
|
# at exactly count==3.
|
|
self.dlg.after(0, self._show_key_mismatch_modal)
|
|
return
|
|
msg = f"Device reported {label}"
|
|
self.dlg.after(0, lambda m=msg: self._set_status(m, "#EF4444"))
|
|
|
|
def _show_key_mismatch_modal(self):
|
|
if not self._dialog_alive():
|
|
return
|
|
try:
|
|
messagebox.showerror(
|
|
"BLE key mismatch",
|
|
"Frames from the M5Stack can't be decrypted because the "
|
|
"host's stored key doesn't match the device's.\n\n"
|
|
"Fix: connect the M5Stack to this computer over USB and "
|
|
"upload your profile. The upload syncs the encryption "
|
|
"key automatically.",
|
|
parent=self.dlg)
|
|
except Exception:
|
|
pass
|
|
|
|
def _apply_ble_status(self, status: str):
|
|
if not self._dialog_alive():
|
|
return
|
|
self._ble_status = status
|
|
# Surface a friendly label next to the buttons.
|
|
labels = {
|
|
"idle": "BLE: idle",
|
|
"scanning": "BLE: scanning...",
|
|
"connecting": "BLE: connecting...",
|
|
"connected": "BLE: connected",
|
|
"disconnected": "BLE: disconnected — auto-retrying",
|
|
"error": "BLE: error",
|
|
}
|
|
self.ble_status_var.set(labels.get(status, f"BLE: {status}"))
|
|
|
|
if status == "connected":
|
|
if self._recording:
|
|
self._set_status("● Recording", "#EF4444")
|
|
else:
|
|
self._set_status(self._idle_status(), "#22C55E")
|
|
elif status in ("disconnected", "error"):
|
|
if self._recording:
|
|
# Capture pauses transparently — _on_key_press already
|
|
# bails on !is_connected(). Show banner so user knows.
|
|
self._set_status("BLE dropped — waiting for reconnect...",
|
|
"#F59E0B")
|
|
elif status in ("scanning", "connecting"):
|
|
if self._recording:
|
|
self._set_status(f"● Recording (BLE: {status}...)",
|
|
"#F59E0B")
|
|
self._refresh_button_states()
|
|
|
|
def _refresh_button_states(self):
|
|
"""Drive button enable/disable from the canonical BLE state."""
|
|
worker_running = self._ble_client is not None
|
|
connected = (self._ble_status == "connected")
|
|
# Open BLE: only when no worker is up.
|
|
self.open_btn.config(state="normal" if not worker_running else "disabled")
|
|
# Close BLE: enabled whenever a worker is alive (lets user
|
|
# cancel a long scan).
|
|
self.close_btn.config(state="normal" if worker_running else "disabled")
|
|
# Record: only when actually connected.
|
|
if connected or self._recording:
|
|
self.rec_btn.config(state="normal")
|
|
else:
|
|
self.rec_btn.config(state="disabled")
|
|
|
|
# ---- Recording lifecycle ----
|
|
|
|
def _toggle_record(self):
|
|
if self._recording:
|
|
self._stop_recording()
|
|
else:
|
|
self._start_recording()
|
|
|
|
def _start_recording(self):
|
|
if not (self._ble_client and self._ble_client.is_connected()):
|
|
messagebox.showinfo(
|
|
"Not connected",
|
|
"Open BLE first and make sure the M5Stack is idle on its "
|
|
"macro selector.",
|
|
parent=self.dlg)
|
|
return
|
|
|
|
self._held_keys.clear()
|
|
self._hook_held_codes.clear()
|
|
# Reset the per-event timestamp anchor on the BLE client so this
|
|
# session's events start at t=0 rather than continuing from the
|
|
# previous session.
|
|
try:
|
|
self._ble_client.reset_session_clock()
|
|
except AttributeError:
|
|
pass
|
|
self._recording = True
|
|
self._record_start_ms = int(time.monotonic() * 1000)
|
|
|
|
self.rec_btn.config(text="■ Stop", bg="#B91C1C")
|
|
self._set_status("● Recording", "#EF4444")
|
|
self._bind_keys()
|
|
|
|
def _stop_recording(self):
|
|
if not self._recording:
|
|
return
|
|
self._recording = False
|
|
|
|
# Synthesize releases for anything still held so playback doesn't
|
|
# leave a key latched. Drain both held sets — _hook_held_codes
|
|
# is the canonical one for hook-captured keys, _held_keys is
|
|
# the legacy fallback for the Tk path.
|
|
now_ms = int(time.monotonic() * 1000) - self._record_start_ms
|
|
for code in list(self._hook_held_codes):
|
|
self._events.append([now_ms, self.ACTION_UP, code])
|
|
if self._ble_client and self._ble_client.is_connected():
|
|
self._ble_client.send_event(self.ACTION_UP, code)
|
|
self._hook_held_codes.clear()
|
|
self._held_keys.clear()
|
|
self._unbind_keys()
|
|
|
|
self.rec_btn.config(text="● Record", bg="#DC2626")
|
|
self._set_status(self._idle_status(), "#F59E0B")
|
|
self._refresh_event_list()
|
|
|
|
def _clear_all(self):
|
|
"""Wipe every recorded event after a confirmation. Bound to the
|
|
"Clear all" button — explicit, scary-red styling because this
|
|
is destructive."""
|
|
if not self._events:
|
|
self._set_status("Nothing to clear.", "#94A3B8")
|
|
return
|
|
if not messagebox.askyesno(
|
|
"Clear all events?",
|
|
"Remove ALL recorded events? This cannot be undone.",
|
|
parent=self.dlg):
|
|
return
|
|
self._events = []
|
|
self._held_keys.clear()
|
|
self._hook_held_codes.clear()
|
|
self._refresh_event_list()
|
|
self._set_status(self._idle_status(), "#F59E0B")
|
|
|
|
# ---- Key capture ----
|
|
#
|
|
# Two parallel capture paths, depending on platform:
|
|
#
|
|
# 1. Tkinter <KeyPress>/<KeyRelease> bindings on the dialog. This
|
|
# is the cross-platform fallback. It misses the Windows key on
|
|
# Windows because the OS swallows VK_LWIN/VK_RWIN before any
|
|
# window sees it.
|
|
#
|
|
# 2. WinKeyboardHook (Windows only). A low-level keyboard hook
|
|
# that sees every keystroke globally, before any window or the
|
|
# Start menu handler. We use this to cover the gaps Tk leaves
|
|
# AND to suppress the Windows key locally so it relays via BLE
|
|
# only.
|
|
#
|
|
# The Tk path stays installed too because the hook can miss IME
|
|
# composition / dead-key sequences that Tk synthesizes from
|
|
# WM_CHAR. The two paths feed into the same recording state via
|
|
# _record_event, which dedupes by HID code.
|
|
|
|
def _bind_keys(self):
|
|
self.dlg.bind("<KeyPress>", self._on_key_press, add="+")
|
|
self.dlg.bind("<KeyRelease>", self._on_key_release, add="+")
|
|
self.dlg.focus_set()
|
|
if (WinKeyboardHook is not None and _winhook_supported()
|
|
and self._win_hook is None):
|
|
try:
|
|
self._win_hook = WinKeyboardHook(
|
|
on_event=self._on_hook_event,
|
|
on_escape=self._on_hook_escape,
|
|
)
|
|
self._win_hook.start(suppress_local=True)
|
|
except Exception as exc:
|
|
print(f"[recorder] win hook install failed: {exc}")
|
|
self._win_hook = None
|
|
|
|
def _unbind_keys(self):
|
|
try:
|
|
self.dlg.unbind("<KeyPress>")
|
|
self.dlg.unbind("<KeyRelease>")
|
|
except tk.TclError:
|
|
pass
|
|
if self._win_hook is not None:
|
|
try:
|
|
self._win_hook.stop()
|
|
except Exception:
|
|
pass
|
|
self._win_hook = None
|
|
self._hook_held_codes.clear()
|
|
|
|
def _on_hook_event(self, action: int, hid: int) -> None:
|
|
"""Called on the hook thread. Marshal to Tk and dispatch."""
|
|
try:
|
|
self.dlg.after(0, lambda a=action, h=hid: self._record_event(a, h))
|
|
except Exception:
|
|
pass
|
|
|
|
def _on_hook_escape(self) -> None:
|
|
"""Hook sees Escape — stop recording (marshalled to Tk)."""
|
|
if not self._recording:
|
|
return
|
|
try:
|
|
self.dlg.after(0, self._stop_recording)
|
|
except Exception:
|
|
pass
|
|
|
|
def _record_event(self, action: int, hid: int) -> None:
|
|
"""Single funnel for both Tk and hook paths. Dedupes auto-repeat
|
|
and pushes the event to the recorder + BLE."""
|
|
if not self._recording:
|
|
return
|
|
if not (self._ble_client and self._ble_client.is_connected()):
|
|
return
|
|
if action == 0: # DOWN
|
|
if hid in self._hook_held_codes:
|
|
return
|
|
self._hook_held_codes.add(hid)
|
|
else: # UP
|
|
if hid not in self._hook_held_codes:
|
|
return
|
|
self._hook_held_codes.discard(hid)
|
|
t_ms = int(time.monotonic() * 1000) - self._record_start_ms
|
|
self._events.append([t_ms, action, hid])
|
|
self._ble_client.send_event(action, hid)
|
|
self._append_event_row(len(self._events) - 1)
|
|
|
|
def _on_key_press(self, event):
|
|
# On Windows the low-level hook with suppress_local=True
|
|
# short-circuits Tk before this fires, so this branch is the
|
|
# non-Windows fallback (and also catches IME-translated
|
|
# WM_CHAR events on Windows that bypass the low-level hook).
|
|
if not self._recording:
|
|
return
|
|
if event.keysym == "Escape":
|
|
self._stop_recording()
|
|
return "break"
|
|
code = TKKEYSYM_TO_HID.get(event.keysym)
|
|
if code is None:
|
|
return "break"
|
|
self._record_event(self.ACTION_DOWN, code)
|
|
return "break"
|
|
|
|
def _on_key_release(self, event):
|
|
if not self._recording:
|
|
return
|
|
code = TKKEYSYM_TO_HID.get(event.keysym)
|
|
if code is None:
|
|
return "break"
|
|
self._record_event(self.ACTION_UP, code)
|
|
return "break"
|
|
|
|
# ---- Event list / editor ----
|
|
|
|
def _refresh_event_list(self):
|
|
for iid in self.tree.get_children():
|
|
self.tree.delete(iid)
|
|
for i in range(len(self._events)):
|
|
self._append_event_row(i)
|
|
|
|
def _append_event_row(self, idx: int):
|
|
t_ms, action, code = self._events[idx]
|
|
action_str = "↓ press" if action == self.ACTION_DOWN else "↑ release"
|
|
label = hid_code_label(code)
|
|
self.tree.insert("", "end", iid=str(idx),
|
|
values=(f"{t_ms:>6d}", action_str, f"{label} (0x{code:02X})"))
|
|
self.tree.see(str(idx))
|
|
|
|
def _on_tree_double_click(self, _event):
|
|
self._edit_selected()
|
|
|
|
def _on_tree_right_click(self, event):
|
|
row = self.tree.identify_row(event.y)
|
|
if row:
|
|
self.tree.selection_set(row)
|
|
menu = tk.Menu(self.tree, tearoff=0)
|
|
menu.add_command(label="Edit...", command=self._edit_selected)
|
|
menu.add_command(label="Insert before",
|
|
command=lambda: self._insert_relative(before=True))
|
|
menu.add_command(label="Insert after",
|
|
command=lambda: self._insert_relative(before=False))
|
|
menu.add_separator()
|
|
menu.add_command(label="Delete", command=self._delete_selected)
|
|
try:
|
|
menu.tk_popup(event.x_root, event.y_root)
|
|
finally:
|
|
menu.grab_release()
|
|
|
|
def _edit_selected(self):
|
|
sel = self.tree.selection()
|
|
if not sel:
|
|
return
|
|
try:
|
|
idx = int(sel[0])
|
|
except ValueError:
|
|
return
|
|
if not (0 <= idx < len(self._events)):
|
|
return
|
|
|
|
def commit(new_ev):
|
|
self._events[idx] = new_ev
|
|
self._refresh_event_list()
|
|
if not self._recording:
|
|
self._set_status(self._idle_status(), "#F59E0B")
|
|
|
|
MacroEventEditorDialog(self.dlg, list(self._events[idx]), commit,
|
|
title=f"Edit event #{idx}")
|
|
|
|
def _insert_relative(self, *, before: bool):
|
|
if self._recording:
|
|
messagebox.showinfo("Stop recording first",
|
|
"Stop recording before inserting events manually.",
|
|
parent=self.dlg)
|
|
return
|
|
sel = self.tree.selection()
|
|
idx = None
|
|
if sel:
|
|
try:
|
|
idx = int(sel[0])
|
|
except ValueError:
|
|
idx = None
|
|
if idx is None:
|
|
idx = len(self._events) - 1 if before else len(self._events)
|
|
|
|
insert_at = idx if before else idx + 1
|
|
if not self._events:
|
|
seed = [0, self.ACTION_DOWN, 0x04]
|
|
else:
|
|
anchor = max(0, min(idx, len(self._events) - 1))
|
|
t_here = self._events[anchor][0]
|
|
neighbor_idx = insert_at - 1 if not before else insert_at
|
|
neighbor_idx = max(0, min(neighbor_idx, len(self._events) - 1))
|
|
t_neighbor = self._events[neighbor_idx][0]
|
|
t_seed = (t_here + t_neighbor) // 2 if t_here != t_neighbor else t_here
|
|
seed = [int(t_seed), self.ACTION_DOWN,
|
|
int(self._events[anchor][2])]
|
|
|
|
def commit(new_ev):
|
|
self._events.insert(insert_at, new_ev)
|
|
self._refresh_event_list()
|
|
|
|
MacroEventEditorDialog(self.dlg, seed, commit,
|
|
title="New event")
|
|
|
|
def _delete_selected(self):
|
|
sel = self.tree.selection()
|
|
if not sel:
|
|
return
|
|
indices = sorted({int(iid) for iid in sel}, reverse=True)
|
|
for i in indices:
|
|
if 0 <= i < len(self._events):
|
|
self._events.pop(i)
|
|
self._refresh_event_list()
|
|
self._set_status(self._idle_status(), "#F59E0B")
|
|
|
|
def _sort_events(self):
|
|
if len(self._events) < 2:
|
|
return
|
|
self._events.sort(key=lambda e: e[0])
|
|
self._refresh_event_list()
|
|
|
|
# ---- Save / Cancel ----
|
|
|
|
def _save(self):
|
|
if self._recording:
|
|
self._stop_recording()
|
|
if not self._events_monotonic():
|
|
if messagebox.askyesno(
|
|
"Events out of order",
|
|
"Some events have timestamps earlier than a preceding event.\n"
|
|
"Playback will fire them back-to-back.\n\n"
|
|
"Sort events by time now?",
|
|
parent=self.dlg):
|
|
self._events.sort(key=lambda e: e[0])
|
|
self.data["name"] = self.name_var.get().strip()
|
|
self.data["events"] = [list(e) for e in self._events]
|
|
self._close_ble()
|
|
try:
|
|
self.on_save()
|
|
finally:
|
|
self.dlg.destroy()
|
|
|
|
def _cancel(self):
|
|
if self._recording:
|
|
self._stop_recording()
|
|
self._close_ble()
|
|
self.dlg.destroy()
|
|
|
|
# ---- Utilities ----
|
|
|
|
def _events_monotonic(self) -> bool:
|
|
prev = -1
|
|
for ev in self._events:
|
|
if ev[0] < prev:
|
|
return False
|
|
prev = ev[0]
|
|
return True
|
|
|
|
def _set_status(self, text: str, color):
|
|
try:
|
|
self.status_var.set(text)
|
|
if color:
|
|
self.status_label.config(fg=color)
|
|
except tk.TclError:
|
|
pass
|
|
|
|
def _dialog_alive(self) -> bool:
|
|
try:
|
|
return bool(self.dlg.winfo_exists())
|
|
except tk.TclError:
|
|
return False
|