460 lines
17 KiB
Python
460 lines
17 KiB
Python
# -- coding: utf-8 --
|
||
"""
|
||
海康相机拍照工具类
|
||
功能：连接海康相机、内存取 BGR 帧、保存 JPG
|
||
使用方式：导入此类后，实例化 HikCameraCapture 并调用相关方法
|
||
"""
|
||
import sys
|
||
import os
|
||
import time
|
||
from ctypes import *
|
||
|
||
import numpy as np
|
||
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
if _HERE not in sys.path:
|
||
sys.path.insert(0, _HERE)
|
||
from MvImport.MvCameraControl_class import *
|
||
|
||
|
||
class HikCameraCapture:
|
||
"""海康相机拍照工具类（仅保存JPG）"""
|
||
|
||
def __init__(self):
|
||
self.cam = None # 相机实例
|
||
self.is_initialized = False # SDK初始化状态
|
||
self.is_connected = False # 相机连接状态
|
||
self.is_grabbing = False
|
||
self.initialize_sdk()
|
||
|
||
def initialize_sdk(self):
|
||
"""初始化SDK"""
|
||
try:
|
||
# 初始化SDK
|
||
MvCamera.MV_CC_Initialize()
|
||
self.is_initialized = True
|
||
return 0
|
||
except Exception as e:
|
||
print(f"SDK初始化失败: {e}")
|
||
return -1
|
||
|
||
def enum_devices(self):
|
||
"""枚举所有可用相机设备"""
|
||
if not self.is_initialized:
|
||
print("请先初始化SDK")
|
||
return None
|
||
|
||
deviceList = MV_CC_DEVICE_INFO_LIST()
|
||
# 枚举所有类型的相机
|
||
tlayerType = (MV_GIGE_DEVICE | MV_USB_DEVICE | MV_GENTL_CAMERALINK_DEVICE
|
||
| MV_GENTL_CXP_DEVICE | MV_GENTL_XOF_DEVICE)
|
||
|
||
ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
|
||
if ret != 0:
|
||
print(f"枚举设备失败! ret[0x{ret:x}]")
|
||
return None
|
||
|
||
if deviceList.nDeviceNum == 0:
|
||
print("未找到任何相机设备!")
|
||
return None
|
||
|
||
# 整理设备信息返回
|
||
devices = []
|
||
for i in range(deviceList.nDeviceNum):
|
||
dev_info = cast(deviceList.pDeviceInfo[i], POINTER(MV_CC_DEVICE_INFO)).contents
|
||
device_info = {
|
||
"index": i,
|
||
"type": self._get_device_type(dev_info.nTLayerType),
|
||
"model": self._get_device_model(dev_info),
|
||
"serial": self._get_device_serial(dev_info)
|
||
}
|
||
devices.append(device_info)
|
||
print(f"\n设备[{i}]:")
|
||
print(f" 类型: {device_info['type']}")
|
||
print(f" 型号: {device_info['model']}")
|
||
print(f" 序列号: {device_info['serial']}")
|
||
|
||
return devices
|
||
|
||
def connect_camera(self, device_index):
|
||
"""
|
||
连接指定索引的相机
|
||
:param device_index: 相机设备索引（从enum_devices获取）
|
||
:return: 0-成功，非0-失败
|
||
"""
|
||
if not self.is_initialized:
|
||
print("请先初始化SDK")
|
||
return -1
|
||
|
||
# 重新枚举设备（确保设备列表最新）
|
||
deviceList = MV_CC_DEVICE_INFO_LIST()
|
||
tlayerType = (MV_GIGE_DEVICE | MV_USB_DEVICE | MV_GENTL_CAMERALINK_DEVICE
|
||
| MV_GENTL_CXP_DEVICE | MV_GENTL_XOF_DEVICE)
|
||
ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
|
||
if ret != 0 or deviceList.nDeviceNum == 0:
|
||
print("枚举设备失败，无法连接相机")
|
||
return -1
|
||
|
||
if int(device_index) >= deviceList.nDeviceNum:
|
||
print(f"设备索引{device_index}超出范围（最大索引：{deviceList.nDeviceNum - 1}）")
|
||
return -1
|
||
|
||
# 创建相机实例
|
||
self.cam = MvCamera()
|
||
stDeviceList = cast(deviceList.pDeviceInfo[int(device_index)], POINTER(MV_CC_DEVICE_INFO)).contents
|
||
|
||
# 创建句柄
|
||
ret = self.cam.MV_CC_CreateHandle(stDeviceList)
|
||
if ret != 0:
|
||
print(f"创建相机句柄失败! ret[0x{ret:x}]")
|
||
return ret
|
||
|
||
# 打开设备
|
||
ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
|
||
if ret != 0:
|
||
print(f"打开相机失败! ret[0x{ret:x}]")
|
||
self.cam.MV_CC_DestroyHandle()
|
||
return ret
|
||
|
||
# GigE相机优化包大小
|
||
if stDeviceList.nTLayerType == MV_GIGE_DEVICE or stDeviceList.nTLayerType == MV_GENTL_GIGE_DEVICE:
|
||
nPacketSize = self.cam.MV_CC_GetOptimalPacketSize()
|
||
if int(nPacketSize) > 0:
|
||
ret_packet = self.cam.MV_CC_SetIntValue("GevSCPSPacketSize", nPacketSize)
|
||
if ret_packet != 0:
|
||
print(f"警告: 设置包大小失败! ret[0x{ret_packet:x}]")
|
||
|
||
# 设置触发模式为关闭
|
||
ret = self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
|
||
if ret != 0:
|
||
print(f"设置触发模式失败! ret[0x{ret:x}]")
|
||
self.disconnect_camera()
|
||
return ret
|
||
|
||
self.is_connected = True
|
||
self.is_grabbing = False
|
||
print("相机连接成功")
|
||
return 0
|
||
|
||
def start_grabbing(self):
|
||
"""开始持续取流（供实时预览）。"""
|
||
if not self.is_connected:
|
||
return -1
|
||
if getattr(self, "is_grabbing", False):
|
||
return 0
|
||
ret = self.cam.MV_CC_StartGrabbing()
|
||
if ret != 0:
|
||
print(f"开始取流失败! ret[0x{ret:x}]")
|
||
return ret
|
||
self.is_grabbing = True
|
||
return 0
|
||
|
||
def stop_grabbing(self):
|
||
"""停止持续取流。"""
|
||
if not self.cam or not getattr(self, "is_grabbing", False):
|
||
return 0
|
||
self.cam.MV_CC_StopGrabbing()
|
||
self.is_grabbing = False
|
||
return 0
|
||
|
||
def grab_bgr_frame(self, timeout_ms=200):
|
||
"""在已取流状态下取一帧，返回 BGR ndarray（供流畅预览，不落盘）。"""
|
||
if not self.is_connected or not self.cam:
|
||
return None
|
||
if not getattr(self, "is_grabbing", False):
|
||
if self.start_grabbing() != 0:
|
||
return None
|
||
stOutFrame = MV_FRAME_OUT()
|
||
memset(byref(stOutFrame), 0, sizeof(stOutFrame))
|
||
ret = self.cam.MV_CC_GetImageBuffer(stOutFrame, int(timeout_ms))
|
||
if ret != 0 or stOutFrame.pBufAddr is None:
|
||
return None
|
||
try:
|
||
info = stOutFrame.stFrameInfo
|
||
w, h = int(info.nWidth), int(info.nHeight)
|
||
if w <= 0 or h <= 0:
|
||
return None
|
||
src_type = info.enPixelType
|
||
need = w * h * 3
|
||
# 已是 BGR8 则直接拷贝
|
||
if src_type == PixelType_Gvsp_BGR8_Packed:
|
||
n = int(info.nFrameLen)
|
||
raw = string_at(stOutFrame.pBufAddr, n)
|
||
arr = np.frombuffer(raw, dtype=np.uint8)
|
||
if arr.size < need:
|
||
return None
|
||
return arr[:need].reshape((h, w, 3)).copy()
|
||
|
||
dst_buf = (c_ubyte * need)()
|
||
stConvert = MV_CC_PIXEL_CONVERT_PARAM_EX()
|
||
memset(byref(stConvert), 0, sizeof(stConvert))
|
||
stConvert.nWidth = w
|
||
stConvert.nHeight = h
|
||
stConvert.pSrcData = cast(stOutFrame.pBufAddr, POINTER(c_ubyte))
|
||
stConvert.nSrcDataLen = info.nFrameLen
|
||
stConvert.enSrcPixelType = src_type
|
||
stConvert.enDstPixelType = PixelType_Gvsp_BGR8_Packed
|
||
stConvert.pDstBuffer = cast(dst_buf, POINTER(c_ubyte))
|
||
stConvert.nDstBufferSize = need
|
||
cret = self.cam.MV_CC_ConvertPixelTypeEx(stConvert)
|
||
if cret != 0:
|
||
print(f"像素转换失败! ret[0x{cret:x}]")
|
||
return None
|
||
out_len = int(stConvert.nDstLen) if stConvert.nDstLen else need
|
||
arr = np.frombuffer(dst_buf, dtype=np.uint8, count=out_len)
|
||
if arr.size < need:
|
||
return None
|
||
return arr[:need].reshape((h, w, 3)).copy()
|
||
except Exception as e:
|
||
print(f"内存取帧失败: {e}")
|
||
return None
|
||
finally:
|
||
try:
|
||
self.cam.MV_CC_FreeImageBuffer(stOutFrame)
|
||
except Exception:
|
||
pass
|
||
|
||
def grab_and_save_jpg(self, save_path, quality=80, timeout_ms=1000):
|
||
"""在已取流状态下抓一帧并保存 JPG。"""
|
||
if not self.is_connected:
|
||
return -1, ""
|
||
if quality < 50 or quality > 99:
|
||
quality = 80
|
||
own_grab = False
|
||
stOutFrame = None
|
||
try:
|
||
if not getattr(self, "is_grabbing", False):
|
||
ret = self.start_grabbing()
|
||
if ret != 0:
|
||
return -1, ""
|
||
own_grab = True
|
||
stOutFrame = MV_FRAME_OUT()
|
||
memset(byref(stOutFrame), 0, sizeof(stOutFrame))
|
||
ret = self.cam.MV_CC_GetImageBuffer(stOutFrame, int(timeout_ms))
|
||
if ret != 0 or stOutFrame.pBufAddr is None:
|
||
raise Exception(f"获取图像帧失败! ret[0x{ret:x}]")
|
||
save_ret, final_path = self._save_jpg_image(stOutFrame, save_path, quality)
|
||
self.cam.MV_CC_FreeImageBuffer(stOutFrame)
|
||
stOutFrame = None
|
||
if save_ret != 0:
|
||
return -1, ""
|
||
return 0, final_path
|
||
except Exception as e:
|
||
print(f"抓帧保存失败: {e}")
|
||
if stOutFrame is not None:
|
||
try:
|
||
self.cam.MV_CC_FreeImageBuffer(stOutFrame)
|
||
except Exception:
|
||
pass
|
||
return -1, ""
|
||
finally:
|
||
if own_grab:
|
||
self.stop_grabbing()
|
||
|
||
def capture_and_save_jpg(self, save_path=None, quality=80):
|
||
"""
|
||
拍摄并保存JPG图片
|
||
:param save_path: 自定义保存路径（如 "test.jpg"），None则使用默认命名
|
||
:param quality: JPG质量 (50-99)
|
||
:return: 0-成功，非0-失败；成功时同时返回图片保存路径
|
||
"""
|
||
if not self.is_connected:
|
||
print("相机未连接，请先连接相机")
|
||
return -1, ""
|
||
|
||
if quality < 50 or quality > 99:
|
||
print("JPG质量范围必须是50-99，自动调整为80")
|
||
quality = 80
|
||
|
||
if save_path is None:
|
||
save_path = f"Image_{int(time.time())}.jpg"
|
||
|
||
was_grabbing = getattr(self, "is_grabbing", False)
|
||
if was_grabbing:
|
||
return self.grab_and_save_jpg(save_path, quality=quality, timeout_ms=2000)
|
||
|
||
try:
|
||
ret = self.cam.MV_CC_StartGrabbing()
|
||
if ret != 0:
|
||
raise Exception(f"开始取流失败! ret[0x{ret:x}]")
|
||
|
||
stOutFrame = MV_FRAME_OUT()
|
||
memset(byref(stOutFrame), 0, sizeof(stOutFrame))
|
||
|
||
ret = self.cam.MV_CC_GetImageBuffer(stOutFrame, 20000)
|
||
if ret != 0 or stOutFrame.pBufAddr is None:
|
||
raise Exception(f"获取图像帧失败! ret[0x{ret:x}]")
|
||
|
||
save_ret, final_path = self._save_jpg_image(stOutFrame, save_path, quality)
|
||
if save_ret != 0:
|
||
raise Exception(f"保存JPG图片失败! ret[0x{save_ret:x}]")
|
||
|
||
print(f"图片保存成功: {final_path}")
|
||
self.cam.MV_CC_FreeImageBuffer(stOutFrame)
|
||
self.cam.MV_CC_StopGrabbing()
|
||
return 0, final_path
|
||
|
||
except Exception as e:
|
||
print(f"拍照保存失败: {e}")
|
||
if 'stOutFrame' in locals():
|
||
try:
|
||
self.cam.MV_CC_FreeImageBuffer(stOutFrame)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.cam.MV_CC_StopGrabbing()
|
||
except Exception:
|
||
pass
|
||
return -1, ""
|
||
|
||
def _save_jpg_image(self, frame_info, save_path, quality):
|
||
"""
|
||
内部方法：保存JPG图片
|
||
:param frame_info: 帧信息
|
||
:param save_path: 保存路径
|
||
:param quality: JPG质量
|
||
:return: 0-成功，非0-失败；最终保存路径
|
||
"""
|
||
# 转换路径为C字符串
|
||
c_file_path = save_path.encode('ascii')
|
||
|
||
# 配置保存参数
|
||
stSaveParam = MV_SAVE_IMAGE_TO_FILE_PARAM_EX()
|
||
stSaveParam.enPixelType = frame_info.stFrameInfo.enPixelType # 相机像素格式
|
||
stSaveParam.nWidth = frame_info.stFrameInfo.nWidth # 宽度
|
||
stSaveParam.nHeight = frame_info.stFrameInfo.nHeight # 高度
|
||
stSaveParam.nDataLen = frame_info.stFrameInfo.nFrameLen # 数据长度
|
||
stSaveParam.pData = frame_info.pBufAddr # 图像数据缓存
|
||
stSaveParam.enImageType = MV_Image_Jpeg # 仅保存JPG
|
||
stSaveParam.pcImagePath = create_string_buffer(c_file_path)
|
||
stSaveParam.iMethodValue = 1
|
||
stSaveParam.nQuality = quality # JPG质量
|
||
|
||
# 执行保存
|
||
mv_ret = self.cam.MV_CC_SaveImageToFileEx(stSaveParam)
|
||
|
||
return mv_ret, save_path
|
||
|
||
def disconnect_camera(self):
|
||
"""断开相机连接并释放资源"""
|
||
if self.cam:
|
||
try:
|
||
self.stop_grabbing()
|
||
except Exception:
|
||
pass
|
||
# 关闭设备
|
||
self.cam.MV_CC_CloseDevice()
|
||
# 销毁句柄
|
||
self.cam.MV_CC_DestroyHandle()
|
||
self.is_connected = False
|
||
print("相机已断开连接")
|
||
|
||
def finalize(self):
|
||
"""反初始化SDK，释放所有资源"""
|
||
if self.is_connected:
|
||
self.disconnect_camera()
|
||
if self.is_initialized:
|
||
MvCamera.MV_CC_Finalize()
|
||
self.is_initialized = False
|
||
print("SDK已反初始化")
|
||
|
||
def _get_device_type(self, tlayer_type):
|
||
"""内部方法：获取设备类型描述"""
|
||
type_map = {
|
||
MV_GIGE_DEVICE: "GigE相机",
|
||
MV_USB_DEVICE: "USB3.0相机",
|
||
MV_GENTL_CAMERALINK_DEVICE: "CameraLink相机",
|
||
MV_GENTL_CXP_DEVICE: "CXP相机",
|
||
MV_GENTL_XOF_DEVICE: "XoF相机",
|
||
MV_GENTL_GIGE_DEVICE: "GENTL GigE相机"
|
||
}
|
||
return type_map.get(tlayer_type, f"未知类型({tlayer_type})")
|
||
|
||
def _get_device_model(self, dev_info):
|
||
"""内部方法：获取设备型号"""
|
||
model = ""
|
||
if dev_info.nTLayerType == MV_GIGE_DEVICE or dev_info.nTLayerType == MV_GENTL_GIGE_DEVICE:
|
||
for per in dev_info.SpecialInfo.stGigEInfo.chModelName:
|
||
if per == 0:
|
||
break
|
||
model += chr(per)
|
||
elif dev_info.nTLayerType == MV_USB_DEVICE:
|
||
for per in dev_info.SpecialInfo.stUsb3VInfo.chModelName:
|
||
if per == 0:
|
||
break
|
||
model += chr(per)
|
||
elif dev_info.nTLayerType == MV_GENTL_CAMERALINK_DEVICE:
|
||
for per in dev_info.SpecialInfo.stCMLInfo.chModelName:
|
||
if per == 0:
|
||
break
|
||
model += chr(per)
|
||
elif dev_info.nTLayerType == MV_GENTL_CXP_DEVICE:
|
||
for per in dev_info.SpecialInfo.stCXPInfo.chModelName:
|
||
if per == 0:
|
||
break
|
||
model += chr(per)
|
||
elif dev_info.nTLayerType == MV_GENTL_XOF_DEVICE:
|
||
for per in dev_info.SpecialInfo.stXoFInfo.chModelName:
|
||
if per == 0:
|
||
break
|
||
model += chr(per)
|
||
return model or "未知型号"
|
||
|
||
def _get_device_serial(self, dev_info):
|
||
"""内部方法：获取设备序列号"""
|
||
serial = ""
|
||
if dev_info.nTLayerType == MV_GIGE_DEVICE or dev_info.nTLayerType == MV_GENTL_GIGE_DEVICE:
|
||
serial = "无"
|
||
elif dev_info.nTLayerType == MV_USB_DEVICE:
|
||
for per in dev_info.SpecialInfo.stUsb3VInfo.chSerialNumber:
|
||
if per == 0:
|
||
break
|
||
serial += chr(per)
|
||
elif dev_info.nTLayerType == MV_GENTL_CAMERALINK_DEVICE:
|
||
for per in dev_info.SpecialInfo.stCMLInfo.chSerialNumber:
|
||
if per == 0:
|
||
break
|
||
serial += chr(per)
|
||
elif dev_info.nTLayerType == MV_GENTL_CXP_DEVICE:
|
||
for per in dev_info.SpecialInfo.stCXPInfo.chSerialNumber:
|
||
if per == 0:
|
||
break
|
||
serial += chr(per)
|
||
elif dev_info.nTLayerType == MV_GENTL_XOF_DEVICE:
|
||
for per in dev_info.SpecialInfo.stXoFInfo.chSerialNumber:
|
||
if per == 0:
|
||
break
|
||
serial += chr(per)
|
||
return serial or "未知序列号"
|
||
|
||
|
||
# 测试示例（外部调用示例）
|
||
if __name__ == "__main__":
|
||
# 1. 创建相机工具类实例
|
||
camera_tool = HikCameraCapture()
|
||
|
||
try:
|
||
# 3. 枚举设备
|
||
devices = camera_tool.enum_devices()
|
||
if not devices:
|
||
sys.exit(1)
|
||
# 4. 选择并连接相机
|
||
device_index = input("请输入要连接的设备索引: ")
|
||
ret = camera_tool.connect_camera(int(device_index))
|
||
if ret != 0:
|
||
sys.exit(1)
|
||
ret, save_path = camera_tool.capture_and_save_jpg(
|
||
save_path="my_capture.jpg",
|
||
quality=90
|
||
)
|
||
|
||
if ret != 0:
|
||
sys.exit(1)
|
||
|
||
except Exception as e:
|
||
print(f"执行失败: {e}")
|
||
finally:
|
||
# 6. 释放所有资源
|
||
camera_tool.finalize()
|
||
|