树莓派 /PC 实现实时摄像头数据共享（Python—OpenCV）

关注

发布于: 3 小时前

使用 Python 和 OpenCV 实现树莓派/PC 实时摄像头数据共享，主要分为服务器和客户端两部分。

服务器 Demo 如下：

#服务器端 import socketimport threadingimport structimport timeimport cv2import numpy class Carame_Accept_Object:    def __init__(self,S_addr_port=("",8880)):        self.resolution=(640,480)       #分辨率        self.img_fps=15                 #每秒传输多少帧数        self.addr_port=S_addr_port        self.Set_Socket(self.addr_port)     #设置套接字    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) #端口可复用        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):    #按格式解码，确定帧数和分辨率    info=struct.unpack('lhh',client.recv(8))    if info[0]>888:        object.img_fps=int(info[0])-888          #获取帧数        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]  #设置传送图像格式、帧数    while(1):        time.sleep(0.1)             #推迟线程运行0.1s        _,object.img=camera.read()  #读取视频每一帧         object.img=cv2.resize(object.img,object.resolution)     #按要求调整图像大小(resolution必须为元组)        _,img_encode=cv2.imencode('.jpg',object.img,img_param)  #按格式生成图片        img_code=numpy.array(img_encode)                        #转换成矩阵        object.img_data=img_code.tostring()                     #生成相应的字符串        try:            #按照相应的格式进行打包发送图片            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()

复制代码

客户端 Demo 如下：

#客户端 import socketimport cv2import threadingimport structimport 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                 #双方确定传输帧数，（888）为校验值        self.interval=0                 #图片播放时间间隔        self.img_fps=100               #每秒传输多少帧数     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):        #按照格式打包发送帧数和分辨率        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]                    #获取读的图片总长度            if buf_size:                try:                    self.buf=b""                #代表bytes类型                    temp_buf=self.buf                    while(buf_size):            #读取每一张图片的长度                        temp_buf=self.client.recv(buf_size)                        buf_size-=len(temp_buf)                        self.buf+=temp_buf      #获取图片                        data = numpy.fromstring(self.buf, dtype='uint8')    #按uint8转换为图像矩阵                        self.image = cv2.imdecode(data, 1)                  #图像解码                        gray = cv2.cvtColor(self.image, cv2.COLOR_BGR2GRAY)                        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]=input("Please input IP:")    camera.addr_port=tuple(camera.addr_port)    camera.Socket_Connect()    camera.Get_Data(camera.interval)