104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
"""
|
|
设备A示例 - 发送端
|
|
模拟第一台设备，定期发送站点数据（用户名、线路名称、站点编号）
|
|
"""
|
|
|
|
from serial_protocol import SerialProtocol, Command
|
|
from data_models import StationData, ResponseData
|
|
import time
|
|
|
|
|
|
def on_receive(cmd: int, data: bytes):
|
|
"""接收数据回调函数"""
|
|
print("\n[设备A] 收到数据:")
|
|
cmd_name = (Command(cmd).name if cmd in Command._value2member_map_
|
|
else 'UNKNOWN')
|
|
print(f" 命令: 0x{cmd:02X} ({cmd_name})")
|
|
|
|
if cmd == Command.ACK:
|
|
print(" >>> 对方已确认接收")
|
|
elif cmd == Command.DATA_RESPONSE:
|
|
# 优先尝试解析字典响应
|
|
response_dict = ResponseData.from_bytes(data)
|
|
if response_dict and isinstance(response_dict, dict):
|
|
print("\n 📥 收到设备B的响应字典:")
|
|
for key, value in response_dict.items():
|
|
print(f" {key}: \"{value}\"")
|
|
else:
|
|
# 尝试解析站点数据
|
|
station_data = StationData.from_bytes(data)
|
|
if station_data:
|
|
print(f" >>> 对方发来站点数据: {station_data}")
|
|
else:
|
|
print(f" >>> 对方发来数据: "
|
|
f"{data.decode('utf-8', errors='ignore')}")
|
|
|
|
|
|
def main():
|
|
# 创建串口协议实例
|
|
# Windows: 'COM1', 'COM2', etc.
|
|
# Linux: '/dev/ttyUSB0', '/dev/ttyS0', etc.
|
|
device = SerialProtocol(port='COM1', baudrate=115200)
|
|
|
|
print("=" * 60)
|
|
print("设备A - 发送端")
|
|
print("=" * 60)
|
|
|
|
# 打开串口
|
|
if not device.open():
|
|
print("❌ 无法打开串口")
|
|
return
|
|
|
|
print(f"✓ 串口已打开: {device.port} @ {device.baudrate}")
|
|
|
|
# 启动接收线程
|
|
device.start_receive(on_receive)
|
|
print("✓ 接收线程已启动")
|
|
|
|
try:
|
|
# 模拟线路数据
|
|
lines = ["1号线", "2号线", "3号线", "环线"]
|
|
users = ["张三", "李四", "王五", "赵六"]
|
|
|
|
# 模拟设备运行
|
|
counter = 0
|
|
while True:
|
|
counter += 1
|
|
print(f"\n{'='*60}")
|
|
print(f"循环 {counter}")
|
|
print('='*60)
|
|
|
|
# 1. 发送心跳包
|
|
if device.send_heartbeat():
|
|
print("✓ 发送心跳包")
|
|
else:
|
|
print("✗ 发送心跳包失败")
|
|
|
|
time.sleep(2)
|
|
|
|
# 2. 发送站点数据
|
|
station_data = StationData(
|
|
username=users[counter % len(users)],
|
|
line_name=lines[counter % len(lines)],
|
|
station_no=counter
|
|
)
|
|
print("\n准备发送站点数据:")
|
|
print(f" {station_data}")
|
|
|
|
if device.send_data(station_data.to_bytes()):
|
|
print("✓ 站点数据发送成功")
|
|
else:
|
|
print("✗ 站点数据发送失败")
|
|
|
|
time.sleep(5)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n用户中断")
|
|
finally:
|
|
device.close()
|
|
print("串口已关闭")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|