343 lines
9.9 KiB
Python
343 lines
9.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
import sys
|
||
import ctypes
|
||
import threading
|
||
import time
|
||
import os
|
||
|
||
# 添加SDK路径
|
||
sys.path.append('Python/MvImport')
|
||
from MvCameraControl_class import *
|
||
from MvErrorDefine_const import *
|
||
from CameraParams_header import *
|
||
import cv2
|
||
import numpy as np
|
||
|
||
# 初始化全局变量
|
||
global deviceList
|
||
deviceList = MV_CC_DEVICE_INFO_LIST()
|
||
global cam
|
||
cam = MvCamera()
|
||
global nSelCamIndex
|
||
nSelCamIndex = 0
|
||
global isOpen
|
||
isOpen = False
|
||
global isGrabbing
|
||
isGrabbing = False
|
||
global bExit
|
||
bExit = False
|
||
|
||
# 将返回的错误码转换为十六进制显示
|
||
def ToHexStr(num):
|
||
chaDic = {10: 'a', 11: 'b', 12: 'c', 13: 'd', 14: 'e', 15: 'f'}
|
||
hexStr = ""
|
||
if num < 0:
|
||
num = num + 2 ** 32
|
||
while num >= 16:
|
||
digit = num % 16
|
||
hexStr = chaDic.get(digit, str(digit)) + hexStr
|
||
num //= 16
|
||
hexStr = chaDic.get(num, str(num)) + hexStr
|
||
return hexStr
|
||
|
||
# Decoding Characters
|
||
def decoding_char(ctypes_char_array):
|
||
"""
|
||
安全地从 ctypes 字符数组中解码出字符串。
|
||
适用于 Python 2.x 和 3.x，以及 32/64 位环境。
|
||
"""
|
||
byte_str = memoryview(ctypes_char_array).tobytes()
|
||
|
||
# 在第一个空字符处截断
|
||
null_index = byte_str.find(b'\x00')
|
||
if null_index != -1:
|
||
byte_str = byte_str[:null_index]
|
||
|
||
# 多编码尝试解码
|
||
for encoding in ['gbk', 'utf-8', 'latin-1']:
|
||
try:
|
||
return byte_str.decode(encoding)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
|
||
# 如果所有编码都失败，使用替换策略
|
||
return byte_str.decode('latin-1', errors='replace')
|
||
|
||
# 枚举相机
|
||
def enum_devices():
|
||
global deviceList
|
||
|
||
deviceList = MV_CC_DEVICE_INFO_LIST()
|
||
n_layer_type = (MV_GIGE_DEVICE | MV_USB_DEVICE)
|
||
ret = MvCamera.MV_CC_EnumDevices(n_layer_type, deviceList)
|
||
if ret != 0:
|
||
print(f"枚举设备失败! ret = :{ToHexStr(ret)}")
|
||
return ret
|
||
|
||
if deviceList.nDeviceNum == 0:
|
||
print("未找到设备")
|
||
return ret
|
||
print(f"找到 {deviceList.nDeviceNum} 台设备!")
|
||
|
||
devList = []
|
||
for i in range(0, deviceList.nDeviceNum):
|
||
mvcc_dev_info = cast(deviceList.pDeviceInfo[i], POINTER(MV_CC_DEVICE_INFO)).contents
|
||
if mvcc_dev_info.nTLayerType == MV_GIGE_DEVICE:
|
||
print(f"\ngige device: [{i}]")
|
||
user_defined_name = decoding_char(mvcc_dev_info.SpecialInfo.stGigEInfo.chUserDefinedName)
|
||
model_name = decoding_char(mvcc_dev_info.SpecialInfo.stGigEInfo.chModelName)
|
||
print(f"device user define name: {user_defined_name}")
|
||
print(f"device model name: {model_name}")
|
||
|
||
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)
|
||
print(f"current ip: {nip1}.{nip2}.{nip3}.{nip4} ")
|
||
devList.append(f"[{i}]GigE: {user_defined_name} {model_name}({nip1}.{nip2}.{nip3}.{nip4})")
|
||
elif mvcc_dev_info.nTLayerType == MV_USB_DEVICE:
|
||
print(f"\nu3v device: [{i}]")
|
||
user_defined_name = decoding_char(mvcc_dev_info.SpecialInfo.stUsb3VInfo.chUserDefinedName)
|
||
model_name = decoding_char(mvcc_dev_info.SpecialInfo.stUsb3VInfo.chModelName)
|
||
print(f"device user define name: {user_defined_name}")
|
||
print(f"device model name: {model_name}")
|
||
|
||
strSerialNumber = ""
|
||
for per in mvcc_dev_info.SpecialInfo.stUsb3VInfo.chSerialNumber:
|
||
if per == 0:
|
||
break
|
||
strSerialNumber = strSerialNumber + chr(per)
|
||
print(f"user serial number: {strSerialNumber}")
|
||
devList.append(f"[{i}]USB: {user_defined_name} {model_name}({strSerialNumber})")
|
||
|
||
# 打印设备列表
|
||
print("\n设备列表:")
|
||
for i, dev in enumerate(devList):
|
||
print(f"{i}: {dev}")
|
||
|
||
return deviceList.nDeviceNum
|
||
|
||
# 打开相机
|
||
def open_device(nIndex):
|
||
global cam
|
||
global deviceList
|
||
global isOpen
|
||
|
||
if isOpen:
|
||
print("相机已经打开!")
|
||
return MV_E_CALLORDER
|
||
|
||
if nIndex < 0 or nIndex >= deviceList.nDeviceNum:
|
||
print("请选择有效的相机索引!")
|
||
return MV_E_CALLORDER
|
||
|
||
# 选择设备并创建句柄
|
||
stDeviceList = cast(deviceList.pDeviceInfo[nIndex], POINTER(MV_CC_DEVICE_INFO)).contents
|
||
cam = MvCamera()
|
||
ret = cam.MV_CC_CreateHandle(stDeviceList)
|
||
if ret != 0:
|
||
print(f"创建设备句柄失败! ret = {ToHexStr(ret)}")
|
||
cam.MV_CC_DestroyHandle()
|
||
return ret
|
||
|
||
# 打开设备
|
||
ret = cam.MV_CC_OpenDevice()
|
||
if ret != 0:
|
||
print(f"打开设备失败! ret = {ToHexStr(ret)}")
|
||
cam.MV_CC_DestroyHandle()
|
||
return ret
|
||
|
||
print("设备打开成功!")
|
||
isOpen = True
|
||
|
||
# 设置触发模式为off
|
||
ret = cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
|
||
if ret != 0:
|
||
print(f"设置触发模式失败! ret = {ToHexStr(ret)}")
|
||
|
||
# 设置像素格式为 BGR8
|
||
ret = cam.MV_CC_SetEnumValue("PixelFormat", PixelType_Gvsp_BGR8_Packed)
|
||
if ret != 0:
|
||
print(f"设置像素格式失败! ret = {ToHexStr(ret)}")
|
||
|
||
# 设置连续采集模式
|
||
ret = cam.MV_CC_SetEnumValue("AcquisitionMode", 2) # Continuous
|
||
if ret != 0:
|
||
print(f"设置连续采集模式失败! ret = {ToHexStr(ret)}")
|
||
|
||
return MV_OK
|
||
|
||
# 开始取流
|
||
def start_grabbing():
|
||
global cam
|
||
global isGrabbing
|
||
global bExit
|
||
|
||
if not isOpen:
|
||
print("相机未打开!")
|
||
return MV_E_CALLORDER
|
||
|
||
if isGrabbing:
|
||
print("已经开始取流!")
|
||
return MV_E_CALLORDER
|
||
|
||
bExit = False
|
||
ret = cam.MV_CC_StartGrabbing()
|
||
if ret != 0:
|
||
print(f"开始取流失败! ret = {ToHexStr(ret)}")
|
||
return ret
|
||
|
||
isGrabbing = True
|
||
print("开始取流成功!")
|
||
|
||
# 启动取图线程
|
||
global hThreadHandle
|
||
hThreadHandle = threading.Thread(target=work_thread)
|
||
hThreadHandle.daemon = True
|
||
hThreadHandle.start()
|
||
|
||
return MV_OK
|
||
|
||
# 停止取流
|
||
def stop_grabbing():
|
||
global cam
|
||
global isGrabbing
|
||
global bExit
|
||
|
||
if not isOpen:
|
||
print("相机未打开!")
|
||
return MV_E_CALLORDER
|
||
|
||
if not isGrabbing:
|
||
print("未开始取流!")
|
||
return MV_E_CALLORDER
|
||
|
||
bExit = True
|
||
time.sleep(0.1)
|
||
|
||
ret = cam.MV_CC_StopGrabbing()
|
||
if ret != 0:
|
||
print(f"停止取流失败! ret = {ToHexStr(ret)}")
|
||
return ret
|
||
|
||
isGrabbing = False
|
||
print("停止取流成功!")
|
||
|
||
return MV_OK
|
||
|
||
# 关闭相机
|
||
def close_device():
|
||
global cam
|
||
global isOpen
|
||
global isGrabbing
|
||
|
||
if isGrabbing:
|
||
stop_grabbing()
|
||
|
||
if isOpen:
|
||
ret = cam.MV_CC_CloseDevice()
|
||
if ret != 0:
|
||
print(f"关闭设备失败! ret = {ToHexStr(ret)}")
|
||
|
||
# 销毁句柄
|
||
cam.MV_CC_DestroyHandle()
|
||
isOpen = False
|
||
print("设备关闭成功!")
|
||
|
||
# 取图线程函数
|
||
def work_thread():
|
||
global cam
|
||
global bExit
|
||
|
||
stOutFrame = MV_FRAME_OUT()
|
||
memset(byref(stOutFrame), 0, sizeof(stOutFrame))
|
||
|
||
while not bExit:
|
||
# 获取图像
|
||
ret = cam.MV_CC_GetImageBuffer(stOutFrame, 1000)
|
||
|
||
if ret == 0:
|
||
# 打印图像信息
|
||
print(f"获取一帧图像: 宽度[{stOutFrame.stFrameInfo.nWidth}], 高度[{stOutFrame.stFrameInfo.nHeight}], 帧数[{stOutFrame.stFrameInfo.nFrameNum}]")
|
||
|
||
# 转换为 OpenCV 格式
|
||
try:
|
||
# 计算数据大小
|
||
data_size = stOutFrame.stFrameInfo.nWidth * stOutFrame.stFrameInfo.nHeight * 3
|
||
|
||
# 从缓冲区复制数据
|
||
frame_data = np.ctypeslib.as_array(stOutFrame.pBufAddr, shape=(data_size,))
|
||
frame_data = frame_data.reshape((stOutFrame.stFrameInfo.nHeight, stOutFrame.stFrameInfo.nWidth, 3))
|
||
|
||
# 显示图像
|
||
cv2.imshow("Camera", frame_data)
|
||
key = cv2.waitKey(1) & 0xFF
|
||
if key == ord('q'):
|
||
bExit = True
|
||
break
|
||
except Exception as e:
|
||
print(f"处理图像失败: {e}")
|
||
finally:
|
||
# 释放缓存
|
||
cam.MV_CC_FreeImageBuffer(stOutFrame)
|
||
else:
|
||
print(f"获取图像失败! ret = {ToHexStr(ret)}")
|
||
time.sleep(0.1)
|
||
|
||
cv2.destroyAllWindows()
|
||
|
||
# 主函数
|
||
def main():
|
||
print("海康相机验证程序")
|
||
print("按 Q 退出显示窗口")
|
||
print("按 Ctrl+C 退出程序")
|
||
|
||
# 初始化 SDK
|
||
MvCamera.MV_CC_Initialize()
|
||
|
||
try:
|
||
# 枚举设备
|
||
device_count = enum_devices()
|
||
if device_count == 0:
|
||
print("未找到设备，程序退出!")
|
||
return
|
||
|
||
# 选择设备
|
||
nIndex = int(input("请输入要打开的相机索引: "))
|
||
|
||
# 打开设备
|
||
ret = open_device(nIndex)
|
||
if ret != MV_OK:
|
||
print("打开设备失败，程序退出!")
|
||
return
|
||
|
||
# 开始取流
|
||
ret = start_grabbing()
|
||
if ret != MV_OK:
|
||
print("开始取流失败，程序退出!")
|
||
close_device()
|
||
return
|
||
|
||
# 等待用户操作
|
||
while True:
|
||
time.sleep(1)
|
||
if not isGrabbing:
|
||
break
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n用户中断程序")
|
||
except Exception as e:
|
||
print(f"程序异常: {e}")
|
||
finally:
|
||
# 清理资源
|
||
if isOpen:
|
||
close_device()
|
||
|
||
# 反初始化 SDK
|
||
MvCamera.MV_CC_Finalize()
|
||
print("SDK反初始化成功")
|
||
print("程序退出")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|