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

首次提交

Browse Source 
将PC上的文件存档到远程仓库。
This commit is contained in:
JingweiCui 2025-01-07 09:07:01 +08:00
commit f7a4a23ab8
13 changed files with 1237 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored Normal file
View File

@ -0,0 +1,5 @@
.venv
__pycache__
build
dist
*.dat

215
data_collector copy.py Normal file
View File

@ -0,0 +1,215 @@
import serial
import time
from collections import deque
import threading
from queue import Queue, Empty
class Frame:
START = b'\xAA\xAA\xAA\xAA'
CMD_POS = (4, 8)
LEN_POS = (8, 12)
ID_POS = (12, 16)
# CHANNEL_DATA_POS = (16, 16 + 4*16)
# CHKSUM_POS = (16 + 4*16, 16 + 4*16 + 1)
def __init__(self, ch_num, frame_data):
self.frame_data = frame_data
self.ch_num = ch_num
self.CHANNEL_DATA_POS = (16, 16 + 4 * ch_num)
self.CHKSUM_POS = (16 + 4 * ch_num, 16 + 4 * ch_num + 1)
self.extract_frame(frame_data)
def get_chksum(self, data):
chksum = 0
for byte in data:
chksum += byte
return chksum & 0xFF
def extract_frame(self, frame_data):
"""提取数据帧"""
self.index = frame_data[self.ID_POS[0]:self.ID_POS[1]]
self.cmd = frame_data[self.CMD_POS[0]:self.CMD_POS[1]]
self.checksum = frame_data[self.CHKSUM_POS[0]:self.CHKSUM_POS[1]]
self.datas = []
for i in range(0, self.ch_num):
channel_data_bytes = frame_data[self.CHANNEL_DATA_POS[0]+i*4:self.CHANNEL_DATA_POS[0] + (i + 1) * 4]
self.datas.append(channel_data_bytes)
def verify(self):
"""验证数据帧"""
self.chksum = self.get_chksum(self.frame_data[0:self.CHANNEL_DATA_POS[1]])
if self.checksum[0] != self.chksum:
return False
else:
return True
class SerialDataLogger:
def __init__(self, port='COM50', baudrate=4000000):
# 初始化参数
self.CHUNK_SIZE = 256
self.BUFFER_SIZE = 1024*50 # 约81KB的缓存区
self.buffer = deque(maxlen=self.BUFFER_SIZE)
self.output_file = "serial_data.dat"
self.process_queue = Queue()
self.is_running = True
self.tag = 0xAA
self.data_remainder = b''
# 初始化串口
self.ser = serial.Serial(
port=port,
baudrate=baudrate,
timeout=1
)
# 创建处理线程
self.process_thread = threading.Thread(target=self.process_thread_func)
self.process_thread.daemon = True
self.process_thread.start()
def read_chunk(self):
"""读取一个数据块"""
try:
chunk = self.ser.read(self.CHUNK_SIZE)
return chunk
except Exception as e:
print(f"读取数据出错: {e}")
return None
def process_thread_func(self):
"""数据处理线程的主函数"""
while self.is_running:
try:
# 从队列中获取数据，设置1秒超时
data = self.process_queue.get(timeout=1)
self.process_buffer(data)
except Empty:
continue
# except Exception as e:
# print(f"处理线程发生错误: {e}")
def process_buffer(self, data):
"""处理缓存区数据，识别有效数据帧"""
# 查找数据帧的起始标志（假设数据帧以0xAA 0x55开始）
frame_start = b'\xAA\xAA\xAA\xAA'
valid_frames = []
data = self.data_remainder + data
print(data)
print(len(data))
# self.save_data(data)
pos = 0
while pos < len(data):
# 查找帧起始标志
start_pos = data.find(frame_start, pos)
if start_pos == -1:
break
# 假设每个数据帧的长度是81字节（根据实际协议修改）
FRAME_LENGTH = 81
if start_pos + FRAME_LENGTH <= len(data):
frame = data[start_pos:start_pos + FRAME_LENGTH]
# 验证数据帧的有效性（根据实际协议添加校验）
if Frame(16, frame).verify():
valid_frames.append(frame)
pos = start_pos + FRAME_LENGTH
else:
# 将剩余的数据保留
print(start_pos)
self.data_remainder = data[start_pos:]
print(self.data_remainder)
break
# 将有效数据帧写入文件
self.save_frames(valid_frames)
def save_frames(self, frames):
"""保存有效数据帧到文件"""
try:
with open(self.output_file, 'a+') as f:
# 写入数据段标签
# 获取时间戳
# 获取当前时间戳（秒，带微秒）
timestamp = time.time()
# 转换为微秒时间戳
microsecond_timestamp = int(timestamp * 1_000)
_tag_string = f':{self.tag:02X} {microsecond_timestamp:08X}\n'
f.write(_tag_string)
for frame in frames:
_format_string = str(self.format_frame(Frame(16, frame)))
f.write(_format_string)
f.write('\n') # 添加换行符分隔不同帧
except Exception as e:
print(f"保存数据出错: {e}")
def add_tag(self, tag, data):
"""添加标签"""
# 时间戳 TAG标记
pass
def format_frame(self, frame: Frame):
"""格式化数据帧"""
"""_summary_
起始符号 INDEX 通道数量 通道1数据 通道2数据 通道3数据 ...
"""
_format_string = f':{frame.index.hex().upper()} {frame.ch_num:02X}'
for data in frame.datas:
_format_string += f' {data.hex().upper()}'
return _format_string
def save_data(self, data):
"""保存有效数据帧到文件"""
try:
with open(self.output_file, 'ab') as f:
f.write(data)
except Exception as e:
print(f"保存数据出错: {e}")
def run(self):
"""主运行循环"""
print("开始接收数据...")
try_counter = 0
try:
while True:
chunk = self.read_chunk()
if chunk:
# 将数据块添加到缓存区
for byte in chunk:
self.buffer.append(byte)
# 如果缓存区满，将数据发送到处理队列
if len(self.buffer) >= self.BUFFER_SIZE:
print("缓存区已满，加入处理队列...")
# 将当前缓存区的数据转换为字节并加入处理队列
self.process_queue.put(bytes(list(self.buffer)))
self.buffer.clear()
try_counter = 0
else:
try_counter += 1
if try_counter > 2:
if len(self.buffer) > 0:
self.process_queue.put(bytes(list(self.buffer)))
self.buffer.clear()
try_counter = 0
except KeyboardInterrupt:
print("\n程序已停止")
finally:
self.is_running = False # 停止处理线程
self.process_thread.join() # 等待处理线程结束
self.ser.close()
if __name__ == "__main__":
# 创建实例并运行
logger = SerialDataLogger(port='COM50', baudrate=4000000)
logger.run()

369
data_collector.py Normal file
View File

@ -0,0 +1,369 @@
import serial
import time
from collections import deque
import threading
from queue import Queue, Empty
from PySide6.QtCore import Signal, QObject
class Frame:
START = b'\xAA\xAA\xAA\xAA'
CMD_POS = (4, 8)
LEN_POS = (8, 12)
ID_POS = (12, 16)
# CHANNEL_DATA_POS = (16, 16 + 4*16)
# CHKSUM_POS = (16 + 4*16, 16 + 4*16 + 1)
def __init__(self, ch_num, frame_data):
self.frame_data = frame_data
self.ch_num = ch_num
self.CHANNEL_DATA_POS = (16, 16 + 4 * ch_num)
self.CHKSUM_POS = (16 + 4 * ch_num, 16 + 4 * ch_num + 1)
self.extract_frame(frame_data)
def get_chksum(self, data):
chksum = 0
for byte in data:
chksum += byte
return chksum & 0xFF
def extract_frame(self, frame_data):
"""提取数据帧"""
self.index = frame_data[self.ID_POS[0]:self.ID_POS[1]]
self.cmd = frame_data[self.CMD_POS[0]:self.CMD_POS[1]]
self.checksum = frame_data[self.CHKSUM_POS[0]:self.CHKSUM_POS[1]]
self.datas = []
for i in range(0, self.ch_num):
channel_data_bytes = frame_data[self.CHANNEL_DATA_POS[0]+i*4:self.CHANNEL_DATA_POS[0] + (i + 1) * 4]
self.datas.append(channel_data_bytes)
def verify(self):
"""验证数据帧"""
self.chksum = self.get_chksum(self.frame_data[0:self.CHANNEL_DATA_POS[1]])
if self.checksum[0] != self.chksum:
return False
else:
return True
class SerialDataLogger(QObject):
signal_some_frames_processed = Signal(int)
START_CMD = b'\xAA\xAA\xAA\xAA\x00\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\xAD'
END_CMD = b'\xAA\xAA\xAA\xAA\x01\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\xAE'
CH_NUM = 24
def __init__(self):
super().__init__()
# 初始化参数
self.CHUNK_SIZE = 256
self.BUFFER_SIZE = 1024*50 # 约81KB的缓存区
self.buffer = deque(maxlen=self.BUFFER_SIZE)
self.output_file = "serial_data.dat"
self.process_queue = Queue()
self.is_running = False
self.collect_thread = None
self.port = None
self.baudrate = None
self.ser = None
self.data_remainder = b''
self.RECORDER_TAG = 0
self.recorded_frames_count = 0
# 创建处理线程
self.process_thread = threading.Thread(target=self.process_thread_func)
self.process_thread.daemon = True
self.process_thread.start()
def set_port(self, port, baudrate):
self.port = port
self.baudrate = baudrate
def read_chunk(self):
"""读取一个数据块"""
try:
chunk = self.ser.read(self.CHUNK_SIZE)
return chunk
except Exception as e:
print(f"读取数据出错: {e}")
return None
def process_thread_func(self):
"""数据处理线程的主函数"""
while True:
try:
if not self.is_running and self.process_queue.empty():
continue
# 从队列中获取数据，设置1秒超时
data = self.process_queue.get(timeout=1)
# print('process_thread_func processing')
self.process_buffer(data)
self.signal_some_frames_processed.emit(self.recorded_frames_count)
except Empty:
continue
except Exception as e:
print(f"处理线程发生错误: {e}")
def process_buffer(self, data):
"""处理缓存区数据，识别有效数据帧"""
# 查找数据帧的起始标志（假设数据帧以0xAA 0x55开始）
frame_start = b'\xAA\xAA\xAA\xAA'
valid_frames = []
data = self.data_remainder + data
# self.save_data(data)
pos = 0
while pos < len(data):
# 查找帧起始标志
start_pos = data.find(frame_start, pos)
if start_pos == -1:
break
# 假设每个数据帧的长度是81字节（根据实际协议修改）
FRAME_LENGTH = 17 + self.CH_NUM * 4
if start_pos + FRAME_LENGTH <= len(data):
frame = data[start_pos:start_pos + FRAME_LENGTH]
# 验证数据帧的有效性（根据实际协议添加校验）
if Frame(self.CH_NUM, frame).verify():
valid_frames.append(frame)
pos = start_pos + FRAME_LENGTH
else:
# 将剩余的数据保留
self.data_remainder = data[start_pos:]
# print(self.data_remainder)
break
# 将有效数据帧写入文件
self.save_frames(valid_frames)
def save_frames(self, frames):
"""保存有效数据帧到文件"""
self.recorded_frames_count += len(frames)
try:
with open(self.output_file, 'a+') as f:
# 写入数据段标签
# 获取时间戳
# 获取当前时间戳（秒，带微秒）
timestamp = time.time()
# 转换为微秒时间戳
microsecond_timestamp = int(timestamp * 1_000)
_tag_string = f':{self.RECORDER_TAG:02X} {microsecond_timestamp:08X}\n'
f.write(_tag_string)
for frame in frames:
_format_string = str(self.format_frame(Frame(self.CH_NUM, frame)))
f.write(_format_string)
f.write('\n') # 添加换行符分隔不同帧
except Exception as e:
print(f"保存数据出错: {e}")
def add_tag(self, tag, data):
"""添加标签"""
# 时间戳 TAG标记
pass
def format_frame(self, frame: Frame):
"""格式化数据帧"""
"""_summary_
起始符号 INDEX 通道数量 通道1数据 通道2数据 通道3数据 ...
"""
_format_string = f':{frame.index.hex().upper()} {frame.ch_num:02X}'
for data in frame.datas:
_format_string += f' {data.hex().upper()}'
return _format_string
def save_data(self, data):
"""保存有效数据帧到文件"""
try:
with open(self.output_file, 'ab') as f:
f.write(data)
except Exception as e:
print(f"保存数据出错: {e}")
def test_connection(self):
"""测试串口连接
Returns:
bool: 连接是否成功
"""
try:
if self.ser and self.ser.is_open:
self.ser.write(self.START_CMD)
return True
else:
self.ser = serial.Serial(
port=self.port,
baudrate=self.baudrate,
timeout=1
)
self.ser.write(self.START_CMD)
return True
except Exception as e:
print(f"连接测试失败: {e}")
return False
def disconnect(self):
"""断开串口连接
Returns:
bool: 断开是否成功
"""
try:
if self.ser and self.ser.is_open:
self.ser.close()
return True
except Exception as e:
print(f"断开连接失败: {e}")
return False
def start(self):
"""启动数据采集"""
try:
if not self.ser or not self.ser.is_open:
return False
# 设置运行标志
self.is_running = True
# 创建并启动采集线程
self.collect_thread = threading.Thread(target=self._run)
# self.collect_thread.daemon = True
self.collect_thread.start()
print("recorder start")
self.recorded_frames_count = 0
# 当前时间格式化为本地时间
formatted = time.strftime("%Y_%m_%d_%H_%M_%S", time.localtime())
self.output_file = f"serial_data_{formatted}.dat"
return True
except Exception as e:
print(f"启动失败: {e}")
self.is_running = False
return False
def stop(self):
"""停止数据采集"""
try:
# 先设置停止标志
self.is_running = False
# 等待采集线程结束
if self.collect_thread and self.collect_thread.is_alive():
self.collect_thread.join(timeout=2.0)
# 确保串口仍然打开时才处理剩余数据
if self.ser and self.ser.is_open:
# 处理剩余的数据
if len(self.buffer) > 0:
self.process_queue.put(bytes(list(self.buffer)))
self.buffer.clear()
# 等待处理队列清空
timeout = time.time() + 2.0
while not self.process_queue.empty() and time.time() < timeout:
time.sleep(0.1)
return True
except Exception as e:
print(f"停止失败: {e}")
return False
def _run(self):
"""内部运行方法，由start()调用"""
print("开始接收数据...")
try_counter = 0
try:
while self.is_running and self.ser and self.ser.is_open: # 添加串口检查
chunk = self.read_chunk()
if chunk:
# 将数据块添加到缓存区
for byte in chunk:
self.buffer.append(byte)
# 如果缓存区满，将数据发送到处理队列
if len(self.buffer) >= self.BUFFER_SIZE:
print("缓存区已满，加入处理队列...")
self.process_queue.put(bytes(list(self.buffer)))
self.buffer.clear()
try_counter = 0
else:
try_counter += 1
if try_counter > 2:
if len(self.buffer) > 0:
self.process_queue.put(bytes(list(self.buffer)))
self.buffer.clear()
try_counter = 0
except Exception as e:
print(f"数据采集错误: {e}")
finally:
# 确保关闭串口
if self.ser and self.ser.is_open:
try:
self.ser.close()
except:
pass
print("数据采集结束")
def cleanup(self):
"""完全清理所有资源"""
try:
# 停止运行
self.is_running = False
# 等待采集线程结束
if hasattr(self, 'collect_thread') and self.collect_thread:
try:
self.collect_thread.join(timeout=2.0)
except:
pass
# 处理剩余数据
if hasattr(self, 'buffer') and len(self.buffer) > 0:
try:
self.process_queue.put(bytes(list(self.buffer)))
self.buffer.clear()
except:
pass
# 等待处理队列清空
if hasattr(self, 'process_queue'):
timeout = time.time() + 2.0
while not self.process_queue.empty() and time.time() < timeout:
try:
self.process_queue.get_nowait()
except:
break
# 关闭串口
if hasattr(self, 'ser') and self.ser:
try:
if self.ser.is_open:
self.ser.close()
except:
pass
# 等待处理线程结束
if hasattr(self, 'process_thread') and self.process_thread:
try:
self.process_thread.join(timeout=2.0)
except:
pass
except Exception as e:
print(f"清理资源时出错: {e}")
if __name__ == "__main__":
# 创建实例并运行
logger = SerialDataLogger(port='COM50', baudrate=4000000)
logger.start()

38
data_collector_16ch_v1.0.4.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='data_collector_16ch_v1.0.4',
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,
)

38
data_collector_24ch_v1.0.3.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='data_collector_24ch_v1.0.3',
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,
)

38
data_collector_24ch_v1.0.4.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='data_collector_24ch_v1.0.4',
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,
)

38
data_collector_v1.0.0.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='data_collector_v1.0.0',
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,
)

38
data_collector_v1.0.1.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='data_collector_v1.0.1',
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,
)

38
data_collector_v1.0.2.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='data_collector_v1.0.2',
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,
)

38
data_collector_v1.0.3.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='data_collector_v1.0.3',
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,
)

374
main.py Normal file
View File

@ -0,0 +1,374 @@
import sys
import serial.tools.list_ports
from PySide6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QComboBox, QPushButton, QLabel,
QLineEdit, QMessageBox)
from PySide6.QtCore import Qt
from PySide6.QtGui import QIntValidator
from data_collector import SerialDataLogger
class SerialUI(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("Data Colllector 24CH V1.0.4")
self.resize(400, 200)
# 设置应用样式
self.setStyleSheet("""
QMainWindow {
background-color: #2B2B2B;
}
QWidget {
background-color: #2B2B2B;
color: #FFFFFF;
}
QLabel {
color: #CCCCCC;
font-size: 14px;
}
QComboBox {
background-color: #3C3F41;
border: 1px solid #646464;
border-radius: 3px;
color: #FFFFFF;
padding: 5px;
min-width: 100px;
}
QComboBox:hover {
border: 1px solid #FF8C00;
}
QComboBox::drop-down {
border: none;
padding-right: 8px;
}
QPushButton {
background-color: #3C3F41;
border: 1px solid #646464;
border-radius: 3px;
color: #FFFFFF;
padding: 5px 15px;
min-height: 25px;
}
QPushButton:hover {
background-color: #434749;
border: 1px solid #FF8C00;
}
QPushButton:pressed {
background-color: #333637;
}
QPushButton:checked {
background-color: #FF8C00;
border: 1px solid #FF8C00;
}
QLineEdit {
background-color: #3C3F41;
border: 1px solid #646464;
border-radius: 3px;
color: #FFFFFF;
padding: 5px;
min-width: 100px;
}
QLineEdit:focus {
border: 1px solid #FF8C00;
}
QLineEdit:disabled {
background-color: #2B2B2B;
color: #808080;
}
QStatusBar {
background-color: #3C3F41;
color: #CCCCCC;
}
#record_btn {
background-color: #FF8C00;
border: none;
font-weight: bold;
}
#record_btn:hover {
background-color: #FFA500;
}
#record_btn:checked {
background-color: #CD3333;
}
""")
# 创建中心部件和布局
central_widget = QWidget()
self.setCentralWidget(central_widget)
layout = QVBoxLayout(central_widget)
# 串口选择区域
port_layout = QHBoxLayout()
port_label = QLabel("串口:")
self.port_combo = QComboBox()
self.refresh_btn = QPushButton("刷新")
self.refresh_btn.clicked.connect(self.refresh_ports)
port_layout.addWidget(port_label)
port_layout.addWidget(self.port_combo)
port_layout.addWidget(self.refresh_btn)
layout.addLayout(port_layout)
# 波特率选择区域
baud_layout = QHBoxLayout()
baud_label = QLabel("波特率:")
self.baud_input = QLineEdit()
self.baud_input.setText("4000000")
self.baud_input.setValidator(QIntValidator(1200, 4000000))
self.baud_input.setPlaceholderText("请输入波特率")
self.connect_btn = QPushButton("连接端口")
self.connect_btn.setCheckable(True)
self.connect_btn.clicked.connect(self.toggle_connection)
baud_layout.addWidget(baud_label)
baud_layout.addWidget(self.baud_input)
baud_layout.addWidget(self.connect_btn)
layout.addLayout(baud_layout)
# 录制控制区域
record_layout = QHBoxLayout()
self.record_btn = QPushButton("开始录制")
self.record_btn.setCheckable(True)
self.record_btn.clicked.connect(self.toggle_recording)
record_layout.addWidget(self.record_btn)
layout.addLayout(record_layout)
# 标签输入区域
tag_layout = QHBoxLayout()
self.tag_input = QLineEdit()
self.tag_input.setText("0") # 设置默认值为0
self.tag_input.setPlaceholderText("输入标签内容")
self.add_label_btn = QPushButton("增加标签")
self.add_label_btn.clicked.connect(self.add_tag)
tag_layout.addWidget(self.tag_input)
tag_layout.addWidget(self.add_label_btn)
layout.addLayout(tag_layout)
# 数据计数显示区域
count_layout = QHBoxLayout()
self.count_label = QLabel("当前记录数量：0")
self.count_label.setAlignment(Qt.AlignCenter)
count_layout.addWidget(self.count_label)
layout.addLayout(count_layout)
# 添加状态栏
self.statusBar().showMessage("就绪")
# 初始化串口列表
self.refresh_ports()
# 初始化录制状态
self.is_recording = False
self.serial_logger = None
# 设置对象名称以便样式表识别
self.record_btn.setObjectName("record_btn")
# 增加组件间距
layout.setSpacing(10)
layout.setContentsMargins(20, 20, 20, 20)
# 为所有水平布局添加间距
for layout_item in [port_layout, baud_layout, record_layout, tag_layout]:
layout_item.setSpacing(10)
# 设置窗口最小尺寸
self.setMinimumSize(450, 250)
# 初始化时禁用录制按钮
self.record_btn.setEnabled(False)
self.serial_logger = SerialDataLogger()
self.serial_logger.signal_some_frames_processed.connect(self.update_count_display)
# 添加退出时的确认标志
self.exit_confirmed = False
def update_count_display(self):
"""更新数据计数显示"""
if self.serial_logger:
count = self.serial_logger.recorded_frames_count
self.count_label.setText(f"当前记录数量：{count}")
def get_recorder_tag(self):
return self.label_input.text()
def refresh_ports(self):
"""刷新可用串口列表"""
self.port_combo.clear()
ports = serial.tools.list_ports.comports()
if ports:
for port in ports:
# 格式化串口信息：端口 - 设备名称 (设备ID)
port_info = f"{port.device} - {port.description}"
if port.serial_number:
port_info += f" ({port.serial_number})"
# 将完整信息显示在下拉框中，但在数据中保存实际的端口名
self.port_combo.addItem(port_info, port.device)
else:
self.statusBar().showMessage("未找到可用串口")
def toggle_connection(self):
"""切换端口连接状态"""
print("toggle_connection")
if self.exit_confirmed and self.connect_btn.isChecked():
self.connect_btn.setChecked(False)
return
if self.connect_btn.isChecked():
port = self.port_combo.currentData()
try:
baudrate = int(self.baud_input.text())
self.serial_logger.set_port(port, baudrate)
if self.serial_logger.test_connection():
self.connect_btn.setText("断开连接")
self.statusBar().showMessage("端口已连接")
# 禁用串口和波特率选择
self.port_combo.setEnabled(False)
self.baud_input.setEnabled(False)
self.refresh_btn.setEnabled(False)
# 启用录制按钮
self.record_btn.setEnabled(True)
self.is_connected = True
else:
raise Exception("端口连接失败")
except ValueError:
QMessageBox.critical(self, "错误", "请输入有效的波特率")
self.connect_btn.setChecked(False)
except Exception as e:
QMessageBox.critical(self, "错误", f"无法连接端口: {str(e)}")
self.connect_btn.setChecked(False)
else:
if self.serial_logger:
if self.serial_logger.disconnect():
self.connect_btn.setText("连接端口")
self.statusBar().showMessage("端口已断开")
# 启用串口和波特率选择
self.port_combo.setEnabled(True)
self.baud_input.setEnabled(True)
self.refresh_btn.setEnabled(True)
# 禁用录制按钮
self.record_btn.setEnabled(False)
self.is_connected = False
else:
QMessageBox.warning(self, "警告", "断开连接时发生错误")
def toggle_recording(self):
"""切换录制状态"""
if not self.is_connected:
QMessageBox.warning(self, "警告", "请先连接端口")
self.record_btn.setChecked(False)
return
# 如果正在退出，不允许开始新的录制
if self.exit_confirmed:
self.record_btn.setChecked(False)
return
if self.record_btn.isChecked():
print('start recording')
self.serial_logger.test_connection()
if self.serial_logger.start():
self.record_btn.setText("停止录制")
self.statusBar().showMessage("正在录制...")
# 禁用连接按钮
self.connect_btn.setEnabled(False)
else:
QMessageBox.critical(self, "错误", "启动数据记录失败")
self.record_btn.setChecked(False)
else:
if self.serial_logger.stop():
self.record_btn.setText("开始录制")
self.statusBar().showMessage("录制已停止")
self.record_btn.setEnabled(True)
# self.record_btn.setChecked(True)
# 启用连接按钮
self.connect_btn.setEnabled(True)
print(self.serial_logger)
else:
QMessageBox.warning(self, "警告", "停止录制时发生错误")
def add_tag(self):
"""添加标签"""
label_text = self.label_input.text().strip()
if not label_text:
return
if self.serial_logger and self.is_recording:
# 这里可以添加向文件写入标签的逻辑
# 例如：self.serial_logger.add_label(label_text)
self.serial_logger.RECORDER_TAG = int(self.tag_input.text())
self.statusBar().showMessage(f"已添加标签: {label_text}")
self.label_input.clear()
else:
QMessageBox.warning(self, "警告", "请先开始录制")
def closeEvent(self, event):
"""窗口关闭事件处理"""
if not self.exit_confirmed:
reply = QMessageBox.question(
self,
'确认退出',
'确定要退出程序吗？',
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No
)
if reply == QMessageBox.Yes:
# 停止更新计时器
if hasattr(self, 'update_timer'):
self.update_timer.stop()
# 如果正在录制，先停止录制
if self.record_btn.isChecked():
self.record_btn.setChecked(False)
self.toggle_recording()
# 如果已连接，断开连接
if self.connect_btn.isChecked():
self.connect_btn.setChecked(False)
self.toggle_connection()
self.exit_confirmed = True
event.accept()
else:
event.ignore()
else:
event.accept()
def cleanup_resources(self):
"""清理所有资源"""
try:
# 清理串口记录器
if self.serial_logger:
try:
# 直接设置停止标志
self.serial_logger.is_running = False
# 关闭串口连接
if hasattr(self.serial_logger, 'ser') and self.serial_logger.ser:
self.serial_logger.ser.close()
# 快速清理处理队列
if hasattr(self.serial_logger, 'process_queue'):
while not self.serial_logger.process_queue.empty():
try:
self.serial_logger.process_queue.get_nowait()
except:
break
self.serial_logger = None
except Exception as e:
print(f"清理串口记录器时出错: {e}")
except Exception as e:
print(f"清理资源时出错: {e}")
if __name__ == '__main__':
app = QApplication(sys.argv)
window = SerialUI()
window.show()
sys.exit(app.exec())

BIN
requirements Normal file
View File

Binary file not shown.

8
serial_logger.py Normal file
View File

@ -0,0 +1,8 @@
def add_label(self, label_text):
"""添加标签到记录文件"""
try:
with open(self.output_file, 'a') as f:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
f.write(f"\n=== LABEL [{timestamp}]: {label_text} ===\n")
except Exception as e:
print(f"添加标签时出错: {e}")