串口关键字抓取
带环形缓冲区
import serial
import time
from datetime import datetime # 用于获取精确到秒的系统时间
class RingBuffer:
"""自定义15字符长度的环形缓冲区"""
def __init__(self, capacity=8):
self.capacity = capacity # 缓冲区容量固定为15
self.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 = None
i = 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 += 1
total_tests = i + j + k
current_time = datetime.now()
success_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
# 计算与上一次有效接收的时间间隔
time_interval = 0
if 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 += 1
total_tests = i + j + k
non_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_time
if elapsed_time >= wait_time - 0.01 and not has_received_data:
j += 1
total_tests = i + j + k
timeout_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 = None
i = 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 += 1
total_tests = i + j + k
success_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 += 1
total_tests = i + j + k
non_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_time
if elapsed_time >= wait_time - 0.01 and not has_received_data: # 新增：仅无数据时算超时
j += 1
total_tests = i + j + k
timeout_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()
浙公网安备 33010602011771号