python如何實(shí)現(xiàn)opencv+scoket網(wǎng)絡(luò)實(shí)時(shí)圖傳

這篇文章主要介紹了python如何實(shí)現(xiàn)opencv+scoket網(wǎng)絡(luò)實(shí)時(shí)圖傳，具有一定借鑒價(jià)值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

服務(wù)器分析：

1.　先通過在服務(wù)器端利用OpenCV捕獲到視頻的每一幀圖片

2.　將這些圖片進(jìn)行壓縮成JPEG格式，這樣能減小圖片大小，便于傳輸

3.　按照提前協(xié)商好的分辨率和幀數(shù)進(jìn)行打包編碼傳輸

4.　利用服務(wù)器端打開端口8880，此時(shí)客戶端連接后，便可以在客戶端中捕獲到服務(wù)器端的視頻。

#服務(wù)端
import socket
import threading
import struct
import time
import cv2
import numpy

class Carame_Accept_Object:
  def __init__(self,S_addr_port=("",8880)):
    self.resolution=(640,480)    #分辨率
    self.img_fps=15         #每秒傳輸多少幀數(shù)
    self.addr_port=S_addr_port
    self.Set_Socket(self.addr_port)

  #設(shè)置套接字
  def Set_Socket(self,S_addr_port):
    self.server=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    self.server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) #端口可復(fù)用
    self.server.bind(S_addr_port)
    self.server.listen(5)
    #print("the process work in the port:%d" % S_addr_port[1])


def check_option(object,client):
  #按格式解碼，確定幀數(shù)和分辨率
  info=struct.unpack('lhh',client.recv(8))
  if info[0]>888:
    object.img_fps=int(info[0])-888     #獲取幀數(shù)
    object.resolution=list(object.resolution)
    # 獲取分辨率
    object.resolution[0]=info[1]
    object.resolution[1]=info[2]
    object.resolution = tuple(object.resolution)
    return 1
  else:
    return 0

def RT_Image(object,client,D_addr):
  if(check_option(object,client)==0):
    return
  camera=cv2.VideoCapture(0)                #從攝像頭中獲取視頻
  img_param=[int(cv2.IMWRITE_JPEG_QUALITY),object.img_fps] #設(shè)置傳送圖像格式、幀數(shù)
  while(1):
    time.sleep(0.1)       #推遲線程運(yùn)行0.1s
    _,object.img=camera.read() #讀取視頻每一幀

    object.img=cv2.resize(object.img,object.resolution)   #按要求調(diào)整圖像大小(resolution必須為元組)
    _,img_encode=cv2.imencode('.jpg',object.img,img_param) #按格式生成圖片
    img_code=numpy.array(img_encode)            #轉(zhuǎn)換成矩陣
    object.img_data=img_code.tostring()           #生成相應(yīng)的字符串
    try:
      #按照相應(yīng)的格式進(jìn)行打包發(fā)送圖片
      client.send(struct.pack("lhh",len(object.img_data),object.resolution[0],object.resolution[1])+object.img_data)
    except:
      camera.release()    #釋放資源
      return

if __name__ == '__main__':
  camera=Carame_Accept_Object()
  while(1):
    client,D_addr=camera.server.accept()
    clientThread=threading.Thread(None,target=RT_Image,args=(camera,client,D_addr,))
    clientThread.start()

客戶端分析：

1.　客戶端連接端口后，首先發(fā)送需要協(xié)商的分辨率和幀數(shù)，以致能夠使傳輸“協(xié)議”一致

2.　客戶端使用線程，對(duì)圖片進(jìn)行收集

3.　對(duì)收到的每一張圖片進(jìn)行解碼，并利用OpenCV播放出來，即可實(shí)現(xiàn)C/S兩端實(shí)時(shí)視頻傳輸。

#客戶端
import socket
import cv2
import threading
import struct
import numpy

class Camera_Connect_Object:
  def __init__(self,D_addr_port=["",8880]):
    self.resolution=[640,480]
    self.addr_port=D_addr_port
    self.src=888+15         #雙方確定傳輸幀數(shù)，（888）為校驗(yàn)值
    self.interval=0         #圖片播放時(shí)間間隔
    self.img_fps=15         #每秒傳輸多少幀數(shù)

  def Set_socket(self):
    self.client=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    self.client.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)

  def Socket_Connect(self):
    self.Set_socket()
    self.client.connect(self.addr_port)
    print("IP is %s:%d" % (self.addr_port[0],self.addr_port[1]))

  def RT_Image(self):
    #按照格式打包發(fā)送幀數(shù)和分辨率
    self.name=self.addr_port[0]+" Camera"
    self.client.send(struct.pack("lhh", self.src, self.resolution[0], self.resolution[1]))
    while(1):
      info=struct.unpack("lhh",self.client.recv(8))
      buf_size=info[0]          #獲取讀的圖片總長(zhǎng)度
      if buf_size:
        try:
          self.buf=b""        #代表bytes類型
          temp_buf=self.buf
          while(buf_size):      #讀取每一張圖片的長(zhǎng)度
            temp_buf=self.client.recv(buf_size)
            buf_size-=len(temp_buf)
            self.buf+=temp_buf   #獲取圖片
            data = numpy.fromstring(self.buf, dtype='uint8')  #按uint8轉(zhuǎn)換為圖像矩陣
            self.image = cv2.imdecode(data, 1)         #圖像解碼
            cv2.imshow(self.name, self.image)          #展示圖片
        except:
          pass;
        finally:
          if(cv2.waitKey(10)==27):    #每10ms刷新一次圖片，按‘ESC'（27）退出
            self.client.close()
            cv2.destroyAllWindows()
            break

  def Get_Data(self,interval):
    showThread=threading.Thread(target=self.RT_Image)
    showThread.start()

if __name__ == '__main__':
  camera=Camera_Connect_Object()
  camera.addr_port[0]="服務(wù)端的ip"
  camera.addr_port=tuple(camera.addr_port)
  camera.Socket_Connect()
  camera.Get_Data(camera.interval

感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享的“python如何實(shí)現(xiàn)opencv+scoket網(wǎng)絡(luò)實(shí)時(shí)圖傳”這篇文章對(duì)大家有幫助，同時(shí)也希望大家多多支持億速云，關(guān)注億速云行業(yè)資訊頻道，更多相關(guān)知識(shí)等著你來學(xué)習(xí)!