"""Serial communication manager for ATOMS3 MacroPad.""" import json import os import time import threading import serial import serial.tools.list_ports import ble_keystore from utils.constants import DEVICE_ID, ESPRESSIF_VID from utils.image_converter import convert_to_rgb565 class SerialManager: def __init__(self): self.ser: serial.Serial | None = None self.port: str | None = None self.connected = False self.is_uploading = False # Last ping response (board, sta_mac, mesh_ch, live_tx, ...). Captured # on connect and refreshed on every ping() so the UI can show which # device is attached and what live transport it's set to. self.device_info: dict | None = None self._lock = threading.Lock() self._on_disconnect = None self._on_connect = None def set_callbacks(self, on_connect=None, on_disconnect=None): self._on_connect = on_connect self._on_disconnect = on_disconnect def scan_and_connect(self) -> bool: """Scan COM ports for ATOMS3 MacroPad device.""" ports = serial.tools.list_ports.comports() for port in ports: if port.vid == ESPRESSIF_VID: try: # Open without DTR/RTS asserted so the ESP32 doesn't reset ser = serial.Serial() ser.port = port.device ser.baudrate = 115200 ser.timeout = 2 ser.dtr = False ser.rts = False ser.open() time.sleep(0.1) # Assert DTR so TinyUSB CDC sees the host as connected ser.dtr = True time.sleep(0.3) ser.reset_input_buffer() ser.write(b'{"cmd":"ping"}\n') line = ser.readline().decode("utf-8", errors="ignore").strip() if line: data = json.loads(line) if data.get("id") == DEVICE_ID: self.ser = ser self.port = port.device self.connected = True self.device_info = data if self._on_connect: self._on_connect(port.device) return True ser.close() except (serial.SerialException, json.JSONDecodeError, OSError): pass return False def disconnect(self): if self.ser and self.ser.is_open: self.ser.close() self.ser = None self.port = None self.connected = False self.device_info = None if self._on_disconnect: self._on_disconnect() def send_command(self, cmd: dict) -> dict | None: """Send a JSON command and wait for response.""" if not self.connected or not self.ser: return None with self._lock: try: msg = json.dumps(cmd) + "\n" self.ser.write(msg.encode("utf-8")) line = self.ser.readline().decode("utf-8", errors="ignore").strip() if line: return json.loads(line) return None except (serial.SerialException, json.JSONDecodeError, OSError): self.connected = False if self._on_disconnect: self._on_disconnect() return None def ping(self) -> dict | None: """Ping the device. Newer firmware also returns 'board' ("atoms3"/"atoms3_lite"), 'sta_mac' (the WiFi/mesh MAC), 'mesh_ch', and 'live_tx' (0 = mesh, 1 = BLE) alongside the legacy id/ver/macros/free fields.""" rsp = self.send_command({"cmd": "ping"}) if rsp: self.device_info = rsp return rsp def live_mode_label(self) -> str | None: """Human label for the connected device's persisted live transport, or None if unknown (older firmware doesn't report it).""" if not self.device_info: return None tx = self.device_info.get("live_tx") if tx is None: return None return "BLE" if tx == 1 else "Mesh" def get_ble_key_info(self) -> dict | None: """Return {'key': bytes, 'tag': str|None} from the device. Newer firmware includes the device tag (with MAC) so the host can store keys per-device. Older firmware returns just the key — tag will be None. """ rsp = self.send_command({"cmd": "get_ble_key"}) if rsp is None or rsp.get("rsp") != "ble_key": return None try: key = bytes.fromhex(rsp["key"]) except (KeyError, ValueError): return None if len(key) != ble_keystore.KEY_LEN: return None tag = rsp.get("tag") if not isinstance(tag, str): tag = None return {"key": key, "tag": tag} def get_ble_key(self) -> bytes | None: """Pull the firmware's BLE payload-encryption key (32 bytes).""" rsp = self.send_command({"cmd": "get_ble_key"}) if rsp and rsp.get("rsp") == "ble_key": try: key = bytes.fromhex(rsp["key"]) except (KeyError, ValueError): return None return key if len(key) == ble_keystore.KEY_LEN else None return None def get_settings(self) -> dict | None: return self.send_command({"cmd": "get_settings"}) def set_setting(self, key: str, value: int) -> bool: rsp = self.send_command({"cmd": "set", "key": key, "val": value}) return rsp is not None and rsp.get("rsp") == "ok" def upload_settings(self, settings) -> bool: """Upload all settings to device.""" ok = True ok = ok and self.set_setting("hold_ms", settings.hold_ms) ok = ok and self.set_setting("orientation", settings.orientation) ok = ok and self.set_setting("type_delay", settings.type_delay) ok = ok and self.set_setting("resume_delay", settings.resume_delay) ok = ok and self.set_setting("combo_pre_ms", settings.combo_pre_ms) ok = ok and self.set_setting("combo_post_ms", settings.combo_post_ms) ok = ok and self.set_setting("probe_timeout_ms", settings.probe_timeout_ms) ok = ok and self.set_setting("media_hold_ms", settings.media_hold_ms) ok = ok and self.set_setting("type_shift_extra_ms", settings.type_shift_extra_ms) ok = ok and self.set_setting("type_settle_ms", settings.type_settle_ms) ok = ok and self.set_setting("pause_margin_left", settings.pause_margin_left) ok = ok and self.set_setting("pause_margin_right", settings.pause_margin_right) ok = ok and self.set_setting("pause_margin_top", settings.pause_margin_top) ok = ok and self.set_setting("pause_margin_bottom", settings.pause_margin_bottom) return ok def upload_macro(self, slot: int, macro, progress_cb=None) -> bool: """Upload a single macro to the device.""" if not self.connected or not self.ser: return False with self._lock: try: nodes = macro.flatten_for_device() node_count = len(nodes) img_data = b"" if macro.image_path and os.path.exists(macro.image_path): img_data = convert_to_rgb565(macro.image_path) begin_cmd = json.dumps({ "cmd": "macro_begin", "slot": slot, "name": macro.name, "label_color": getattr(macro, "label_color", "white"), "node_count": node_count, "img_size": len(img_data), }) + "\n" self.ser.write(begin_cmd.encode("utf-8")) rsp = self._read_response() if not rsp or rsp.get("rsp") != "ready": return False if progress_cb: progress_cb(0.1) if img_data: chunk_size = 128 # Small chunks for reliable TinyUSB CDC transfer total_chunks = (len(img_data) + chunk_size - 1) // chunk_size for ci, i in enumerate(range(0, len(img_data), chunk_size)): chunk = img_data[i:i + chunk_size] self.ser.write(f"CHUNK:{len(chunk)}\n".encode("utf-8")) self.ser.flush() self.ser.write(chunk) self.ser.flush() ack = self.ser.readline().decode("utf-8", errors="ignore").strip() if ack != "OK": return False if progress_cb: progress_cb(0.1 + 0.3 * (ci + 1) / total_chunks) self.ser.write(b"IMG_DONE\n") self.ser.flush() rsp = self._read_response() if not rsp or rsp.get("rsp") != "img_ok": return False # Device needs a moment to settle after the image write time.sleep(0.2) self.ser.reset_input_buffer() if progress_cb: progress_cb(0.4) for i, node in enumerate(nodes): node_cmd = json.dumps({ "cmd": "node", "idx": i, "type": node["type"], "data": node["data"], }) + "\n" self.ser.write(node_cmd.encode("utf-8")) rsp = self._read_response() if not rsp or rsp.get("rsp") != "ok": return False if progress_cb and node_count > 0: progress_cb(0.4 + 0.4 * (i + 1) / node_count) end_cmd = json.dumps({"cmd": "macro_end", "slot": slot}) + "\n" self.ser.write(end_cmd.encode("utf-8")) rsp = self._read_response() if not rsp or rsp.get("rsp") != "ok": return False if progress_cb: progress_cb(1.0) return True except (serial.SerialException, OSError): self.connected = False if self._on_disconnect: self._on_disconnect() return False def upload_all(self, project, progress_cb=None, subroutine_macros=None) -> bool: """Upload entire project (settings + all macros + subroutines) to device. Deletes all existing macros on the device first, then re-uploads everything. """ self.is_uploading = True try: return self._upload_all_impl(project, progress_cb, subroutine_macros) finally: self.is_uploading = False def _upload_all_impl(self, project, progress_cb=None, subroutine_macros=None) -> bool: if not self.connected: return False ping_rsp = self.ping() if not ping_rsp: return False # Sync the BLE payload-encryption key from firmware. Older firmware # without this command returns None — leave any cached key alone. # Newer firmware also returns the device tag (with MAC) so the # host can keep keys for multiple ATOMS3s without clobbering each # other on every upload. info = self.get_ble_key_info() if info is not None: try: # Always update the legacy single-key file (back-compat). ble_keystore.save_key(info["key"]) # And the per-MAC store, if the device reported its tag. if info["tag"]: ble_keystore.save_key_for_mac(info["tag"], info["key"]) except OSError as exc: print(f"[BLE] Failed to persist key: {exc}") device_macro_count = ping_rsp.get("macros", 0) for slot in range(device_macro_count): self.delete_macro(slot) time.sleep(0.05) self.send_command({"cmd": "sub_clear"}) time.sleep(0.05) sub_macros = subroutine_macros or [] total = len(project.macros) + len(sub_macros) + 2 # +2: delete phase, settings step = 1 if progress_cb: progress_cb(step / total) if not self.upload_settings(project.settings): return False step += 1 if progress_cb: progress_cb(step / total) for i, macro in enumerate(project.macros): def macro_progress(p, _step=step): if progress_cb: progress_cb((_step + p) / total) if not self.upload_macro(i, macro, macro_progress): return False step += 1 if progress_cb: progress_cb(step / total) order = list(range(len(project.macros))) self.send_command({"cmd": "macro_reorder", "order": order}) for i, sub_macro in enumerate(sub_macros): nodes = sub_macro.flatten_for_device() node_count = len(nodes) begin_rsp = self.send_command({ "cmd": "sub_begin", "slot": i, "name": sub_macro.name, "node_count": node_count, }) if not begin_rsp or begin_rsp.get("rsp") != "ready": # Older firmware without sub-routine support replies non-ready step += 1 if progress_cb: progress_cb(step / total) continue node_ok = True for j, node in enumerate(nodes): rsp = self.send_command({ "cmd": "sub_node", "idx": j, "type": node["type"], "data": node["data"], }) if not rsp or rsp.get("rsp") != "ok": node_ok = False break end_rsp = self.send_command({"cmd": "sub_end", "slot": i}) if not node_ok or not end_rsp or end_rsp.get("rsp") != "ok": return False step += 1 if progress_cb: progress_cb(step / total) return True def delete_macro(self, slot: int) -> bool: rsp = self.send_command({"cmd": "macro_delete", "slot": slot}) return rsp is not None and rsp.get("rsp") == "ok" def rs232_open(self, baud: int, data_bits: int = 8, stop_bits: str = "1", parity: str = "none") -> bool: rsp = self.send_command({ "cmd": "rs232_open", "baud": baud, "data_bits": data_bits, "stop_bits": stop_bits, "parity": parity, }) return rsp is not None and rsp.get("rsp") == "ok" def rs232_close(self) -> bool: rsp = self.send_command({"cmd": "rs232_close"}) return rsp is not None and rsp.get("rsp") == "ok" def rs232_send(self, data: bytes) -> bool: """Send raw bytes over the RS232 port (host-side terminal).""" if not isinstance(data, (bytes, bytearray)): data = str(data).encode("utf-8", errors="replace") hex_str = data.hex() rsp = self.send_command({"cmd": "rs232_send", "hex": hex_str}) return rsp is not None and rsp.get("rsp") == "ok" def rs232_poll(self) -> bytes: """Return any bytes received since the last poll (may be empty).""" rsp = self.send_command({"cmd": "rs232_poll"}) if rsp and rsp.get("rsp") == "rx": hex_str = rsp.get("hex", "") if hex_str: try: return bytes.fromhex(hex_str) except ValueError: return b"" return b"" def get_log(self) -> list | None: rsp = self.send_command({"cmd": "get_log"}) if rsp and rsp.get("rsp") == "log": return rsp.get("entries", []) return None def clear_log(self) -> bool: rsp = self.send_command({"cmd": "clear_log"}) return rsp is not None and rsp.get("rsp") == "ok" def get_ble_log(self) -> list | None: """Pull the device's in-memory BLE debug ring buffer.""" rsp = self.send_command({"cmd": "get_ble_log"}) if rsp and rsp.get("rsp") == "ble_log": return rsp.get("entries", []) return None def clear_ble_log(self) -> bool: rsp = self.send_command({"cmd": "clear_ble_log"}) return rsp is not None and rsp.get("rsp") == "ok" def _read_response(self, timeout: float = 10.0) -> dict | None: """Read a JSON response line with timeout.""" old_timeout = self.ser.timeout self.ser.timeout = timeout try: line = self.ser.readline().decode("utf-8", errors="ignore").strip() if line: return json.loads(line) return None except (json.JSONDecodeError, serial.SerialException): return None finally: self.ser.timeout = old_timeout