diff --git a/include/heating.hpp b/include/heating.hpp index c98e11d..db3f850 100644 --- a/include/heating.hpp +++ b/include/heating.hpp @@ -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(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); diff --git a/scripts/test_gui.py b/scripts/test_gui.py new file mode 100644 index 0000000..2a0c38c --- /dev/null +++ b/scripts/test_gui.py @@ -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() diff --git a/src/main.cpp b/src/main.cpp index f407ce7..6e7d12a 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -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};