forked from EmbeddedTeam/app_photomagnetic
1e6d6b9b30
Extend the UART protocol to support writing and reading the heating state, and reading individual pole NTC sensors and the heating pad NTC. Cache NTC readings for query on demand. Move PID updates to a workqueue to avoid blocking the timer IRQ context.
344 lines
12 KiB
Python
344 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for app_photomagnetic communication protocol.
|
|
|
|
Frame format (SimpleProtocal, per DTS config):
|
|
H0(0x7E) H1(0xE7) CMD(1B) LEN(1B) DATA[0..N] CRC_LO CRC_HI
|
|
CRC = CRC-16 Modbus over CMD+LEN+DATA, LSB-MSB order
|
|
|
|
Commands:
|
|
0x00 TEMP D→H (2 bytes: val1 val2)
|
|
0x01 GET_ID H→D / D→H (20 bytes HWID)
|
|
0x02 RUNNING_STATE H→D (1 byte: 0=Standby 1=Running)
|
|
0x03 W_HEATING_STATE H→D (1 byte: 0=stop, >0=target °C)
|
|
0x04 R_HEATING_STATE H→D / D→H (3 bytes: state val1 val2)
|
|
0x64 R_TEMP_POLE_NTC H→D (1 byte index) / D→H (2 bytes: val1 val2)
|
|
0x65 R_HEATING_PAD_NTC H→D (0 bytes) / D→H (2 bytes: val1 val2)
|
|
|
|
Usage:
|
|
python test.py /dev/ttyUSB0 --monitor
|
|
python test.py /dev/ttyUSB0 --heating 42
|
|
python test.py /dev/ttyUSB0 --read-pole-ntc 3
|
|
python test.py /dev/ttyUSB0 --read-heating-ntc
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
import time
|
|
|
|
try:
|
|
import serial
|
|
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)
|
|
|
|
CMD_TEMP = 0x00
|
|
CMD_GET_ID = 0x01
|
|
CMD_RUNNING_STATE = 0x02
|
|
CMD_W_HEATING = 0x03
|
|
CMD_R_HEATING = 0x04
|
|
CMD_READ_POLE_NTC = 0x64
|
|
CMD_READ_HEATING_NTC = 0x65
|
|
|
|
HEATING_STATE_NAMES = {0: "OFF", 1: "ON"}
|
|
|
|
|
|
# ── CRC-16 Modbus ─────────────────────────────────────────────
|
|
def crc16_modbus(data: bytes) -> int:
|
|
crc = 0xFFFF
|
|
for byte in data:
|
|
crc ^= byte
|
|
for _ in range(8):
|
|
if crc & 1:
|
|
crc = (crc >> 1) ^ 0xA001
|
|
else:
|
|
crc >>= 1
|
|
return crc & 0xFFFF
|
|
|
|
|
|
# ── Frame helpers ─────────────────────────────────────────────
|
|
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
|
|
|
|
|
|
# ── Serial reader ─────────────────────────────────────────────
|
|
class ProtocolReader:
|
|
def __init__(self, ser: serial.Serial):
|
|
self._ser = ser
|
|
self._buf = bytearray()
|
|
|
|
def read_frame(self, timeout: float = 1.0) -> bytes | None:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if len(self._buf) >= 2:
|
|
idx = self._buf.find(HEADER)
|
|
if idx > 0:
|
|
del self._buf[:idx]
|
|
elif idx < 0:
|
|
self._buf.clear()
|
|
|
|
if len(self._buf) >= HEADER_SIZE and self._buf[:HEADER_SIZE] == HEADER:
|
|
if len(self._buf) >= HEADER_SIZE + 2 + 2:
|
|
length = self._buf[HEADER_SIZE + 1]
|
|
total = HEADER_SIZE + 2 + length + 2
|
|
if len(self._buf) >= total:
|
|
frame = bytes(self._buf[:total])
|
|
del self._buf[:total]
|
|
return frame
|
|
elif len(self._buf) >= HEADER_SIZE:
|
|
del self._buf[0]
|
|
continue
|
|
|
|
try:
|
|
chunk = self._ser.read(self._ser.in_waiting or 1)
|
|
if chunk:
|
|
self._buf.extend(chunk)
|
|
except (serial.SerialTimeoutException, serial.SerialException):
|
|
pass
|
|
return None
|
|
|
|
|
|
def _read_reply(ser: serial.Serial, timeout: float = 2.0) -> tuple[int, bytes] | None:
|
|
reader = ProtocolReader(ser)
|
|
frame = reader.read_frame(timeout)
|
|
if frame is None:
|
|
print("✗ No response (timeout)")
|
|
return None
|
|
result = parse_frame(frame)
|
|
if result is None:
|
|
print(f"✗ Invalid frame: {frame.hex()}")
|
|
return None
|
|
return result
|
|
|
|
|
|
def _temp_from_data(data: bytes) -> float:
|
|
return data[0] + data[1] / 100.0 if len(data) >= 2 else 0.0
|
|
|
|
|
|
# ── Command handlers ──────────────────────────────────────────
|
|
def cmd_get_id(ser: serial.Serial, timeout: float = 2.0):
|
|
print("→ Sending GET_ID...")
|
|
ser.write(build_frame(CMD_GET_ID))
|
|
r = _read_reply(ser, timeout)
|
|
if r is None:
|
|
return
|
|
cmd, data = r
|
|
if cmd != CMD_GET_ID:
|
|
print(f"✗ Unexpected cmd=0x{cmd:02X}")
|
|
return
|
|
print(f"✓ Device ID ({len(data)} bytes): {data.hex(' ')}")
|
|
|
|
|
|
def cmd_set_state(ser: serial.Serial, state: int):
|
|
names = {0: "Standby", 1: "Running", 2: "Pause", 3: "Error"}
|
|
name = names.get(state, "Unknown")
|
|
print(f"→ Setting state: {name} (0x{state:02X})")
|
|
ser.write(build_frame(CMD_RUNNING_STATE, bytes([state])))
|
|
print("✓ Sent")
|
|
|
|
|
|
def cmd_write_heating(ser: serial.Serial, target: int):
|
|
if target == 0:
|
|
print("→ Stopping heating")
|
|
else:
|
|
print(f"→ Starting heating to {target}°C")
|
|
ser.write(build_frame(CMD_W_HEATING, bytes([target])))
|
|
print("✓ Sent")
|
|
|
|
|
|
def cmd_read_heating(ser: serial.Serial, timeout: float = 2.0):
|
|
print("→ Querying heating state...")
|
|
ser.write(build_frame(CMD_R_HEATING))
|
|
r = _read_reply(ser, timeout)
|
|
if r is None:
|
|
return
|
|
cmd, data = r
|
|
if cmd != CMD_R_HEATING:
|
|
print(f"✗ Unexpected cmd=0x{cmd:02X}")
|
|
return
|
|
if len(data) < 3:
|
|
print(f"✗ Short data: {data.hex()}")
|
|
return
|
|
state = HEATING_STATE_NAMES.get(data[0], f"Unknown({data[0]})")
|
|
print(f"✓ Heating: {state}, temp={_temp_from_data(data[1:3]):.2f}°C")
|
|
|
|
|
|
def cmd_read_pole_ntc(ser: serial.Serial, index: int, timeout: float = 2.0):
|
|
print(f"→ Reading pole NTC #{index}...")
|
|
ser.write(build_frame(CMD_READ_POLE_NTC, bytes([index])))
|
|
r = _read_reply(ser, timeout)
|
|
if r is None:
|
|
return
|
|
cmd, data = r
|
|
if cmd != CMD_READ_POLE_NTC:
|
|
print(f"✗ Unexpected cmd=0x{cmd:02X}")
|
|
return
|
|
if len(data) < 2:
|
|
print(f"✗ Short data: {data.hex()}")
|
|
return
|
|
print(f"✓ Pole NTC #{index}: {_temp_from_data(data):.2f}°C")
|
|
|
|
|
|
def cmd_read_heating_ntc(ser: serial.Serial, timeout: float = 2.0):
|
|
print("→ Reading heating pad NTC...")
|
|
ser.write(build_frame(CMD_READ_HEATING_NTC))
|
|
r = _read_reply(ser, timeout)
|
|
if r is None:
|
|
return
|
|
cmd, data = r
|
|
if cmd != CMD_READ_HEATING_NTC:
|
|
print(f"✗ Unexpected cmd=0x{cmd:02X}")
|
|
return
|
|
if len(data) < 2:
|
|
print(f"✗ Short data: {data.hex()}")
|
|
return
|
|
print(f"✓ Heating pad NTC: {_temp_from_data(data):.2f}°C")
|
|
|
|
|
|
def cmd_monitor(ser: serial.Serial):
|
|
print("Monitoring... (Ctrl+C to stop)")
|
|
reader = ProtocolReader(ser)
|
|
try:
|
|
while True:
|
|
frame = reader.read_frame(timeout=0.5)
|
|
if frame is None:
|
|
continue
|
|
result = parse_frame(frame)
|
|
if result is None:
|
|
print(f"⚠ Bad frame: {frame.hex()}")
|
|
continue
|
|
|
|
cmd, data = result
|
|
if cmd == CMD_TEMP:
|
|
t = _temp_from_data(data)
|
|
print(f"🌡 Temp: {t:.2f}°C (raw: {data.hex(' ')})")
|
|
elif cmd == CMD_GET_ID:
|
|
print(f"🆔 Device ID: {data.hex(' ')}")
|
|
elif cmd == CMD_RUNNING_STATE:
|
|
print(f"🔁 Running state: 0x{data.hex()}")
|
|
elif cmd == CMD_R_HEATING:
|
|
if len(data) >= 3:
|
|
state = HEATING_STATE_NAMES.get(data[0], f"?{data[0]}")
|
|
t = _temp_from_data(data[1:3])
|
|
print(f"🔥 Heating: {state}, temp={t:.2f}°C")
|
|
else:
|
|
print(f"🔥 Heating: {data.hex()}")
|
|
elif cmd == CMD_READ_POLE_NTC:
|
|
t = _temp_from_data(data)
|
|
print(f"📡 Pole NTC reply: {t:.2f}°C")
|
|
elif cmd == CMD_READ_HEATING_NTC:
|
|
t = _temp_from_data(data)
|
|
print(f"🔥 Heating pad NTC reply: {t:.2f}°C")
|
|
else:
|
|
print(f"📦 cmd=0x{cmd:02X} data={data.hex()}")
|
|
except KeyboardInterrupt:
|
|
print("\nDone.")
|
|
|
|
|
|
def cmd_loop(ser: serial.Serial, count: int = 0):
|
|
i = 0
|
|
print("Looping GET_ID... (Ctrl+C to stop)")
|
|
try:
|
|
while count == 0 or i < count:
|
|
cmd_get_id(ser, timeout=1.0)
|
|
i += 1
|
|
time.sleep(0.5)
|
|
except KeyboardInterrupt:
|
|
print("\nDone.")
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Test script for app_photomagnetic comm protocol"
|
|
)
|
|
parser.add_argument("port", help="Serial port")
|
|
parser.add_argument("--baud", type=int, default=115200)
|
|
parser.add_argument("--get-id", action="store_true")
|
|
parser.add_argument(
|
|
"--state", type=lambda x: int(x, 0), metavar="N",
|
|
help="Set running state")
|
|
parser.add_argument(
|
|
"--heating", type=int, metavar="TEMP",
|
|
help="Write heating target (0=stop)")
|
|
parser.add_argument(
|
|
"--read-heating", action="store_true",
|
|
help="Read heating state + temp")
|
|
parser.add_argument(
|
|
"--read-pole-ntc", type=int, metavar="IDX",
|
|
help="Read pole NTC by index (0..7)")
|
|
parser.add_argument(
|
|
"--read-heating-ntc", action="store_true",
|
|
help="Read heating pad NTC temperature")
|
|
parser.add_argument("--monitor", action="store_true")
|
|
parser.add_argument(
|
|
"--loop", type=int, nargs="?", const=0, metavar="N",
|
|
help="Loop GET_ID N times")
|
|
parser.add_argument(
|
|
"--raw", type=lambda x: bytes.fromhex(x), metavar="HEX",
|
|
help="Send raw CMD+DATA")
|
|
|
|
args = parser.parse_args()
|
|
|
|
ser = serial.Serial(args.port, args.baud, timeout=0.1)
|
|
print(f"Connected to {args.port} @ {args.baud} baud")
|
|
|
|
try:
|
|
if args.get_id:
|
|
cmd_get_id(ser)
|
|
elif args.state is not None:
|
|
cmd_set_state(ser, args.state)
|
|
elif args.heating is not None:
|
|
cmd_write_heating(ser, args.heating)
|
|
elif args.read_heating:
|
|
cmd_read_heating(ser)
|
|
elif args.read_pole_ntc is not None:
|
|
cmd_read_pole_ntc(ser, args.read_pole_ntc)
|
|
elif args.read_heating_ntc:
|
|
cmd_read_heating_ntc(ser)
|
|
elif args.monitor:
|
|
cmd_monitor(ser)
|
|
elif args.loop is not None:
|
|
cmd_loop(ser, args.loop)
|
|
elif args.raw is not None:
|
|
payload = args.raw
|
|
c = payload[0]
|
|
d = payload[1:] if len(payload) > 1 else b""
|
|
print(f"→ Sending CMD=0x{c:02X} data={d.hex()}")
|
|
ser.write(build_frame(c, d))
|
|
print("✓ Sent")
|
|
else:
|
|
print("Use --get-id, --state, --heating, --read-heating, "
|
|
"--read-pole-ntc, --read-heating-ntc, --monitor, --loop, --raw")
|
|
finally:
|
|
ser.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|