138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
|
# vision/camera.py
|
|||
|
|
import cv2
|
|||
|
|
|
|||
|
|
|
|||
|
|
class CameraController:
|
|||
|
|
def __init__(self):
|
|||
|
|
self.camera = None
|
|||
|
|
self.camera_type = "ip"
|
|||
|
|
self.camera_ip = "192.168.1.51"
|
|||
|
|
self.camera_port = 554
|
|||
|
|
self.camera_username = "admin"
|
|||
|
|
self.camera_password = "XJ123456"
|
|||
|
|
self.camera_channel = 1
|
|||
|
|
|
|||
|
|
def set_config(self, camera_type="ip", ip=None, port=None, username=None, password=None, channel=1):
|
|||
|
|
"""
|
|||
|
|
设置摄像头配置
|
|||
|
|
"""
|
|||
|
|
self.camera_type = camera_type
|
|||
|
|
if ip:
|
|||
|
|
self.camera_ip = ip
|
|||
|
|
if port:
|
|||
|
|
self.camera_port = port
|
|||
|
|
if username:
|
|||
|
|
self.camera_username = username
|
|||
|
|
if password:
|
|||
|
|
self.camera_password = password
|
|||
|
|
self.camera_channel = channel
|
|||
|
|
|
|||
|
|
def setup_capture(self, camera_index=0):
|
|||
|
|
"""
|
|||
|
|
设置摄像头捕获
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
rtsp_url = f"rtsp://{self.camera_username}:{self.camera_password}@{self.camera_ip}:{self.camera_port}/streaming/channels/{self.camera_channel}01"
|
|||
|
|
self.camera = cv2.VideoCapture(rtsp_url)
|
|||
|
|
|
|||
|
|
if not self.camera.isOpened():
|
|||
|
|
print(f"无法打开网络摄像头: {rtsp_url}")
|
|||
|
|
return False
|
|||
|
|
print(f"网络摄像头初始化成功，地址: {rtsp_url}")
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"摄像头设置失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def capture_frame_exec(self):
|
|||
|
|
"""捕获当前帧并返回numpy数组，设置5秒总超时"""
|
|||
|
|
try:
|
|||
|
|
if self.camera is None:
|
|||
|
|
print("摄像头未初始化")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 设置总超时时间为5秒
|
|||
|
|
total_timeout = 5.0 # 5秒总超时时间
|
|||
|
|
start_time = time.time()
|
|||
|
|
|
|||
|
|
# 跳20帧，获取最新图像
|
|||
|
|
frames_skipped = 0
|
|||
|
|
while frames_skipped < 20:
|
|||
|
|
# 检查总超时
|
|||
|
|
if time.time() - start_time > total_timeout:
|
|||
|
|
print("捕获图像总超时")
|
|||
|
|
return None
|
|||
|
|
self.camera.grab()
|
|||
|
|
time.sleep(0.05) # 稍微增加延迟，确保有新帧到达
|
|||
|
|
frames_skipped += 1
|
|||
|
|
|
|||
|
|
# 尝试读取帧，使用同一超时计时器
|
|||
|
|
read_attempts = 0
|
|||
|
|
max_read_attempts = 3
|
|||
|
|
if self.camera.grab():
|
|||
|
|
while read_attempts < max_read_attempts:
|
|||
|
|
# 使用同一个超时计时器检查
|
|||
|
|
if time.time() - start_time > total_timeout:
|
|||
|
|
print("捕获图像总超时")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
ret, frame = self.camera.retrieve()
|
|||
|
|
if ret:
|
|||
|
|
return frame
|
|||
|
|
else:
|
|||
|
|
print(f"尝试读取图像帧失败，重试 ({read_attempts+1}/{max_read_attempts})")
|
|||
|
|
read_attempts += 1
|
|||
|
|
# 短暂延迟后重试
|
|||
|
|
time.sleep(0.05)
|
|||
|
|
|
|||
|
|
print("多次尝试后仍无法捕获有效图像帧")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"图像捕获失败: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def capture_frame(self):
|
|||
|
|
"""捕获当前帧并返回numpy数组"""
|
|||
|
|
try:
|
|||
|
|
if self.camera is None:
|
|||
|
|
# self.set_config()
|
|||
|
|
self.setup_capture()
|
|||
|
|
|
|||
|
|
|
|||
|
|
frame = self.capture_frame_exec()
|
|||
|
|
if frame is not None:
|
|||
|
|
return frame
|
|||
|
|
else:
|
|||
|
|
print("无法捕获图像帧")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"图像捕获失败: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def capture_frame_bak(self):
|
|||
|
|
"""捕获当前帧并返回numpy数组"""
|
|||
|
|
try:
|
|||
|
|
if self.camera is None:
|
|||
|
|
print("摄像头未初始化")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
ret, frame = self.camera.read()
|
|||
|
|
if ret:
|
|||
|
|
return frame
|
|||
|
|
else:
|
|||
|
|
print("无法捕获图像帧")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"图像捕获失败: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def release(self):
|
|||
|
|
"""释放摄像头资源"""
|
|||
|
|
if self.camera is not None:
|
|||
|
|
self.camera.release()
|
|||
|
|
self.camera = None
|
|||
|
|
|
|||
|
|
def __del__(self):
|
|||
|
|
"""析构函数，确保资源释放"""
|
|||
|
|
self.release()
|