111 lines
4.4 KiB
Python
111 lines
4.4 KiB
Python
import sys
|
||
import time
|
||
from PySide6.QtCore import QCoreApplication, QTimer
|
||
from PySide6.QtSerialPort import QSerialPort, QSerialPortInfo
|
||
from br_com_serial import BrComSerial
|
||
from br_com_message import BrComMessage
|
||
|
||
class TestBrComSerial:
|
||
def __init__(self):
|
||
self.app = QCoreApplication(sys.argv)
|
||
self.com_serial = BrComSerial()
|
||
self.test_data = bytearray()
|
||
self.received_messages = []
|
||
self.test_complete = False
|
||
|
||
# 连接信号
|
||
self.com_serial.signal_message.connect(self.on_message_received)
|
||
self.com_serial.signal_error.connect(self.on_error)
|
||
|
||
# 创建测试定时器
|
||
self.timer = QTimer()
|
||
self.timer.timeout.connect(self.run_test)
|
||
self.timer.start(100) # 每100ms执行一次测试
|
||
|
||
def generate_test_message(self, op_type, addr, value=None):
|
||
"""生成测试消息"""
|
||
msg = bytearray([0xAA]) # 消息头
|
||
msg.extend([0x00, 0x08]) # 长度字段（8字节）
|
||
msg.append(op_type) # 操作类型
|
||
msg.extend(addr.to_bytes(4, 'big')) # 地址（4字节）
|
||
if value is not None:
|
||
msg.extend(value.to_bytes(4, 'big')) # 值（4字节）
|
||
return msg
|
||
|
||
def simulate_serial_data(self):
|
||
"""模拟串口数据"""
|
||
# 生成一些测试消息
|
||
messages = [
|
||
self.generate_test_message(BrComMessage.OP_READ, 100),
|
||
self.generate_test_message(BrComMessage.OP_READ, 101),
|
||
self.generate_test_message(BrComMessage.OP_WRITE, 102, 123),
|
||
self.generate_test_message(BrComMessage.OP_READ, 103),
|
||
]
|
||
|
||
# 模拟数据分片发送
|
||
for msg in messages:
|
||
# 将消息分成多个部分发送
|
||
parts = [msg[i:i+2] for i in range(0, len(msg), 2)]
|
||
for part in parts:
|
||
self.com_serial.serial_port.write(part)
|
||
time.sleep(0.01) # 模拟传输延迟
|
||
|
||
def on_message_received(self, msg):
|
||
"""处理接收到的消息"""
|
||
print(f"Received message: op={msg.op}, addr={msg.addr}, value={msg.value}")
|
||
self.received_messages.append(msg)
|
||
|
||
def on_error(self, error_str):
|
||
"""处理错误"""
|
||
print(f"Error: {error_str}")
|
||
|
||
def run_test(self):
|
||
"""运行测试"""
|
||
if not self.test_complete:
|
||
# 模拟串口数据
|
||
self.simulate_serial_data()
|
||
|
||
# 检查接收到的消息
|
||
if len(self.received_messages) >= 4: # 期望收到4条消息
|
||
print("\nTest Results:")
|
||
print(f"Total messages received: {len(self.received_messages)}")
|
||
|
||
# 验证消息内容
|
||
for i, msg in enumerate(self.received_messages):
|
||
print(f"Message {i+1}: op={msg.op}, addr={msg.addr}, value={msg.value}")
|
||
|
||
# 验证消息顺序和内容
|
||
expected_messages = [
|
||
(BrComMessage.OP_READ, 100),
|
||
(BrComMessage.OP_READ, 101),
|
||
(BrComMessage.OP_WRITE, 102),
|
||
(BrComMessage.OP_READ, 103)
|
||
]
|
||
|
||
for i, (expected_op, expected_addr) in enumerate(expected_messages):
|
||
if i < len(self.received_messages):
|
||
msg = self.received_messages[i]
|
||
if msg.op == expected_op and msg.addr == expected_addr:
|
||
print(f"Message {i+1} matches expected values")
|
||
else:
|
||
print(f"Message {i+1} does not match expected values")
|
||
print(f"Expected: op={expected_op}, addr={expected_addr}")
|
||
print(f"Received: op={msg.op}, addr={msg.addr}")
|
||
|
||
self.test_complete = True
|
||
self.com_serial.close()
|
||
self.app.quit()
|
||
|
||
def run(self):
|
||
"""运行测试程序"""
|
||
# 创建虚拟串口
|
||
port_info = QSerialPortInfo.availablePorts()[0] # 使用第一个可用串口
|
||
self.com_serial.serial_port.setPort(port_info)
|
||
self.com_serial.serial_port.setBaudRate(115200)
|
||
self.com_serial.serial_port.open()
|
||
|
||
return self.app.exec()
|
||
|
||
if __name__ == "__main__":
|
||
test = TestBrComSerial()
|
||
sys.exit(test.run()) |