串口关键字抓取

带环形缓冲区

import serial
import time
from datetime import datetime  # 用于获取精确到秒的系统时间class RingBuffer:"""自定义15字符长度的环形缓冲区"""def __init__(self, capacity=8):self.capacity = capacity  # 缓冲区容量固定为15self.buffer = []  # 存储字符的列表self.last_valid_time = None  # 记录上一次有效接收（U-Boot）的时间戳def add_char(self, char):"""添加字符到缓冲区，超出容量则移除最旧字符"""if len(self.buffer) >= self.capacity:self.buffer.pop(0)  # 移除头部最旧字符self.buffer.append(char)  # 新增字符到尾部def get_buffer_str(self):"""返回缓冲区拼接后的字符串"""return ''.join(self.buffer)def reset(self):"""清空缓冲区"""self.buffer = []def serial_monitor_demo(port="COM3", bps=38400, timeout=5,  # 串口单次读取超时（与wait_time保持一致）wait_time=100000,  # 每次接收的最大等待时间target_str="U-Boot",  # 指定接收的目标ASCII字符串
):ser = Nonei = 0  # 成功接收目标字符串计数j = 0  # 超时计数k = 0  # 接收非目标字符串计数total_tests = 0  # 总测试数（i+j+k）# 初始化15字符长度的环形缓冲区ring_buffer = RingBuffer(capacity=15)try:# 1. 初始化并打开串口（仅打开一次，持续保持连接）ser = serial.Serial(port=port, baudrate=bps, timeout=wait_time,parity=serial.PARITY_NONE,stopbits=serial.STOPBITS_ONE,bytesize=serial.EIGHTBITS,xonxoff=False,  # 关闭软件流控rtscts=False    # 关闭硬件流控)if ser.isOpen():open_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"RMC_AC压测_测试复位引脚台阶_程序\n[{open_time}] 串口 {port} 打开成功（波特率：{bps}），持续等待接收目标ASCII字符串：{target_str}（单次超时时间设定为：{wait_time}s）\n")# 2. 持续等待接收数据（串口保持打开，循环接收）while True:received_data = b""  # 存储单次接收的字节数据start_time = time.time()  # 记录本次接收开始时间has_received_data = False  # 标记是否接收到任何数据# 单次接收等待逻辑（超时时间=wait_time）while (time.time() - start_time) < wait_time:if ser.in_waiting > 0:received_data += ser.read(ser.in_waiting)has_received_data = True  # 标记已接收到数据# ASCII解码（忽略非ASCII无效字符）received_str = received_data.decode("ascii", errors="ignore")# 将接收字符逐个加入环形缓冲区for char in received_str:ring_buffer.add_char(char)# 检查缓冲区中是否包含目标字符串buffer_str = ring_buffer.get_buffer_str()if target_str in buffer_str:i += 1total_tests = i + j + kcurrent_time = datetime.now()success_time = current_time.strftime("%Y-%m-%d %H:%M:%S")# 计算与上一次有效接收的时间间隔time_interval = 0if ring_buffer.last_valid_time is not None:time_interval = (current_time - ring_buffer.last_valid_time).total_seconds()# 更新上一次有效接收时间ring_buffer.last_valid_time = current_time# 打印有效接收信息（含时间间隔）print(f"[{success_time}] RMC成功拉起，拉起数：{i} 总测试数：{total_tests} | 与上一次有效接收间隔：{time_interval:.1f}秒")ring_buffer.reset()  # 清空环形缓冲区ser.flushInput()  # 清空串口缓冲区#time.sleep(0.3)  # 隔离尾段字符break  # 退出本次等待，进入下一次接收循环# 3. 判断是否接收到非目标字符串if has_received_data and target_str not in received_data.decode("ascii", errors="ignore"):k += 1total_tests = i + j + knon_target_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"[{non_target_time}] ---非目标字符串---，非目标数：{k} 超时数：{j} 总测试数：{total_tests}")# 4. 判断本次接收是否超时（仅无任何数据时计数超时）elapsed_time = time.time() - start_timeif elapsed_time >= wait_time - 0.01 and not has_received_data:j += 1total_tests = i + j + ktimeout_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"[{timeout_time}] ---超时警告---：本次等待{wait_time}s未接收到任何数据，超时数：{j} 非目标字符串数：{k} 总测试数：{total_tests}")#time.sleep(0.1)  # 降低CPU占用except KeyboardInterrupt:# 手动终止程序（Ctrl+C）stop_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"\n[{stop_time}] ---程序手动终止---")print(f"最终统计：成功拉起数：{i}，非目标字符串数：{k}，超时数：{j}，总测试数：{total_tests}")# 新增：计算各场景占比（便于分析）if total_tests > 0:print(f"成功率：{i/total_tests:.2%}，非目标占比：{k/total_tests:.2%}，超时占比：{j/total_tests:.2%}")except Exception as e:# 串口操作异常exception_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"\n[{exception_time}] ---串口操作异常---：{str(e)}")print(f"异常时统计：成功拉起数：{i}，非目标字符串数：{k}，超时数：{j} 总测试数：{total_tests}")finally:# 程序终止时必关串口（释放资源）if ser and ser.isOpen():ser.close()final_close_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"[{final_close_time}] 串口 {port} 已关闭")# 执行函数
if __name__ == "__main__":serial_monitor_demo()

 

延时等待：

import serial
import time
from datetime import datetime  # 用于获取精确到秒的系统时间def serial_monitor_demo(port="COM3", bps=38400, timeout=5,  # 串口单次读取超时（与wait_time保持一致）wait_time=0.3,  # 每次接收的最大等待时间（1秒）target_str="U-Boot",  # 指定接收的目标ASCII字符串（可修改）#write_data=""  # 初始写入数据（不需要可注释）
):ser = Nonei = 0  # 成功接收目标字符串计数j = 0  # 超时计数k = 0  # 接收非目标字符串计数（新增）total_tests = 0  # 总测试数（i+j+k）try:# 1. 初始化并打开串口（仅打开一次，持续保持连接）ser = serial.Serial(port=port, baudrate=bps, timeout=wait_time,parity=serial.PARITY_NONE,stopbits=serial.STOPBITS_ONE,bytesize=serial.EIGHTBITS,xonxoff=False,  # 关闭软件流控rtscts=False    # 关闭硬件流控)if ser.isOpen():open_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"RMC_AC压测_测试复位引脚台阶_程序\n[{open_time}] 串口 {port} 打开成功（波特率：{bps}），持续等待接收目标ASCII字符串：{target_str}（单次超时时间设定为：{wait_time}s）\n")# 可选：仅发送一次初始数据（不需要可注释）# write_bytes = write_data.encode("gbk")  # 中文用gbk，ASCII用encode("ascii")# ser.write(write_bytes)# send_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")# print(f"[{send_time}] 已发送初始数据：{write_data}\n")# 2. 持续等待接收数据（串口保持打开，循环接收）while True:received_data = b""  # 存储单次接收的字节数据start_time = time.time()  # 记录本次接收开始时间has_received_data = False  # 标记是否接收到任何数据（新增）# 单次接收等待逻辑（超时时间=wait_time）while (time.time() - start_time) < wait_time:if ser.in_waiting > 0:received_data += ser.read(ser.in_waiting)has_received_data = True  # 标记已接收到数据# ASCII解码（忽略非ASCII无效字符）received_str = received_data.decode("ascii", errors="ignore")# 检查目标字符串if target_str in received_str:i += 1total_tests = i + j + ksuccess_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"[{success_time}] RMC成功拉起，拉起数：{i} 非目标字符串数：{k} 超时数：{j} 总测试数：{total_tests}")ser.flushInput()  # 清空缓冲区，准备下一次接收time.sleep(0.3)  # 隔离尾段字符break  # 退出本次等待，进入下一次接收循环# 3. 新增：判断是否接收到非目标字符串if has_received_data and target_str not in received_data.decode("ascii", errors="ignore"):k += 1total_tests = i + j + knon_target_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")# 打印非目标字符串详情（含实际接收内容，便于调试）received_str = received_data.decode("ascii", errors="ignore")#print(f"[{non_target_time}] ---非目标字符串---：接收内容：{received_str.strip()}，非目标数：{k} 超时数：{j} 总测试数：{total_tests}")print(f"[{non_target_time}] ---非目标字符串---，非目标数：{k} 超时数：{j} 总测试数：{total_tests}")# 4. 判断本次接收是否超时（仅无任何数据时计数超时）elapsed_time = time.time() - start_timeif elapsed_time >= wait_time - 0.01 and not has_received_data:  # 新增：仅无数据时算超时j += 1total_tests = i + j + ktimeout_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"[{timeout_time}] ---超时警告---：本次等待{wait_time}s未接收到任何数据，超时数：{j} 非目标字符串数：{k} 总测试数：{total_tests}")time.sleep(0.1)  # 降低CPU占用，不影响接收响应except KeyboardInterrupt:# 手动终止程序（Ctrl+C）stop_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"\n[{stop_time}] ---程序手动终止---")print(f"最终统计：成功拉起数：{i}，非目标字符串数：{k}，超时数：{j}，总测试数：{total_tests}")# 新增：计算各场景占比（便于分析）if total_tests > 0:print(f"成功率：{i/total_tests:.2%}，非目标占比：{k/total_tests:.2%}，超时占比：{j/total_tests:.2%}")except Exception as e:# 串口操作异常exception_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"\n[{exception_time}] ---串口操作异常---：{str(e)}")print(f"异常时统计：成功拉起数：{i}，非目标字符串数：{k}，超时数：{j}，总测试数：{total_tests}")finally:# 程序终止时必关串口（释放资源）if ser and ser.isOpen():ser.close()final_close_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")print(f"[{final_close_time}] 串口 {port} 已关闭")# 执行函数
if __name__ == "__main__":serial_monitor_demo()