forked from EmbeddedTeam/app_photomagnetic
feat: Parameterize heating start with configurable duty cycle
This commit is contained in:
zhangyisong
parent
f425f0acab
commit
e0908bc8f0
@ -32,8 +32,13 @@ public:
|
||||
s_active = true;
|
||||
}
|
||||
|
||||
static auto StartFullEnergy() -> void {
|
||||
pwm_set_pulse_dt(&s_pwm_spec, s_pwm_spec.period);
|
||||
/// Start PWM with raw duty cycle (0.0 ~ 1.0).
|
||||
static auto Start(float duty_cycle) -> void {
|
||||
duty_cycle = std::clamp(duty_cycle, 0.0f, 1.0f);
|
||||
const uint32_t pulse =
|
||||
static_cast<uint32_t>(s_pwm_spec.period * duty_cycle);
|
||||
pwm_set_pulse_dt(&s_pwm_spec, pulse);
|
||||
s_active = true;
|
||||
}
|
||||
|
||||
static auto Stop() -> void {
|
||||
@ -42,7 +47,14 @@ public:
|
||||
pwm_set_pulse_dt(&s_pwm_spec, 0);
|
||||
}
|
||||
|
||||
static auto CurrentTemp() -> sensor_value { return s_current_temp; }
|
||||
static auto CurrentTemp() -> sensor_value {
|
||||
if (sensor_sample_fetch(s_temp_sensor) == 0) {
|
||||
sensor_value temp;
|
||||
sensor_channel_get(s_temp_sensor, SENSOR_CHAN_AMBIENT_TEMP, &temp);
|
||||
s_current_temp = temp;
|
||||
}
|
||||
return s_current_temp;
|
||||
}
|
||||
|
||||
private:
|
||||
static constexpr uint32_t kPeriodNs = PWM_KHZ(5);
|
||||
|
||||
586
scripts/test_gui.py
Normal file
586
scripts/test_gui.py
Normal file
@ -0,0 +1,586 @@
|
||||
#!/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
|
||||
from enum import IntEnum
|
||||
|
||||
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_POLE_NTC = 0x64
|
||||
READ_HEATING_NTC = 0x65
|
||||
|
||||
HEATING_STATE_NAMES = {0: "OFF", 1: "ON"}
|
||||
STATE_NAMES = {0: "Standby", 1: "Running", 2: "Pause", 3: "Error"}
|
||||
|
||||
|
||||
# ── 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 ───────────────────────────────────────────
|
||||
class PhotomagneticGUI:
|
||||
def __init__(self):
|
||||
self.root = tk.Tk()
|
||||
self.root.title("Photomagnetic Communication Tool")
|
||||
self.root.geometry("850x700")
|
||||
self.root.minsize(700, 550)
|
||||
|
||||
self.worker = SerialWorker(self)
|
||||
self._monitor_after_id = None
|
||||
|
||||
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)
|
||||
ttk.Button(conn_frame, text="↻", width=3,
|
||||
command=self._scan_ports).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")
|
||||
baud_combo = ttk.Combobox(conn_frame, textvariable=self._baud_var,
|
||||
values=("9600", "19200", "38400", "57600",
|
||||
"115200", "230400", "460800"),
|
||||
width=10)
|
||||
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)
|
||||
|
||||
# ── 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 + 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.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)
|
||||
|
||||
# ── 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"),
|
||||
("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))
|
||||
|
||||
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_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("")
|
||||
|
||||
# ── Monitor ──────────────────────────────────────────
|
||||
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(" ")
|
||||
|
||||
# ── 0x00 TEMP: 设备主动推送, 只更新 live value, 不输出 log ──
|
||||
if cmd == Cmd.TEMP:
|
||||
t = temp_from_data(data)
|
||||
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)
|
||||
self._append_log(f"🆔 Device ID ({len(data)}B): {raw_hex}")
|
||||
|
||||
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})")
|
||||
|
||||
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 _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()
|
||||
@ -48,7 +48,7 @@ auto main(void) -> int {
|
||||
// auto wdt = app::WatchDogConfig{};
|
||||
StatusLed::Flash(1s);
|
||||
ther::Com::Init();
|
||||
ther::HeatingPad::StartFullEnergy();
|
||||
ther::HeatingPad::Start(1.0f);
|
||||
// ther::HeatingPad::Start(sensor_value{42, 0});
|
||||
sensor_trigger tri{.type = SENSOR_TRIG_DATA_READY,
|
||||
.chan = SENSOR_CHAN_AMBIENT_TEMP};
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user