forked from EmbeddedTeam/app_photomagnetic
27ae21bbea
- Add UID encoding/decoding with flash storage and CRC16 validation - Add write UID (0x07) and LED strip color (0x08) protocol commands - Add debug LED strip configuration option - Switch MCUboot signing from RSA-2048 to ECDSA-P256 - Add SMP serial client for firmware upload over UART - Add firmware version output on boot - Update upgrade documentation with ECDSA key generation steps
938 lines
37 KiB
Python
938 lines
37 KiB
Python
#!/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('<H', crc)
|
||
try:
|
||
with serial.Serial(port, 115200, timeout=0.1) as s:
|
||
s.write(upgrade_frame)
|
||
self._append_log(f" Sent: {upgrade_frame.hex()}")
|
||
except Exception as e:
|
||
self._append_log(f"Failed to send upgrade command: {e}")
|
||
return
|
||
|
||
# Step 3: Wait for device to reboot into MCUboot
|
||
self._append_log("Waiting for device to enter bootloader...")
|
||
time.sleep(1.2)
|
||
|
||
# Step 4: 进程内 SMP 上传(不再调用外部 smpmgr 命令)
|
||
# MCUboot serial recovery 固定把镜像写到 slot0(主槽),
|
||
# 上传完 reset 即可,无需 test/confirm。
|
||
from smp_client import SMPClient
|
||
client = SMPClient(port, 115200, log=lambda m: None)
|
||
try:
|
||
client.open()
|
||
self._append_log("Connected to bootloader, uploading...")
|
||
self._upgrade_last_pct = -1
|
||
ok = client.upload_image(firmware,
|
||
progress_callback=self._upgrade_progress)
|
||
if not ok:
|
||
self._append_log(
|
||
"⚠ 上传失败。设备可能已停在 bootloader,重新点升级即可重传;"
|
||
"或 west flash -d build 重烧应用")
|
||
return
|
||
self._append_log("Upload complete (100%)")
|
||
|
||
self._append_log("Resetting device...")
|
||
client.reset()
|
||
time.sleep(1)
|
||
finally:
|
||
client.close()
|
||
|
||
self._append_log("Upgrade complete! Device will reboot with new firmware")
|
||
|
||
# Step 6: Wait for device to boot new firmware
|
||
self._append_log("Waiting for device to boot...")
|
||
time.sleep(4)
|
||
|
||
# Step 7: Reconnect
|
||
self._connect()
|
||
|
||
except Exception as e:
|
||
self._append_log(f"Upgrade failed: {e}")
|
||
finally:
|
||
def _restore():
|
||
self._upgrade_btn.configure(state=tk.NORMAL)
|
||
for w in (self._connect_btn, self._port_combo,
|
||
self._baud_combo, self._scan_btn):
|
||
w.configure(state=tk.NORMAL)
|
||
self.root.after(0, _restore)
|
||
|
||
threading.Thread(target=upgrade_thread, daemon=True).start()
|
||
|
||
def _upgrade_progress(self, offset: int, total: int):
|
||
"""SMP upload progress callback (runs in upgrade thread)."""
|
||
pct = offset * 100 // total if total else 0
|
||
if pct // 10 != getattr(self, '_upgrade_last_pct', -1) // 10 or pct >= 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()
|