#!/usr/bin/env python3 """ GUI test tool for app_photomagnetic communication protocol. Built with Python native tkinter + threading. """ import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import threading import time import sys import os import struct import base64 from enum import IntEnum # Add scripts directory to path for smp_client import sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from smp_client import SMPClient try: import serial import serial.tools.list_ports except ImportError: print("Please install pyserial: pip install pyserial", file=sys.stderr) sys.exit(1) # ── Protocol constants ──────────────────────────────────────── HEADER = b"\x7e\xe7" HEADER_SIZE = len(HEADER) class Cmd(IntEnum): TEMP = 0x00 GET_ID = 0x01 RUNNING_STATE = 0x02 W_HEATING = 0x03 R_HEATING = 0x04 READ_VERSION = 0x06 WRITE_UID = 0x07 SET_LED = 0x08 # 调试用:灯带 RGB(仅调试固件支持) READ_POLE_NTC = 0x64 READ_HEATING_NTC = 0x65 HEATING_STATE_NAMES = {0: "OFF", 1: "ON"} STATE_NAMES = {0: "Standby", 1: "Running", 2: "Pause", 3: "Error"} # UID 编码规则(与 doc/UID编码规则.md 一致) UID_TYPE_NAMES = { 0x01: "冲击波", 0x02: "光磁/热疗", 0x03: "外置按摩头", 0xF0: "通用(未分类)", 0xFF: "未知", } UID_DEV_NAMES = { (0x01, 0x01): "标准冲击波手柄", (0x01, 0x02): "高能型冲击波手柄", (0x02, 0x01): "光磁手柄", (0x02, 0x02): "热疗手柄", (0x02, 0x03): "增强型光磁手柄", (0x03, 0x01): "捶打按摩头", (0x03, 0x02): "揉捏按摩头", (0x03, 0x03): "热敷按摩头", (0x03, 0x04): "EMS 电磁按摩头", (0x03, 0x05): "DMS 深层肌肉刺激仪", (0x03, 0x06): "负压吸力按摩头", (0x03, 0xF0): "通用按摩头(未分类)", (0x03, 0xFF): "未知设备(自动探测)", } # ── Protocol helpers (from test.py) ─────────────────────────── def crc16_modbus(data: bytes) -> int: crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1 return crc & 0xFFFF def build_frame(cmd: int, payload: bytes = b"") -> bytes: body = bytes([cmd, len(payload)]) + payload crc = crc16_modbus(body) return HEADER + body + bytes([crc & 0xFF, (crc >> 8) & 0xFF]) def parse_frame(frame: bytes) -> tuple[int, bytes] | None: if len(frame) < HEADER_SIZE + 2 + 2: return None if frame[:HEADER_SIZE] != HEADER: return None cmd = frame[HEADER_SIZE] length = frame[HEADER_SIZE + 1] expected = HEADER_SIZE + 2 + length + 2 if len(frame) != expected: return None data = frame[HEADER_SIZE + 2: HEADER_SIZE + 2 + length] crc_body = frame[HEADER_SIZE: HEADER_SIZE + 2 + length] actual_crc = crc16_modbus(crc_body) wire_crc = (frame[expected - 1] << 8) | frame[expected - 2] if actual_crc != wire_crc: return None return cmd, data def temp_from_data(data: bytes) -> float: return data[0] + data[1] / 100.0 if len(data) >= 2 else 0.0 # ── Serial I/O thread ───────────────────────────────────────── class SerialWorker: def __init__(self, gui_callback): self.ser: serial.Serial | None = None self._lock = threading.Lock() self._running = False self._monitoring = False self._reader_thread: threading.Thread | None = None self._gui = gui_callback # on_frame(cmd, data) called from reader thread @property def is_open(self) -> bool: return self.ser is not None and self.ser.is_open def open(self, port: str, baud: int) -> str | None: """Open serial port. Returns error string or None on success.""" try: ser = serial.Serial(port, baud, timeout=0.05) with self._lock: self.ser = ser self._running = True self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True) self._reader_thread.start() return None except serial.SerialException as e: return str(e) def close(self): with self._lock: self._running = False self._monitoring = False if self.ser: try: self.ser.close() except Exception: pass self.ser = None if self._reader_thread: self._reader_thread.join(timeout=2) self._reader_thread = None def send(self, cmd: int, payload: bytes = b"") -> bool: """Send a frame. Returns True if sent.""" with self._lock: if not self.ser or not self.ser.is_open: return False try: self.ser.write(build_frame(cmd, payload)) return True except serial.SerialException: return False def send_raw(self, data: bytes) -> bool: """Send raw bytes directly.""" with self._lock: if not self.ser or not self.ser.is_open: return False try: self.ser.write(data) return True except serial.SerialException: return False @property def monitoring(self) -> bool: return self._monitoring @monitoring.setter def monitoring(self, value: bool): self._monitoring = value def _reader_loop(self): buf = bytearray() while self._running: # Read data with self._lock: if not self.ser or not self.ser.is_open: time.sleep(0.05) continue try: chunk = self.ser.read(self.ser.in_waiting or 1) except serial.SerialException: chunk = b"" if not chunk: time.sleep(0.01) continue buf.extend(chunk) # Try to extract frames while len(buf) >= HEADER_SIZE + 2 + 2: # Find header idx = buf.find(HEADER) if idx > 0: del buf[:idx] continue if idx < 0: buf.clear() break length = buf[HEADER_SIZE + 1] total = HEADER_SIZE + 2 + length + 2 if len(buf) < total: break # wait for more frame = bytes(buf[:total]) del buf[:total] result = parse_frame(frame) if result: self._gui.on_frame_received(result[0], result[1]) else: self._gui.on_log(f"⚠ Bad frame: {frame.hex()}") # ── GUI Application ─────────────────────────────────────────── def read_version_file(filepath): """Read Zephyr VERSION file and return version dict""" version = {'major': 0, 'minor': 0, 'patchlevel': 0, 'tweak': 0, 'extraversion': ''} try: with open(filepath, 'r') as f: for line in f: line = line.strip() if '=' not in line or line.startswith('#'): continue key, val = line.split('=', 1) key = key.strip() val = val.strip().strip('"').strip("'") if key == 'VERSION_MAJOR': version['major'] = int(val) if val else 0 elif key == 'VERSION_MINOR': version['minor'] = int(val) if val else 0 elif key == 'PATCHLEVEL': version['patchlevel'] = int(val) if val else 0 elif key == 'VERSION_TWEAK': version['tweak'] = int(val) if val else 0 elif key == 'EXTRAVERSION': version['extraversion'] = val except (FileNotFoundError, ValueError): pass return version def format_version(version): """Format version dict to string like '0.0.9'""" v = f"{version['major']}.{version['minor']}.{version['patchlevel']}" if version['tweak']: v += f"+{version['tweak']}" if version['extraversion']: v += f"-{version['extraversion']}" return v class PhotomagneticGUI: def __init__(self): # Read version from VERSION file version_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'VERSION') self._version = read_version_file(version_path) self._version_str = format_version(self._version) self.root = tk.Tk() self.root.title(f"Photomagnetic Communication Tool v{self._version_str}") self.root.geometry("850x700") self.root.minsize(700, 550) self.worker = SerialWorker(self) self._monitor_after_id = None self._last_temp_time = 0.0 self._temp_interval = 0.0 self._build_ui() # ── UI Build ─────────────────────────────────────────── def _build_ui(self): # ── Top: Serial connection ── conn_frame = ttk.LabelFrame(self.root, text="Serial Connection", padding=8) conn_frame.pack(fill=tk.X, padx=8, pady=4) ttk.Label(conn_frame, text="Port:").grid(row=0, column=0, sticky=tk.W) self._port_var = tk.StringVar() self._port_combo = ttk.Combobox(conn_frame, textvariable=self._port_var, width=25) self._port_combo.grid(row=0, column=1, padx=4, sticky=tk.W) self._scan_btn = ttk.Button(conn_frame, text="↻", width=3, command=self._scan_ports) self._scan_btn.grid(row=0, column=2, padx=2) ttk.Label(conn_frame, text="Baud:").grid(row=0, column=3, padx=(12, 0), sticky=tk.W) self._baud_var = tk.StringVar(value="115200") self._baud_combo = ttk.Combobox(conn_frame, textvariable=self._baud_var, values=("9600", "19200", "38400", "57600", "115200", "230400", "460800"), width=10) self._baud_combo.grid(row=0, column=4, padx=4, sticky=tk.W) self._connect_btn = ttk.Button(conn_frame, text="Connect", command=self._toggle_connect) self._connect_btn.grid(row=0, column=5, padx=(12, 0)) self._status_lbl = ttk.Label(conn_frame, text="Disconnected", foreground="gray") self._status_lbl.grid(row=0, column=6, padx=8, sticky=tk.W) conn_frame.columnconfigure(6, weight=1) # Version display version_frame = ttk.Frame(self.root) version_frame.pack(fill=tk.X, padx=8, pady=0) ttk.Label(version_frame, text="Tool Version:", font=("Consolas", 9), foreground="#666").pack(side=tk.LEFT) ttk.Label(version_frame, text=f"v{self._version_str}", font=("Consolas", 9, "bold"), foreground="#333").pack(side=tk.LEFT, padx=(2, 12)) ttk.Label(version_frame, text="Firmware Version:", font=("Consolas", 9), foreground="#666").pack(side=tk.LEFT) self._fw_version_lbl = ttk.Label(version_frame, text="N/A", font=("Consolas", 9, "bold"), foreground="#333") self._fw_version_lbl.pack(side=tk.LEFT, padx=2) # ── Main content: left (controls) + right (display) ── main_frame = ttk.Frame(self.root) main_frame.pack(fill=tk.BOTH, expand=True, padx=8, pady=2) left = ttk.Frame(main_frame) left.pack(side=tk.LEFT, fill=tk.BOTH, expand=False) right = ttk.Frame(main_frame) right.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=(6, 0)) # ── Left: Command panel ── cmd_frame = ttk.LabelFrame(left, text="Commands", padding=8) cmd_frame.pack(fill=tk.X) # Row 1: GET_ID + Version + State r1 = ttk.Frame(cmd_frame) r1.pack(fill=tk.X, pady=2) ttk.Button(r1, text="Get Device ID", width=14, command=self._cmd_get_id).pack(side=tk.LEFT) ttk.Button(r1, text="Read Version", width=12, command=self._cmd_read_version).pack(side=tk.LEFT, padx=4) ttk.Label(r1, text="State:").pack(side=tk.LEFT, padx=(10, 2)) self._state_combo = ttk.Combobox(r1, values=["Standby", "Running", "Pause", "Error"], state="readonly", width=10) self._state_combo.current(0) self._state_combo.pack(side=tk.LEFT) ttk.Button(r1, text="Set", width=4, command=self._cmd_set_state).pack(side=tk.LEFT, padx=4) # Row 2: Heating r2 = ttk.Frame(cmd_frame) r2.pack(fill=tk.X, pady=2) ttk.Label(r2, text="Heating:").pack(side=tk.LEFT) self._heating_var = tk.IntVar(value=0) self._heating_spin = tk.Spinbox(r2, from_=0, to=60, textvariable=self._heating_var, width=5, state="readonly") self._heating_spin.pack(side=tk.LEFT, padx=2) ttk.Label(r2, text="°C").pack(side=tk.LEFT) ttk.Button(r2, text="Set / Stop", width=10, command=self._cmd_set_heating).pack(side=tk.LEFT, padx=6) ttk.Button(r2, text="Read State", width=10, command=self._cmd_read_heating).pack(side=tk.LEFT) # Row 3: NTC r3 = ttk.Frame(cmd_frame) r3.pack(fill=tk.X, pady=2) ttk.Label(r3, text="Pole NTC #").pack(side=tk.LEFT) self._ntc_idx_var = tk.IntVar(value=0) ntc_spin = tk.Spinbox(r3, from_=0, to=7, textvariable=self._ntc_idx_var, width=3, state="readonly") ntc_spin.pack(side=tk.LEFT, padx=2) ttk.Button(r3, text="Read Pole NTC", width=12, command=self._cmd_read_pole_ntc).pack(side=tk.LEFT, padx=6) ttk.Button(r3, text="Read Pad NTC", width=12, command=self._cmd_read_heating_ntc).pack(side=tk.LEFT) # Row 4: Monitor + Raw r4 = ttk.Frame(cmd_frame) r4.pack(fill=tk.X, pady=2) self._monitor_btn = ttk.Button(r4, text="▶ Monitor", width=12, command=self._toggle_monitor) self._monitor_btn.pack(side=tk.LEFT) ttk.Label(r4, text="Raw Hex:").pack(side=tk.LEFT, padx=(10, 2)) self._raw_var = tk.StringVar() ttk.Entry(r4, textvariable=self._raw_var, width=18).pack(side=tk.LEFT, padx=2) ttk.Button(r4, text="Send", width=5, command=self._cmd_raw).pack(side=tk.LEFT, padx=2) # Row 5: Loop r5 = ttk.Frame(cmd_frame) r5.pack(fill=tk.X, pady=2) self._loop_btn = ttk.Button(r5, text="▶ Loop GET_ID", width=14, command=self._toggle_loop) self._loop_btn.pack(side=tk.LEFT) ttk.Separator(r5, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=8) self._upgrade_btn = ttk.Button(r5, text="⬆ Upgrade Firmware", width=18, command=self._cmd_upgrade) self._upgrade_btn.pack(side=tk.LEFT, padx=4) # Row 6: UID 写入(生产烧录自定义 UID,10 字节编码) r6 = ttk.Frame(cmd_frame) r6.pack(fill=tk.X, pady=2) ttk.Label(r6, text="UID(10B):").pack(side=tk.LEFT) self._uid_var = tk.StringVar() uid_entry = ttk.Entry(r6, textvariable=self._uid_var, width=24) uid_entry.pack(side=tk.LEFT, padx=2) ttk.Button(r6, text="Write UID", width=10, command=self._cmd_write_uid).pack(side=tk.LEFT, padx=2) ttk.Button(r6, text="Use Chip UID", width=12, command=self._cmd_clear_uid).pack(side=tk.LEFT, padx=2) # Row 7: LED 调色调试(需调试固件 CONFIG_APP_DEBUG_LED_STRIP=y, # 生产固件无 0x08 命令) r7 = ttk.Frame(cmd_frame) r7.pack(fill=tk.X, pady=2) ttk.Label(r7, text="LED:").pack(side=tk.LEFT) self._led_r_var = tk.IntVar(value=255) self._led_g_var = tk.IntVar(value=255) self._led_b_var = tk.IntVar(value=255) # 允许键盘直接输入(0~255),上下箭头仍可用 vcmd = (self.root.register(self._validate_led_channel), "%P") for var in (self._led_r_var, self._led_g_var, self._led_b_var): tk.Spinbox(r7, from_=0, to=255, textvariable=var, width=4, validate="key", validatecommand=vcmd).pack( side=tk.LEFT, padx=1) self._led_swatch = tk.Label(r7, text=" ", bg="#FFFFFF", width=3, relief=tk.SUNKEN) self._led_swatch.pack(side=tk.LEFT, padx=4) ttk.Button(r7, text="Set", width=4, command=self._cmd_set_led).pack(side=tk.LEFT, padx=2) ttk.Button(r7, text="Off", width=4, command=self._cmd_led_off).pack(side=tk.LEFT, padx=2) # ── Right top: Live display ── disp_frame = ttk.LabelFrame(right, text="Live Values", padding=6) disp_frame.pack(fill=tk.X) self._disp_labels = {} entries = [ ("Temperature", "--- °C", "temp"), ("Sample Rate", "---", "sample_rate"), ("Heating State", "---", "heat_state"), ("Heating Temp.", "--- °C", "heat_temp"), ("Device ID", "---", "dev_id"), ("Running State", "---", "run_state"), ] for i, (label, default, key) in enumerate(entries): ttk.Label(disp_frame, text=label + ":").grid(row=i, column=0, sticky=tk.W, padx=4) lbl = ttk.Label(disp_frame, text=default, font=("Consolas", 11, "bold"), foreground="#333") lbl.grid(row=i, column=1, sticky=tk.W, padx=4) self._disp_labels[key] = lbl # Pole NTC sub-frame self._ntc_labels = [] ntc_header = ttk.Label(disp_frame, text="Pole NTCs:") ntc_header.grid(row=len(entries), column=0, sticky=tk.W, padx=4, pady=(4, 0)) ntc_row = ttk.Frame(disp_frame) ntc_row.grid(row=len(entries), column=1, sticky=tk.W, padx=4, pady=(4, 0)) for i in range(8): lbl = ttk.Label(ntc_row, text=f"{i}:---", font=("Consolas", 10), width=8) lbl.pack(side=tk.LEFT) self._ntc_labels.append(lbl) disp_frame.columnconfigure(1, weight=1) # ── Bottom: Log ── log_frame = ttk.LabelFrame(right, text="Log", padding=4) log_frame.pack(fill=tk.BOTH, expand=True, pady=(4, 0)) log_toolbar = ttk.Frame(log_frame) log_toolbar.pack(fill=tk.X, pady=(0, 2)) ttk.Button(log_toolbar, text="Clear", width=6, command=self._clear_log).pack(side=tk.RIGHT) self._log_text = scrolledtext.ScrolledText(log_frame, height=12, font=("Consolas", 9), wrap=tk.WORD, state=tk.DISABLED) self._log_text.pack(fill=tk.BOTH, expand=True) self._scan_ports() # ── Serial connection ──────────────────────────────── def _scan_ports(self): ports = [p.device for p in serial.tools.list_ports.comports()] self._port_combo["values"] = ports if ports and not self._port_var.get(): self._port_var.set(ports[0]) def _toggle_connect(self): if self.worker.is_open: self._disconnect() else: self._connect() def _connect(self): port = self._port_var.get().strip() if not port: messagebox.showerror("Error", "Select a serial port") return try: baud = int(self._baud_var.get()) except ValueError: messagebox.showerror("Error", "Invalid baud rate") return err = self.worker.open(port, baud) if err: messagebox.showerror("Connection Error", err) return self._connect_btn.configure(text="Disconnect") self._status_lbl.configure(text=f"Connected @ {baud}", foreground="green") self.on_log(f"Connected to {port} @ {baud} baud") def _disconnect(self): self._stop_monitor() self._stop_loop() self.worker.close() self._connect_btn.configure(text="Connect") self._status_lbl.configure(text="Disconnected", foreground="gray") self.on_log("Disconnected") def _check_connected(self) -> bool: if not self.worker.is_open: self.on_log("✗ Not connected") return False return True # ── Commands ───────────────────────────────────────── def _cmd_get_id(self): if not self._check_connected(): return self.worker.send(Cmd.GET_ID) self.on_log("→ GET_ID sent") def _cmd_read_version(self): """Request firmware version from device""" if not self._check_connected(): return self.worker.send(Cmd.READ_VERSION) self.on_log("→ VERSION request sent (0x06)") print(f"[DEBUG] VERSION request sent: cmd=0x{Cmd.READ_VERSION:02X}", flush=True) def _cmd_set_state(self): if not self._check_connected(): return idx = self._state_combo.current() self.worker.send(Cmd.RUNNING_STATE, bytes([idx])) self.on_log(f"→ Set state: {self._state_combo.get()} ({idx})") def _cmd_set_heating(self): if not self._check_connected(): return target = self._heating_var.get() self.worker.send(Cmd.W_HEATING, bytes([target])) if target == 0: self.on_log("→ Stop heating") else: self.on_log(f"→ Set heating to {target}°C") def _cmd_read_heating(self): if not self._check_connected(): return self.worker.send(Cmd.R_HEATING) self.on_log("→ Read heating state") def _cmd_read_pole_ntc(self): if not self._check_connected(): return idx = self._ntc_idx_var.get() self.worker.send(Cmd.READ_POLE_NTC, bytes([idx])) self.on_log(f"→ Read pole NTC #{idx}") def _cmd_read_heating_ntc(self): if not self._check_connected(): return self.worker.send(Cmd.READ_HEATING_NTC) self.on_log("→ Read heating pad NTC") def _cmd_raw(self): if not self._check_connected(): return hex_str = self._raw_var.get().strip() if not hex_str: return try: payload = bytes.fromhex(hex_str.replace(" ", "")) except ValueError: self.on_log(f"✗ Invalid hex: {hex_str}") return cmd = payload[0] data = payload[1:] if len(payload) > 1 else b"" self.worker.send(cmd, data) self.on_log(f"→ Raw: CMD=0x{cmd:02X} data={data.hex()}") self._raw_var.set("") # ── UID ──────────────────────────────────────────────── def _cmd_write_uid(self): """写生产 UID(10 字节 = 20 个 hex 字符,编码见 doc/UID编码规则.md)""" if not self._check_connected(): return hex_str = self._uid_var.get().strip().replace(" ", "") if not hex_str: self.on_log("✗ UID 为空") return try: uid = bytes.fromhex(hex_str) except ValueError: self.on_log("✗ UID 不是合法 hex(20 个 hex 字符 = 10 字节)") return if len(uid) != 10: self.on_log(f"✗ UID 长度错误: {len(uid)} 字节,应为 10 字节") return self.worker.send(Cmd.WRITE_UID, uid) self.on_log(f"→ Write UID: {uid.hex(' ').upper()} (10B)") def _cmd_clear_uid(self): """清除自定义 UID,恢复芯片 UID""" if not self._check_connected(): return self.worker.send(Cmd.WRITE_UID, b"") self.on_log("→ Clear UID (restore chip UID)") self._uid_var.set("") # ── LED 调色调试 ──────────────────────────────────────── def _validate_led_channel(self, new_text: str) -> bool: """RGB 输入框键盘校验:允许空串(编辑中)或 0~255 的整数""" if new_text == "": return True return (new_text.isascii() and new_text.isdigit() and int(new_text) <= 255) def _cmd_set_led(self): """调试用:灯带整体设为 RGB 颜色(0x08,需 CONFIG_APP_DEBUG_LED_STRIP=y) """ if not self._check_connected(): return try: r, g, b = (self._led_r_var.get(), self._led_g_var.get(), self._led_b_var.get()) except tk.TclError: r = g = b = 0 # 输入框为空(编辑中)时按 0 处理 r, g, b = max(0, min(255, r)), max(0, min(255, g)), max(0, min(255, b)) self.worker.send(Cmd.SET_LED, bytes([r, g, b])) self._led_swatch.configure(bg=f"#{r:02X}{g:02X}{b:02X}") self.on_log(f"→ Set LED RGB({r},{g},{b}) #{r:02X}{g:02X}{b:02X}") def _cmd_led_off(self): """调试用:熄灭灯带""" if not self._check_connected(): return self.worker.send(Cmd.SET_LED, b"\x00\x00\x00") self._led_swatch.configure(bg="#000000") self.on_log("→ LED off") # ── Monitor ────────────────────────────────────────── def _cmd_upgrade(self): """Upgrade firmware: send upgrade cmd -> wait reboot -> SMP upload in-process""" if not self.worker.is_open: self._append_log("Please connect serial port first") return from tkinter import filedialog firmware = filedialog.askopenfilename( title="Select firmware file", filetypes=[("Signed Binary", "*.signed.bin"), ("Binary", "*.bin"), ("All", "*.*")], initialdir=os.path.join(os.getcwd(), "build", "app_photomagnetic", "zephyr") ) if not firmware: return self._append_log(f"Upgrade start: {os.path.basename(firmware)}") self._upgrade_btn.configure(state=tk.DISABLED) # 升级期间锁定连接相关控件:禁止用户中途 Connect/换端口,否则会抢占 # 串口导致 smpmgr 上传中断、slot0 被擦除一半 → 设备变砖。 for w in (self._connect_btn, self._port_combo, self._baud_combo, self._scan_btn): w.configure(state=tk.DISABLED) def upgrade_thread(): try: port = self._port_var.get() # Step 1: Close GUI serial connection first(释放协议串口) self._append_log("Preparing serial port...") self._disconnect() # Step 2: Send upgrade command using separate serial connection self._append_log("-> Sending upgrade command...") body = bytes([0xFF, 0x01, 0x01]) crc = crc16_modbus(body) upgrade_frame = HEADER + body + struct.pack('= 100: self._upgrade_last_pct = pct self.root.after(0, self._append_log, f" Uploading... {offset}/{total} ({pct}%)") def _toggle_monitor(self): if self.worker.monitoring: self._stop_monitor() else: self._start_monitor() def _start_monitor(self): if not self._check_connected(): return self.worker.monitoring = True self._monitor_btn.configure(text="■ Stop Monitor") self.on_log("▶ Monitoring started") def _stop_monitor(self): self.worker.monitoring = False self._monitor_btn.configure(text="▶ Monitor") if self._monitor_after_id: self.root.after_cancel(self._monitor_after_id) self._monitor_after_id = None # ── Loop ───────────────────────────────────────────── def _toggle_loop(self): if self._loop_running: self._stop_loop() else: self._start_loop() def _start_loop(self): if not self._check_connected(): return self._loop_running = True self._loop_btn.configure(text="■ Stop Loop") self.on_log("▶ Loop GET_ID started") self._loop_tick() def _stop_loop(self): self._loop_running = False self._loop_btn.configure(text="▶ Loop GET_ID") if hasattr(self, '_loop_after_id') and self._loop_after_id: self.root.after_cancel(self._loop_after_id) self._loop_after_id = None def _loop_tick(self): if not self._loop_running or not self.worker.is_open: self._stop_loop() return self.worker.send(Cmd.GET_ID) self._loop_after_id = self.root.after(1000, self._loop_tick) # ── Callbacks from SerialWorker (called in reader thread) ── def on_frame_received(self, cmd: int, data: bytes): """Called from reader thread. Schedule UI update.""" self.root.after(0, self._handle_frame, cmd, data) def on_log(self, msg: str): """Thread-safe log write.""" self.root.after(0, self._append_log, msg) # ── Frame handling (runs in main thread via after) ─── def _handle_frame(self, cmd: int, data: bytes): raw_hex = data.hex(" ") print(f"[DEBUG] Frame received: cmd=0x{cmd:02X}, data={raw_hex}", flush=True) # ── 0x00 TEMP: 设备主动推送, 只更新 live value, 不输出 log ── if cmd == Cmd.TEMP: t = temp_from_data(data) now = time.time() if self._last_temp_time > 0: self._temp_interval = now - self._last_temp_time hz = 1.0 / self._temp_interval if self._temp_interval > 0 else 0 self._disp_labels["sample_rate"].configure( text=f"{self._temp_interval*1000:.0f}ms ({hz:.1f}Hz)") self._last_temp_time = now self._disp_labels["temp"].configure(text=f"{t:.2f} °C") elif cmd == Cmd.GET_ID: id_str = data.hex(" ") id_show = id_str[:47] + "..." if len(id_str) > 50 else id_str self._disp_labels["dev_id"].configure(text=id_show) if len(data) == 10: # 生产 UID:[设备类型][产品代次][生产年月 4B][序列号 4B] dev_type, gen = data[0], data[1] ym = data[2:6].hex() serial = data[6:10].hex() tname = UID_TYPE_NAMES.get(dev_type, f"0x{dev_type:02X}") dev = UID_DEV_NAMES.get((dev_type, gen)) desc = f"{tname}·代{gen}" + (f" ({dev})" if dev else "") self._append_log( f"🆔 Device ID (10B UID): {raw_hex} | {desc} | " f"年月={ym} 序列号={serial}") else: self._append_log(f"🆔 Device ID ({len(data)}B): {raw_hex}") elif cmd == Cmd.READ_VERSION: # 固件回复: ASCII 字符串,如 b"0.0.14" ver = data.decode("utf-8", "replace").strip("\x00 \r\n") self._fw_version_lbl.configure(text=ver if ver else "?") self._append_log(f"ℹ️ Firmware version: {ver if ver else '(empty)'}") elif cmd == Cmd.RUNNING_STATE: state = data[0] if data else -1 name = STATE_NAMES.get(state, f"Unknown({state})") self._disp_labels["run_state"].configure(text=name) self._append_log(f"🔁 State: {name}") elif cmd == Cmd.W_HEATING: target = data[0] if data else 0 if target == 0: self._append_log(f"✓ Heating stopped (ACK)") else: self._append_log(f"✓ Heating set to {target}°C (ACK)") elif cmd == Cmd.R_HEATING: if len(data) >= 3: s = HEATING_STATE_NAMES.get(data[0], f"?{data[0]}") t = temp_from_data(data[1:3]) self._disp_labels["heat_state"].configure(text=s) self._disp_labels["heat_temp"].configure(text=f"{t:.2f} °C") self._append_log(f"🔥 Heating: {s}, temp={t:.2f}°C") else: self._append_log(f"🔥 Heating: {raw_hex}") elif cmd == Cmd.READ_POLE_NTC: if len(data) >= 2: t = temp_from_data(data) self._append_log(f"📡 Pole NTC reply: {t:.2f}°C ({raw_hex})") else: self._append_log(f"📡 Pole NTC: {raw_hex}") elif cmd == Cmd.READ_HEATING_NTC: t = temp_from_data(data) self._append_log(f"🔥 Pad NTC: {t:.2f}°C ({raw_hex})") elif cmd == Cmd.WRITE_UID: status = data[0] if data else 0xFF if status == 0x00: self._append_log("✓ UID 写入成功") # 刷新 ID 显示 self.worker.send(Cmd.GET_ID) elif status == 0x01: self._append_log("✗ UID 写入失败(flash 错误)") elif status == 0x02: self._append_log("✗ UID 长度错误(需 12 字节或空=清除)") else: self._append_log(f"📦 UID ACK: status=0x{status:02X}") else: self._append_log(f"📦 CMD=0x{cmd:02X} data={raw_hex}") def _set_ntc_display(self, idx: int, temp: float): if 0 <= idx < len(self._ntc_labels): self._ntc_labels[idx].configure(text=f"{idx}:{temp:.1f}") def _clear_log(self): self._log_text.configure(state=tk.NORMAL) self._log_text.delete(1.0, tk.END) self._log_text.configure(state=tk.DISABLED) def _append_log(self, msg: str): self._log_text.configure(state=tk.NORMAL) self._log_text.insert(tk.END, msg + "\n") self._log_text.see(tk.END) self._log_text.configure(state=tk.DISABLED) # ── Run ────────────────────────────────────────────── def run(self): self.root.mainloop() self._disconnect() if __name__ == "__main__": app = PhotomagneticGUI() app.run()