hik_driver.py

# -- coding: utf-8 --


import cv2
import numpy as np
import sys
import os
import logging

logger = logging.getLogger(__name__)

try:
    from MVS.Samples.aarch64.Python.MvImport.MvCameraControl_class import *
    HIKVISION_AVAILABLE = True
except Exception as e:
    logger.error(f"Failed to import HIK Vision camera driver: {e}")
    HIKVISION_AVAILABLE = False

g_bExit = False


def hik_init():
    if not HIKVISION_AVAILABLE:
        raise RuntimeError("HIKVision camera support is not available. The SDK library might be missing.")
    
    try:
        SDKVersion = MvCamera.MV_CC_GetSDKVersion()
        logger.info("SDKVersion[0x%x]" % SDKVersion)

        deviceList = MV_CC_DEVICE_INFO_LIST()
        tlayerType = MV_GIGE_DEVICE | MV_USB_DEVICE

        # ch:枚举设备 | en:Enum device
        ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
        if ret != 0:
            logger.error("enum devices fail! ret[0x%x]" % ret)
            raise RuntimeError(f"Failed to enumerate devices (error code: {ret})")

        if deviceList.nDeviceNum == 0:
            logger.error("find no device!")
            raise RuntimeError("No HIKVision devices found")

        logger.info("Find %d devices!" % deviceList.nDeviceNum)

        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("\ngige device: [%d]" % i)
                strModeName = ""
                for per in mvcc_dev_info.SpecialInfo.stGigEInfo.chModelName:
                    strModeName = strModeName + chr(per)
                print("device model name: %s" % strModeName)

                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("current ip: %d.%d.%d.%d\n" % (nip1, nip2, nip3, nip4))
            elif mvcc_dev_info.nTLayerType == MV_USB_DEVICE:
                print("\nu3v device: [%d]" % i)
                strModeName = ""
                for per in mvcc_dev_info.SpecialInfo.stUsb3VInfo.chModelName:
                    if per == 0:
                        break
                    strModeName = strModeName + chr(per)
                print("device model name: %s" % strModeName)

                strSerialNumber = ""
                for per in mvcc_dev_info.SpecialInfo.stUsb3VInfo.chSerialNumber:
                    if per == 0:
                        break
                    strSerialNumber = strSerialNumber + chr(per)
                print("user serial number: %s" % strSerialNumber)

        nConnectionNum = 0

        # ch:创建相机实例 | en:Creat Camera Object
        cam = MvCamera()

        # ch:选择设备并创建句柄| en:Select device and create handle
        stDeviceList = cast(deviceList.pDeviceInfo[int(
            nConnectionNum)], POINTER(MV_CC_DEVICE_INFO)).contents

        ret = cam.MV_CC_CreateHandle(stDeviceList)

        if ret != 0:
            print("create handle fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:打开设备 | en:Open device
        ret = cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
        if ret != 0:
            print("open device fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:开始取流 | en:Start grab image
        ret = cam.MV_CC_StartGrabbing()
        if ret != 0:
            print("start grabbing fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:获取数据包大小 | en:Get payload size
        stParam = MVCC_INTVALUE()
        memset(byref(stParam), 0, sizeof(MVCC_INTVALUE))

        ret = cam.MV_CC_GetIntValue("PayloadSize", stParam)
        if ret != 0:
            print("get payload size fail! ret[0x%x]" % ret)
            sys.exit()

        nPayloadSize = stParam.nCurValue
        data_buf = (c_ubyte * nPayloadSize)()  # image buffer
        stFrameInfo = MV_FRAME_OUT_INFO_EX()
        memset(byref(stFrameInfo), 0, sizeof(stFrameInfo))

        info_lst = [cam, data_buf, nPayloadSize, stFrameInfo]
        # check frame readability
        for i in range(5):
            ret = cam.MV_CC_GetOneFrameTimeout(
                data_buf, nPayloadSize, stFrameInfo, 5000) # modified the nMsec=1000 to 5000, 1000 failed frequently
            if ret != 0:
                print("pipline broke while testing frame readability", ret)
                sys.exit()

        return info_lst
        # -------------------------------------------
    except Exception as e:
        logger.error(f"Failed to initialize HIK Vision camera: {e}")
        raise


def read_hik_frame(info_lst):
    cam = info_lst[0]
    data_buf = info_lst[1]
    nPayloadSize = info_lst[2]
    stFrameInfo = info_lst[3]

    ret = cam.MV_CC_GetOneFrameTimeout(
        data_buf, nPayloadSize, stFrameInfo, 1000)
    if ret == 0:
        # print("get one frame: Width[%d], Height[%d], nFrameNum[%d], enPixelType[%d]" % (
        #     stFrameInfo.nWidth, stFrameInfo.nHeight, stFrameInfo.nFrameNum, stFrameInfo.enPixelType))

        image = np.asarray(data_buf).reshape(
            (stFrameInfo.nHeight, stFrameInfo.nWidth, -1))
        rgb_image = cv2.cvtColor(image, cv2.COLOR_BAYER_RG2RGB)

        # cv2.imshow("show", rgb_image)
        # k = cv2.waitKey(1) & 0xff
        return rgb_image
    else:
        print("no data[0x%x] --- Hik" % ret)


def hik_close(info_lst):
    cam = info_lst[0]
    data_buf = info_lst[1]
    # ch:停止取流 | en:Stop grab image
    ret = cam.MV_CC_StopGrabbing()
    if ret != 0:
        print("stop grabbing fail! ret[0x%x]" % ret)
        del data_buf
        sys.exit()

    # ch:关闭设备 | Close device
    ret = cam.MV_CC_CloseDevice()
    if ret != 0:
        print("close deivce fail! ret[0x%x]" % ret)
        del data_buf
        sys.exit()

    # ch:销毁句柄 | Destroy handle
    ret = cam.MV_CC_DestroyHandle()
    if ret != 0:
        print("destroy handle fail! ret[0x%x]" % ret)
        del data_buf
        sys.exit()
    del data_buf


# info_lst = hik_init()
# while True:
#     read_hik_frame(info_lst)
#
# hik_close(info_lst)