"""RS232 Terminal — talk directly to the RS232 port on the M5Stack over USB. The Python app sends JSON commands to the M5Stack (over USB CDC) which relays them to the Atomic RS232 Base. See the ``rs232_*`` command handlers in ``firmware/MacroPad/serial_protocol.h`` for the firmware side. """ from __future__ import annotations import queue import threading import tkinter as tk from datetime import datetime from tkinter import filedialog, messagebox, ttk from utils.constants import ( RS232_BAUD_RATES, RS232_DATA_BITS, RS232_STOP_BITS, RS232_PARITY, RS232_LINE_ENDINGS, ) # KVM quick-send presets: (button label, bytes to send) KVM_PRESETS = [(f"PC {i}", f"X{i},1$") for i in range(1, 9)] DISPLAY_MODES = [ ("ASCII", "ascii"), ("ASCII + Hex", "ascii_hex"), ("Hex only", "hex"), ] class RS232Terminal(tk.Toplevel): """Modal window for direct RS232 interaction via the M5Stack.""" POLL_INTERVAL_MS = 150 # how often the UI drains the RX queue BG_POLL_SLEEP = 0.10 # how often the background thread polls the device def __init__(self, parent, serial_manager, on_close=None): super().__init__(parent) self.serial_manager = serial_manager self._on_close_cb = on_close self.title("RS232 Terminal") self.geometry("760x620") self.minsize(640, 480) self.configure(bg="#2D2D3D") self._rx_queue: "queue.Queue[bytes]" = queue.Queue() self._poll_stop = threading.Event() self._poll_thread: threading.Thread | None = None self._is_open = False self._baud_var = tk.StringVar(value="115200") self._data_bits_var = tk.StringVar(value="8") self._stop_bits_var = tk.StringVar(value="1") self._parity_var = tk.StringVar(value="none") self._line_ending_var = tk.StringVar(value="None") self._display_mode_var = tk.StringVar(value="ASCII + Hex") self._echo_var = tk.BooleanVar(value=True) self._timestamps_var = tk.BooleanVar(value=False) self._auto_scroll_var = tk.BooleanVar(value=True) self._build_ui() self._set_status("Not open", "#E74C3C") self._ui_poll_job = None self.protocol("WM_DELETE_WINDOW", self._on_close) def _build_ui(self): lbl = dict(bg="#2D2D3D", fg="white", font=("Segoe UI", 9)) hint = dict(bg="#2D2D3D", fg="#888888", font=("Segoe UI", 8)) entry = dict(bg="#1E1E2E", fg="white", insertbackground="white", relief="flat", font=("Segoe UI", 10)) btn = dict(bg="#3D3D5C", fg="white", font=("Segoe UI", 9), relief="flat") # ---- Connection settings ---- conn_frame = tk.LabelFrame(self, text="Connection", bg="#2D2D3D", fg="white", font=("Segoe UI", 9, "bold"), bd=1, relief="groove") conn_frame.pack(fill="x", padx=10, pady=(10, 4)) row1 = tk.Frame(conn_frame, bg="#2D2D3D") row1.pack(fill="x", padx=6, pady=4) tk.Label(row1, text="Baud:", **lbl).pack(side="left") ttk.Combobox(row1, textvariable=self._baud_var, values=[str(b) for b in RS232_BAUD_RATES], width=8).pack(side="left", padx=(2, 10)) tk.Label(row1, text="Data:", **lbl).pack(side="left") ttk.Combobox(row1, textvariable=self._data_bits_var, values=[str(d) for d in RS232_DATA_BITS], state="readonly", width=3).pack(side="left", padx=(2, 10)) tk.Label(row1, text="Stop:", **lbl).pack(side="left") ttk.Combobox(row1, textvariable=self._stop_bits_var, values=RS232_STOP_BITS, state="readonly", width=4).pack(side="left", padx=(2, 10)) tk.Label(row1, text="Parity:", **lbl).pack(side="left") ttk.Combobox(row1, textvariable=self._parity_var, values=RS232_PARITY, state="readonly", width=6).pack(side="left", padx=(2, 10)) self._open_btn = tk.Button(row1, text="Open", bg="#27AE60", fg="white", font=("Segoe UI", 9, "bold"), relief="flat", padx=14, command=self._toggle_open) self._open_btn.pack(side="right", padx=(6, 2)) self._status_dot = tk.Canvas(row1, width=12, height=12, bg="#2D2D3D", highlightthickness=0) self._dot_id = self._status_dot.create_oval(2, 2, 10, 10, fill="#888", outline="") self._status_dot.pack(side="right", padx=(4, 0)) self._status_label = tk.Label(row1, text="Not open", **lbl) self._status_label.pack(side="right") # ---- Display options ---- opts = tk.Frame(self, bg="#2D2D3D") opts.pack(fill="x", padx=10, pady=(0, 4)) tk.Label(opts, text="Line ending (sent):", **lbl).pack(side="left") le_display = [label for label, _ in RS232_LINE_ENDINGS] ttk.Combobox(opts, textvariable=self._line_ending_var, values=le_display, state="readonly", width=10).pack(side="left", padx=(2, 10)) tk.Label(opts, text="Display:", **lbl).pack(side="left") ttk.Combobox(opts, textvariable=self._display_mode_var, values=[label for label, _ in DISPLAY_MODES], state="readonly", width=14).pack(side="left", padx=(2, 10)) tk.Checkbutton(opts, text="Local echo", variable=self._echo_var, bg="#2D2D3D", fg="white", selectcolor="#1E1E2E", activebackground="#2D2D3D", activeforeground="white", font=("Segoe UI", 9)).pack(side="left", padx=6) tk.Checkbutton(opts, text="Timestamps", variable=self._timestamps_var, bg="#2D2D3D", fg="white", selectcolor="#1E1E2E", activebackground="#2D2D3D", activeforeground="white", font=("Segoe UI", 9)).pack(side="left", padx=6) tk.Checkbutton(opts, text="Auto-scroll", variable=self._auto_scroll_var, bg="#2D2D3D", fg="white", selectcolor="#1E1E2E", activebackground="#2D2D3D", activeforeground="white", font=("Segoe UI", 9)).pack(side="left", padx=6) # ---- Output area ---- out_frame = tk.Frame(self, bg="#2D2D3D") out_frame.pack(fill="both", expand=True, padx=10, pady=(6, 4)) self._output = tk.Text(out_frame, wrap="word", bg="#12121A", fg="white", insertbackground="white", font=("Consolas", 10), relief="flat", bd=0) vsb = tk.Scrollbar(out_frame, orient="vertical", command=self._output.yview) self._output.configure(yscrollcommand=vsb.set, state="disabled") self._output.pack(side="left", fill="both", expand=True) vsb.pack(side="right", fill="y") self._output.tag_config("tx", foreground="#7BED9F") self._output.tag_config("rx", foreground="#74B9FF") self._output.tag_config("meta", foreground="#888888", font=("Consolas", 9, "italic")) self._output.tag_config("err", foreground="#FF6B6B") self._output.tag_config("hex", foreground="#DFE4EA") # ---- Send row ---- send_row = tk.Frame(self, bg="#2D2D3D") send_row.pack(fill="x", padx=10, pady=(2, 6)) self._cmd_var = tk.StringVar() cmd_entry = tk.Entry(send_row, textvariable=self._cmd_var, **entry) cmd_entry.pack(side="left", fill="x", expand=True, padx=(0, 4)) cmd_entry.bind("", self._send_clicked) tk.Button(send_row, text="Send", bg="#27AE60", fg="white", font=("Segoe UI", 9, "bold"), relief="flat", padx=14, command=self._send_clicked).pack(side="left", padx=(0, 4)) tk.Button(send_row, text="Clear", **btn, padx=10, command=self._clear).pack(side="left", padx=(0, 2)) tk.Button(send_row, text="Save Log…", **btn, padx=10, command=self._save_log).pack(side="left", padx=2) # ---- KVM quick-sends ---- kvm_frame = tk.LabelFrame(self, text="KVM Quick-Send", bg="#2D2D3D", fg="white", font=("Segoe UI", 9, "bold"), bd=1, relief="groove") kvm_frame.pack(fill="x", padx=10, pady=(0, 6)) kvm_inner = tk.Frame(kvm_frame, bg="#2D2D3D") kvm_inner.pack(padx=6, pady=6) tk.Label(kvm_inner, text="Select PC:", **lbl).pack(side="left", padx=(0, 6)) for label, payload in KVM_PRESETS: tk.Button(kvm_inner, text=label, bg="#4A4A6A", fg="white", font=("Segoe UI", 9), relief="flat", padx=10, pady=2, command=lambda p=payload: self._send_preset(p)).pack(side="left", padx=2) tk.Label(kvm_frame, text="Presets send the literal string (no line ending is appended).", **hint).pack(anchor="w", padx=8, pady=(0, 4)) cmd_entry.focus_set() def _toggle_open(self): if self._is_open: self._close_port() else: self._open_port() def _open_port(self): if not self.serial_manager.connected: messagebox.showerror("RS232 Terminal", "The MacroPad is not connected over USB.\n\n" "Connect the device and try again.", parent=self) return try: baud = int(self._baud_var.get()) except ValueError: messagebox.showerror("RS232 Terminal", "Baud must be a number.", parent=self) return try: data_bits = int(self._data_bits_var.get()) except ValueError: data_bits = 8 ok = self.serial_manager.rs232_open( baud=baud, data_bits=data_bits, stop_bits=self._stop_bits_var.get(), parity=self._parity_var.get(), ) if not ok: messagebox.showerror("RS232 Terminal", "Device refused to open the RS232 port.\n" "Check that firmware is up to date.", parent=self) return self._is_open = True self._open_btn.config(text="Close", bg="#D94A4A") self._set_status(f"Open @ {baud}", "#27AE60") cfg = f"{baud} {data_bits}{self._parity_var.get()[0].upper()}{self._stop_bits_var.get()}" self._append_meta(f"--- Opened RS232 @ {cfg} ---") self._start_polling() def _close_port(self): self._stop_polling() try: self.serial_manager.rs232_close() except Exception: pass self._is_open = False self._open_btn.config(text="Open", bg="#27AE60") self._set_status("Not open", "#E74C3C") self._append_meta("--- Closed ---") def _set_status(self, text, color): self._status_label.config(text=text, fg=color) try: self._status_dot.itemconfig(self._dot_id, fill=color) except tk.TclError: pass def _start_polling(self): import time self._poll_stop.clear() def loop(): while not self._poll_stop.is_set(): try: data = self.serial_manager.rs232_poll() if data: self._rx_queue.put(data) except Exception: # Keep the thread alive across transient errors; a real # disconnect simply yields empty polls on the next tick. pass time.sleep(self.BG_POLL_SLEEP) self._poll_thread = threading.Thread(target=loop, daemon=True) self._poll_thread.start() self._schedule_ui_drain() def _stop_polling(self): self._poll_stop.set() if self._ui_poll_job: try: self.after_cancel(self._ui_poll_job) except Exception: pass self._ui_poll_job = None def _schedule_ui_drain(self): self._drain_rx_queue() if not self._poll_stop.is_set(): self._ui_poll_job = self.after(self.POLL_INTERVAL_MS, self._schedule_ui_drain) def _drain_rx_queue(self): while True: try: data = self._rx_queue.get_nowait() except queue.Empty: break self._append_rx(data) def _line_ending_bytes(self) -> bytes: disp = self._line_ending_var.get() for label, code in RS232_LINE_ENDINGS: if label == disp: return {"none": b"", "cr": b"\r", "lf": b"\n", "crlf": b"\r\n"}.get(code, b"") return b"" def _send_clicked(self, event=None): if not self._is_open: messagebox.showwarning("RS232 Terminal", "Open the port first.", parent=self) return text = self._cmd_var.get() payload = text.encode("utf-8", errors="replace") + self._line_ending_bytes() if self._send_bytes(payload): self._cmd_var.set("") def _send_preset(self, payload_str: str): if not self._is_open: messagebox.showwarning("RS232 Terminal", "Open the port first.", parent=self) return self._send_bytes(payload_str.encode("ascii", errors="replace")) def _send_bytes(self, data: bytes) -> bool: ok = self.serial_manager.rs232_send(data) if ok: if self._echo_var.get(): self._append_tx(data) else: self._append_err("!! send failed") return ok def _prefix(self) -> str: if self._timestamps_var.get(): return datetime.now().strftime("[%H:%M:%S.%f]")[:-3] + " " return "" def _format_payload(self, data: bytes, mode: str) -> str: if mode == "ascii": return self._to_printable(data) if mode == "hex": return " ".join(f"{b:02x}" for b in data) ascii_part = self._to_printable(data) hex_part = " ".join(f"{b:02x}" for b in data) return f"{ascii_part} [{hex_part}]" @staticmethod def _to_printable(data: bytes) -> str: out = [] for b in data: if b == 0x0D: out.append("\\r") elif b == 0x0A: out.append("\\n") elif b == 0x09: out.append("\\t") elif 0x20 <= b <= 0x7E: out.append(chr(b)) else: out.append(f"\\x{b:02x}") return "".join(out) def _mode_code(self) -> str: disp = self._display_mode_var.get() for label, code in DISPLAY_MODES: if label == disp: return code return "ascii_hex" def _append_line(self, text: str, tag: str): self._output.config(state="normal") self._output.insert("end", text + "\n", (tag,)) self._output.config(state="disabled") if self._auto_scroll_var.get(): self._output.see("end") def _append_tx(self, data: bytes): body = self._format_payload(data, self._mode_code()) self._append_line(f"{self._prefix()}> {body}", "tx") def _append_rx(self, data: bytes): body = self._format_payload(data, self._mode_code()) self._append_line(f"{self._prefix()}< {body}", "rx") def _append_meta(self, text: str): self._append_line(text, "meta") def _append_err(self, text: str): self._append_line(text, "err") def _clear(self): self._output.config(state="normal") self._output.delete("1.0", "end") self._output.config(state="disabled") def _save_log(self): path = filedialog.asksaveasfilename( parent=self, title="Save RS232 log", defaultextension=".log", filetypes=[("Log file", "*.log"), ("Text", "*.txt"), ("All files", "*.*")], ) if not path: return try: with open(path, "w", encoding="utf-8") as f: f.write(self._output.get("1.0", "end")) except OSError as e: messagebox.showerror("Save Log", f"Could not save:\n{e}", parent=self) def _on_close(self): try: if self._is_open: self._close_port() except Exception: pass self.destroy() if self._on_close_cb: try: self._on_close_cb() except Exception: pass