389 lines
13 KiB
Python
389 lines
13 KiB
Python
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_channels(self, channel_num):
|
||
"""设置通道数
|
||
Args:
|
||
channels (int): 通道数量
|
||
"""
|
||
self.CH_NUM = channel_num
|
||
|
||
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
|
||
# 从队列中获取数据，设置0.1秒超时
|
||
data = self.process_queue.get(timeout=0.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 send_cmd(self, cmd):
|
||
"""发送命令"""
|
||
try:
|
||
self.ser.write(cmd)
|
||
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"./records/signal_record_{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=0.1)
|
||
|
||
|
||
|
||
# 确保串口仍然打开时才处理剩余数据
|
||
if self.ser and self.ser.is_open:
|
||
# 处理剩余的数据
|
||
if len(self.buffer) > 0:
|
||
self.process_queue.put(bytes(list(self.buffer)))
|
||
self.buffer.clear()
|
||
|
||
# 等待处理队列清空
|
||
while not self.process_queue.empty():
|
||
time.sleep(0.1)
|
||
|
||
print('stop !')
|
||
|
||
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()
|