#!/usr/bin/env python # -*- coding: utf-8 -*- """ 海康工业相机管理类 - 支持多相机选择和操作 """ import time from MvImport.MvCameraControl_class import * class HikCameraManager: """海康相机管理器""" def __init__(self): self.cam = None self.is_initialized = False def initialize(self): """初始化SDK""" ret = MvCamera.MV_CC_Initialize() if ret != 0: raise Exception(f"初始化SDK失败! 错误码: {hex(ret)}") self.is_initialized = True print("SDK初始化成功") def list_cameras(self): """ 列出所有可用相机 返回: 相机信息列表 """ if not self.is_initialized: self.initialize() deviceList = MV_CC_DEVICE_INFO_LIST() ret = MvCamera.MV_CC_EnumDevices(MV_GIGE_DEVICE | MV_USB_DEVICE, deviceList) if ret != 0: raise Exception(f"枚举设备失败! 错误码: {hex(ret)}") if deviceList.nDeviceNum == 0: print("未找到相机设备") return [] cameras = [] for i in range(deviceList.nDeviceNum): mvcc_dev_info = cast(deviceList.pDeviceInfo[i], POINTER(MV_CC_DEVICE_INFO)).contents camera_info = { 'index': i, 'device_info': mvcc_dev_info, 'type': None, 'model': '', 'serial': '', 'ip': '', 'user_defined_name': '' } if mvcc_dev_info.nTLayerType == MV_GIGE_DEVICE: camera_info['type'] = 'GigE' # 型号 model_name = "" for per in mvcc_dev_info.SpecialInfo.stGigEInfo.chModelName: if per == 0: break model_name += chr(per) camera_info['model'] = model_name.strip() # 序列号 serial = "" for per in mvcc_dev_info.SpecialInfo.stGigEInfo.chSerialNumber: if per == 0: break serial += chr(per) camera_info['serial'] = serial.strip() # IP地址 nip1 = ((mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0xff000000) >> 24) nip2 = ((mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0x00ff0000) >> 16) nip3 = ((mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0x0000ff00) >> 8) nip4 = (mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0x000000ff) camera_info['ip'] = f"{nip1}.{nip2}.{nip3}.{nip4}" # 用户自定义名称 user_name = "" for per in mvcc_dev_info.SpecialInfo.stGigEInfo.chUserDefinedName: if per == 0: break user_name += chr(per) camera_info['user_defined_name'] = user_name.strip() elif mvcc_dev_info.nTLayerType == MV_USB_DEVICE: camera_info['type'] = 'USB' # 型号 model_name = "" for per in mvcc_dev_info.SpecialInfo.stUsb3VInfo.chModelName: if per == 0: break model_name += chr(per) camera_info['model'] = model_name.strip() # 序列号 serial = "" for per in mvcc_dev_info.SpecialInfo.stUsb3VInfo.chSerialNumber: if per == 0: break serial += chr(per) camera_info['serial'] = serial.strip() # 用户自定义名称 user_name = "" for per in mvcc_dev_info.SpecialInfo.stUsb3VInfo.chUserDefinedName: if per == 0: break user_name += chr(per) camera_info['user_defined_name'] = user_name.strip() cameras.append(camera_info) return cameras def print_cameras(self): """打印所有相机信息""" cameras = self.list_cameras() if not cameras: print("未找到相机") return print(f"\n找到 {len(cameras)} 个相机:") print("=" * 80) for cam in cameras: print(f"\n[{cam['index']}] {cam['type']} 相机") print(f" 型号: {cam['model']}") print(f" 序列号: {cam['serial']}") if cam['ip']: print(f" IP地址: {cam['ip']}") if cam['user_defined_name']: print(f" 自定义名称: {cam['user_defined_name']}") print("=" * 80) def open_camera(self, index=None, serial=None, ip=None, model=None): """ 打开指定相机 参数: index: 相机索引（0, 1, 2...） serial: 相机序列号 ip: 相机IP地址（仅GigE相机） model: 相机型号 优先级: serial > ip > model > index """ cameras = self.list_cameras() if not cameras: raise Exception("未找到相机设备") # 根据条件选择相机 selected_camera = None if serial: # 按序列号查找 for cam in cameras: if cam['serial'] == serial: selected_camera = cam print(f"通过序列号选择相机: {serial}") break if not selected_camera: raise Exception(f"未找到序列号为 {serial} 的相机") elif ip: # 按IP地址查找（仅GigE） for cam in cameras: if cam['ip'] == ip: selected_camera = cam print(f"通过IP地址选择相机: {ip}") break if not selected_camera: raise Exception(f"未找到IP地址为 {ip} 的相机") elif model: # 按型号查找 for cam in cameras: if model in cam['model']: selected_camera = cam print(f"通过型号选择相机: {model}") break if not selected_camera: raise Exception(f"未找到型号包含 {model} 的相机") else: # 按索引选择（默认第一个） if index is None: index = 0 if index >= len(cameras): raise Exception(f"相机索引 {index} 超出范围（共 {len(cameras)} 个相机）") selected_camera = cameras[index] print(f"通过索引选择相机: {index}") # 打印选中的相机信息 print(f"选中相机: [{selected_camera['type']}] {selected_camera['model']} (序列号: {selected_camera['serial']})") # 创建句柄 self.cam = MvCamera() ret = self.cam.MV_CC_CreateHandle(selected_camera['device_info']) if ret != 0: raise Exception(f"创建句柄失败! 错误码: {hex(ret)}") # 打开设备 ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0) if ret != 0: # 尝试控制模式 ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Control, 0) if ret != 0: self.cam.MV_CC_DestroyHandle() raise Exception(f"打开设备失败! 错误码: {hex(ret)}") print("相机打开成功") return selected_camera def capture_image(self, filename="captured_image", storageAddress = "D:pic/", save_bmp=True, save_jpg=True, timeout=3000): """ 采集一帧图像 参数: filename: 文件名（不含扩展名） save_bmp: 是否保存BMP格式 save_jpg: 是否保存JPG格式 timeout: 超时时间（毫秒） """ if not self.cam: raise Exception("相机未打开") # 设置触发模式为连续采集 ret = self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF) if ret != 0: print(f"警告: 设置触发模式失败! 错误码: {hex(ret)}") # 开始取流 ret = self.cam.MV_CC_StartGrabbing() if ret != 0: raise Exception(f"开始取流失败! 错误码: {hex(ret)}") print("开始采集...") time.sleep(1) # 等待相机稳定 # 获取图像 stFrameInfo = MV_FRAME_OUT() memset(byref(stFrameInfo), 0, sizeof(stFrameInfo)) ret = self.cam.MV_CC_GetImageBuffer(stFrameInfo, timeout) if ret != 0: self.cam.MV_CC_StopGrabbing() raise Exception(f"获取图像失败! 错误码: {hex(ret)}") print(f"成功获取图像: {stFrameInfo.stFrameInfo.nWidth}x{stFrameInfo.stFrameInfo.nHeight}") # 转换并保存图像 saved_files = [] if save_bmp: bmp_file = f"{storageAddress}{filename}.bmp" if self._save_image(stFrameInfo, bmp_file, MV_Image_Bmp): saved_files.append(bmp_file) if save_jpg: jpg_file = f"{storageAddress}{filename}.jpg" if self._save_image(stFrameInfo, jpg_file, MV_Image_Jpeg): saved_files.append(jpg_file) # 释放图像缓存 self.cam.MV_CC_FreeImageBuffer(stFrameInfo) # 停止取流 self.cam.MV_CC_StopGrabbing() return saved_files def _save_image(self, stFrameInfo, filename, image_type): """内部方法：保存图像""" stConvertParam = MV_SAVE_IMAGE_PARAM_EX() memset(byref(stConvertParam), 0, sizeof(stConvertParam)) nBufSize = stFrameInfo.stFrameInfo.nWidth * stFrameInfo.stFrameInfo.nHeight * 3 + 2048 pBuf = (c_ubyte * nBufSize)() stConvertParam.nWidth = stFrameInfo.stFrameInfo.nWidth stConvertParam.nHeight = stFrameInfo.stFrameInfo.nHeight stConvertParam.pData = stFrameInfo.pBufAddr stConvertParam.nDataLen = stFrameInfo.stFrameInfo.nFrameLen stConvertParam.enPixelType = stFrameInfo.stFrameInfo.enPixelType stConvertParam.pImageBuffer = cast(pBuf, POINTER(c_ubyte)) stConvertParam.nBufferSize = nBufSize stConvertParam.enImageType = image_type stConvertParam.nJpgQuality = 90 ret = self.cam.MV_CC_SaveImageEx2(stConvertParam) if ret != 0: print(f"图像转换失败! 错误码: {hex(ret)}") return False try: with open(filename, "wb") as f: f.write(bytearray(pBuf[0:stConvertParam.nImageLen])) print(f"✓ 图像已保存: {filename}") return True except Exception as e: print(f"保存文件失败: {e}") return False def close_camera(self): """关闭相机""" if self.cam: self.cam.MV_CC_CloseDevice() self.cam.MV_CC_DestroyHandle() self.cam = None print("相机已关闭") def finalize(self): """清理SDK资源""" self.close_camera() if self.is_initialized: MvCamera.MV_CC_Finalize() self.is_initialized = False print("SDK已清理") def __enter__(self): """支持with语句""" self.initialize() return self def __exit__(self, exc_type, exc_val, exc_tb): """支持with语句""" self.finalize() # 使用示例 if __name__ == "__main__": try: with HikCameraManager() as manager: # 1. 列出所有相机 manager.print_cameras() # 2. 选择相机的几种方式： # 方式1: 通过索引选择（默认第一个） # manager.open_camera(index=0) # 方式2: 通过序列号选择 # manager.open_camera(serial="DA7598570") # 方式3: 通过IP地址选择（GigE相机） manager.open_camera(ip="192.168.4.22") # 方式4: 通过型号选择 # manager.open_camera(model="MV-CS200") # 3. 采集图像 saved_files = manager.capture_image("test") print(f"\n保存的文件: {saved_files}") except Exception as e: print(f"错误: {e}") import traceback traceback.print_exc()