【v5Lite】识别+串口

前言

一、版本一

''''''
#Author ：susocool
#Creattime:2024/10/16
#FileName:4-线程带异常处理
#Description: 1、帧数显示、加入异常处理2、线程处理-推理部分放在另一个线程中
'''
import sys
sys.path.append('/home/pi/Desktop/Lite')
import torch
import cv2
import numpy as np
from models.experimental import attempt_load
from utils.general import non_max_suppression, scale_coords
from utils.plots import plot_one_box
from queue import Queue, Empty
import threading
import time
import sys 
print(sys)# 加载模型权重，指定CPU
try:model = attempt_load('/home/pi/Desktop/Lite/v5lite-s.pt',map_location=torch.device('cpu'))model.eval()
except Exception as e:print(f"加载模型失败: {e}")exit()# 初始化摄像头
cap = cv2.VideoCapture(0)
if not cap.isOpened():print("无法打开摄像头")exit()# 创建一个队列来存储从摄像头捕获的帧
frame_queue = Queue()
# 创建一个队列来存储推理结果
result_queue = Queue()# 定义推理线程函数
def inference_thread(model, frame_queue, result_queue):while True:try:# 从队列中获取帧frame = frame_queue.get(timeout=1)  # 设置超时以避免阻塞if frame is None:  # 如果收到None，则退出线程break# 将图像转换为 YOLOv5 模型所需的格式global imgimg = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)img = torch.from_numpy(img).permute(2, 0, 1).float().div(255.0).unsqueeze(0)# 进行推理with torch.no_grad():pred = model(img)[0]# 非极大值抑制pred = non_max_suppression(pred, conf_thres=0.5, iou_thres=0.5)# 将结果放入结果队列result_queue.put(pred)except Empty:# 如果队列为空，则继续下一次循环（通常不会发生，因为设置了超时）continueexcept Exception as e:print(f"推理线程出现异常: {e}")# 启动推理线程
inference_thread_obj = threading.Thread(target=inference_thread, args=(model, frame_queue, result_queue))
inference_thread_obj.start()# 用于计算帧率的变量
frame_count = 0
start_time = time.time()try:while True:# 设置摄像头的分辨率,防止其因为分辨率不能正常显示画面cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)# 捕获摄像头的帧ret, frame = cap.read()if not ret:print("无法获取摄像头的的数据")break# 将帧放入队列以供推理线程处理try:frame_queue.put(frame)except Exception as e:print(f"将帧放入队列时出现异常: {e}")# 尝试从结果队列中获取推理结果try:pred = result_queue.get_nowait()  # 使用nowait以避免阻塞，如果没有结果则跳过绘制# 处理预测结果for det in pred:if det is not None and len(det):det[:, :4] = scale_coords(img.shape[2:], det[:, :4], frame.shape).round()for *xyxy, conf, cls in det:label = f'{model.names[int(cls)]} {conf:.2f}'label_onlyName = f'{model.names[int(cls)]}'plot_one_box(xyxy, frame, label=label, color=(0, 255, 0), line_thickness=3)print(label_onlyName)except Empty:# print("Empty为空")# 如果没有结果，则继续下一次循环passexcept Exception as e:print(f"处理推理结果时出现异常: {e}")# 显示结果frame = cv2.resize(frame, (640, 480)) frame_count += 1elapsed_time = time.time() - start_timefps = frame_count / elapsed_timecv2.putText( frame, f"FPS: {fps:.2f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2, cv2.LINE_AA )cv2.imshow('v5lite_ResultShow', frame)if cv2.waitKey(1) & 0xFF == ord('q'):# 发送None到帧队列以通知推理线程停止frame_queue.put(None)break
finally:# 确保在退出前释放摄像头和关闭窗口cap.release()cv2.destroyAllWindows()# 等待推理线程结束inference_thread_obj.join()

二、版本二

这个版本运行的时候，识别框一直不出来。现在找到原因了
↓
两个线程，一个用于推理，另一个用于串口发送。推理线程将结果和原始帧放入 result_queue，而串口发送线程也从 result_queue 中获取数据。但是，主线程也尝试从 result_queue 中获取数据以显示结果。会导致竞争条件，其中数据可能在显示之前就被串口发送线程取走了。

import sys
sys.path.append('/home/pi/Desktop/Lite')import torch
import cv2
import numpy as np
from models.experimental import attempt_load
from utils.general import non_max_suppression, scale_coords
from utils.plots import plot_one_box
from queue import Queue, Empty
import threading
import time
import serial# 初始化串口
try:ser = serial.Serial('/dev/ttyAMA0', baudrate=115200, timeout=1)
except Exception as e:print(f"无法打开串口: {e}")exit()# 加载模型，指定CPU
try:model = attempt_load('/home/pi/Desktop/Lite/v5lite-s.pt', map_location=torch.device('cpu'))model.eval()
except Exception as e:print(f"加载模型失败: {e}")exit()# 初始化摄像头
cap = cv2.VideoCapture(0)
if not cap.isOpened():print("无法打开摄像头")exit()# 创建一个队列来存储从摄像头获取的帧
frame_queue = Queue()
# 创建一个队列来存储推理结果
result_queue = Queue()# 定义推理线程函数
def inference_thread(model, frame_queue, result_queue):while True:try:# 从队列中获取帧frame = frame_queue.get(timeout=1)  # 设置超时时间以避免阻塞if frame is None:  # 如果获取到None，则退出线程break# 将图像转换为 YOLOv5 模型所需的格式global imgimg = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)img = torch.from_numpy(img).permute(2, 0, 1).float().div(255.0).unsqueeze(0)# 进行推理with torch.no_grad():pred = model(img)[0]# 非极大值抑制pred = non_max_suppression(pred, conf_thres=0.5, iou_thres=0.5)# 将结果放入结果队列result_queue.put((pred, frame))  # 将原始 frame 传递给串口发送线程except Empty:continueexcept Exception as e:print(f"推理线程出现异常: {e}")# 定义串口发送线程函数
def serial_send_thread(result_queue):while True:try:pred, frame = result_queue.get(timeout=1)  # 设置超时时间以避免阻塞if pred is None:  # 如果获取到None，则退出线程breakfor det in pred:if det is not None and len(det):det[:, :4] = scale_coords(frame.shape, det[:, :4], frame.shape).round()for *xyxy, conf, cls in det:# label = f'{model.names[int(cls)]} {conf:.2f}'# plot_one_box(xyxy, frame, label=label, color=(0, 255, 0), line_thickness=3)label_onlyName = f'{model.names[int(cls)]}'ser.write('@'.encode())ser.write(label_onlyName.encode())  # 发送数据到串口ser.write('\r\n'.encode())print(f"Sent to serial: {label_onlyName}")result_queue.task_done()  # 标记任务完成except Empty:continueexcept Exception as e:print(f"串口发送线程出现异常: {e}")# 启动推理线程
inference_thread_obj = threading.Thread(target=inference_thread, args=(model, frame_queue, result_queue))
inference_thread_obj.start()# 启动串口发送线程
serial_send_thread_obj = threading.Thread(target=serial_send_thread, args=(result_queue,))
serial_send_thread_obj.start()# 用于计算帧率的变量
frame_count = 0
start_time = time.time()try:while True:# 设置摄像头画面的分辨率（避免因分辨率过高导致显示界面卡顿）cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)# 获取摄像头的帧ret, frame = cap.read()if not ret:print("无法获取摄像头的帧数据")break# 将帧放入队列以供推理线程处理try:frame_queue.put(frame)except Exception as e:print(f"将帧放入队列时出错: {e}")# 尝试从结果队列中获取推理结果try:pred = result_queue.get_nowait()  # 使用nowait以避免阻塞# 处理预测结果for det in pred:if det is not None and len(det):det[:, :4] = scale_coords(frame.shape, det[:, :4], frame.shape).round()for *xyxy, conf, cls in det:label = f'{model.names[int(cls)]} {conf:.2f}'plot_one_box(xyxy, frame, label=label, color=(0, 255, 0), line_thickness=3)except Empty:# 如果没有结果，则继续下一次循环passexcept Exception as e:print(f"处理推理结果时出现异常: {e}")# 显示结果frame = cv2.resize(frame, (640, 480))frame_count += 1elapsed_time = time.time() - start_timefps = frame_count / elapsed_timecv2.putText(frame, f"FPS: {fps:.2f}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2, cv2.LINE_AA)cv2.imshow('v5lite_ResultShow', frame)if cv2.waitKey(1) & 0xFF == ord('q'):# 发送None到帧队列以通知推理线程退出frame_queue.put(None)break
finally:# 确保在退出之前释放摄像头和关闭窗口cap.release()cv2.destroyAllWindows()ser.close()  # 关闭串口# 等待推理线程结束inference_thread_obj.join()serial_send_thread_obj.join()

总结

这篇文章依旧没有总结