397 lines
13 KiB
Python
397 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""JDY-28M Mesh 继电器测试工具。
|
||
|
||
本工具通过电脑串口连接一个处于 Mesh 指令模式的 JDY-28M 模块，提供以下
|
||
测试功能：
|
||
|
||
1. 广播查询网络内设备短地址；
|
||
2. 将收到的设备短地址整理成编号列表；
|
||
3. 按编号选择目标设备；
|
||
4. 发送继电器吸合、断开和状态查询命令；
|
||
5. 自动计算 CRC-16/MODBUS；
|
||
6. 自动解析 JDY-28M 外层接收帧和设备业务应答。
|
||
|
||
业务帧格式：
|
||
|
||
CMD | LEN | DATA | CRC16_H | CRC16_L
|
||
|
||
主机发送给 JDY-28M 的 Mesh 指令格式：
|
||
|
||
41 54 2B AA A1 | DST_H DST_L | A0 | BUSINESS | 0D 0A
|
||
|
||
电脑串口收到的 Mesh 数据格式：
|
||
|
||
F1 DD 09 | SRC_H SRC_L | DST_H DST_L | BUSINESS | 0A
|
||
|
||
依赖：
|
||
|
||
python -m pip install pyserial
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import time
|
||
from dataclasses import dataclass
|
||
|
||
import serial
|
||
from serial.tools import list_ports
|
||
|
||
|
||
# JDY-28M Mesh 主机发送帧的固定部分。
|
||
TX_PREFIX = bytes((0x41, 0x54, 0x2B, 0xAA, 0xA1))
|
||
TX_DATA_FLAG = 0xA0
|
||
|
||
# JDY-28M Mesh 接收帧的固定部分。
|
||
# 根据当前JDY-28M实测，接收帧固定头为F1 DD 0A。
|
||
RX_PREFIXES = (
|
||
bytes((0xF1, 0xDD, 0x08)),
|
||
bytes((0xF1, 0xDD, 0x09)),
|
||
bytes((0xF1, 0xDD, 0x0A)),
|
||
)
|
||
|
||
# Mesh 广播目标地址。
|
||
BROADCAST_ADDRESS = 0xFFFF
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MeshFrame:
|
||
"""一帧已经拆分完成的 JDY-28M Mesh 接收数据。"""
|
||
|
||
source: int
|
||
destination: int
|
||
business: bytes
|
||
|
||
|
||
def crc16_modbus(data: bytes) -> int:
|
||
"""计算 CRC-16/MODBUS，并返回16位整数结果。
|
||
|
||
算法参数与设备固件一致：多项式为0xA001，初始值为0xFFFF，
|
||
输入和输出均按 MODBUS 反射规则处理。返回值内部按整数保存，
|
||
发送时由调用方拆成高字节、低字节。
|
||
"""
|
||
|
||
crc = 0xFFFF
|
||
for value in data:
|
||
crc ^= value
|
||
for _ in range(8):
|
||
if crc & 0x0001:
|
||
crc = (crc >> 1) ^ 0xA001
|
||
else:
|
||
crc >>= 1
|
||
return crc & 0xFFFF
|
||
|
||
|
||
def build_business(command: int, data: bytes = b"") -> bytes:
|
||
"""构造业务帧，并按高字节在前追加CRC16。"""
|
||
|
||
if not 0 <= command <= 0xFF:
|
||
raise ValueError("命令必须是0~255")
|
||
if len(data) > 0xFF:
|
||
raise ValueError("业务数据不能超过255字节")
|
||
|
||
body = bytes((command, len(data))) + data
|
||
crc = crc16_modbus(body)
|
||
return body + bytes((crc >> 8, crc & 0xFF))
|
||
|
||
|
||
def build_mesh_command(destination: int, command: int, data: bytes = b"") -> bytes:
|
||
"""封装一帧发送给JDY-28M的Mesh指令。"""
|
||
|
||
if not 0 <= destination <= 0xFFFF:
|
||
raise ValueError("目标地址必须是0x0000~0xFFFF")
|
||
|
||
business = build_business(command, data)
|
||
return (
|
||
TX_PREFIX
|
||
+ bytes((destination >> 8, destination & 0xFF, TX_DATA_FLAG))
|
||
+ business
|
||
+ bytes((0x0D, 0x0A))
|
||
)
|
||
|
||
|
||
def parse_mesh_frame(buffer: bytearray) -> MeshFrame | None:
|
||
"""从接收缓存中提取一帧完整 Mesh 数据。
|
||
|
||
返回值为：
|
||
|
||
- 成功提取时返回 MeshFrame；
|
||
- 当前缓存还不完整时返回 None，同时保留缓存内容；
|
||
- 发现无效前导数据时丢弃无效字节，继续等待下一帧。
|
||
"""
|
||
|
||
while True:
|
||
matches = [(buffer.find(prefix), prefix) for prefix in RX_PREFIXES]
|
||
matches = [(index, prefix) for index, prefix in matches if index >= 0]
|
||
if not matches:
|
||
# 只保留可能构成半个帧头的末尾字节，防止缓存无限增长。
|
||
# 没有完整帧头时，保留最长帧头长度减1的尾部数据，
|
||
# 以便下一次串口读取时拼接出跨读取边界的帧头。
|
||
keep = min(len(buffer), max(len(prefix) for prefix in RX_PREFIXES) - 1)
|
||
if keep:
|
||
buffer[:] = buffer[-keep:]
|
||
else:
|
||
buffer.clear()
|
||
return None
|
||
|
||
prefix_index, _ = min(matches, key=lambda item: item[0])
|
||
if prefix_index:
|
||
del buffer[:prefix_index]
|
||
|
||
# 固定头3字节 + 源地址2字节 + 目标地址2字节。
|
||
if len(buffer) < 7:
|
||
return None
|
||
|
||
# 业务数据以0A结束。模块接收侧可能只显示0A，也可能保留0D 0A；
|
||
# 这里统一以0A作为接收帧结束字节。
|
||
tail_index = buffer.find(b"\x0A", 7)
|
||
if tail_index < 0:
|
||
return None
|
||
|
||
frame_bytes = bytes(buffer[:tail_index + 1])
|
||
del buffer[:tail_index + 1]
|
||
|
||
source = (frame_bytes[3] << 8) | frame_bytes[4]
|
||
destination = (frame_bytes[5] << 8) | frame_bytes[6]
|
||
business = frame_bytes[7:-1]
|
||
# 部分模块接收侧会显示0D 0A，0D属于外层结束符，不属于业务数据。
|
||
if business.endswith(b"\x0D"):
|
||
business = business[:-1]
|
||
return MeshFrame(source, destination, business)
|
||
|
||
|
||
def validate_business(business: bytes) -> tuple[int, bytes] | None:
|
||
"""校验业务帧长度和CRC，成功时返回命令号及DATA。"""
|
||
|
||
if len(business) < 4:
|
||
return None
|
||
|
||
command = business[0]
|
||
data_length = business[1]
|
||
if len(business) != data_length + 4:
|
||
return None
|
||
|
||
body = business[:-2]
|
||
received_crc = (business[-2] << 8) | business[-1]
|
||
if crc16_modbus(body) != received_crc:
|
||
return None
|
||
|
||
return command, business[2:2 + data_length]
|
||
|
||
|
||
def receive_frames(port: serial.Serial, duration: float) -> list[MeshFrame]:
|
||
"""在指定时间窗口内读取并解析所有完整 Mesh 帧。"""
|
||
|
||
frames: list[MeshFrame] = []
|
||
buffer = bytearray()
|
||
deadline = time.monotonic() + duration
|
||
|
||
while time.monotonic() < deadline:
|
||
waiting = port.in_waiting
|
||
if waiting:
|
||
received_bytes = port.read(waiting)
|
||
# 调试阶段完整打印串口收到的每一段原始数据，
|
||
# 包括无法组成完整Mesh帧或格式校验失败的数据。
|
||
print_hex("串口原始接收", received_bytes)
|
||
buffer.extend(received_bytes)
|
||
while True:
|
||
frame = parse_mesh_frame(buffer)
|
||
if frame is None:
|
||
break
|
||
frames.append(frame)
|
||
else:
|
||
time.sleep(0.005)
|
||
|
||
return frames
|
||
|
||
|
||
def print_hex(label: str, data: bytes) -> None:
|
||
"""以空格分隔的十六进制形式打印串口帧。"""
|
||
|
||
print(f"{label}: {data.hex(' ').upper()}")
|
||
|
||
|
||
def send_command(
|
||
port: serial.Serial,
|
||
destination: int,
|
||
command: int,
|
||
data: bytes = b"",
|
||
) -> None:
|
||
"""构造并发送一条Mesh业务命令。"""
|
||
|
||
frame = build_mesh_command(destination, command, data)
|
||
port.write(frame)
|
||
port.flush()
|
||
print_hex("已发送", frame)
|
||
|
||
|
||
def discover_devices(port: serial.Serial, timeout: float) -> list[int]:
|
||
"""广播查询设备地址，并返回去重后的设备地址列表。"""
|
||
|
||
port.reset_input_buffer()
|
||
frame = build_mesh_command(BROADCAST_ADDRESS, 0x05)
|
||
port.write(frame)
|
||
port.flush()
|
||
print_hex("地址查询广播", frame)
|
||
print(f"等待设备回复 {timeout:.1f} 秒...")
|
||
|
||
addresses: set[int] = set()
|
||
for received in receive_frames(port, timeout):
|
||
result = validate_business(received.business)
|
||
if result is None:
|
||
continue
|
||
command, data = result
|
||
if command == 0x85 and len(data) == 2:
|
||
address = (data[0] << 8) | data[1]
|
||
addresses.add(address)
|
||
print(
|
||
f"收到地址应答：来源={received.source:04X}，"
|
||
f"设备地址={address:04X}"
|
||
)
|
||
|
||
return sorted(addresses)
|
||
|
||
|
||
def select_device(devices: list[int]) -> int | None:
|
||
"""显示设备列表并让用户选择一个设备。"""
|
||
|
||
if not devices:
|
||
print("当前没有已发现的设备，请先执行地址广播查询。")
|
||
return None
|
||
|
||
print("设备列表：")
|
||
for index, address in enumerate(devices, start=1):
|
||
print(f" {index}. {address:04X}")
|
||
|
||
value = input("请输入设备编号：").strip()
|
||
try:
|
||
index = int(value)
|
||
except ValueError:
|
||
print("设备编号无效。")
|
||
return None
|
||
|
||
if not 1 <= index <= len(devices):
|
||
print("设备编号超出范围。")
|
||
return None
|
||
return devices[index - 1]
|
||
|
||
|
||
def wait_for_response(port: serial.Serial, timeout: float, target: int) -> None:
|
||
"""接收并解析指定设备的业务应答。"""
|
||
|
||
for received in receive_frames(port, timeout):
|
||
result = validate_business(received.business)
|
||
if result is None:
|
||
print_hex("收到CRC或长度错误帧", received.business)
|
||
continue
|
||
|
||
command, data = result
|
||
if received.source != target:
|
||
print(f"收到其他设备应答：{received.source:04X}")
|
||
|
||
if command == 0x81 and len(data) == 1:
|
||
state = "吸合" if data[0] else "断开"
|
||
print(f"继电器控制成功：{state}（DATA={data[0]:02X}）")
|
||
elif command == 0x82 and len(data) == 1:
|
||
state = "吸合" if data[0] else "断开"
|
||
print(f"继电器当前状态：{state}")
|
||
elif command == 0xE0 and len(data) == 1:
|
||
print(f"设备返回错误：错误码 0x{data[0]:02X}")
|
||
else:
|
||
print(f"收到应答 CMD=0x{command:02X}, DATA={data.hex(' ').upper()}")
|
||
return
|
||
|
||
print("等待设备应答超时。")
|
||
|
||
|
||
def run_menu(port: serial.Serial, response_timeout: float) -> None:
|
||
"""运行交互式菜单。"""
|
||
|
||
devices: list[int] = []
|
||
|
||
while True:
|
||
print("\n===== STM8 Mesh 继电器测试工具 =====")
|
||
print("1. 广播查询设备地址")
|
||
print("2. 显示设备列表")
|
||
print("3. 选择设备并吸合继电器")
|
||
print("4. 选择设备并断开继电器")
|
||
print("5. 选择设备并查询继电器状态")
|
||
print("0. 退出")
|
||
|
||
choice = input("请选择：").strip()
|
||
if choice == "0":
|
||
return
|
||
if choice == "1":
|
||
devices = discover_devices(port, response_timeout)
|
||
print("发现设备：", ", ".join(f"{value:04X}" for value in devices) or "无")
|
||
continue
|
||
if choice == "2":
|
||
if devices:
|
||
print("、".join(f"{index}. {value:04X}" for index, value in enumerate(devices, 1)))
|
||
else:
|
||
print("设备列表为空。")
|
||
continue
|
||
if choice in {"3", "4", "5"}:
|
||
target = select_device(devices)
|
||
if target is None:
|
||
continue
|
||
|
||
if choice == "3":
|
||
send_command(port, target, 0x01, bytes((0x01,)))
|
||
elif choice == "4":
|
||
send_command(port, target, 0x01, bytes((0x00,)))
|
||
else:
|
||
send_command(port, target, 0x02)
|
||
wait_for_response(port, response_timeout, target)
|
||
continue
|
||
|
||
print("未知选项。")
|
||
|
||
|
||
def select_serial_port() -> str:
|
||
"""扫描电脑可用串口并让用户选择一个串口。
|
||
|
||
同一个串口在不同操作系统上的描述字段可能不同，因此同时显示设备路径、
|
||
设备名称、描述和硬件信息，方便用户识别USB转串口适配器。
|
||
"""
|
||
|
||
ports = sorted(list_ports.comports(), key=lambda item: item.device)
|
||
if not ports:
|
||
raise RuntimeError("没有扫描到可用串口，请检查USB转串口设备和驱动。")
|
||
|
||
print("可用串口：")
|
||
for index, item in enumerate(ports, start=1):
|
||
description = item.description or "无描述"
|
||
manufacturer = item.manufacturer or "未知厂商"
|
||
print(f" {index}. {item.device} | {description} | {manufacturer}")
|
||
|
||
while True:
|
||
value = input("请选择串口编号：").strip()
|
||
try:
|
||
index = int(value)
|
||
except ValueError:
|
||
print("请输入有效的数字编号。")
|
||
continue
|
||
|
||
if 1 <= index <= len(ports):
|
||
return ports[index - 1].device
|
||
print("串口编号超出范围，请重新选择。")
|
||
|
||
|
||
def main() -> None:
|
||
"""解析命令行参数并启动测试菜单。"""
|
||
|
||
parser = argparse.ArgumentParser(description="JDY-28M Mesh继电器测试工具")
|
||
parser.add_argument("--port", help="直接指定电脑串口；不指定时启动后扫描并选择")
|
||
parser.add_argument("--baudrate", type=int, default=9600, help="串口波特率，默认9600")
|
||
parser.add_argument("--timeout", type=float, default=0.2, help="串口读取超时，默认0.2秒")
|
||
parser.add_argument("--response-timeout", type=float, default=10.0, help="设备应答等待时间，默认10秒")
|
||
args = parser.parse_args()
|
||
|
||
selected_port = args.port or select_serial_port()
|
||
with serial.Serial(selected_port, args.baudrate, timeout=args.timeout) as port:
|
||
print(f"已打开串口：{selected_port}，波特率：{args.baudrate}")
|
||
run_menu(port, args.response_timeout)
|
||
if __name__ == "__main__":
|
||
main()
|