引言

做了下实验楼的关于Python 实现局域网视频聊天工具 ，感觉还不错，作为练习opencv和socket的小脚本入门了。

1. 内容简介

• 本实验实现简易的视频通信工具
• 在视频通信的基础上加入语音
• 用户可以选择通信的质量，即画质、停顿等参数
• 支持IPv6

2. 实验知识点

本课程项目完成过程中将学习：

• Python 基于 OpenCV 对摄像头信息的捕获和压缩
• Python 关于 线程 和 socket 通信的一些基础技巧
• Python 基于 PyAudio 对语音信息的捕获和压缩

其中将重点介绍 socket 传输过程中对数据的压缩和处理。

3.实验环境

• python 3.5
• opencv-python 3.4.1.15
• numpy 1.14.5
• PyAudio 0.2.11

实验步骤

环境搭建

首先关于实验所需要用到的包，opencv-python和PyAudio并不是太好装，我简单讲一下我的安装步骤。

因为我自己电脑Windows下的是anaconda，关于opencv的python第三方包如果是在conda环境下查找，会发现找不到，那么使用pip安装同样会出现四次retry，然后安装失败。所以建议最好创建一个python的虚拟环境，便于直接install，而不需要找什么本地whl文件，或者换源等等，比较麻烦。

另外就是PyAudio==0.2.11，这个包是已经很久没有更新过了，但这并不意味着它适配不了新的python环境，已经编译不了了。它的安装百度到的答案，或者说pypi推荐的安装方式都是pip install PyAudio == 0.2.11，但就最新的python3.6到3.7版本来讲，我是没有安装成功的，原因应该都是报这个错吧：

 error: Microsoft Visual C++ 14.0 is required. Get it with "Build Tools for Visual Studio":  

于是继续百度，发现推荐的都是去找PyAudio-0.2.11-cp36-cp36m-win_amd64.whl文件，然后我也去找了，但可能是我python是3.7的原因，我并没有安装成功。报错为：

>>> pip install D:DownloadsPyAudio-0.2.11-cp36-cp36m-win_amd64.whl  ERROR: PyAudio-0.2.11-cp36-cp36m-win_amd64.whl is not a supported wheel on this platform. 

这里卡了很久，再次发现国内博客都是抄来抄去的，中间还有其它事情，直到第二天搭梯子谷歌了一下，终于找到了答案：

pip install pipwin # 安装当前pip环境下的所有必要依赖的包  pipwin install pyaudio # 成功安装 

当然还有一种方式，是去GitHub将包的源码拷贝到本地重命名，不过我不喜欢这么做。

https://github.com/intxcc/pyaudio_portaudio/releases

代码实现

先为双方的通信设计 Server 类和 Client类，两个类均继承 threading.Thread并实现双向的C / S连接，只需要分别实现 init、__del__和run方法，之后对象调用.start()方法即可在独立线程中执行run方法中的内容。首先Client类需要存储远端的IP地址和端口，而Server类需要存储本地服务器监听的端口号。用户还应当可以指定通信双方使用的协议版本，即基于IPv4 还是IPv6 的TCP连接。因此Server类的初始化需要传入两个参数（端口、版本），Client类的初始化需要三个参数（远端IP、端口、版本）。新建文件vchat.py，在其中定义基础的两个类，另外捕获视频流的任务应当由Client类完成，一起完善为：

from socket import * import threading class Video_Server(threading.Thread): def __init__(self, port, version) :         threading.Thread.__init__(self)         self.setDaemon(True) # 设置守护线程，必须在t.start()之前设置         self.ADDR = ('', port) if version == 4: # ipv4版本，client要与之对应             self.sock = socket(AF_INET ,SOCK_STREAM) # 创建TCP连接，ipv4版本 else:             self.sock = socket(AF_INET6 ,SOCK_STREAM) # 创建TCP连接，ipv6版本 def __del__(self):         self.sock.close() # TODO def run(self): print("server starts...")         self.sock.bind(self.ADDR)         self.sock.listen(1) # 操作系统可以挂起的最大连接数，如果同一时间的连接数超过1，拒绝其他的连接         conn, addr = self.sock.accept() # 接收客户端的请求，且获取新socket对象和客户端信息 print("remote client success connected...") # TODO class Video_Client(threading.Thread): def __init__(self ,ip, port, version):         threading.Thread.__init__(self)         self.setDaemon(True)         self.ADDR = (ip, port) if version == 4:             self.sock = socket(AF_INET, SOCK_STREAM) else:             self.sock = socket(AF_INET6, SOCK_STREAM)         self.cap = cv2.VideoCapture(0) # 创建一个成员变量cap def __del__(self) :         self.sock.close()         self.cap.release() def run(self): print("client starts...") while True: try:                 self.sock.connect(self.ADDR) break except:                 time.sleep(3) continue print("client connected...") while self.cap.isOpened(): # 如果连接上了，进入循环读相应的文件，这里也可以try一下，没有连接上接收异常，这样的语法不好排错             ret, frame = self.cap.read() # 读 

已经捕获到数据，接下来要发送字节流。首先我们继续编写Client，为其添加发送数据功能的实现。这里只改动了run方法。在捕获到帧后，我们使用pickle.dumps方法对其打包，并用sock.sendall方法发送。注意发送过程中我们用struct.pack方法为每批数据加了一个头，用于接收方确认接受数据的长度。

 def run(self): while True: try:                 self.sock.connect(self.ADDR) break except:                 time.sleep(3) continue print("client connected...") while self.cap.isOpened():             ret, frame = self.cap.read()              data = pickle.dumps(frame) try: """     将len(data)等参数的值进行一层包装，包装的方法由fmt指定。被包装的参数必须严格符合fmt。最后返回一个包装后的字符串。     """                 self.sock.sendall(struct.pack("L", len(data)) + data) except: break 

下面编写Server，在服务器端连接成功后，应当创建一个窗口用于显示接收到的视频。因为连接不一定创建成功，因此cv.destroyAllWindows()被放在一个try…catch块中防止出现错误。在接收数据过程中，我们使用payload_size记录当前从缓冲区读入的数据长度，这个长度通过struct.calcsize(‘L’)来读取。使用该变量的意义在于缓冲区中读出的数据可能不足一个帧，也可能由多个帧构成。为了准确提取每一帧，我们用payload_size区分帧的边界。在从缓冲区读出的数据流长度超过payload_size时，剩余部分和下一次读出的数据流合并，不足payload_size时将合并下一次读取的数据流到当前帧中。在接收完完整的一帧后，显示在创建的窗口中。同时我们为窗口创建一个键盘响应，当按下Esc 或 q键时退出程序。

另外当用户指定使用低画质通信，我们应当对原始数据做变换，最简单的方式即将捕获的每一帧按比例缩放，同时降低传输的帧速，在代码中体现为resize，该函数的第二个参数为缩放中心，后两个参数为缩放比例，并且根据用户指定的等级，不再传输捕获的每一帧，而是间隔几帧传输一帧。为了防止用户指定的画质过差，代码中限制了最坏情况下的缩放比例为0.3，最大帧间隔为3。此外，我们在发送每一帧的数据前使用zlib.compress对其压缩，尽量降低带宽负担。

所以完整代码为：

class Video_Server(threading.Thread): def __init__(self, port, version) :         threading.Thread.__init__(self)         self.setDaemon(True)         self.ADDR = ('', port) if version == 4:             self.sock = socket(AF_INET ,SOCK_STREAM) else:             self.sock = socket(AF_INET6 ,SOCK_STREAM) def __del__(self):         self.sock.close() try:             cv2.destroyAllWindows() except: pass def run(self): print("VIDEO server starts...")         self.sock.bind(self.ADDR)         self.sock.listen(1)         conn, addr = self.sock.accept() print("remote VIDEO client success connected...")         data = "".encode("utf-8")         payload_size = struct.calcsize("L") # 结果为4         cv2.namedWindow('Remote', cv2.WINDOW_NORMAL) while True: while len(data) < payload_size:                 data += conn.recv(81920)             packed_size = data[:payload_size]             data = data[payload_size:]             msg_size = struct.unpack("L", packed_size)[0] while len(data) < msg_size:                 data += conn.recv(81920)             zframe_data = data[:msg_size]             data = data[msg_size:]             frame_data = zlib.decompress(zframe_data)             frame = pickle.loads(frame_data)             cv2.imshow('Remote', frame) if cv2.waitKey(1) & 0xFF == 27: break class Video_Client(threading.Thread): def __init__(self ,ip, port, level, version):         threading.Thread.__init__(self)         self.setDaemon(True)         self.ADDR = (ip, port) if level <= 3:             self.interval = level         else:             self.interval = 3         self.fx = 1 / (self.interval + 1) if self.fx < 0.3: # 限制最大帧间隔为3帧             self.fx = 0.3 if version == 4:             self.sock = socket(AF_INET, SOCK_STREAM) else:             self.sock = socket(AF_INET6, SOCK_STREAM)         self.cap = cv2.VideoCapture(0) def __del__(self) :         self.sock.close()         self.cap.release() def run(self): print("VIDEO client starts...") while True: try:                 self.sock.connect(self.ADDR) break except:                 time.sleep(3) continue print("VIDEO client connected...") while self.cap.isOpened():             ret, frame = self.cap.read()             sframe = cv2.resize(frame, (0,0), fx=self.fx, fy=self.fx)             data = pickle.dumps(sframe)             zdata = zlib.compress(data, zlib.Z_BEST_COMPRESSION) try:                 self.sock.sendall(struct.pack("L", len(zdata)) + zdata) except: break for i in range(self.interval):                 self.cap.read() 

里面有些代码上面的大段应该解释得比较清楚，如果还有不明白的，可以看作者的推荐，还有我找到的一些资料链接：

struct.pack()、struct.unpack()和struct.calcsize()

Python 网络编程

加入音频的捕获和传输

在完成视频通信的基础上，整体框架对于音频通信可以直接挪用，只需要修改其中捕获视频/音频的代码和服务器解码播放的部分。这里我们使用 PyAudio 库处理音频，在 Linux 下你也可以选择 sounddevice。关于sounddevice这里不做过多介绍，可以在这里看到它最新版本的文档。将vchat.py复制一份，重命名为achat.py，简单修改几处，最终音频捕获、传输的完整代码如下。将上面代码中的Server和Client分别加上Video和Audio前缀以区分，同时显示给用户的print输出语句也做了一定修改，对于视频加上VIDEO前缀，音频加上AUDIO前缀。

from socket import * import threading import pyaudio import wave import sys import zlib import struct import pickle import time import numpy as np  CHUNK = 1024 FORMAT = pyaudio.paInt16    # 格式 CHANNELS = 2 # 输入/输出通道数 RATE = 44100 # 音频数据的采样频率 RECORD_SECONDS = 0.5 # 记录秒 class Audio_Server(threading.Thread): def __init__(self, port, version) :         threading.Thread.__init__(self)         self.setDaemon(True)         self.ADDR = ('', port) if version == 4:             self.sock = socket(AF_INET ,SOCK_STREAM) else:             self.sock = socket(AF_INET6 ,SOCK_STREAM)         self.p = pyaudio.PyAudio() # 实例化PyAudio,并于下面设置portaudio参数         self.stream = None def __del__(self):         self.sock.close() # 关闭套接字 if self.stream is not None:             self.stream.stop_stream() # 暂停播放 / 录制             self.stream.close() # 终止流         self.p.terminate() # 终止会话 def run(self): print("Video server starts...")         self.sock.bind(self.ADDR)         self.sock.listen(1)         conn, addr = self.sock.accept() print("remote Video client success connected...")         data = "".encode("utf-8")         payload_size = struct.calcsize("L") # 返回对应于格式字符串fmt的结构，L为4         self.stream = self.p.open(format=FORMAT,                                   channels=CHANNELS,                                   rate=RATE,                                   output=True,                                   frames_per_buffer = CHUNK                                   ) while True: while len(data) < payload_size:                 data += conn.recv(81920)             packed_size = data[:payload_size]             data = data[payload_size:]             msg_size = struct.unpack("L", packed_size)[0] while len(data) < msg_size:                 data += conn.recv(81920)             frame_data = data[:msg_size]             data = data[msg_size:]             frames = pickle.loads(frame_data) for frame in frames:                 self.stream.write(frame, CHUNK) class Audio_Client(threading.Thread): def __init__(self ,ip, port, version):         threading.Thread.__init__(self)         self.setDaemon(True)         self.ADDR = (ip, port) if version == 4:             self.sock = socket(AF_INET, SOCK_STREAM) else:             self.sock = socket(AF_INET6, SOCK_STREAM)         self.p = pyaudio.PyAudio()         self.stream = None print("AUDIO client starts...") def __del__(self) :         self.sock.close() if self.stream is not None:             self.stream.stop_stream()             self.stream.close()         self.p.terminate() def run(self): while True: try:                 self.sock.connect(self.ADDR) break except:                 time.sleep(3) continue print("AUDIO client connected...")         self.stream = self.p.open(format=FORMAT,                               channels=CHANNELS,                              rate=RATE, input=True,                              frames_per_buffer=CHUNK) while self.stream.is_active():             frames = [] for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):                 data = self.stream.read(CHUNK)                 frames.append(data)             senddata = pickle.dumps(frames) try:                 self.sock.sendall(struct.pack("L", len(senddata)) + senddata) except: break 

Play and Record Sound with Python

PyAudio Documentation¶

以上为参考文档，下面就需要编写启动这两个文件的主函数了。

编写程序入口 main.py

为了提供用户参数解析，代码使用了argparse。你可能对此前几个类中初始化方法的self.setDaemon(True)有疑惑。这个方法的调用使每个线程在主线程结束之后自动退出，保证程序不会出现崩溃且无法销毁的情况。在main.py中，我们通过每隔1s做一次线程的保活检查，如果视频/音频中出现阻塞/故障，主线程会终止。

import sys import time import argparse from vchat import Video_Server, Video_Client from achat import Audio_Server, Audio_Client  parser = argparse.ArgumentParser()  parser.add_argument('--host', type=str, default='127.0.0.1') parser.add_argument('--port', type=int, default=10087) parser.add_argument('--level', type=int, default=1) parser.add_argument('-v', '--version', type=int, default=4)  args = parser.parse_args()  IP = args.host PORT = args.port VERSION = args.version LEVEL = args.level  if __name__ == '__main__':     vclient = Video_Client(IP, PORT, LEVEL, VERSION)     vserver = Video_Server(PORT, VERSION)     aclient = Audio_Client(IP, PORT+1, VERSION)     aserver = Audio_Server(PORT+1, VERSION)     vclient.start()     aclient.start()     time.sleep(1) # make delay to start server     vserver.start()     aserver.start() while True:         time.sleep(1) if not vserver.isAlive() or not vclient.isAlive(): print("Video connection lost...")             sys.exit(0) if not aserver.isAlive() or not aclient.isAlive(): print("Audio connection lost...")             sys.exit(0) 

关于argparse模块，我记得在之前的一篇文章中提到过，现在基本都忘了。。。链接为：

Python利用argparse模块图片转字符文件

如果有不清楚，可以进去看看我对参数画了思维导图。

总结

因为环境都没有提供摄像头，如果需运行，要修改一下代码，让程序从一个本地视频文件读取，模拟摄像头的访问。将Video_Client中self.cap = cv2.VideoCapture(0)改为self.cap = cv2.VideoCapture(‘xxx.mp4’)，即从本地视频xxx.mp4中读取。我这里拿了一个B站的MAD来试验，另外，当程序启动时，电脑接收到的声音，因为自己与自己远程，都会有回声。

那么我在Windows环境下运行为：