cuijingwei/rnc_tuning
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[2.0.5]

Browse Source
This commit is contained in:
JingweiCui 2025-11-03 13:53:12 +08:00
parent dcf1116207
commit d08b62ff54
13 changed files with 785 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -2,3 +2,4 @@
dist
build
.idea
main.dist

38
BrisonusRNCTunningTool_V2.0.6.spec Normal file
View File

@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='BrisonusRNCTunningTool_V2.0.6',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=False,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)

4
Brisonus_Params_2.0.5.spec
View File

@ -3,7 +3,7 @@
a = Analysis(
['main.py'],
pathex=['.\\venv\\Lib\\site-packages\\'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
@ -29,7 +29,7 @@ exe = EXE(
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
console=False,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,

183
br_com_serial.py Normal file
View File

@ -0,0 +1,183 @@
from PySide6.QtCore import QObject, Signal, Slot, QThread, QMutex, QWaitCondition
from PySide6.QtSerialPort import QSerialPort, QSerialPortInfo
from br_com_message import BrComMessage
import time
import queue
class SerialReaderThread(QThread):
"""串口数据读取线程"""
def __init__(self, serial_port, buffer_queue):
super().__init__()
self.serial_port = serial_port
self.buffer_queue = buffer_queue
self.running = True
self.buffer = bytearray()
self.max_buffer_size = 50 * 1024 * 1024 # 50MB
self.buffer_mutex = QMutex()
def run(self):
while self.running:
if not self.serial_port.isOpen():
time.sleep(0.1)
continue
if self.serial_port.waitForReadyRead(100): # 等待100ms
data = self.serial_port.readAll()
if data:
self.buffer_mutex.lock()
try:
# 检查缓冲区大小
if len(self.buffer) >= self.max_buffer_size:
print("Warning: Buffer overflow, clearing buffer")
self.buffer.clear()
# 将数据添加到缓冲区
self.buffer.extend(data.data())
# 将完整的数据块放入队列
self.buffer_queue.put(self.buffer)
# 清空缓冲区，准备接收下一块数据
self.buffer = bytearray()
finally:
self.buffer_mutex.unlock()
else:
# 如果没有数据，短暂休眠
time.sleep(0.01)
def stop(self):
self.running = False
class MessageProcessorThread(QThread):
"""消息处理线程"""
def __init__(self, buffer_queue, message_queue):
super().__init__()
self.buffer_queue = buffer_queue
self.message_queue = message_queue
self.running = True
self.accumulated_data = bytearray()
self.max_accumulated_size = 50 * 1024 * 1024 # 50MB
self.data_mutex = QMutex()
def run(self):
while self.running:
try:
# 从缓冲区队列获取数据，设置超时
data = self.buffer_queue.get(timeout=0.1)
if data:
self.data_mutex.lock()
try:
# 检查累积数据大小
if len(self.accumulated_data) >= self.max_accumulated_size:
print("Warning: Accumulated data overflow, clearing data")
self.accumulated_data.clear()
# 将新数据添加到累积数据中
self.accumulated_data.extend(data)
# 处理累积的数据，查找完整的消息
while len(self.accumulated_data) > 0:
# 查找消息头（假设消息头是0xAA）
start_index = self.accumulated_data.find(0xAA)
if start_index == -1:
# 没有找到消息头，清空数据
self.accumulated_data.clear()
break
# 移除消息头之前的数据
self.accumulated_data = self.accumulated_data[start_index:]
# 检查是否有足够的数据来获取消息长度
if len(self.accumulated_data) < 4: # 假设消息头+长度至少4字节
break
# 获取消息长度（假设长度字段在消息头后的2个字节）
message_length = (self.accumulated_data[1] << 8) | self.accumulated_data[2]
# 检查消息长度是否合理
if message_length > self.max_accumulated_size:
print(f"Warning: Invalid message length {message_length}, skipping")
self.accumulated_data = self.accumulated_data[1:] # 跳过当前字节
continue
# 检查是否有完整的消息
if len(self.accumulated_data) < message_length:
break
# 提取完整的消息
message_data = self.accumulated_data[:message_length]
self.accumulated_data = self.accumulated_data[message_length:]
# 将消息放入消息队列
self.message_queue.put(message_data)
finally:
self.data_mutex.unlock()
except queue.Empty:
# 队列超时，继续循环
continue
def stop(self):
self.running = False
class BrComSerial(QObject):
signal_connected = Signal()
signal_disconnected = Signal()
signal_error = Signal(str)
signal_message = Signal(BrComMessage)
def __init__(self, parent=None):
super().__init__(parent)
self.serial_port = QSerialPort()
self.serial_port.errorOccurred.connect(self.on_error)
# 创建数据缓冲队列和消息队列，设置最大大小
self.buffer_queue = queue.Queue(maxsize=1000) # 限制队列大小
self.message_queue = queue.Queue(maxsize=1000) # 限制队列大小
# 创建并启动数据读取线程
self.reader_thread = SerialReaderThread(self.serial_port, self.buffer_queue)
self.reader_thread.start()
# 创建并启动消息处理线程
self.processor_thread = MessageProcessorThread(self.buffer_queue, self.message_queue)
self.processor_thread.start()
# 启动消息处理定时器
self.timer = QTimer()
self.timer.timeout.connect(self.process_messages)
self.timer.start(10) # 每10ms检查一次消息队列
def process_messages(self):
"""处理消息队列中的消息"""
while not self.message_queue.empty():
try:
message_data = self.message_queue.get_nowait()
# 处理消息（这里可以添加具体的消息处理逻辑）
msg = BrComMessage.from_bytes(message_data)
if msg:
self.signal_message.emit(msg)
except queue.Empty:
break
except Exception as e:
print(f"Error processing message: {e}")
def on_error(self, error):
"""处理串口错误"""
if error == QSerialPort.NoError:
return
error_str = f"Serial port error: {error}"
print(error_str)
self.signal_error.emit(error_str)
if self.serial_port.isOpen():
self.serial_port.close()
self.signal_disconnected.emit()
def close(self):
"""关闭串口和线程"""
self.reader_thread.stop()
self.processor_thread.stop()
self.reader_thread.wait()
self.processor_thread.wait()
if self.serial_port.isOpen():
self.serial_port.close()

232
br_widget_arr_details.py
View File

@ -8,10 +8,26 @@ from br_widget_basic_plot import BrWidgetBasicPlot
class BrWidgetArrDetails(QWidget, Ui_Widget):
signal_read: SignalInstance = Signal(list)
signal_write: SignalInstance = Signal(list)
signal_stop: SignalInstance = Signal()
def __init__(self, parent=None):
super().__init__(parent)
self.setupUi(self)
self.param_data: ParamDataArr = None
self.is_reading = False
self.is_writing = False
self.read_count = 0 # 读取计数器
self.total_read_count = 0 # 总读取数量
self.write_count = 0 # 写入计数器
self.total_write_count = 0 # 总写入数量
self.processed_read_addresses = set() # 已处理的读取地址
self.processed_write_addresses = set() # 已处理的写入地址
# 初始化按钮状态
self.pushButton_read.setEnabled(True)
self.pushButton_write.setEnabled(True)
self.pushButton_read_stop.setEnabled(False)
self.pushButton_write_stop.setEnabled(False)
self.updating = False
self.tableWidget_val.setColumnCount(2)
@ -22,6 +38,8 @@ class BrWidgetArrDetails(QWidget, Ui_Widget):
self.tableWidget_setval.itemChanged.connect(self.on_table_setval_itemChanged)
self.pushButton_read_stop.clicked.connect(self.on_pushButton_read_stop_clicked)
self.pushButton_write_stop.clicked.connect(self.on_pushButton_write_stop_clicked)
def on_table_setval_itemChanged(self, item: QTableWidgetItem):
if self.updating:
@ -81,23 +99,84 @@ class BrWidgetArrDetails(QWidget, Ui_Widget):
for i in range(0, len(self.param_data.setval_list)):
self.tableWidget_setval.setItem(0, i, QTableWidgetItem(str(self.param_data.setval_list[i])))
def reset_read_state(self):
"""重置读取状态"""
print('Resetting read state')
self.is_reading = False
self.read_count = 0
self.total_read_count = 0
self.processed_read_addresses.clear()
self.pushButton_read.setEnabled(True)
self.pushButton_write.setEnabled(True)
self.pushButton_read_stop.setEnabled(False)
self.pushButton_write_stop.setEnabled(False)
def reset_write_state(self):
"""重置写入状态"""
print('Resetting write state')
self.is_writing = False
self.write_count = 0
self.total_write_count = 0
self.processed_write_addresses.clear()
self.pushButton_write.setEnabled(True)
self.pushButton_read.setEnabled(True)
self.pushButton_write_stop.setEnabled(False)
self.pushButton_read_stop.setEnabled(False)
@Slot()
def on_pushButton_read_clicked(self):
print('[read] button clicked!')
print(f'Starting read operation for array size: {self.param_data.size}')
# 确保之前的状态被清理
self.reset_read_state()
self.reset_write_state()
# 设置新的读取状态
self.is_reading = True
self.read_count = 0
self.total_read_count = self.param_data.size
# 设置按钮状态
self.pushButton_read.setEnabled(False)
self.pushButton_write.setEnabled(False)
self.pushButton_read_stop.setEnabled(True)
self.pushButton_write_stop.setEnabled(False)
# 生成一组读取命令
_cmd_list = []
print(f'Expected addresses to read: ', end='')
for i in range(0, self.param_data.size):
addr = self.param_data.addr + i
_cmd_list.append(
BrComMessage(
BrComMessage.OP_READ,
self.param_data.addr + i
addr
)
)
print(f'{addr} ', end='')
print() # 换行
print(f'Initialized read operation: count={self.read_count}, total={self.total_read_count}')
self.signal_read.emit(_cmd_list)
@Slot()
def on_pushButton_write_clicked(self):
print('[write] button clicked!')
# 确保之前的状态被清理
self.reset_read_state()
self.reset_write_state()
# 设置新的写入状态
self.is_writing = True
self.write_count = 0
self.total_write_count = self.param_data.size
# 设置按钮状态
self.pushButton_write.setEnabled(False)
self.pushButton_read.setEnabled(False)
self.pushButton_write_stop.setEnabled(True)
self.pushButton_read_stop.setEnabled(False)
print(f'Initialized write operation: count={self.write_count}, total={self.total_write_count}')
_cmd_list = []
for i in range(0, self.param_data.size):
@ -110,6 +189,25 @@ class BrWidgetArrDetails(QWidget, Ui_Widget):
)
self.signal_write.emit(_cmd_list)
@Slot()
def on_pushButton_read_stop_clicked(self):
print('[read stop] button clicked!')
self.reset_read_state()
self.signal_stop.emit()
@Slot()
def on_pushButton_write_stop_clicked(self):
print('[write stop] button clicked!')
self.reset_write_state()
self.signal_stop.emit()
def operation_completed(self):
"""操作完成时调用此方法"""
if self.is_reading:
self.reset_read_state()
if self.is_writing:
self.reset_write_state()
@Slot()
def on_pushButton_valplot_clicked(self):
print('[valpolt] button clicked!')
@ -122,6 +220,138 @@ class BrWidgetArrDetails(QWidget, Ui_Widget):
def on_pushButton_setvalplot_clicked(self):
print('[setvalplot] button clicked!')
def resend_missing_read_commands(self):
"""重新发送未处理地址的读取命令"""
if not self.is_reading:
return
array_start = self.param_data.addr
array_end = self.param_data.addr + self.param_data.size
expected_addresses = set(range(array_start, array_end))
missing_addresses = expected_addresses - self.processed_read_addresses
if missing_addresses:
print(f'Resending read commands for missing addresses: {sorted(missing_addresses)}')
_cmd_list = []
for addr in sorted(missing_addresses):
_cmd_list.append(
BrComMessage(
BrComMessage.OP_READ,
addr
)
)
self.signal_read.emit(_cmd_list)
return True
return False
def check_read_complete(self, addr):
"""检查读取是否完成"""
print(f'Checking completion for addr: {addr}')
if not self.is_reading:
print(f'Not in reading state, ignoring addr: {addr}')
return
# 检查地址是否在当前数组范围内
array_start = self.param_data.addr
array_end = self.param_data.addr + self.param_data.size
print(f'Array range: {array_start} to {array_end-1}, checking addr: {addr}')
if array_start <= addr < array_end and addr not in self.processed_read_addresses:
self.processed_read_addresses.add(addr)
self.read_count += 1
print(f'Valid address {addr}, incrementing counter. Progress: {self.read_count}/{self.total_read_count}')
# 找出未处理的地址
expected_addresses = set(range(array_start, array_end))
missing_addresses = expected_addresses - self.processed_read_addresses
print(f'Missing addresses: {sorted(missing_addresses)}')
# 如果有未处理的地址且已经读取了一定数量（比如80%）的地址，尝试重发
if missing_addresses and self.read_count >= self.total_read_count * 0.8:
print(f'Detected missing addresses at {self.read_count}/{self.total_read_count}, attempting retry')
if self.resend_missing_read_commands():
print('Resent read commands for missing addresses')
return
# 如果已经读取了预期数量的地址
if self.read_count >= self.total_read_count:
if not missing_addresses:
print(f'All data read (count={self.read_count}), completing operation')
self.reset_read_state()
self.signal_stop.emit()
print('Read operation fully completed')
else:
if addr in self.processed_read_addresses:
print(f'Address {addr} already processed')
else:
print(f'Address {addr} outside current array range ({array_start}-{array_end-1})')
def resend_missing_write_commands(self):
"""重新发送未处理地址的写入命令"""
if not self.is_writing:
return
array_start = self.param_data.addr
array_end = self.param_data.addr + self.param_data.size
expected_addresses = set(range(array_start, array_end))
missing_addresses = expected_addresses - self.processed_write_addresses
if missing_addresses:
print(f'Resending write commands for missing addresses: {sorted(missing_addresses)}')
_cmd_list = []
for addr in sorted(missing_addresses):
_cmd_list.append(
BrComMessage(
BrComMessage.OP_WRITE,
addr,
self.param_data.getval_list[addr - array_start]
)
)
self.signal_write.emit(_cmd_list)
return True
return False
def check_write_complete(self, addr):
"""检查写入是否完成"""
print(f'Checking completion for addr: {addr}')
if not self.is_writing:
print(f'Not in writing state, ignoring addr: {addr}')
return
# 检查地址是否在当前数组范围内
array_start = self.param_data.addr
array_end = self.param_data.addr + self.param_data.size
print(f'Array range: {array_start} to {array_end-1}, checking addr: {addr}')
if array_start <= addr < array_end and addr not in self.processed_write_addresses:
self.processed_write_addresses.add(addr)
self.write_count += 1
print(f'Valid address {addr}, incrementing counter. Progress: {self.write_count}/{self.total_write_count}')
# 找出未处理的地址
expected_addresses = set(range(array_start, array_end))
missing_addresses = expected_addresses - self.processed_write_addresses
print(f'Missing addresses: {sorted(missing_addresses)}')
# 如果有未处理的地址且已经写入了一定数量（比如80%）的地址，尝试重发
if missing_addresses and self.write_count >= self.total_write_count * 0.8:
print(f'Detected missing addresses at {self.write_count}/{self.total_write_count}, attempting retry')
if self.resend_missing_write_commands():
print('Resent write commands for missing addresses')
return
# 如果已经写入了预期数量的地址
if self.write_count >= self.total_write_count:
if not missing_addresses:
print(f'All data written (count={self.write_count}), completing operation')
self.reset_write_state()
self.signal_stop.emit()
print('Write operation fully completed')
else:
if addr in self.processed_write_addresses:
print(f'Address {addr} already processed')
else:
print(f'Address {addr} outside current array range ({array_start}-{array_end-1})')
if __name__ == "__main__":
from PySide6.QtWidgets import QApplication

BIN
main.exe Normal file
View File

Binary file not shown.

54
main.py
View File

@ -49,6 +49,20 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.setupUi(self)
# 初始化日志区域折叠状态
self.log_expanded = True
self.textEdit_original_height = self.textEdit.height()
self.pushButton_toggle_log.clicked.connect(self.toggle_log_area)
# 添加自动折叠定时器
self.auto_fold_timer = QTimer()
self.auto_fold_timer.setInterval(5000) # 30秒后自动折叠
self.auto_fold_timer.timeout.connect(self.auto_fold_log)
self.auto_fold_timer.start()
# 连接textEdit的textChanged信号以重置定时器
self.textEdit.textChanged.connect(self.reset_auto_fold_timer)
self.param_data_list = []
self.data_manager = BrDataManager()
@ -56,7 +70,7 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.init_tableview()
# self.tableview_load_data()
self.setWindowTitle('Brisonus RNC Tunning Tool 2.0.4')
self.setWindowTitle('BrisonusRNCTunningTool_V2.0.6')
self.com = QSerialPort()
@ -84,6 +98,7 @@ class MainWindow(QMainWindow, Ui_MainWindow):
# 定义widget_page_arr的slot
self.widget_page_arr.signal_read.connect(self.on_arr_read)
self.widget_page_arr.signal_write.connect(self.on_arr_write)
self.widget_page_arr.signal_stop.connect(self.on_arr_stop)
# var arr 两个页面都处于隐藏状态
self.widget_page_arr.hide()
@ -131,6 +146,15 @@ class MainWindow(QMainWindow, Ui_MainWindow):
self.cmd_server.cmd_queue.put(_cmd_item)
print('arr widget signal/callback on_read has been called!')
def on_arr_stop(self):
"""处理数组操作的停止信号"""
# 清空命令队列
while not self.cmd_server.cmd_queue.empty():
self.cmd_server.cmd_queue.get()
self.statusbar.showMessage('操作已停止')
# 通知widget操作已完成
self.widget_page_arr.operation_completed()
def init_comports(self):
self.lineEdit_baudrate.setText(str(default_config['baud_rate']))
self.lineEdit_pollingInterval.setText(str(default_config['polling_interval']))
@ -285,8 +309,12 @@ Polling interval: 100ms
# 更新var page的显示值
self.statusbar.showMessage('地址[%d]读取完成!' % _msg.addr)
self.update_details_page(self.selected_row)
# 检查数组读取是否完成
self.widget_page_arr.check_read_complete(_msg.addr)
if _msg.operation == BrComMessage.RES_WRITE:
self.statusbar.showMessage('地址[%d]写入完成!' % _msg.addr)
# 检查数组写入是否完成
self.widget_page_arr.check_write_complete(_msg.addr)
current_datetime = datetime.now()
formatted_datetime = current_datetime.strftime("[%Y-%m-%d %H:%M:%S]")
@ -322,6 +350,30 @@ Polling interval: 100ms
else:
event.ignore()
def toggle_log_area(self):
"""切换日志区域的展开/折叠状态"""
if self.log_expanded:
# 折叠
self.textEdit.setMaximumHeight(0)
self.pushButton_toggle_log.setText("︿")
else:
# 展开
self.textEdit.setMaximumHeight(16777215) # 恢复默认最大高度
self.pushButton_toggle_log.setText("﹀")
# 重置自动折叠定时器
self.reset_auto_fold_timer()
self.log_expanded = not self.log_expanded
def auto_fold_log(self):
"""自动折叠日志区域"""
if self.log_expanded:
self.toggle_log_area()
def reset_auto_fold_timer(self):
"""重置自动折叠定时器"""
self.auto_fold_timer.stop()
self.auto_fold_timer.start()
# 按装订区域中的绿色按钮以运行脚本。
if __name__ == '__main__':
app = QApplication()

38
main.spec Normal file
View File

@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='main',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=False,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)

20
rnc_tunning.py
View File

@ -142,6 +142,24 @@ class Ui_MainWindow(object):
self.verticalLayout_2.addWidget(self.widget)
# 添加日志区域的水平布局
self.horizontalLayout_log = QHBoxLayout()
self.horizontalLayout_log.setObjectName(u"horizontalLayout_log")
# 添加日志标题和折叠按钮
self.label_log = QLabel(self.centralwidget)
self.label_log.setObjectName(u"label_log")
self.horizontalLayout_log.addWidget(self.label_log)
self.pushButton_toggle_log = QPushButton(self.centralwidget)
self.pushButton_toggle_log.setObjectName(u"pushButton_toggle_log")
self.pushButton_toggle_log.setMaximumWidth(20)
self.horizontalLayout_log.addWidget(self.pushButton_toggle_log)
self.horizontalLayout_log.addStretch()
self.verticalLayout_2.addLayout(self.horizontalLayout_log)
self.textEdit = QTextEdit(self.centralwidget)
self.textEdit.setObjectName(u"textEdit")
@ -185,5 +203,7 @@ class Ui_MainWindow(object):
self.pushButton_loadfile.setText(QCoreApplication.translate("MainWindow", u"\u52a0\u8f7d\u6587\u4ef6", None))
self.pushButton_update.setText(QCoreApplication.translate("MainWindow", u"\u5237\u65b0", None))
self.pushButton_copy.setText(QCoreApplication.translate("MainWindow", u"\u590d\u5236\u4fe1\u606f", None))
self.label_log.setText(QCoreApplication.translate("MainWindow", u"\u65e5\u5fd7\u4fe1\u606f", None))
self.pushButton_toggle_log.setText(QCoreApplication.translate("MainWindow", u"﹀", None))
# retranslateUi

111
test_br_com_serial.py Normal file
View File

@ -0,0 +1,111 @@
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())

94
test_device_simulator.py Normal file
View File

@ -0,0 +1,94 @@
import sys
import time
import random
from PySide6.QtCore import QCoreApplication, QTimer
from PySide6.QtSerialPort import QSerialPort, QSerialPortInfo
from br_com_message import BrComMessage
class DeviceSimulator:
def __init__(self):
self.app = QCoreApplication(sys.argv)
self.serial_port = QSerialPort()
self.running = True
# 创建发送定时器
self.timer = QTimer()
self.timer.timeout.connect(self.send_data)
self.timer.start(100) # 每100ms发送一次数据
# 测试数据配置
self.test_addresses = [100, 101, 102, 103, 104, 105]
self.test_values = [0, 100, 200, 300, 400, 500]
def generate_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 send_data(self):
"""发送数据"""
if not self.serial_port.isOpen():
return
try:
# 随机选择操作类型和地址
op_type = random.choice([BrComMessage.OP_READ, BrComMessage.OP_WRITE])
addr = random.choice(self.test_addresses)
value = random.choice(self.test_values) if op_type == BrComMessage.OP_WRITE else None
# 生成消息
msg = self.generate_message(op_type, addr, value)
# 模拟数据分片发送
parts = [msg[i:i+2] for i in range(0, len(msg), 2)]
for part in parts:
self.serial_port.write(part)
# 随机延迟，模拟真实设备
time.sleep(random.uniform(0.001, 0.01))
print(f"Sent: op={op_type}, addr={addr}, value={value}")
except Exception as e:
print(f"Error sending data: {e}")
def run(self):
"""运行模拟器"""
# 查找可用串口
available_ports = QSerialPortInfo.availablePorts()
if not available_ports:
print("No available serial ports found!")
return 1
# 使用第一个可用串口
port_info = available_ports[0]
print(f"Using port: {port_info.portName()}")
self.serial_port.setPort(port_info)
self.serial_port.setBaudRate(115200)
if not self.serial_port.open():
print(f"Failed to open port {port_info.portName()}")
return 1
print("Device simulator started")
return self.app.exec()
def close(self):
"""关闭模拟器"""
self.running = False
if self.serial_port.isOpen():
self.serial_port.close()
if __name__ == "__main__":
simulator = DeviceSimulator()
try:
sys.exit(simulator.run())
except KeyboardInterrupt:
print("\nSimulator stopped by user")
simulator.close()
sys.exit(0)

12
ui_arr_details.py
View File

@ -77,6 +77,11 @@ class Ui_Widget(object):
self.verticalLayout_2.addWidget(self.pushButton_read)
self.pushButton_read_stop = QPushButton(Widget)
self.pushButton_read_stop.setObjectName(u"pushButton_read_stop")
self.pushButton_read_stop.setEnabled(False)
self.verticalLayout_2.addWidget(self.pushButton_read_stop)
self.pushButton_valplot = QPushButton(Widget)
self.pushButton_valplot.setObjectName(u"pushButton_valplot")
@ -115,6 +120,11 @@ class Ui_Widget(object):
self.verticalLayout_3.addWidget(self.pushButton_write)
self.pushButton_write_stop = QPushButton(Widget)
self.pushButton_write_stop.setObjectName(u"pushButton_write_stop")
self.pushButton_write_stop.setEnabled(False)
self.verticalLayout_3.addWidget(self.pushButton_write_stop)
self.pushButton_setvalplot = QPushButton(Widget)
self.pushButton_setvalplot.setObjectName(u"pushButton_setvalplot")
@ -163,9 +173,11 @@ class Ui_Widget(object):
self.label_2.setText(QCoreApplication.translate("Widget", u"\u53d8\u91cf\u63cf\u8ff0", None))
self.label_5.setText(QCoreApplication.translate("Widget", u"\u5f53\u524d\u503c", None))
self.pushButton_read.setText(QCoreApplication.translate("Widget", u"Read", None))
self.pushButton_read_stop.setText(QCoreApplication.translate("Widget", u"Stop", None))
self.pushButton_valplot.setText(QCoreApplication.translate("Widget", u"Plot", None))
self.label_6.setText(QCoreApplication.translate("Widget", u"\u8bbe\u5b9a\u503c", None))
self.pushButton_write.setText(QCoreApplication.translate("Widget", u"Write", None))
self.pushButton_write_stop.setText(QCoreApplication.translate("Widget", u"Stop", None))
self.pushButton_setvalplot.setText(QCoreApplication.translate("Widget", u"Plot", None))
self.label_details.setText(QCoreApplication.translate("Widget", u"\u7a7a", None))
# retranslateUi

2
ui_basic_plot.py
View File

@ -42,7 +42,7 @@ class Ui_Dialog(object):
self.verticalLayout.addWidget(self.listView)
self.verticalLayout.setStretch()
self.verticalLayout.setStretch(2, 1)
self.horizontalLayout.addLayout(self.verticalLayout)