"""Main application window for ATOMS3 MacroPad.""" import tkinter as tk from tkinter import messagebox from models.node_graph import Project from models.profile_manager import ProfileManager from models.backup_manager import BackupManager from serial_manager import SerialManager from node_editor.canvas import NodeCanvas from widgets.macro_list import MacroListPanel from widgets.properties_panel import PropertiesPanel from widgets.toolbar import Toolbar from widgets.ble_variables_window import BLEVariablesWindow from widgets.backup_dialog import BackupDialog from widgets.rs232_terminal import RS232Terminal from ble_server import BLEVariableClient from utils.constants import APP_NAME, APPDATA_DIR import os class MacroPadApp(tk.Tk): """Main application window with three-pane layout.""" def __init__(self): super().__init__() self.title(APP_NAME) self.geometry("1200x700") self.minsize(900, 500) self.configure(bg="#1E1E2E") # Dark title bar on Windows try: self.update_idletasks() import ctypes hwnd = ctypes.windll.user32.GetParent(self.winfo_id()) DWMWA_USE_IMMERSIVE_DARK_MODE = 20 ctypes.windll.dwmapi.DwmSetWindowAttribute( hwnd, DWMWA_USE_IMMERSIVE_DARK_MODE, ctypes.byref(ctypes.c_int(1)), ctypes.sizeof(ctypes.c_int)) except Exception: pass os.makedirs(APPDATA_DIR, exist_ok=True) self.serial_manager = SerialManager() self.serial_manager.set_callbacks( on_connect=lambda port: self.after(0, lambda p=port: self.toolbar.set_connected(p)), on_disconnect=lambda: self.after(0, self.toolbar.set_disconnected), ) self.profile_manager = ProfileManager() self.project = self.profile_manager.startup_load() self.backup_manager = BackupManager() self._autosave_job = None self._backup_job = None self._rs232_terminal = None self._port_watch_job = None # Read by _scan_ports' exception path, which can fire on the very first # call from _start_port_watcher before any successful scan self._last_ports = frozenset() self.ble_client = BLEVariableClient() self._build_ui() self._load_project() self._auto_connect() self._start_ble_client() self._start_port_watcher() self.protocol("WM_DELETE_WINDOW", self._on_close) def _build_ui(self): self.toolbar = Toolbar( self, serial_manager=self.serial_manager, on_settings_change=self._on_change, on_profile_switch=self._switch_profile, on_profile_new=self._new_profile, on_profile_delete=self._delete_profile, on_ble_variables=self._open_ble_variables, on_backups=self._open_backups, on_rs232_terminal=self._open_rs232_terminal, on_bt_keyboard=self._open_bt_keyboard, on_settings_open=self.pause_port_watcher, on_settings_close=self.resume_port_watcher, profile_manager=self.profile_manager, ) self.toolbar.pack(fill="x") content = tk.Frame(self, bg="#1E1E2E") content.pack(fill="both", expand=True) self.macro_list = MacroListPanel( content, on_select=self._on_macro_selected, on_change=self._on_change, ) self.macro_list.pack(side="left", fill="y") self.properties = PropertiesPanel( content, on_change=self._on_change, ) self.properties.pack(side="right", fill="y") self.node_canvas = NodeCanvas( content, on_node_select=self._on_node_selected, on_change=self._on_change, ) self.node_canvas.pack(side="left", fill="both", expand=True) self.macro_list.set_profile_manager(self.profile_manager) self.properties.set_node_canvas(self.node_canvas) self.properties.set_macro_list(self.macro_list) self.properties.set_profile_manager(self.profile_manager) self.properties.set_project(self.project) # Skipped when an Entry/Text/Combobox has focus so native copy/paste still works self.bind_all("", self._on_ctrl_c, add="+") self.bind_all("", self._on_ctrl_v, add="+") def _load_project(self): self.toolbar.set_project(self.project) self.toolbar.set_profiles( self.profile_manager.profile_names(), self.profile_manager.active_name, ) self.macro_list.set_project(self.project) # Profile switch reassigns self.project, so panels that cached a # Project handle at startup (text-node variable list, pause-editor # margins) must be re-pointed at the new instance self.properties.set_project(self.project) self.ble_client.update_variables(self.project.ble_variables) if self.project.macros: self._on_macro_selected(0) else: self.node_canvas.load_macro(None) self.properties.show_node(None) def _on_macro_selected(self, index): if index < 0 or index >= len(self.project.macros): self.node_canvas.load_macro(None) self.properties.show_node(None) return macro = self.project.macros[index] self.node_canvas.load_macro(macro) self.properties.show_node(None) start_id = self.node_canvas.find_start_node() if start_id: self.node_canvas.scroll_to_node(start_id) def _on_node_selected(self, node_widget): self.properties.show_node(node_widget) def _is_text_focus(self) -> bool: """Return True if a text input widget currently has focus.""" focused = self.focus_get() if focused is None: return False cls = focused.__class__.__name__ return cls in ("Entry", "Text", "Spinbox") or "Combobox" in cls def _on_ctrl_c(self, event): if self._is_text_focus(): return self.node_canvas.copy_selected() return "break" def _on_ctrl_v(self, event): if self._is_text_focus(): return self.node_canvas.paste() return "break" def _on_change(self): """Schedule debounced autosave and throttled auto-backup.""" if self._autosave_job: self.after_cancel(self._autosave_job) self._autosave_job = self.after(2000, self._autosave) # If a backup is already pending, let its existing timer fire — it # will pick up the latest state. This caps backups at one per 5s. if self._backup_job is None: self._backup_job = self.after(5000, self._run_backup) def _autosave(self): self._autosave_job = None try: self.profile_manager.save_current() except Exception as e: print(f"Autosave error: {e}") def _run_backup(self): self._backup_job = None try: self.profile_manager.save_current() self.backup_manager.create_backup() except Exception as e: print(f"Auto-backup error: {e}") def _open_backups(self): BackupDialog(self, self.backup_manager, on_restore=self._restore_backup) def _open_rs232_terminal(self): """Open the RS232 Terminal dialog (talks through the M5Stack over USB).""" if not self.serial_manager.connected: messagebox.showwarning( "RS232 Terminal", "Connect the MacroPad over USB first.", ) return if self._is_rs232_terminal_open(): self._rs232_terminal.lift() self._rs232_terminal.focus_set() return def cleared(): self._rs232_terminal = None self._rs232_terminal = RS232Terminal(self, self.serial_manager, on_close=cleared) def _is_rs232_terminal_open(self) -> bool: if self._rs232_terminal is None: return False try: return bool(self._rs232_terminal.winfo_exists()) except tk.TclError: return False def _restore_backup(self, path) -> bool: """Restore the app state from a backup zip.""" # Cancel pending jobs so they don't write over the restored state if self._autosave_job: self.after_cancel(self._autosave_job) self._autosave_job = None if self._backup_job: self.after_cancel(self._backup_job) self._backup_job = None try: ok = self.backup_manager.restore_backup(path) if not ok: return False self.profile_manager = ProfileManager() self.project = self.profile_manager.startup_load() self.toolbar.profile_manager = self.profile_manager self.properties.set_profile_manager(self.profile_manager) self.macro_list.set_profile_manager(self.profile_manager) self._load_project() self.ble_client.update_variables(self.project.ble_variables) return True except Exception as e: print(f"Restore error: {e}") return False def _switch_profile(self, name: str): if self._autosave_job: self.after_cancel(self._autosave_job) self._autosave_job = None self.profile_manager.save_current() self.project = self.profile_manager.switch(name) self._load_project() def _new_profile(self, name: str, copy_current: bool): try: self.project = self.profile_manager.new_profile(name, copy_current) except ValueError as e: messagebox.showerror("New Profile", str(e)) return self._load_project() def _delete_profile(self, name: str): try: self.project = self.profile_manager.delete_profile(name) except ValueError as e: messagebox.showerror("Delete Profile", str(e)) return self._load_project() def _auto_connect(self): """Try to connect to device on startup.""" import threading def try_connect(): self.serial_manager.scan_and_connect() if self.serial_manager.connected: self.after(0, lambda: self.toolbar.set_connected(self.serial_manager.port)) threading.Thread(target=try_connect, daemon=True).start() def _scan_ports(self): """Return (set of device names, True if any Espressif-VID port present).""" import serial.tools.list_ports from utils.constants import ESPRESSIF_VID try: ports = list(serial.tools.list_ports.comports()) except Exception: return self._last_ports, False devices = frozenset(p.device for p in ports) has_esp = any(p.vid == ESPRESSIF_VID for p in ports) return devices, has_esp def _start_port_watcher(self): self._last_ports, _ = self._scan_ports() self._reconnect_in_flight = False self._port_watch_job = self.after(3000, self._poll_ports) def pause_port_watcher(self): if self._port_watch_job is not None: try: self.after_cancel(self._port_watch_job) except Exception: pass self._port_watch_job = None def resume_port_watcher(self): if self._port_watch_job is None: self._start_port_watcher() def _poll_ports(self): self._port_watch_job = None try: # Upload and RS232 terminal both need exclusive serial access gated = self.serial_manager.is_uploading or self._is_rs232_terminal_open() if not gated: current, has_esp_port = self._scan_ports() # `connected` only flips on a failed send, so if the user # unplugged while idle the flag is still True. Force-clear it # so the reconnect path below can run. if (self.serial_manager.connected and self.serial_manager.port and self.serial_manager.port not in current): self.serial_manager.disconnect() self._last_ports = current # Poll-retry rather than edge-trigger: the first attempt right # after plug-in often races the ESP32's boot and times out, # and no further port-set change would re-trigger us. if (not self.serial_manager.connected and has_esp_port and not self._reconnect_in_flight): self._attempt_reconnect() finally: self._port_watch_job = self.after(3000, self._poll_ports) def _attempt_reconnect(self): import threading self._reconnect_in_flight = True def worker(): try: self.serial_manager.scan_and_connect() finally: self.after(0, lambda: setattr(self, "_reconnect_in_flight", False)) threading.Thread(target=worker, daemon=True).start() def _start_ble_client(self): def on_status(status): self.after(0, lambda s=status: self.toolbar.set_ble_status(s)) # BLE callbacks run on the BLE thread — read-only ones use the # project directly; mutating ones bounce through self.after for # Tk-safe state changes. def get_vars(scope, mac): if scope == "device": return dict(self.project.get_device(mac)) return dict(self.project.get_universal()) def set_device_vars(mac, vars_): def apply(): self.project.set_device(mac, vars_) self._on_change() self.after(0, apply) def prompt_request(mac, names): import threading done = threading.Event() result = {"value": None} def open_dialog(): from widgets.request_variable_dialog import RequestVariableDialog def on_submit(edited): result["value"] = edited done.set() current = dict(self.project.get_device(mac)) RequestVariableDialog( self, mac, names, current, on_submit=on_submit, play_sound=True, ) self.after(0, open_dialog) # No timeout — the device-side request_ble wait is also indefinite. # Cancel via the device's side button if needed. done.wait() return result["value"] def on_device_seen(mac): def announce(): self.project.get_device(mac) self._on_change() self.after(0, announce) self.ble_client.start( self.project.ble_variables, on_status=on_status, get_vars=get_vars, set_device_vars=set_device_vars, prompt_request=prompt_request, on_device_seen=on_device_seen, ) def make_ble_live_client(self): """Factory for a fresh BLELiveKeystrokeClient. The macro recorder dialog calls this every time the user clicks Open BLE. A new client per session keeps the lifecycle simple and avoids any cross-session replay-state leakage on the worker. """ from ble_live import BLELiveKeystrokeClient return BLELiveKeystrokeClient() def pause_var_sync_ble(self): """Stop the variable-sync BLE client so the live-record client has the radio to itself. Both clients filter by the same SERVICE_UUID, so without this they race for connect attempts and the var-sync side blocks waiting for a JSON hello that the live flow never sends. """ try: self.ble_client.stop() except Exception as exc: print(f"[BLE] pause var-sync error: {exc}") def resume_var_sync_ble(self): """Re-start the variable-sync BLE client after a live-record session ends. Re-wires the same callbacks _start_ble_client installed at boot so device pushes/pulls keep flowing.""" try: # Drop the old client (which has a dead worker thread) and # construct a fresh one; reuses BLEVariableClient.start's # standard setup path. from ble_server import BLEVariableClient self.ble_client = BLEVariableClient() self._start_ble_client() except Exception as exc: print(f"[BLE] resume var-sync error: {exc}") def _open_bt_keyboard(self): """Open the multi-device live-keystroke streaming window. Prompts for a transport first (direct BLE fan-out, or ESP-NOW hub), then opens the streamer in that mode. In hub mode the window borrows our USB serial link to switch the plugged-in device into hub mode and drives the whole fleet through it; in BLE mode it links to each device directly. Either way we hand over the serial manager plus the port-watcher and var-sync pause/resume hooks so nothing else contends for the port (or the BLE radio) while it's live.""" from widgets.bt_keyboard_window import ( BtKeyboardWindow, choose_keyboard_mode, ) if (getattr(self, "_bt_kbd_window", None) is not None and self._bt_kbd_window.winfo_exists()): self._bt_kbd_window.lift() self._bt_kbd_window.focus_set() return mode = choose_keyboard_mode(self) if mode is None: return # user cancelled the picker self._bt_kbd_window = BtKeyboardWindow( self, serial_manager=self.serial_manager, mode=mode, pause_var_sync=self.pause_var_sync_ble, resume_var_sync=self.resume_var_sync_ble, pause_port_watcher=self.pause_port_watcher, resume_port_watcher=self.resume_port_watcher, ) def _open_ble_variables(self): def on_save(variables: dict, comments: dict): self.project.ble_variables = variables self.project.ble_comments = comments self.ble_client.update_variables(variables) self._on_change() BLEVariablesWindow( self, self.project.ble_variables, on_save=on_save, ble_comments=self.project.ble_comments, ) def _on_close(self): # Cancel pending jobs so they don't race shutdown if self._port_watch_job: try: self.after_cancel(self._port_watch_job) except Exception: pass self._port_watch_job = None if self._autosave_job: try: self.after_cancel(self._autosave_job) except Exception: pass self._autosave_job = None pending_backup = self._backup_job is not None if pending_backup: try: self.after_cancel(self._backup_job) except Exception: pass self._backup_job = None try: self.profile_manager.save_current() except Exception: pass # If a backup was pending, take one now to capture the final changes if pending_backup: try: self.backup_manager.create_backup() except Exception: pass try: self.ble_client.stop() except Exception: pass if self.serial_manager.connected: self.serial_manager.disconnect() self.destroy()