建设网站的内容规划,沧州市网站设计,word和wordpress,网站建设所需知识在监控和视频分析领域#xff0c;实时采样视频流中的图像数据是十分重要的。本文将介绍一个基于Python和Tkinter构建的RTSP流图片采样助手的设计与实现#xff0c;旨在简化RTSP流的采样过程#xff0c;并支持根据用户定义的特殊标签进行筛选。
项目概述
该项目的主要功能包…在监控和视频分析领域实时采样视频流中的图像数据是十分重要的。本文将介绍一个基于Python和Tkinter构建的RTSP流图片采样助手的设计与实现旨在简化RTSP流的采样过程并支持根据用户定义的特殊标签进行筛选。
项目概述
该项目的主要功能包括
从多个RTSP流中实时采样图像。根据用户定义的特殊标签筛选并保存图像。提供友好的图形用户界面GUI以便于用户操作。
技术栈
Python: 主要编程语言。OpenCV: 用于视频流处理和图像处理。Tkinter: 用于创建图形用户界面。PyTorch: 用于加载和使用深度学习模型进行目标检测。Subprocess: 用于ping测试IP地址的连通性。
关键功能实现
1、GUI设计
使用Tkinter创建用户界面用户可以输入多个IP地址、密码、特殊标签和取样间隔时间。以下是创建GUI的代码片段
root tk.Tk()
root.title(RTSP流图片取样助手)tk.Label(root, textIP地址每行一个).pack()
ip_entry scrolledtext.ScrolledText(root, height15, width50)
ip_entry.pack()tk.Label(root, text密码每行一个).pack()
password_entry scrolledtext.ScrolledText(root, height15, width50)
password_entry.pack()tk.Label(root, text特殊标签以逗号分隔).pack()
special_items_entry tk.Entry(root, width50)
special_items_entry.pack()
special_items_entry.insert(0, person)tk.Label(root, text取样间隔时间秒).pack()
interval_entry tk.Entry(root, width10)
interval_entry.pack()
interval_entry.insert(0, 5)model_var StringVar(root)
model_var.set(未选择模型)
model_button Button(root, text选择模型, commandselect_model)
model_button.pack()status_text tk.Text(root, height10, width50)
status_text.pack()start_button Button(root, text开始采样, commandlambda: start_detection_thread(stop_event, status_text))
start_button.pack()stop_button Button(root, text停止采样, commandlambda: stop_detection(stop_event))
stop_button.pack()root.mainloop()2、视频流处理
通过OpenCV打开RTSP流并读取视频帧。以下是处理视频流的代码片段
cap cv2.VideoCapture(rtsp_url)
if not cap.isOpened():print(fError: Could not open video stream {rtsp_url})returnwhile not stop_event.is_set():ret, frame cap.read()if not ret:print(Error: Failed to read frame from the video stream.)break3、目标检测
集成深度学习模型进行目标检测。在每一帧中我们使用模型识别物体并绘制边框。以下是模型推理的代码
if model is not None:boxes detect(imgsz, conf_thres, iou_thres, model, infer_frame, stride, device)save_frame Falsefor box in boxes:if float(box[4]) conf_thres:x1, y1, x2, y2, label_id int(box[0]), int(box[1]), int(box[2]), int(box[3]), int(box[5])label _names[int(label_id)]plot_one_box(x1, y1, x2, y2, plot_frame, conf_or_proportionconf_thres, label_cls_idlabel, line_thickness3, colorglobalColors[int(label_id)])if label in special_items:save_frame True4.、图像保存逻辑
根据用户输入的特殊标签筛选图像并保存。以下是相关代码
current_time time.time()
if (model is None or save_frame) and (current_time - last_save_time) sample_interval:frame_filename os.path.join(ip_folder, fdetected_{int(current_time)}.jpg)cv2.imwrite(frame_filename, infer_frame)print(fSaved frame: {frame_filename})last_save_time current_time5.、多线程处理
为了能够同时处理多个RTSP流我们使用线程来管理每个流的采样。这样可以确保主线程始终响应用户的操作。以下是创建线程的示例代码
threads []
for index, (ip, password) in enumerate(zip(ip_list, password_list)):rtsp_url rtsp_base_url.format(admin, password, ip)thread threading.Thread(targetstart, args(rtsp_url, index, imgsz, conf_thres, iou_thres, model, device, half, stride, special_items, _names, globalColors, save_special_items_name, stop_event, status_text, sample_interval, ip))thread.start()threads.append(thread)for thread in threads:thread.join()6、 停止采样
用户可以点击“停止采样”按钮来中断正在进行的采样操作。以下是实现该功能的代码
def stop_detection(stop_event):stop_event.set()start_detection_thread.running False运行示例
运行程序后用户需要在GUI中输入以下信息
IP地址: 监控摄像头的RTSP流地址每行一个。密码: 对应的RTSP流密码每行一个。特殊标签: 用户希望采样的物体标签以逗号分隔。取样间隔时间: 图像采样的时间间隔单位秒。点击“开始采样”后程序将开始处理指定的RTSP流并根据设置保存图像。如果需要停止采样只需点击“停止采样”按钮。
源码
1、主程序
import subprocess
from tools import * # 确保你有 tools.py 文件包含所需的函数
import cv2
import os
import torch
import random
import time
import tkinter as tk
from tkinter import scrolledtext, messagebox, StringVar, Button, filedialog
import threading
import warningswarnings.filterwarnings(ignore)def ping_ip(ip):Ping an IP address and return True if it is reachable, else False.try:output subprocess.check_output([ping, -n, 1, ip], stderrsubprocess.STDOUT, universal_newlinesTrue)return Trueexcept subprocess.CalledProcessError:return Falsedef start(rtsp_url, idx, imgsz, conf_thres, iou_thres, model, device, half, stride, special_items, _names, globalColors, save_special_items_name, stop_event, status_text, sample_interval, ip_address):if not ping_ip(ip_address):messagebox.showerror(Error, f无法连接到 IP 地址: {ip_address})returncap cv2.VideoCapture(rtsp_url)if not cap.isOpened():print(fError: Could not open video stream {rtsp_url})returnstatus_text.insert(tk.END, f正在采样视频流: {rtsp_url}_{idx}\n)last_save_time time.time()ip_folder os.path.join(save_special_items_name, f{ip_address}_{str(idx)})os.makedirs(ip_folder, exist_okTrue)while not stop_event.is_set():ret, frame cap.read()if not ret:print(Error: Failed to read frame from the video stream.)breakinfer_frame frame.copy()plot_frame frame.copy()if model is not None:boxes detect(imgsz, conf_thres, iou_thres, model, infer_frame, stride, device)save_frame Falsefor box in boxes:if float(box[4]) conf_thres:x1, y1, x2, y2, label_id int(box[0]), int(box[1]), int(box[2]), int(box[3]), int(box[5])label _names[int(label_id)]plot_one_box(x1, y1, x2, y2, plot_frame, conf_or_proportionconf_thres, label_cls_idlabel, line_thickness3, colorglobalColors[int(label_id)])if label in special_items:save_frame Truecurrent_time time.time()if (model is None or save_frame) and (current_time - last_save_time) sample_interval:frame_filename os.path.join(ip_folder, fdetected_{int(current_time)}.jpg)cv2.imwrite(frame_filename, infer_frame)print(fSaved frame: {frame_filename})last_save_time current_timecv2.imshow(fDetection - {rtsp_url}_{idx}, plot_frame)if cv2.waitKey(1) 0xFF ord(q):breakcap.release()cv2.destroyAllWindows()status_text.insert(tk.END, f视频流 {rtsp_url}_{idx} 已停止采样。\n)def run_detection(ip_list, password_list, special_items, model_path, stop_event, status_text, sample_interval):rtsp_base_url rtsp://{}:{}{}:554/Streaming/Channels/101conf_thres 0.25iou_thres 0.5imgsz 640save_special_items_name special_items_datasetsdevice select_device(0 if torch.cuda.is_available() else cpu)half device.type ! cpumodel Noneif model_path:model torch.load(model_path, map_locationdevice)[model].float()model.to(device).eval()if half:model.half()if not os.path.exists(save_special_items_name):os.makedirs(save_special_items_name)img torch.zeros((1, 3, imgsz, imgsz), devicedevice)if model is not None:_ model(img.half() if half else img) if device.type ! cpu else None_names model.module.names if hasattr(model, module) else model.namesglobalColors [[random.randint(0, 255) for _ in range(3)] for _ in _names]stride max(int(model.stride.max()), 32)else:_names []globalColors []stride 32threads []for index, (ip, password) in enumerate(zip(ip_list, password_list)):rtsp_url rtsp_base_url.format(admin, password, ip)thread threading.Thread(targetstart, args(rtsp_url, index, imgsz, conf_thres, iou_thres, model, device, half, stride, special_items, _names, globalColors, save_special_items_name, stop_event, status_text, sample_interval, ip))thread.start()threads.append(thread)for thread in threads:thread.join()start_detection_thread.running Falsedef start_detection_thread(stop_event, status_text):if hasattr(start_detection_thread, running) and start_detection_thread.running:messagebox.showwarning(Warning, Detection is already running.)returnstop_event.clear() # 重置 stop_eventstart_detection_thread.running Truestatus_text.delete(1.0, tk.END)status_text.insert(tk.END, 开始采样...\n)ip_list ip_entry.get(1.0, tk.END).strip().split(\n)password_list password_entry.get(1.0, tk.END).strip().split(\n)special_items special_items_entry.get().strip().split(,)model_path model_var.get() if model_var.get() ! 未选择模型 else Nonesample_interval int(interval_entry.get())if len(ip_list) ! len(password_list):messagebox.showerror(Error, IP addresses and passwords must match.)start_detection_thread.running Falsereturndetection_thread threading.Thread(targetrun_detection, args(ip_list, password_list, special_items, model_path, stop_event, status_text, sample_interval))detection_thread.start()status_text.insert(tk.END, 采样正在进行中...\n)def stop_detection(stop_event):stop_event.set()start_detection_thread.running Falsedef select_model():model_path filedialog.askopenfilename(title选择模型文件, filetypes[(PyTorch Model, *.pt)])if model_path:model_var.set(model_path)if __name__ __main__:# Tkinter GUIroot tk.Tk()root.title(RTSP流图片取样助手)tk.Label(root, textIP地址每行一个).pack()ip_entry scrolledtext.ScrolledText(root, height15, width50)ip_entry.pack()tk.Label(root, text密码每行一个).pack()password_entry scrolledtext.ScrolledText(root, height15, width50)password_entry.pack()tk.Label(root, text特殊标签以逗号分隔).pack()special_items_entry tk.Entry(root, width50)special_items_entry.pack()special_items_entry.insert(0, person)tk.Label(root, text取样间隔时间秒).pack()interval_entry tk.Entry(root, width10)interval_entry.pack()interval_entry.insert(0, 5)model_var StringVar(root)model_var.set(未选择模型)model_button Button(root, text选择模型, commandselect_model)model_button.pack()stop_event threading.Event()status_text tk.Text(root, height10, width50)status_text.pack()start_button Button(root, text开始采样, commandlambda: start_detection_thread(stop_event, status_text))start_button.pack()stop_button Button(root, text停止采样, commandlambda: stop_detection(stop_event))stop_button.pack()root.mainloop()2、工具程序
# !/usr/bin/python3
# -*- coding:utf-8 -*-
# cython: language_level3
import os.path
import random
import shutil
import time
from pathlib import Path
import cv2
import numpy as np
import torch
from tqdm import tqdm
from utils.augmentations import letterbox
from utils.general import non_max_suppression, scale_boxes
from utils.torch_utils import select_device
NUM_THREADS min(8, max(1, os.cpu_count() - 1)) # number of YOLO multiprocessing threadsdef img_transpose(img0, img_size, stride):assert img0 is not None, Image Not Found img letterbox(img0, img_size, stridestride)[0]img img[:, :, ::-1].transpose(2, 0, 1)img np.ascontiguousarray(img)return imgdef calculate_box_area(x1, y1, x2, y2):return (x2 - x1) * (y2 - y1)def detect(img_size, conf_thres, iou_thres, model, img0, stride, device):imgsz img_sizeimg img_transpose(img0, imgsz, stride)img torch.from_numpy(img).to(device) # 移动到与模型相同的设备img img.float() # 确保是全精度img / 255.0if img.ndimension() 3:img img.unsqueeze(0)pred model(img, augmentFalse)[0]pred non_max_suppression(pred, conf_thres, iou_thres, classesNone, agnosticFalse)for i, det in enumerate(pred):if det is not None and len(det):det[:, :4] scale_boxes(img.shape[2:], det[:, :4], img0.shape).round()return detdef plot_one_box(x1, y1, x2, y2, img, conf_or_proportionNone, label_cls_idNone, line_thicknessNone, colorNone):tl line_thickness or round(0.002 * (img.shape[0] img.shape[1]) / 2) 1c1, c2 (x1, y1), (x2, y2)cv2.rectangle(img, c1, c2, color, thicknesstl, lineTypecv2.LINE_AA)if label_cls_id:full_label str(label_cls_id) (f_{conf_or_proportion} if conf_or_proportion is not None else )tf max(tl - 1, 1) # font thicknesst_size cv2.getTextSize(full_label, 0, fontScaletl / 3, thicknesstf)[0]# Compute the size of the label background based on the full label text sizec2 c1[0] t_size[0], c1[1] - t_size[1] - 3cv2.rectangle(img, c1, (c2[0], c1[1] - t_size[1] - 3), color, -1, cv2.LINE_AA)# Draw label text above the rectangle backgroundbottom_left_corner_of_text (c1[0], c1[1] - 2)cv2.putText(img, full_label, bottom_left_corner_of_text, 0, tl / 3, [225, 255, 255], thicknesstf,lineTypecv2.LINE_AA)3、代码依赖
将以上两个代码放在yolov5-7.0的根目录运行即可
4、程序打包参考我的博客《使用 PyInstaller 打包 Python 应用程序时解决 FileNotFoundError 的问题》
spec代码
# -*- mode: python ; coding: utf-8 -*-block_cipher Nonea Analysis([pic.py],pathex[C:\\Users\\linds\\anaconda3\\envs\\py36\\Lib\\site-packages\\torch\\lib],binaries[(r.\utils\general.pyc, r.\utils ),(rC:\\Users\\linds\\anaconda3\\envs\\py36\\Lib\\site-packages\\torch\\lib\\*, torch\\lib),(rC:\\Users\\linds\\anaconda3\\envs\\py36\\Lib\\site-packages\\torchvision\\*.dll, torchvision)],datas[(rutils/general.py, utils), (utils/general.pyc, utils)],hiddenimports[torch, torchvision, PIL], # 添加隐藏导入hookspath[],hooksconfig{},runtime_hooks[],excludes[],win_no_prefer_redirectsFalse,win_private_assembliesFalse,cipherblock_cipher,noarchiveFalse)for d in a.datas:if cp36-win_amd64.pyd in d[0]:a.datas.remove(d)breakfor d in a.datas:if cp36-win_amd64.pyd in d[0]:a.datas.remove(d)breakpyz PYZ(a.pure, a.zipped_data,cipherblock_cipher)exe EXE(pyz,a.scripts,a.binaries,a.zipfiles,a.datas, [],namepic,debugFalse,bootloader_ignore_signalsFalse,stripFalse,upxTrue,upx_exclude[],runtime_tmpdirNone,consoleTrue,disable_windowed_tracebackFalse,target_archNone,codesign_identityNone,entitlements_fileNone )5、软件截图 总结
本项目展示了如何使用Python、OpenCV和Tkinter构建一个功能强大的RTSP流图片采样助手。该工具可用于监控、视频分析和机器学习等多个领域能够帮助用户实时采样并保存感兴趣的图像数据。
通过这个项目您可以更好地理解视频流处理、目标检测以及多线程编程的基本概念。希望这个项目能为您提供灵感欢迎随时反馈与交流
