#!/usr/bin/env python3 """ CAN IO Board GUI ================ Simple Tkinter front-end for the XIAO ESP32-S3 CAN IO board, talking through a second ESP32 running SLCAN firmware (any SLCAN/LAWICEL adapter works). Features -------- - Serial port picker (with refresh) + bitrate + base ID - Live view of the 4 inputs and 2 outputs (green = active) - Per-output ON / OFF / Toggle buttons + "Request status" - Frame log with decoded status frames Protocol (must match the firmware's config.h / PROTOCOL.md) ------------------------------------------------------------ - STATUS (board -> bus, ID = base + 0): data[0] input bitmap, data[1] output bitmap, data[2] reason, data[3] reserved, data[4:8] uptime seconds (uint32 LE) - COMMAND (bus -> board, ID = base + 1): data[0] = 0x00 GET_STATUS 0x01 SET_OUTPUT (data[1] index, data[2] 0/1) 0x02 SET_ALL (data[1] mask, data[2] values) 0x03 TOGGLE (data[1] index) Usage ----- pip install -r requirements.txt python can_io_gui.py """ import queue import threading import time import tkinter as tk from tkinter import ttk, messagebox try: import can from serial.tools import list_ports except ImportError as exc: # pragma: no cover raise SystemExit( f"Missing dependency: {exc.name}\n" "Install requirements first: pip install -r requirements.txt" ) # -------------------------------------------------------------------------- # Protocol constants (keep in sync with firmware include/config.h) # -------------------------------------------------------------------------- DEFAULT_BASE_ID = 0x100 NUM_INPUTS = 4 NUM_OUTPUTS = 2 CMD_GET_STATUS = 0x00 CMD_SET_OUTPUT = 0x01 CMD_SET_ALL = 0x02 CMD_TOGGLE = 0x03 REASONS = {0: "periodic", 1: "change", 2: "request", 3: "boot"} BITRATES = [125000, 250000, 500000, 1000000] DEFAULT_BITRATE = 500000 DEFAULT_TTY_BAUD = 115200 # ignored by native-USB SLCAN adapters STALE_AFTER_S = 3.5 # no status for this long -> mark stale COLOR_ON = "#2ecc71" COLOR_OFF = "#4a4a4a" COLOR_STALE = "#b3771e" class CanIoGui: """Main application window.""" def __init__(self, root: tk.Tk): self.root = root self.root.title("CAN IO Board") self.root.resizable(False, False) self.bus = None # can.Bus when connected self.rx_thread = None self.rx_running = threading.Event() self.rx_queue = queue.Queue() # RX thread -> GUI thread self.last_status_time = None self.inputs = 0 self.outputs = 0 self.active_base = DEFAULT_BASE_ID # parsed once per connect self._build_ui() self._refresh_ports() self.root.protocol("WM_DELETE_WINDOW", self._on_close) self.root.after(50, self._poll_rx_queue) # ------------------------------------------------------------------ UI -- def _build_ui(self): pad = {"padx": 6, "pady": 4} # --- connection bar ------------------------------------------------- conn = ttk.LabelFrame(self.root, text="SLCAN connection") conn.grid(row=0, column=0, columnspan=2, sticky="ew", **pad) ttk.Label(conn, text="Port:").grid(row=0, column=0, sticky="e") self.port_combo = ttk.Combobox(conn, width=34, state="readonly") self.port_combo.grid(row=0, column=1, **pad) ttk.Button(conn, text="⟳", width=3, command=self._refresh_ports).grid(row=0, column=2) ttk.Label(conn, text="Bitrate:").grid(row=0, column=3, sticky="e") self.bitrate_combo = ttk.Combobox( conn, width=8, state="readonly", values=[str(b) for b in BITRATES]) self.bitrate_combo.set(str(DEFAULT_BITRATE)) self.bitrate_combo.grid(row=0, column=4, **pad) ttk.Label(conn, text="Base ID:").grid(row=1, column=0, sticky="e") self.base_id_entry = ttk.Entry(conn, width=8) self.base_id_entry.insert(0, f"0x{DEFAULT_BASE_ID:03X}") self.base_id_entry.grid(row=1, column=1, sticky="w", **pad) self.connect_btn = ttk.Button(conn, text="Connect", command=self._toggle_connection) self.connect_btn.grid(row=1, column=4, sticky="ew", **pad) self.conn_label = ttk.Label(conn, text="disconnected", foreground="gray") self.conn_label.grid(row=1, column=2, columnspan=2) # --- inputs ---------------------------------------------------------- in_frame = ttk.LabelFrame(self.root, text="Inputs") in_frame.grid(row=1, column=0, sticky="nsew", **pad) self.input_leds = [] for i in range(NUM_INPUTS): led = self._make_led(in_frame, f"IN{i + 1}", row=0, col=i) self.input_leds.append(led) # --- outputs --------------------------------------------------------- out_frame = ttk.LabelFrame(self.root, text="Outputs") out_frame.grid(row=1, column=1, sticky="nsew", **pad) self.output_leds = [] for i in range(NUM_OUTPUTS): led = self._make_led(out_frame, f"OUT{i + 1}", row=0, col=i) self.output_leds.append(led) btns = ttk.Frame(out_frame) btns.grid(row=2, column=i, padx=4, pady=2) ttk.Button(btns, text="On", width=4, command=lambda i=i: self._set_output(i, True) ).pack(side="left") ttk.Button(btns, text="Off", width=4, command=lambda i=i: self._set_output(i, False) ).pack(side="left") ttk.Button(btns, text="Toggle", width=7, command=lambda i=i: self._toggle_output(i) ).pack(side="left") # --- status row -------------------------------------------------- status_bar = ttk.Frame(self.root) status_bar.grid(row=2, column=0, columnspan=2, sticky="ew", **pad) ttk.Button(status_bar, text="Request status", command=self._request_status).pack(side="left") self.uptime_label = ttk.Label(status_bar, text="uptime: —") self.uptime_label.pack(side="left", padx=12) self.fresh_label = ttk.Label(status_bar, text="") self.fresh_label.pack(side="left", padx=12) # --- log ----------------------------------------------------------- log_frame = ttk.LabelFrame(self.root, text="Bus log") log_frame.grid(row=3, column=0, columnspan=2, sticky="nsew", **pad) self.log_text = tk.Text(log_frame, width=88, height=12, state="disabled", font=("Courier", 9)) self.log_text.pack(side="left", fill="both", expand=True) scroll = ttk.Scrollbar(log_frame, command=self.log_text.yview) scroll.pack(side="right", fill="y") self.log_text.configure(yscrollcommand=scroll.set) def _make_led(self, parent, label, row, col): """Create one round indicator + caption; return the canvas.""" canvas = tk.Canvas(parent, width=34, height=34, highlightthickness=0) canvas.grid(row=row, column=col, padx=10, pady=(6, 0)) oval = canvas.create_oval(4, 4, 30, 30, fill=COLOR_OFF, outline="#222") canvas.oval = oval ttk.Label(parent, text=label).grid(row=row + 1, column=col) return canvas def _set_led(self, canvas, on, stale=False): color = COLOR_STALE if stale else (COLOR_ON if on else COLOR_OFF) canvas.itemconfig(canvas.oval, fill=color) # ------------------------------------------------------------ connection -- def _refresh_ports(self): ports = list_ports.comports() values = [f"{p.device} — {p.description}" for p in ports] self.port_combo["values"] = values if values and not self.port_combo.get(): self.port_combo.current(0) def _selected_port(self): sel = self.port_combo.get() return sel.split(" — ")[0].strip() if sel else None @property def status_id(self): return self.active_base + 0 @property def command_id(self): return self.active_base + 1 def _toggle_connection(self): if self.bus: self._disconnect() else: self._connect() def _connect(self): port = self._selected_port() if not port: messagebox.showwarning("No port", "Select a serial port first.") return try: # Base ID entry is interpreted as hex ("0x100" or "100"). self.active_base = int( self.base_id_entry.get().strip().replace("0x", ""), 16) except ValueError: messagebox.showerror("Bad base ID", "Base ID must be hex, e.g. 0x100") return try: self.bus = can.Bus( interface="slcan", channel=port, ttyBaudrate=DEFAULT_TTY_BAUD, bitrate=int(self.bitrate_combo.get()), ) except Exception as exc: messagebox.showerror("Connection failed", str(exc)) self.bus = None return self.rx_running.set() self.rx_thread = threading.Thread(target=self._rx_loop, daemon=True) self.rx_thread.start() self.connect_btn.config(text="Disconnect") self.conn_label.config(text=f"connected: {port}", foreground="green") self._log(f"connected to {port} @ {self.bitrate_combo.get()} bit/s") # Ask the board for its current state right away. self.root.after(300, self._request_status) def _disconnect(self): self.rx_running.clear() if self.rx_thread: self.rx_thread.join(timeout=1.0) self.rx_thread = None if self.bus: try: self.bus.shutdown() except Exception: pass self.bus = None self.connect_btn.config(text="Connect") self.conn_label.config(text="disconnected", foreground="gray") self.last_status_time = None self._log("disconnected") # ------------------------------------------------------------------- RX -- def _rx_loop(self): """Background thread: read frames, hand them to the GUI thread.""" while self.rx_running.is_set(): try: msg = self.bus.recv(timeout=0.2) except Exception as exc: self.rx_queue.put(("error", str(exc))) break if msg is not None: self.rx_queue.put(("frame", msg)) def _poll_rx_queue(self): """GUI thread: apply everything the RX thread queued up.""" try: while True: kind, payload = self.rx_queue.get_nowait() if kind == "frame": self._handle_frame(payload) elif kind == "error": self._log(f"RX error: {payload}") self._disconnect() except queue.Empty: pass self._update_freshness() self.root.after(50, self._poll_rx_queue) def _handle_frame(self, msg): data = bytes(msg.data) if (not msg.is_extended_id and msg.arbitration_id == self.status_id and len(data) >= 8): self.inputs, self.outputs = data[0], data[1] reason = REASONS.get(data[2], f"?{data[2]}") uptime = int.from_bytes(data[4:8], "little") self.last_status_time = time.monotonic() for i, led in enumerate(self.input_leds): self._set_led(led, (self.inputs >> i) & 1) for i, led in enumerate(self.output_leds): self._set_led(led, (self.outputs >> i) & 1) h, rem = divmod(uptime, 3600) m, s = divmod(rem, 60) self.uptime_label.config(text=f"uptime: {h:d}:{m:02d}:{s:02d}") # Heartbeats every second would flood the log; only log events. if reason != "periodic": self._log(f"RX 0x{msg.arbitration_id:03X} " f"{data.hex(' ')} (status: {reason})") else: self._log(f"RX 0x{msg.arbitration_id:03X} {data.hex(' ')}") def _update_freshness(self): """Gray out the view when the board stops talking.""" if self.bus is None: self.fresh_label.config(text="") return if self.last_status_time is None: self.fresh_label.config(text="waiting for board...", foreground="orange") return age = time.monotonic() - self.last_status_time if age > STALE_AFTER_S: self.fresh_label.config( text=f"STALE — last status {age:.0f}s ago", foreground="red") for led in self.input_leds + self.output_leds: self._set_led(led, False, stale=True) else: self.fresh_label.config(text="live", foreground="green") # ------------------------------------------------------------------- TX -- def _send(self, data, what): if not self.bus: messagebox.showinfo("Not connected", "Connect to an SLCAN " "device first.") return try: msg = can.Message(arbitration_id=self.command_id, data=bytes(data), is_extended_id=False) self.bus.send(msg, timeout=0.5) self._log(f"TX 0x{self.command_id:03X} " f"{bytes(data).hex(' ')} ({what})") except Exception as exc: self._log(f"TX error: {exc}") def _request_status(self): self._send([CMD_GET_STATUS], "get status") def _set_output(self, index, on): self._send([CMD_SET_OUTPUT, index, 1 if on else 0], f"OUT{index + 1} {'on' if on else 'off'}") def _toggle_output(self, index): self._send([CMD_TOGGLE, index], f"toggle OUT{index + 1}") # ------------------------------------------------------------------ misc -- def _log(self, text): stamp = time.strftime("%H:%M:%S") self.log_text.configure(state="normal") self.log_text.insert("end", f"{stamp} {text}\n") # Cap the log at ~500 lines so long sessions stay snappy. if int(self.log_text.index("end-1c").split(".")[0]) > 500: self.log_text.delete("1.0", "100.0") self.log_text.see("end") self.log_text.configure(state="disabled") def _on_close(self): self._disconnect() self.root.destroy() def main(): root = tk.Tk() try: # nicer widgets where available ttk.Style().theme_use("clam") except tk.TclError: pass CanIoGui(root) root.mainloop() if __name__ == "__main__": main()