duan8Detection/ROI/test_roi_rtsp.py
2026-01-07 15:40:08 +08:00

164 lines
4.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# test_roi_rtsp.py
import cv2
import numpy as np
from roi_filter import draw_roi_on_frame
import sys
import os
def test_roi_with_rtsp():
print("="*50)
print("ROI区域RTSP流验证测试")
print("请检查红色矩形框是否出现在RTSP流图像上")
print("'q'退出")
print("="*50)
# RTSP流地址 - 根据您的配置
rtsp_url = "rtsp://10.0.0.50:8554/camera_test/scene1"
# 创建视频捕获对象
cap = cv2.VideoCapture(rtsp_url)
# 设置一些参数以优化RTSP流
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # 减少缓冲延迟
if not cap.isOpened():
print(f"❌ 无法连接到RTSP流: {rtsp_url}")
print("请检查RTSP流地址是否正确")
return
print(f"✅ 成功连接到RTSP流: {rtsp_url}")
print("等待视频流开始...")
frame_count = 0
while True:
ret, frame = cap.read()
if not ret:
print("❌ 无法读取视频帧可能RTSP流已断开")
break
frame_count += 1
# 应用ROI区域绘制
frame_with_roi = draw_roi_on_frame(frame, "../config.yaml")
# 添加状态信息
status_text = f'Frame: {frame_count} | ROI Test Active'
cv2.putText(frame_with_roi, status_text, (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
cv2.putText(frame_with_roi, 'Press Q to quit', (10, 60),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
# 显示带有ROI的帧
cv2.imshow('ROI RTSP Verification', frame_with_roi)
# 按'q'退出
if cv2.waitKey(1) & 0xFF == ord('q'):
break
# 释放资源
cap.release()
cv2.destroyAllWindows()
print("✅ RTSP验证测试结束")
def test_roi_with_rtsp_advanced():
"""
增强版RTSP验证包含更多调试信息
"""
print("="*50)
print("增强版ROI区域RTSP流验证测试")
print("'q'退出,按'r'重新连接")
print("="*50)
rtsp_url = "rtsp://10.0.0.50:8554/camera_test/scene1"
cap = None
reconnect_count = 0
while True:
if cap is None or not cap.isOpened():
print(f"尝试连接RTSP流 (重连次数: {reconnect_count})...")
cap = cv2.VideoCapture(rtsp_url)
# 设置RTSP参数
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
cap.set(cv2.CAP_PROP_FPS, 30)
if not cap.isOpened():
print(f"连接失败等待3秒后重试...")
cv2.waitKey(3000)
reconnect_count += 1
continue
else:
print(f"✅ 成功连接到RTSP流: {rtsp_url}")
reconnect_count = 0
ret, frame = cap.read()
if not ret:
print("⚠️ 视频流断开,尝试重连...")
cap.release()
cap = None
cv2.waitKey(1000)
continue
# 应用ROI区域绘制
frame_with_roi = draw_roi_on_frame(frame, "../config.yaml")
# 添加状态信息
status_text = f'RTSP: Connected | ROI Test | Frame: {cv2.getTickCount() % 10000}'
cv2.putText(frame_with_roi, status_text, (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
cv2.putText(frame_with_roi, 'Press Q to quit, R to reconnect', (10, 60),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
cv2.imshow('Advanced ROI RTSP Test', frame_with_roi)
key = cv2.waitKey(1) & 0xFF
if key == ord('q'):
break
elif key == ord('r'):
print("🔄 手动重连RTSP流...")
cap.release()
cap = None
if cap:
cap.release()
cv2.destroyAllWindows()
print("✅ 增强版RTSP验证测试结束")
if __name__ == "__main__":
# 检查配置文件是否存在
config_path = "../config.yaml"
if not os.path.exists(config_path):
print(f"⚠️ 配置文件 {config_path} 不存在,将创建示例配置")
sample_config = '''roi_filter:
enabled: true
regions:
- name: "control_lever_area"
x_min: 0.1
x_max: 0.3
y_min: 0.7
y_max: 0.9
description: "控制摇杆区域"
enabled: true
- name: "valve_area"
x_min: 0.7
x_max: 0.8
y_min: 0.6
y_max: 0.8
description: "阀门区域"
enabled: true
input:
video_path: "rtsp://10.0.0.50:8554/camera_test/scene1"
width: 640
height: 480'''
with open(config_path, 'w', encoding='utf-8') as f:
f.write(sample_config)
print(f"✅ 已创建示例配置文件: {config_path}")
# 选择测试模式
if len(sys.argv) > 1 and sys.argv[1] == "advanced":
test_roi_with_rtsp_advanced()
else:
test_roi_with_rtsp()