本文實(shí)例為大家分享了python實(shí)現(xiàn)opencv+scoket網(wǎng)絡(luò)實(shí)時(shí)圖傳的具體代碼，供大家參考，具體內(nèi)容如下

服務(wù)器分析：

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

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

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

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

#服務(wù)端import socketimport threadingimport structimport timeimport cv2import numpyclass 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 0def 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() #釋放資源 returnif __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 socketimport cv2import threadingimport structimport numpyclass 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

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。