一、背景介紹

1、最近項(xiàng)目中需要給客戶對(duì)接?？禉C(jī)器人工業(yè)相機(jī) MV-CU060-10GM；
2、客戶要求通過部署的管理平臺(tái)，可以在頁面上實(shí)現(xiàn)如下功能：

1）相機(jī)視頻流開始預(yù)覽；
2）相機(jī)視頻流停止預(yù)覽；
3）相機(jī)拍照功能。

需求背景：客戶需要對(duì)生產(chǎn)的產(chǎn)品進(jìn)行定期抽樣質(zhì)檢，其中涉及到外觀檢測(cè)，比如，樣品的表面清潔度、外觀等指標(biāo)。所以，需要先通過管理平臺(tái)點(diǎn)擊相機(jī)的“預(yù)覽”按鈕，進(jìn)行預(yù)覽相機(jī)拍攝的實(shí)時(shí)效果，當(dāng)客戶認(rèn)為清晰度和角度滿足條件時(shí)，才會(huì)點(diǎn)擊“拍照”按鈕進(jìn)行拍攝，從而保證質(zhì)檢的照片是有意義和有實(shí)用價(jià)值的。

二、調(diào)研歷程

由于項(xiàng)目團(tuán)隊(duì)同事之前沒有做過工業(yè)相機(jī)視頻和拍照的相關(guān)開發(fā)，于是乎，就開啟了“漫長”而“煎熬”的調(diào)研之路（斷斷續(xù)續(xù)持續(xù)了1個(gè)多月）。

最終于2022年12月6日，通過 Python “完美”實(shí)現(xiàn)了上述的三個(gè)功能。
特地寫下此篇博客，供需要的網(wǎng)友參考，避免少走很多彎路。

1、?？倒I(yè)相機(jī)官網(wǎng)

https://www.hikrobotics.com/cn/machinevision/productdetail?id=8518&pageNumber=13&pageSize=20

2、官網(wǎng)示例

可以在?？禉C(jī)器人官網(wǎng)提供的客戶端工具 MVS，“幫助”–> “Development”，點(diǎn)擊“Development”會(huì)跳轉(zhuǎn)到安裝目錄，從“Samples”中獲取官方提供的一些簡單示例。

如果已經(jīng)安裝了 MVS，直接進(jìn)入 C:\Program Files (x86)\MVS\Development\Samples 目錄即可看到，目前支持 C#、C++、Java、OpenCV、Python、VB等語言。

本人主要使用Java和Python，所以，本篇博文主要從Java和Python兩種語言調(diào)研了實(shí)現(xiàn)方案。

1）Java示例

SaveImage 有獲取圖片的示例，但是，沒有視頻流獲取并顯示的示例。（參考示例相對(duì)較少）

2）Python示例

GrabImage 目錄下有獲取圖片的示例，Recording 目錄下有獲取視頻流的示例，但是沒有將視頻流返回前端的示例。（參考示例相對(duì)較多）

3、網(wǎng)上博客參考

1）RTSP（Runtime Stream Protocol）協(xié)議方向

（很遺憾，此路不通?。。。?br> 因?yàn)槲覀冇玫?海康機(jī)器人工業(yè)相機(jī) MV-CU060-10GM 這款相機(jī)，不支持 RTSP 協(xié)議。

如下博客適用于 ?？低晹z像頭，并不適用于 海康工業(yè)相機(jī)，如果是使用?？低晹z像頭的小伙伴可以參考下。

參考博客：?？低晹z像頭對(duì)接SDK實(shí)時(shí)預(yù)覽功能和抓拍功能，懶癌福利，可直接CV

2）Java實(shí)現(xiàn)方向

（不能完全滿足客戶需求，此路不全通?。。。?/p>

使用Java目前參考官網(wǎng)的示例，實(shí)現(xiàn)了圖片抓取，并上傳的功能，但是沒有實(shí)現(xiàn)視頻流的實(shí)時(shí)獲取和顯示的功能。

如果需求只是獲取圖片，不要求視頻流實(shí)時(shí)顯示，可以通過Java就可以實(shí)現(xiàn)。

3）Python實(shí)現(xiàn)方向

（目前，網(wǎng)上沒有直接能完全滿足上述三個(gè)需求的，本人通過借鑒、整合，結(jié)合Flask框架實(shí)現(xiàn)的。）

參考博客如下：
python語言下使用opencv接口cv2.VideoCapture()接口調(diào)用?？禉C(jī)器人工業(yè)相機(jī) （此篇博文可以重點(diǎn)看?。。。?/p>

通過python調(diào)用?？低暪I(yè)攝像頭并進(jìn)行圖像存儲(chǔ)，同時(shí)使用opencv實(shí)時(shí)圖像顯示（數(shù)據(jù)流問題已解決）

python調(diào)用?？倒I(yè)相機(jī)并用opencv顯示（整體實(shí)現(xiàn)）（此篇博文可以重點(diǎn)看?。。。?/p>

pyQT5 學(xué)習(xí)使用 筆記 六 pyQt5+opencv 顯示海康GIGE相機(jī)動(dòng)態(tài)視頻流 (該方式雖然實(shí)現(xiàn)了視頻的實(shí)時(shí)顯示，但是，無法被給前端直接調(diào)用)

web實(shí)時(shí)顯示攝像頭圖像（python） （此篇博文可以重點(diǎn)看！?。。?/p>

4）Flask 的相關(guān)教程和博客

• https://www.w3cschool.cn/flask/
• https://dormousehole.readthedocs.io/en/latest/
• https://blog.csdn.net/weixin_44239541/article/details/89390139
• https://zhuanlan.zhihu.com/p/104273184
• https://blog.csdn.net/tulan_xiaoxin/article/details/79132214

三、Python 代碼實(shí)現(xiàn)

1、Python 環(huán)境

1）Python 版本

個(gè)人使用的是 Python 3.8.5 版本（建議使用該版本或者更高版本）
Python 官網(wǎng)地址：https://www.python.org/
安裝和配置可以參考：https://www.runoob.com/python3/python3-install.html

2）Python 虛擬環(huán)境

個(gè)人建議給該工程單獨(dú)創(chuàng)建一個(gè)Python虛擬環(huán)境（如：hikrobotEnv），后續(xù)現(xiàn)場(chǎng)如果出現(xiàn)無法連接外網(wǎng)的情況下，可以直接Copy虛擬環(huán)境部署，比較方便。
當(dāng)然，如果不是特殊情況，多個(gè)項(xiàng)目工程可以共用一個(gè)虛擬環(huán)境，可以使用 virtualenv、anaconda、PyCharm 等創(chuàng)建虛擬環(huán)境。
以 virtualenv 為例，創(chuàng)建 hikrobotEnv 虛擬環(huán)境的命令如下：

virtualenv hikrobotEnv --python=python3.8.5 

如果沒有安裝 virtualenv，可以通過如下命令安裝：

pip install virtualenv

關(guān)于 virtualenv 的相關(guān)介紹和使用，可以參考博客：Python虛擬環(huán)境Virtualenv詳解

2、測(cè)試代碼

1）抓取圖片測(cè)試代碼

前提條件：將 *C:\Program Files (x86)\MVS\Development\Samples\Python* 目錄下的 MvImport 目錄，copy到自己的工程目錄下，創(chuàng)建測(cè)試文件 TestGrabImage.py

代碼如下：

# -- coding: utf-8 --
import cv2
import sys
import copy
import msvcrt
import numpy as np

from ctypes import *

sys.path.append("./MvImport")
from MvCameraControl_class import *

if __name__ == "__main__":

    deviceList = MV_CC_DEVICE_INFO_LIST()
    tlayerType = MV_GIGE_DEVICE | MV_USB_DEVICE

    # ch:枚舉設(shè)備 | en:Enum device
    ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
    if ret != 0:
        print ("enum devices fail! ret[0x%x]" % ret)
        sys.exit()

    if deviceList.nDeviceNum == 0:
        print ("find no device!")
        sys.exit()

    print ("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

    if int(nConnectionNum) >= deviceList.nDeviceNum:
        print ("intput error!")
        sys.exit()

    # ch:創(chuàng)建相機(jī)實(shí)例 | en:Creat Camera Object
    cam = MvCamera()

    # ch:選擇設(shè)備并創(chuàng)建句柄 | 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:打開設(shè)備 | 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:探測(cè)網(wǎng)絡(luò)最佳包大小(只對(duì)GigE相機(jī)有效) | en:Detection network optimal package size(It only works for the GigE camera)
    if stDeviceList.nTLayerType == MV_GIGE_DEVICE:
        nPacketSize = cam.MV_CC_GetOptimalPacketSize()
        if int(nPacketSize) > 0:
            ret = cam.MV_CC_SetIntValue("GevSCPSPacketSize", nPacketSize)
            if ret != 0:
                print ("Warning: Set Packet Size fail! ret[0x%x]" % ret)
        else:
            print ("Warning: Get Packet Size fail! ret[0x%x]" % nPacketSize)

    # ch:設(shè)置觸發(fā)模式為off | en:Set trigger mode as off
    ret = cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
    if ret != 0:
        print ("set trigger mode fail! ret[0x%x]" % ret)
        sys.exit()

    # ch:獲取數(shù)據(jù)包大小 | 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

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

    stDeviceList = MV_FRAME_OUT_INFO_EX()
    memset(byref(stDeviceList), 0, sizeof(stDeviceList))
    data_buf = (c_ubyte * nPayloadSize)()

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

        nRGBSize = stDeviceList.nWidth * stDeviceList.nHeight * 3
        stConvertParam=MV_SAVE_IMAGE_PARAM_EX()
        stConvertParam.nWidth = stDeviceList.nWidth
        stConvertParam.nHeight = stDeviceList.nHeight
        stConvertParam.pData = data_buf
        stConvertParam.nDataLen = stDeviceList.nFrameLen
        stConvertParam.enPixelType = stDeviceList.enPixelType
        stConvertParam.nImageLen = stConvertParam.nDataLen
        stConvertParam.nJpgQuality = 70
        stConvertParam.enImageType = MV_Image_Jpeg
        stConvertParam.pImageBuffer = (c_ubyte * nRGBSize)()
        stConvertParam.nBufferSize = nRGBSize
        # ret = cam.MV_CC_ConvertPixelType(stConvertParam)
        print(stConvertParam.nImageLen)
        ret = cam.MV_CC_SaveImageEx2(stConvertParam)
        if ret != 0:
            print ("convert pixel fail ! ret[0x%x]" % ret)
            del data_buf
            sys.exit()
        file_path = "AfterConvert_RGB2.jpg"
        file_open = open(file_path.encode('ascii'), 'wb+')
        img_buff = (c_ubyte * stConvertParam.nImageLen)()
        cdll.msvcrt.memcpy(byref(img_buff), stConvertParam.pImageBuffer, stConvertParam.nImageLen)
        file_open.write(img_buff)
    print ("Save Image succeed!")


    # 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:關(guān)閉設(shè)備 | 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

在python 虛擬環(huán)境中，執(zhí)行 python TestGrabImage.py 運(yùn)行后，會(huì)在當(dāng)前目錄下生成一個(gè)名為 AfterConvert_RGB2.jpg 的圖片文件。

如果運(yùn)行過程中提示模塊不存在，可以通過 pip install 命令安裝相應(yīng)的模塊，建議使用清華源，如 安裝 opencv-python，命令如下：

pip install opencv-python==4.1.2.30 -i https://pypi.tuna.tsinghua.edu.cn/simple

2）Python+Qt 實(shí)現(xiàn)視頻流實(shí)時(shí)顯示測(cè)試代碼

前提條件：將 *C:\Program Files (x86)\MVS\Development\Samples\Python* 目錄下的 MvImport 目錄，copy到自己的工程目錄下，創(chuàng)建測(cè)試文件 TestVideoStream.py

代碼如下：

# -- coding: utf-8 --
import cv2
from PyQt5 import QtCore, QtGui, QtWidgets
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtGui import *
import numpy as np
 
#from CameraControl_header import MV_CC_DEVICE_INFO_LIST
#from mainWindow import Ui_MainWindow  # 導(dǎo)入創(chuàng)建的GUI類
import sys
import threading
import msvcrt
from ctypes import *

sys.path.append("./MvImport")
from MvCameraControl_class import *
from Ui_MainWindow  import *
from CameraParams_header import *
 
class mywindow(QtWidgets.QMainWindow, Ui_MainWindow):
    sendAddDeviceName = pyqtSignal() #定義一個(gè)添加設(shè)備列表的信號(hào)。
    deviceList = MV_CC_DEVICE_INFO_LIST()
    g_bExit = False
    # ch:創(chuàng)建相機(jī)實(shí)例 | en:Creat Camera Object
    cam = MvCamera()
 
    def connect_and_emit_sendAddDeviceName(self):
        # Connect the sendAddDeviceName signal to a slot.
        self.sendAddDeviceName.connect(self.SelectDevice)
        # Emit the signal.
        self.sendAddDeviceName.emit()
 
    def __init__(self):
        super(mywindow, self).__init__()
        self.setupUi(self)
        self.connect_and_emit_sendAddDeviceName()
        self.butopenCam.clicked.connect(lambda:self.openCam(self.camSelect.currentData()))
        self.butcloseCam.clicked.connect(self.closeCam)
 
        # setting main window geometry
        desktop_geometry = QtWidgets.QApplication.desktop()  # 獲取屏幕大小
        main_window_width = desktop_geometry.width()  # 屏幕的寬
        main_window_height = desktop_geometry.height()  # 屏幕的高
        rect = self.geometry()  # 獲取窗口界面大小
        window_width = rect.width()  # 窗口界面的寬
        window_height = rect.height()  # 窗口界面的高
        x = (main_window_width - window_width) // 2  # 計(jì)算窗口左上角點(diǎn)橫坐標(biāo)
        y = (main_window_height - window_height) // 2  # 計(jì)算窗口左上角點(diǎn)縱坐標(biāo)
        self.setGeometry(x, y, window_width, window_height)  # 設(shè)置窗口界面在屏幕上的位置
        # 無邊框以及背景透明一般不會(huì)在主窗口中用到，一般使用在子窗口中，例如在子窗口中顯示gif提示載入信息等等
       # self.setWindowFlags(Qt.FramelessWindowHint)
       # self.setAttribute(Qt.WA_TranslucentBackground)
 
    #打開攝像頭。
    def openCam(self,camid):
        self.g_bExit = False
        # ch:選擇設(shè)備并創(chuàng)建句柄 | en:Select device and create handle
        stDeviceList = cast(self.deviceList.pDeviceInfo[int(camid)], POINTER(MV_CC_DEVICE_INFO)).contents
        ret = self.cam.MV_CC_CreateHandle(stDeviceList)
        if ret != 0:
            print("create handle fail! ret[0x%x]" % ret)
            sys.exit()
        # ch:打開設(shè)備 | en:Open device
 
        ret = self.cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
        if ret != 0:
            print("open device fail! ret[0x%x]" % ret)
            sys.exit()
 
        # ch:探測(cè)網(wǎng)絡(luò)最佳包大小(只對(duì)GigE相機(jī)有效) | en:Detection network optimal package size(It only works for the GigE camera)
        if stDeviceList.nTLayerType == MV_GIGE_DEVICE:
            nPacketSize = self.cam.MV_CC_GetOptimalPacketSize()
            if int(nPacketSize) > 0:
                ret = self.cam.MV_CC_SetIntValue("GevSCPSPacketSize", nPacketSize)
                if ret != 0:
                    print("Warning: Set Packet Size fail! ret[0x%x]" % ret)
            else:
                print("Warning: Get Packet Size fail! ret[0x%x]" % nPacketSize)
 
        # ch:設(shè)置觸發(fā)模式為off | en:Set trigger mode as off
        ret = self.cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
        if ret != 0:
            print("set trigger mode fail! ret[0x%x]" % ret)
            sys.exit()
            # ch:獲取數(shù)據(jù)包大小 | en:Get payload size
        stParam = MVCC_INTVALUE()
        memset(byref(stParam), 0, sizeof(MVCC_INTVALUE))
 
        ret = self.cam.MV_CC_GetIntValue("PayloadSize", stParam)
        if ret != 0:
            print("get payload size fail! ret[0x%x]" % ret)
            sys.exit()
        nPayloadSize = stParam.nCurValue
 
        # ch:開始取流 | en:Start grab image
        ret = self.cam.MV_CC_StartGrabbing()
        if ret != 0:
            print("start grabbing fail! ret[0x%x]" % ret)
            sys.exit()
 
        data_buf = (c_ubyte * nPayloadSize)()
 
        try:
            hThreadHandle = threading.Thread(target=self.work_thread, args=(self.cam, data_buf, nPayloadSize))
            hThreadHandle.start()
        except:
            print("error: unable to start thread")
 
    #關(guān)閉相機(jī)
    def closeCam(self):
        self.g_bExit=True
        # ch:停止取流 | en:Stop grab image
        ret = self.cam.MV_CC_StopGrabbing()
        if ret != 0:
            print("stop grabbing fail! ret[0x%x]" % ret)
            sys.exit()
 
        # ch:關(guān)閉設(shè)備 | Close device
        ret = self.cam.MV_CC_CloseDevice()
        if ret != 0:
            print("close deivce fail! ret[0x%x]" % ret)
 
        # ch:銷毀句柄 | Destroy handle
        ret = self.cam.MV_CC_DestroyHandle()
        if ret != 0:
            print("destroy handle fail! ret[0x%x]" % ret)
 
    def work_thread(self,cam=0, pData=0, nDataSize=0):
        stFrameInfo = MV_FRAME_OUT_INFO_EX()
        memset(byref(stFrameInfo), 0, sizeof(stFrameInfo))
        while True:
            QIm = np.asarray(pData)  # 將c_ubyte_Array轉(zhuǎn)化成ndarray得到（3686400，）
            QIm = QIm.reshape((2048, 3072, 1))  # 根據(jù)自己分辨率進(jìn)行轉(zhuǎn)化
                # print(temp)
                # print(temp.shape)
            QIm = cv2.cvtColor(QIm, cv2.COLOR_BGR2RGB)  # 這一步獲取到的顏色不對(duì)，因?yàn)槟J(rèn)是BRG，要轉(zhuǎn)化成RGB，顏色才正常
            pyrD1=cv2.pyrDown(QIm) #向下取樣
            pyrD2 = cv2.pyrDown(pyrD1)  # 向下取樣
            image_height, image_width, image_depth = pyrD2.shape  # 讀取圖像高寬深度
            pyrD3 = QImage(pyrD2, image_width, image_height,  image_width * image_depth,QImage.Format_RGB888)
            self.label.setPixmap(QPixmap.fromImage(pyrD3))
            #cv2.namedWindow("result", cv2.WINDOW_AUTOSIZE)
            #cv2.imshow("result", temp)
            #if cv2.waitKey(1) & 0xFF == ord('q'):
            #    break
 
            ret = cam.MV_CC_GetOneFrameTimeout(pData, nDataSize, stFrameInfo, 1000)
            if ret == 0:
                print("get one frame: Width[%d], Height[%d], nFrameNum[%d]" % (
                    stFrameInfo.nWidth, stFrameInfo.nHeight, stFrameInfo.nFrameNum))
            else:
                print("no data[0x%x]" % ret)
            if self.g_bExit == True:
                del pData
                break
 
 
    #獲得所有相機(jī)的列表存入cmbSelectDevice中
    def SelectDevice(self):
        '''選擇所有能用的相機(jī)到列表中，
             gige相機(jī)需要配合 sdk 得到。
        '''
        #得到相機(jī)列表
 
        tlayerType = MV_GIGE_DEVICE | MV_USB_DEVICE
    # ch:枚舉設(shè)備 | en:Enum device
        ret = MvCamera.MV_CC_EnumDevices(tlayerType, self.deviceList)
        if ret != 0:
            print("enum devices fail! ret[0x%x]" % ret)
            sys.exit()
        if self.deviceList.nDeviceNum == 0:
            print("find no device!")
            sys.exit()
 
        print("Find %d devices!" % self.deviceList.nDeviceNum)
        for i in range(0, self.deviceList.nDeviceNum):
            mvcc_dev_info = cast(self.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)
                self.camSelect.addItem(strModeName,i) #寫入設(shè)備列表。
 
    def pushbutton_function(self):
        #do some things
        Img=cv2.imread('JP1.JPG') #通過opencv讀入一張圖片
        image_height, image_width, image_depth=Img.shape #讀取圖像高寬深度
        QIm=cv2.cvtColor(Img,cv2.COLOR_BGR2RGB)
        QIm=QImage(QIm.data, image_width, image_height,  image_width * image_depth,QImage.Format_RGB888)
        self.label.setPixmap(QPixmap.fromImage(QIm))
 
if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    window = mywindow()
    window.show()
    sys.exit(app.exec_())

上述代碼中用到的 Ui_MainWindow.py，放到 MvImport 目錄下。

代碼如下：

# -*- coding: utf-8 -*-
 
# Form implementation generated from reading ui file 'mainWindow.ui'
#
# Created by: PyQt5 UI code generator 5.15.0
#
# WARNING: Any manual changes made to this file will be lost when pyuic5 is
# run again.  Do not edit this file unless you know what you are doing.
 
 
from PyQt5 import QtCore, QtGui, QtWidgets
 
 
class Ui_MainWindow(object):
    def setupUi(self, MainWindow):
        MainWindow.setObjectName("MainWindow")
        MainWindow.resize(589, 530)
        self.centralwidget = QtWidgets.QWidget(MainWindow)
        self.centralwidget.setObjectName("centralwidget")
        self.verticalLayout = QtWidgets.QVBoxLayout(self.centralwidget)
        self.verticalLayout.setObjectName("verticalLayout")
        self.camSelect = QtWidgets.QComboBox(self.centralwidget)
        self.camSelect.setObjectName("camSelect")
        self.verticalLayout.addWidget(self.camSelect)
        self.label = QtWidgets.QLabel(self.centralwidget)
        self.label.setObjectName("label")
        self.verticalLayout.addWidget(self.label)
        self.butopenCam = QtWidgets.QPushButton(self.centralwidget)
        self.butopenCam.setObjectName("butopenCam")
        self.verticalLayout.addWidget(self.butopenCam)
        self.butcloseCam = QtWidgets.QPushButton(self.centralwidget)
        self.butcloseCam.setObjectName("butcloseCam")
        self.verticalLayout.addWidget(self.butcloseCam)
        MainWindow.setCentralWidget(self.centralwidget)
        self.menubar = QtWidgets.QMenuBar(MainWindow)
        self.menubar.setGeometry(QtCore.QRect(0, 0, 589, 23))
        self.menubar.setObjectName("menubar")
        MainWindow.setMenuBar(self.menubar)
        self.statusbar = QtWidgets.QStatusBar(MainWindow)
        self.statusbar.setObjectName("statusbar")
        MainWindow.setStatusBar(self.statusbar)
 
        self.retranslateUi(MainWindow)
        QtCore.QMetaObject.connectSlotsByName(MainWindow)
 
    def retranslateUi(self, MainWindow):
        _translate = QtCore.QCoreApplication.translate
        MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
        self.label.setText(_translate("MainWindow", "TextLabel"))
        self.butopenCam.setText(_translate("MainWindow", "打開相機(jī)"))
        self.butcloseCam.setText(_translate("MainWindow", "關(guān)閉相機(jī)"))

在python 虛擬環(huán)境中，執(zhí)行 python TestVideoStream.py 運(yùn)行，可以看到如下視頻中的效果：

如果運(yùn)行過程中提示模塊不存在，可以通過 pip install 命令安裝相應(yīng)的模塊，建議使用清華源，如 安裝 opencv-python，命令如下：

pip install opencv-python==4.1.2.30 -i https://pypi.tuna.tsinghua.edu.cn/simple

通過視頻中的效果可以看出：雖然實(shí)現(xiàn)了視頻流的實(shí)時(shí)獲取和顯示，但是無法直接對(duì)接前端展示，不夠友好。

四、最終代碼（完整版）

之前的兩個(gè)測(cè)試代碼，僅作為測(cè)試參考，并非最終代碼?。?！

此段落中的代碼，才是最終代碼，可以重點(diǎn)參考！??！

使用 Python + OpenCV + Flask 實(shí)現(xiàn)，可以滿足視頻實(shí)時(shí)獲取，并返回通過GET請(qǐng)求返回給前端進(jìn)行實(shí)時(shí)顯示，也可以抓取圖片，保存上傳。

1、添加依賴文件

將 *C:\Program Files (x86)\MVS\Development\Samples\Python* 目錄下的 MvImport 目錄，copy到自己的工程目錄下。

2、安裝插件 DirectShow

1）進(jìn)入第三方插件 DirectShow 路徑

cd C:\Program Files (x86)\MVS\Development\ThirdPartyPlatformAdapter\DirectShow\x64\MvDSS2

2）安裝 DirectShow

使用 管理員權(quán)限 運(yùn)行 InstallDSSvc_x64.bat

說明：本人使用的是64位Windows操作系統(tǒng)，運(yùn)行代碼也是基于64位運(yùn)行，故而使用該版本；32位系統(tǒng)在上一目錄中也存在，大家可以根據(jù)實(shí)際情況進(jìn)行安裝。

3、安裝依賴模塊

1）安裝 opencv-python

pip install opencv-python==4.1.2.30 -i https://pypi.tuna.tsinghua.edu.cn/simple

個(gè)人使用的是 Python 3.8.5，opencv-python 對(duì)應(yīng)使用 4.1.2.30 版本即可。
如果使用的 opencv-python 的版本過高，可能報(bào)如下錯(cuò)誤：

cv2.error: Unknown C++ exception from OpenCV code.

2）安裝 Flask

pip install flask -i https://pypi.tuna.tsinghua.edu.cn/simple

也可以指定 Flask 的版本，如：

pip install flask=2.2.2 -i https://pypi.tuna.tsinghua.edu.cn/simple

4、核心代碼

工程代碼結(jié)構(gòu)如下：

0）index.html

在工程中創(chuàng)建 templates 目錄，用于存放 index.hml

<html>
  <head>
  </head>
  <body>
      <h1>拍照預(yù)覽</h1>
    <img src="{{ url_for('startPreview') }}" width="50%">
  </body>
</html>

1）JsonResponse.py

用于規(guī)范返回給前端的數(shù)據(jù)類型，代碼如下：

# -- coding: utf-8 --

class JsonResponse(object):
    """
    統(tǒng)一的json返回格式
    """

    def __init__(self, code, msg, data):
        self.code = code
        self.msg = msg
        self.data = data

    @classmethod
    def success(cls, code=0, msg='success', data=None):
        return cls(code, msg, data)

    @classmethod
    def error(cls, code=-1, msg='error', data=None):
        return cls(code, msg, data)

    def to_dict(self):
        return {
            "code": self.code,
            "msg": self.msg,
            "data": self.data
        }

2）JsonFlask.py

用于重定義數(shù)據(jù)返回格式，代碼如下：

# -- coding: utf-8 --

from flask import Flask, jsonify
from JsonResponse import *

class JsonFlask(Flask):
    def make_response(self, rv):
        """視圖函數(shù)可以直接返回: list、dict、None"""
        if rv is None or isinstance(rv, (list, dict)):
            rv = JsonResponse.success(rv)

        if isinstance(rv, JsonResponse):
            rv = jsonify(rv.to_dict())

        return super().make_response(rv)

3）HikRobotCamera.py

核心代碼文件，實(shí)現(xiàn)如下功能：
開始預(yù)覽視頻流、停止預(yù)覽視頻流、獲取圖片、記錄日志等功能。

代碼如下：

# -- coding: utf-8 --
import cv2
from flask import Flask, render_template, Response

import sys
import msvcrt
import base64
import datetime
import logging

sys.path.append("./MvImport")
from MvCameraControl_class import *
from JsonResponse import *
from JsonFlask import *

logging.basicConfig(level=logging.DEBUG,#控制臺(tái)打印的日志級(jí)別
                    filename='hikrobot.log',
                    filemode='a',##模式，有w和a，w就是寫模式，每次都會(huì)重新寫日志，覆蓋之前的日志
                    #a是追加模式，默認(rèn)如果不寫的話，就是追加模式
                    format=
                    '%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s'
                    #日志格式
                    )

# 這里配置一下 template_folder為當(dāng)前目錄，不然可以找不到 index.html
app = JsonFlask(__name__, template_folder='.')

# index
@app.route('/')
def index():
    return render_template('./templates/index.html')

# 獲取碼流
def generate(cap):
    # 捕獲異常信息
    try:
        while True:
            # 如果是關(guān)閉相機(jī)，先退出取視頻流的循環(huán)
            global open
            if (not open):
                break;
            retgrab = cap.grab()
            if retgrab == True:
                logging.debug("Grab true")
            ret1, frame = cap.retrieve()
            # print(type(frame))
            if frame is None:
                logging.error("frame is None")
                continue
            ret1, jpeg = cv2.imencode('.jpg', frame)
            jpg_frame = jpeg.tobytes()
            yield (b'--frame\r\n'
                b'Content-Type: image/jpeg\r\n\r\n' + jpg_frame + b'\r\n')
    except Exception as e:
        logging.error("generate error: %s" % str(e))


# 開始預(yù)覽
@app.route('/startPreview')
def startPreview():
    logging.info("======================================")
    logging.info("start to preview video stream, current_time: " + str(datetime.datetime.now()))
    # 全局變量，用于控制獲取視頻流的開關(guān)狀態(tài)
    global open
    open = True

    # 全局變量，獲取視頻連接
    global cap
    cap = cv2.VideoCapture(1)

    if False == cap.isOpened():
        logging.error("can't open camera")
        quit()
    else:
        logging.info("start to open camera")

    logging.info("open camera ok")

    # 分辨率設(shè)置 3072*2048（海康機(jī)器人工業(yè)相機(jī) MV-CU060-10GM）
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, 3072)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 2048)
    # 幀率配置
    cap.set(cv2.CAP_PROP_FPS, 15)   
    return Response(generate(cap), mimetype='multipart/x-mixed-replace;boundary=frame')

# 停止預(yù)覽
@app.route('/stopPreview')
def stopPreview():
    logging.info("======================================")
    logging.info("stop to preview video stream, current_time: " + str(datetime.datetime.now()))
    logging.info("start to close camera")

    # 全局變量，用于停止循環(huán)
    global open
    open = False

    logging.info("release resources start")
    # 全局變量，用于釋放相機(jī)資源
    try:
        global cap
        cap.release()
        cv2.destroyAllWindows()
    except Exception as e:
        logging.error("stopPreview error: %s" % str(e))
    logging.info("release resources end")
    logging.info("camera closed successfully, current_time: " + str(datetime.datetime.now()))
    logging.info("======================================")
    return "stop to preview"


@app.route('/openAndSave')
def openAndSave():
    logging.info("======================================")
    logging.info("start to grab image, current_time: " + str(datetime.datetime.now()))
    code = 100000
    msg = "連接相機(jī)時(shí)發(fā)生錯(cuò)誤"
    # img_base64 = None
    try:
        deviceList = MV_CC_DEVICE_INFO_LIST()
        tlayerType = MV_GIGE_DEVICE | MV_USB_DEVICE

        # ch:枚舉設(shè)備 | en:Enum device
        ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
        if ret != 0:
            logging.error("enum devices fail! ret[0x%x]" % ret)
            sys.exit()

        if deviceList.nDeviceNum == 0:
            logging.error("find no device!")
            sys.exit()

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

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

        nConnectionNum = 0

        if int(nConnectionNum) >= deviceList.nDeviceNum:
            logging.error("intput error!")
            sys.exit()

        # ch:創(chuàng)建相機(jī)實(shí)例 | en:Creat Camera Object
        cam = MvCamera()

        # ch:選擇設(shè)備并創(chuàng)建句柄 | 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:
            logging.error("create handle fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:打開設(shè)備 | en:Open device
        ret = cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
        if ret != 0:
            logging.error("open device fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:探測(cè)網(wǎng)絡(luò)最佳包大小(只對(duì)GigE相機(jī)有效) | en:Detection network optimal package size(It only works for the GigE camera)
        if stDeviceList.nTLayerType == MV_GIGE_DEVICE:
            nPacketSize = cam.MV_CC_GetOptimalPacketSize()
            if int(nPacketSize) > 0:
                ret = cam.MV_CC_SetIntValue("GevSCPSPacketSize", nPacketSize)
                if ret != 0:
                    logging.warn("Warning: Set Packet Size fail! ret[0x%x]" % ret)
            else:
                logging.warn("Warning: Get Packet Size fail! ret[0x%x]" % nPacketSize)

        # ch:設(shè)置觸發(fā)模式為off | en:Set trigger mode as off
        ret = cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
        if ret != 0:
            logging.error("set trigger mode fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:獲取數(shù)據(jù)包大小 | en:Get payload size
        stParam = MVCC_INTVALUE()
        memset(byref(stParam), 0, sizeof(MVCC_INTVALUE))

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

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

        stDeviceList = MV_FRAME_OUT_INFO_EX()
        memset(byref(stDeviceList), 0, sizeof(stDeviceList))
        data_buf = (c_ubyte * nPayloadSize)()
        
        ret = cam.MV_CC_GetOneFrameTimeout(byref(data_buf), nPayloadSize, stDeviceList, 1000)
        if ret == 0:
            logging.info("get one frame: Width[%d], Height[%d], nFrameNum[%d]" % (stDeviceList.nWidth, stDeviceList.nHeight, stDeviceList.nFrameNum))

            nRGBSize = stDeviceList.nWidth * stDeviceList.nHeight * 3
            stConvertParam=MV_SAVE_IMAGE_PARAM_EX()
            stConvertParam.nWidth = stDeviceList.nWidth
            stConvertParam.nHeight = stDeviceList.nHeight
            stConvertParam.pData = data_buf
            stConvertParam.nDataLen = stDeviceList.nFrameLen
            stConvertParam.enPixelType = stDeviceList.enPixelType
            stConvertParam.nImageLen = stConvertParam.nDataLen
            stConvertParam.nJpgQuality = 70
            stConvertParam.enImageType = MV_Image_Jpeg
            stConvertParam.pImageBuffer = (c_ubyte * nRGBSize)()
            stConvertParam.nBufferSize = nRGBSize
            # ret = cam.MV_CC_ConvertPixelType(stConvertParam)
            logging.info("nImageLen: %d" % stConvertParam.nImageLen)
            ret = cam.MV_CC_SaveImageEx2(stConvertParam)
            if ret != 0:
                logging.error("convert pixel fail ! ret[0x%x]" % ret)
                del data_buf
                sys.exit()
            #file_path = "AfterConvert_RGB2.jpg"
            #file_open = open(file_path, 'wb+')
            #file_open = open(file_path.encode('utf8'), 'wb')
            img_buff = (c_ubyte * stConvertParam.nImageLen)()
            cdll.msvcrt.memcpy(byref(img_buff), stConvertParam.pImageBuffer, stConvertParam.nImageLen)
            #file_open.write(img_buff)
            
            # 對(duì)返回的圖片進(jìn)行 base64 格式轉(zhuǎn)換
            img_base64 = "data:image/jpg;base64," + str(base64.b64encode(img_buff)).split("'")[1]
            code = 200
            msg = "success"
        logging.info("Save Image succeed!")

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

        # ch:關(guān)閉設(shè)備 | Close device
        ret = cam.MV_CC_CloseDevice()
        if ret != 0:
            logging.error("close deivce fail! ret[0x%x]" % ret)
            del data_buf
            sys.exit()

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

        del data_buf
    except Exception as e:
        logging.error("openAndSave error: %s" % str(e))
    # print("openAndSave finished, current_time: " + str(datetime.datetime.now()))  
  
    return JsonResponse(code, msg, img_base64)

# 執(zhí)行web服務(wù), 端口號(hào)可自行修訂
logging.info("start to run camera app, current_time: " + str(datetime.datetime.now()))
app.run(host='0.0.0.0', port=65432, debug=True, threaded=True)

5、運(yùn)行代碼

進(jìn)入Python虛擬環(huán)境，執(zhí)行如下命令：

python HikRobotCamera.py

6、功能驗(yàn)證

1）開始預(yù)覽

可以將 http://127.0.0.1:65432/startPreview 在前端用 img 標(biāo)簽 顯示視頻的實(shí)時(shí)預(yù)覽效果。

http://127.0.0.1:65432/startPreview

2）停止預(yù)覽

由于使用的這款?？禉C(jī)器人工業(yè)相機(jī)（MV-CU060-10GM）只能創(chuàng)建一個(gè)連接，所以，當(dāng)預(yù)覽完實(shí)時(shí)視頻，需要調(diào)用該接口釋放相機(jī)資源，避免資源被長期占用。

http://127.0.0.1:65432/stopPreview

3）獲取圖片

返回 base64 格式 的圖片，前端可以直接接收顯示，調(diào)用上傳接口保存。

http://127.0.0.1:65432/openAndSave

7、運(yùn)行效果

由于網(wǎng)速問題和相機(jī)沒有光圈，相機(jī)的拍攝效果有點(diǎn)不清晰。運(yùn)行效果如下：

8、日志查看

工程目錄下回生產(chǎn)日志文件 hikrobot.log，內(nèi)容如下：

9、彩蛋

HikRobotCamera.py 的改良優(yōu)化版，對(duì)獲取圖片方法做了簡化，更為簡潔。代碼如下：文章來源地址http://www.zghlxwxcb.cn/news/detail-473271.html

# -- coding: utf-8 --
import cv2
from flask import Flask, render_template, Response

import sys
import msvcrt
import base64
import datetime
import logging

sys.path.append("./MvImport")
from MvCameraControl_class import *
from JsonResponse import *
from JsonFlask import *

logging.basicConfig(level=logging.DEBUG,#控制臺(tái)打印的日志級(jí)別
                    filename='hikrobot.log',
                    filemode='a',##模式，有w和a，w就是寫模式，每次都會(huì)重新寫日志，覆蓋之前的日志
                    #a是追加模式，默認(rèn)如果不寫的話，就是追加模式
                    format=
                    '%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s'
                    #日志格式
                    )

# 這里配置一下 template_folder為當(dāng)前目錄，不然可以找不到 index.html
app = JsonFlask(__name__, template_folder='.')

# index
@app.route('/')
def index():
    return render_template('./templates/index.html')

# 獲取碼流
def generate(cap):
    # 捕獲異常信息
    try:
        while True:
            # 如果是關(guān)閉相機(jī)，先退出取視頻流的循環(huán)
            global open
            if (not open):
                break;
            retgrab = cap.grab()
            if retgrab == True:
                logging.debug("Grab true")
            ret1, frame = cap.retrieve()
            # print(type(frame))
            if frame is None:
                logging.error("frame is None")
                continue
            ret1, jpeg = cv2.imencode('.jpg', frame)
            jpg_frame = jpeg.tobytes()
            yield (b'--frame\r\n'
                b'Content-Type: image/jpeg\r\n\r\n' + jpg_frame + b'\r\n')
    except Exception as e:
        logging.error("generate error: %s" % str(e))


# 開始預(yù)覽
@app.route('/startPreview')
def startPreview():
    logging.info("================== startPreview start ====================")
    logging.info("start to preview video stream, current_time: " + str(datetime.datetime.now()))
    # 全局變量，用于控制獲取視頻流的開關(guān)狀態(tài)
    try:
        global open
        open = True

        # 全局變量，獲取視頻連接
        global cap
        cap = cv2.VideoCapture(1)

        if False == cap.isOpened():
            logging.error("startPreview -- can't open camera")
            quit()
        else:
            logging.info("startPreview -- start to open camera")

        logging.info("startPreview -- open camera ok")

        # 分辨率設(shè)置 3072*2048（?？禉C(jī)器人工業(yè)相機(jī) MV-CU060-10GM）
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, 3072)
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 2048)
        # 幀率配置
        cap.set(cv2.CAP_PROP_FPS, 15)   
        response = Response(generate(cap), mimetype='multipart/x-mixed-replace;boundary=frame')
    except Exception as e:
        logging.error("startPreview error: %s" % str(e))
    return response

# 停止預(yù)覽
@app.route('/stopPreview')
def stopPreview():
    logging.info("================== stopPreview start ====================")
    logging.info("stop to preview video stream, current_time: " + str(datetime.datetime.now()))
    logging.info("start to close camera")

    # 全局變量，用于停止循環(huán)
    global open
    open = False

    logging.info("release resources start")
    # 全局變量，用于釋放相機(jī)資源
    try:
        global cap
        cap.release()
        cv2.destroyAllWindows()
    except Exception as e:
        logging.error("stopPreview error: %s" % str(e))
    logging.info("release resources end")
    logging.info("camera closed successfully, current_time: " + str(datetime.datetime.now()))
    logging.info("=================== stopPreview end ===================")
    return "stop to preview"

# 獲取base64圖片
@app.route('/grabImage')
def grabImage():
    code = 100000
    msg = "連接相機(jī)時(shí)發(fā)生錯(cuò)誤"
    # 捕獲異常信息
    try:
        print("grabImage -- stopPreview start")
        stopPreview()
        print("grabImage -- stopPreview end")

        # 全局變量，獲取視頻連接
        global cap
        cap = cv2.VideoCapture(1)

        if False == cap.isOpened():
            logging.error("can't open camera")
            quit()
        else:
            logging.info("grabImage -- start to open camera")

        logging.info("grabImage -- open camera ok")

        # 分辨率設(shè)置 3072*2048（?？禉C(jī)器人工業(yè)相機(jī) MV-CU060-10GM）
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, 3072)
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 2048)
        # 幀率配置
        cap.set(cv2.CAP_PROP_FPS, 15)   
        
        retgrab1 = cap.grab()
        if retgrab1 == True:
            logging.debug("grabImage -- Grab true")
        ret1, frame = cap.retrieve()
        ret1, jpeg = cv2.imencode('.jpg', frame)
        img_buff = jpeg.tobytes()

        # 對(duì)返回的圖片進(jìn)行 base64 格式轉(zhuǎn)換
        img_base64 = "data:image/jpg;base64," + str(base64.b64encode(img_buff)).split("'")[1]
        code = 200
        msg = "success"

        stopPreview()
    except Exception as e:
        logging.error("generate error: %s" % str(e))

    return JsonResponse(code, msg, img_base64)

# 獲取base64圖片
@app.route('/hik/openAndSave')
def openAndSave():
    code = 100000
    msg = "連接相機(jī)時(shí)發(fā)生錯(cuò)誤"
    # 捕獲異常信息
    try:
        logging.info("================= openAndSave start =====================")
        # 拍照之前，先停止預(yù)覽
        stopPreview()
        logging.info("start to grab image, current_time: " + str(datetime.datetime.now()))

        # 全局變量，獲取視頻連接
        global cap
        cap = cv2.VideoCapture(1)

        if False == cap.isOpened():
            logging.error("openAndSave -- can't open camera")
            quit()
        else:
            logging.info("openAndSave -- start to open camera")

        logging.info("openAndSave -- open camera ok")

        # 分辨率設(shè)置 3072*2048（海康機(jī)器人工業(yè)相機(jī) MV-CU060-10GM）
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, 3072)
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 2048)
        # 幀率配置
        cap.set(cv2.CAP_PROP_FPS, 15)   
        
        retgrab1 = cap.grab()
        if retgrab1 == True:
            logging.debug("openAndSave -- Grab true")
        ret1, frame = cap.retrieve()
        ret1, jpeg = cv2.imencode('.jpg', frame)
        img_buff = jpeg.tobytes()

        # 對(duì)返回的圖片進(jìn)行 base64 格式轉(zhuǎn)換
        img_base64 = "data:image/jpg;base64," + str(base64.b64encode(img_buff)).split("'")[1]
        code = 200
        msg = "success"

        stopPreview()
        logging.info("================= openAndSave end =====================")
    except Exception as e:
        logging.error("openAndSave error: %s" % str(e))

    return JsonResponse(code, msg, img_base64)

@app.route('/hik/openAndSave2')
def openAndSave2():
    # 拍照之前，先停止預(yù)覽
    stopPreview()
    logging.info("======================================")
    logging.info("start to grab image, current_time: " + str(datetime.datetime.now()))
    code = 100000
    msg = "連接相機(jī)時(shí)發(fā)生錯(cuò)誤"
    # img_base64 = None
    try:
        deviceList = MV_CC_DEVICE_INFO_LIST()
        tlayerType = MV_GIGE_DEVICE | MV_USB_DEVICE

        # ch:枚舉設(shè)備 | en:Enum device
        ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
        if ret != 0:
            logging.error("enum devices fail! ret[0x%x]" % ret)
            sys.exit()

        if deviceList.nDeviceNum == 0:
            logging.error("find no device!")
            sys.exit()

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

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

        nConnectionNum = 0

        if int(nConnectionNum) >= deviceList.nDeviceNum:
            logging.error("intput error!")
            sys.exit()

        # ch:創(chuàng)建相機(jī)實(shí)例 | en:Creat Camera Object
        cam = MvCamera()

        # ch:選擇設(shè)備并創(chuàng)建句柄 | 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:
            logging.error("create handle fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:打開設(shè)備 | en:Open device
        ret = cam.MV_CC_OpenDevice(MV_ACCESS_Exclusive, 0)
        if ret != 0:
            logging.error("open device fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:探測(cè)網(wǎng)絡(luò)最佳包大小(只對(duì)GigE相機(jī)有效) | en:Detection network optimal package size(It only works for the GigE camera)
        if stDeviceList.nTLayerType == MV_GIGE_DEVICE:
            nPacketSize = cam.MV_CC_GetOptimalPacketSize()
            if int(nPacketSize) > 0:
                ret = cam.MV_CC_SetIntValue("GevSCPSPacketSize", nPacketSize)
                if ret != 0:
                    logging.warn("Warning: Set Packet Size fail! ret[0x%x]" % ret)
            else:
                logging.warn("Warning: Get Packet Size fail! ret[0x%x]" % nPacketSize)

        # ch:設(shè)置觸發(fā)模式為off | en:Set trigger mode as off
        ret = cam.MV_CC_SetEnumValue("TriggerMode", MV_TRIGGER_MODE_OFF)
        if ret != 0:
            logging.error("set trigger mode fail! ret[0x%x]" % ret)
            sys.exit()

        # ch:獲取數(shù)據(jù)包大小 | en:Get payload size
        stParam = MVCC_INTVALUE()
        memset(byref(stParam), 0, sizeof(MVCC_INTVALUE))

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

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

        stDeviceList = MV_FRAME_OUT_INFO_EX()
        memset(byref(stDeviceList), 0, sizeof(stDeviceList))
        data_buf = (c_ubyte * nPayloadSize)()
        
        ret = cam.MV_CC_GetOneFrameTimeout(byref(data_buf), nPayloadSize, stDeviceList, 1000)
        if ret == 0:
            logging.info("get one frame: Width[%d], Height[%d], nFrameNum[%d]" % (stDeviceList.nWidth, stDeviceList.nHeight, stDeviceList.nFrameNum))

            nRGBSize = stDeviceList.nWidth * stDeviceList.nHeight * 3
            stConvertParam=MV_SAVE_IMAGE_PARAM_EX()
            stConvertParam.nWidth = stDeviceList.nWidth
            stConvertParam.nHeight = stDeviceList.nHeight
            stConvertParam.pData = data_buf
            stConvertParam.nDataLen = stDeviceList.nFrameLen
            stConvertParam.enPixelType = stDeviceList.enPixelType
            stConvertParam.nImageLen = stConvertParam.nDataLen
            stConvertParam.nJpgQuality = 70
            stConvertParam.enImageType = MV_Image_Jpeg
            stConvertParam.pImageBuffer = (c_ubyte * nRGBSize)()
            stConvertParam.nBufferSize = nRGBSize
            # ret = cam.MV_CC_ConvertPixelType(stConvertParam)
            logging.info("nImageLen: %d" % stConvertParam.nImageLen)
            ret = cam.MV_CC_SaveImageEx2(stConvertParam)
            if ret != 0:
                logging.error("convert pixel fail ! ret[0x%x]" % ret)
                del data_buf
                sys.exit()
            #file_path = "AfterConvert_RGB2.jpg"
            #file_open = open(file_path, 'wb+')
            #file_open = open(file_path.encode('utf8'), 'wb')
            img_buff = (c_ubyte * stConvertParam.nImageLen)()
            cdll.msvcrt.memcpy(byref(img_buff), stConvertParam.pImageBuffer, stConvertParam.nImageLen)
            #file_open.write(img_buff)
            
            # 對(duì)返回的圖片進(jìn)行 base64 格式轉(zhuǎn)換
            img_base64 = "data:image/jpg;base64," + str(base64.b64encode(img_buff)).split("'")[1]
            code = 200
            msg = "success"
        logging.info("Save Image succeed!")

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

        # ch:關(guān)閉設(shè)備 | Close device
        ret = cam.MV_CC_CloseDevice()
        if ret != 0:
            logging.error("close deivce fail! ret[0x%x]" % ret)
            del data_buf
            sys.exit()

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

        del data_buf
    except Exception as e:
        logging.error("openAndSave error: %s" % str(e))
    # print("openAndSave finished, current_time: " + str(datetime.datetime.now()))  
  
    return JsonResponse(code, msg, img_base64)

# 執(zhí)行web服務(wù), 端口號(hào)可自行修訂
logging.info("start to run camera app, current_time: " + str(datetime.datetime.now()))
if __name__ == '__main__':
    try:
        app.run(host='0.0.0.0', port=65432, debug=True, threaded=True)
    except Exception as e:
        logging.error("app error: %s" % str(e))

到了這里，關(guān)于Python 實(shí)現(xiàn)?？禉C(jī)器人工業(yè)相機(jī) MV-CU060-10GM 的實(shí)時(shí)顯示視頻流及拍照功能的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！