428 lines
16 KiB
Python
428 lines
16 KiB
Python
"""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("<Return>", 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
|