树莓派/PC实现实时摄像头数据共享(Python—OpenCV)
        【摘要】  使用Python和OpenCV实现树莓派/PC实时摄像头数据共享,主要分为服务器和客户端两部分。 
服务器Demo如下: 
#服务器端 import socketimport threadingimport structimport timeimport cv2import numpy class Carame_Accept_Object: def __init__(self...
    
    
    
    使用Python和OpenCV实现树莓派/PC实时摄像头数据共享,主要分为服务器和客户端两部分。
服务器Demo如下:
  
   - 
    
     
    
    
     
      #服务器端
     
    
- 
    
     
    
    
      
     
    
- 
    
     
    
    
     
      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 #每秒传输多少帧数
     
    
- 
    
     
    
    
     
       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 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 #双方确定传输帧数,(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)
     
    
 运行效果还是比较流畅的,因为传输网络视频流,所以画质欠佳。

树莓派视频传输改进方式见:https://blog.csdn.net/m0_38106923/article/details/82628061
树莓派视频传输最优方法:https://blog.csdn.net/m0_38106923/article/details/86562451
文章来源: blog.csdn.net,作者:不脱发的程序猿,版权归原作者所有,如需转载,请联系作者。
原文链接:blog.csdn.net/m0_38106923/article/details/81974373
        【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
            cloudbbs@huaweicloud.com
        
        
        
        
        
        
        - 点赞
- 收藏
- 关注作者
 
             
           
评论(0)