# -- 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()