1.環境html
操做系統爲 Ubuntu 16.04python
python 3.5
opencv-python 3.4.1.15
numpy 1.14.5
PyAudio 0.2.11服務器
2. 內容簡介
本實驗實現簡易的視頻通訊工具
在視頻通訊的基礎上加入語音
用戶能夠選擇通訊的質量，即畫質、停頓等參數
支持IPv6網絡
3.安裝環境app
$ sudo pip3 install numpy
$ sudo pip3 install opencv_python
這一步下載了咱們須要的opencv-python和numpy兩個包。框架
剩下的PyAudio，因爲本虛擬環境的部分問題，咱們單獨分開下載。socket
$ sudo apt-get install portaudio19-dev python-all-dev python3-all-dev
$ sudo pip3 install pyaudio==0.2.11
如今，咱們的實驗環境就搭好了。ide
4. 實驗原理
實驗實現了簡易的視頻通訊工具，基於 OpenCV 和 PyAudio，使用 TCP 協議通訊，通訊雙方創建雙向 CS 鏈接，
雙方均維護一個客戶端和一個服務器端。在捕獲視頻信息後，根據用戶指定的參數對畫面作壓縮並傳輸。函數
實驗步驟
接下來咱們分步驟講解本實驗。工具
4.1 實現雙向 C/S 鏈接
先爲雙方的通訊設計 Server 類和 Client類，兩個類均繼承 threading.Thread，只須要分別實現 __init__、__del__和run方法，
以後對象調用.start()方法便可在獨立線程中執行run方法中的內容。首先Client類須要存儲遠端的IP地址和端口，
而Server類須要存儲本地服務器監聽的端口號。用戶還應當能夠指定通訊雙方使用的協議版本，即基於IPv4 仍是IPv6 的TCP鏈接。
所以Server類的初始化須要傳入兩個參數（端口、版本），Client類的初始化須要三個參數（遠端IP、端口、版本）。新建文件vchat.py，
在其中定義基礎的兩個類以下。
1 from socket import *
2 import threading
3 class Video_Server(threading.Thread):
4 def __init__(self, port, version) :
5 threading.Thread.__init__(self)
6 self.setDaemon(True)
7 self.ADDR = ('', port)
8 if version == 4:
9 self.sock = socket(AF_INET ,SOCK_STREAM)
10 else:
11 self.sock = socket(AF_INET6 ,SOCK_STREAM)
12 def __del__(self):
13 self.sock.close()
14 # TODO
15 def run(self):
16 print("server starts...")
17 self.sock.bind(self.ADDR)
18 self.sock.listen(1)
19 conn, addr = self.sock.accept()
20 print("remote client success connected...")
21 # TODO
22
23 class Video_Client(threading.Thread):
24 def __init__(self ,ip, port, version):
25 threading.Thread.__init__(self)
26 self.setDaemon(True)
27 self.ADDR = (ip, port)
28 if version == 4:
29 self.sock = socket(AF_INET, SOCK_STREAM)
30 else:
31 self.sock = socket(AF_INET6, SOCK_STREAM)
32 def __del__(self) :
33 self.sock.close()
34 # TODO
35 def run(self):
36 print("client starts...")
37 while True:
38 try:
39 self.sock.connect(self.ADDR)
40 break
41 except:
42 time.sleep(3)
43 continue
44 print("client connected...")
45 # TODO
複製代碼
4.2 實現攝像頭數據流捕獲
OpenCV 爲 Python 提供的接口很是簡單而且易於理解。捕獲視頻流的任務應當由Client類完成，
下面完善Client的run函數。在下面的代碼中，咱們爲類添加了一個成員變量cap，它用來捕獲默認攝像頭的輸出。
複製代碼
1 class Video_Client(threading.Thread):
2 def __init__(self ,ip, port, version):
3 threading.Thread.__init__(self)
4 self.setDaemon(True)
5 self.ADDR = (ip, port)
6 if version == 4:
7 self.sock = socket(AF_INET, SOCK_STREAM)
8 else:
9 self.sock = socket(AF_INET6, SOCK_STREAM)
10 self.cap = cv2.VideoCapture(0)
11 def __del__(self) :
12 self.sock.close()
13 self.cap.release()
14 def run(self):
15 print("client starts...")
16 while True:
17 try:
18 self.sock.connect(self.ADDR)
19 break
20 except:
21 time.sleep(3)
22 continue
23 print("client connected...")
24 while self.cap.isOpened():
25 ret, frame = self.cap.read()
26 # TODO
複製代碼
4.3 發送捕獲到的數據到服務器
已經捕獲到數據，接下來要發送字節流。首先咱們繼續編寫Client，爲其添加發送數據功能的實現。這裏只改動了run方法。
在捕獲到幀後，咱們使用pickle.dumps方法對其打包，並用sock.sendall方法發送。
注意發送過程當中咱們用struct.pack方法爲每批數據加了一個頭，用於接收方確認接受數據的長度。
複製代碼
1 def run(self):
2 while True:
3 try:
4 self.sock.connect(self.ADDR)
5 break
6 except:
7 time.sleep(3)
8 continue
9 print("client connected...")
10 while self.cap.isOpened():
11 ret, frame = self.cap.read()
12 data = pickle.dumps(frame)
13 try:
14 self.sock.sendall(struct.pack("L", len(data)) + data)
15 except:
16 break
複製代碼
下面編寫Server，在服務器端鏈接成功後，應當建立一個窗口用於顯示接收到的視頻。由於鏈接不必定建立成功，
所以cv.destroyAllWindows()被放在一個try..catch塊中防止出現錯誤。在接收數據過程當中，
咱們使用payload_size記錄當前從緩衝區讀入的數據長度，這個長度經過struct.calcsize('L')來讀取。
使用該變量的意義在於緩衝區中讀出的數據可能不足一個幀，也可能由多個幀構成。爲了準確提取每一幀，咱們用payload_size區分幀的邊界。
在從緩衝區讀出的數據流長度超過payload_size時，剩餘部分和下一次讀出的數據流合併，
不足payload_size時將合併下一次讀取的數據流到當前幀中。在接收完完整的一幀後，顯示在建立的窗口中。同時咱們爲窗口建立一個鍵盤響應，
當按下Esc 或 q鍵時退出程序。
複製代碼
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("server starts...")
self.sock.bind(self.ADDR)
self.sock.listen(1)
conn, addr = self.sock.accept()
print("remote client success connected...")
data = "".encode("utf-8")
payload_size = struct.calcsize("L")
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
複製代碼
4.4 視頻縮放和數據壓縮
如今的服務器和客戶端已經能夠運行，你能夠在代碼中建立一個Client類實例和一個Server類實例，並將IP地址設爲127.0.0.1，
端口設爲任意合法的（0-65535）且不衝突的值，版本設爲IPv4。執行代碼等同於本身和本身通訊。若是網絡情況很差，
你也許會發現本身和本身的通訊也有卡頓現象。爲了使畫面質量、延遲可以和現實網絡情況相匹配，咱們須要容許用戶指定通訊中畫面的質量，
同時咱們的代碼應當自己具備壓縮數據的能力，以儘量利用帶寬。
當用戶指定使用低畫質通訊，咱們應當對原始數據作變換，最簡單的方式即將捕獲的每一幀按比例縮放，同時下降傳輸的幀速，
在代碼中體現爲resize，該函數的第二個參數爲縮放中心，後兩個參數爲縮放比例，而且根據用戶指定的等級，再也不傳輸捕獲的每一幀，
而是間隔幾幀傳輸一幀。爲了防止用戶指定的畫質過差，代碼中限制了最壞狀況下的縮放比例爲0.3，最大幀間隔爲3。此外，
咱們在發送每一幀的數據前使用zlib.compress對其壓縮，儘可能下降帶寬負擔。
複製代碼
1 class Video_Client(threading.Thread):
2 def __init__(self ,ip, port, level, version):
3 threading.Thread.__init__(self)
4 self.setDaemon(True)
5 self.ADDR = (ip, port)
6 if level <= 3:
7 self.interval = level
8 else:
9 self.interval = 3
10 self.fx = 1 / (self.interval + 1)
11 if self.fx < 0.3:
12 self.fx = 0.3
13 if version == 4:
14 self.sock = socket(AF_INET, SOCK_STREAM)
15 else:
16 self.sock = socket(AF_INET6, SOCK_STREAM)
17 self.cap = cv2.VideoCapture(0)
18 def __del__(self) :
19 self.sock.close()
20 self.cap.release()
21 def run(self):
22 print("VEDIO client starts...")
23 while True:
24 try:
25 self.sock.connect(self.ADDR)
26 break
27 except:
28 time.sleep(3)
29 continue
30 print("VEDIO client connected...")
31 while self.cap.isOpened():
32 ret, frame = self.cap.read()
33 sframe = cv2.resize(frame, (0,0), fx=self.fx, fy=self.fx)
34 data = pickle.dumps(sframe)
35 zdata = zlib.compress(data, zlib.Z_BEST_COMPRESSION)
36 try:
37 self.sock.sendall(struct.pack("L", len(zdata)) + zdata)
38 except:
39 break
40 for i in range(self.interval):
41 self.cap.read()
複製代碼
服務器端最終代碼以下，增長了對接收到數據的解壓縮處理。
複製代碼
1 class Video_Server(threading.Thread):
2 def __init__(self, port, version) :
3 threading.Thread.__init__(self)
4 self.setDaemon(True)
5 self.ADDR = ('', port)
6 if version == 4:
7 self.sock = socket(AF_INET ,SOCK_STREAM)
8 else:
9 self.sock = socket(AF_INET6 ,SOCK_STREAM)
10 def __del__(self):
11 self.sock.close()
12 try:
13 cv2.destroyAllWindows()
14 except:
15 pass
16 def run(self):
17 print("VEDIO server starts...")
18 self.sock.bind(self.ADDR)
19 self.sock.listen(1)
20 conn, addr = self.sock.accept()
21 print("remote VEDIO client success connected...")
22 data = "".encode("utf-8")
23 payload_size = struct.calcsize("L")
24 cv2.namedWindow('Remote', cv2.WINDOW_NORMAL)
25 while True:
26 while len(data) < payload_size:
27 data += conn.recv(81920)
28 packed_size = data[:payload_size]
29 data = data[payload_size:]
30 msg_size = struct.unpack("L", packed_size)[0]
31 while len(data) < msg_size:
32 data += conn.recv(81920)
33 zframe_data = data[:msg_size]
34 data = data[msg_size:]
35 frame_data = zlib.decompress(zframe_data)
36 frame = pickle.loads(frame_data)
37 cv2.imshow('Remote', frame)
38 if cv2.waitKey(1) & 0xFF == 27:
39 break
複製代碼
4.5 加入音頻的捕獲和傳輸
在完成視頻通訊的基礎上，總體框架對於音頻通訊能夠直接挪用，只須要修改其中捕獲視頻/音頻的代碼和服務器解碼播放的部分。
這裏咱們使用 PyAudio 庫處理音頻，在 Linux 下你也能夠選擇 sounddevice。關於sounddevice這裏不作過多介紹，
將vchat.py複製一份，重命名爲achat.py，簡單修改幾處，最終音頻捕獲、傳輸的完整代碼以下。
我將上面代碼中的Server和Client分別加上Video和Audio前綴以區分，同時顯示給用戶的print輸出語句也作了必定修改，
對於視頻加上VIDEO前綴，音頻加上AUDIO前綴。若是你對代碼中使用到的 PyAudio 提供的庫函數有所疑問，
複製代碼
1 class Audio_Server(threading.Thread):
2 def __init__(self, port, version) :
3 threading.Thread.__init__(self)
4 self.setDaemon(True)
5 self.ADDR = ('', port)
6 if version == 4:
7 self.sock = socket(AF_INET ,SOCK_STREAM)
8 else:
9 self.sock = socket(AF_INET6 ,SOCK_STREAM)
10 self.p = pyaudio.PyAudio()
11 self.stream = None
12 def __del__(self):
13 self.sock.close()
14 if self.stream is not None:
15 self.stream.stop_stream()
16 self.stream.close()
17 self.p.terminate()
18 def run(self):
19 print("AUDIO server starts...")
20 self.sock.bind(self.ADDR)
21 self.sock.listen(1)
22 conn, addr = self.sock.accept()
23 print("remote AUDIO client success connected...")
24 data = "".encode("utf-8")
25 payload_size = struct.calcsize("L")
26 self.stream = self.p.open(format=FORMAT,
27 channels=CHANNELS,
28 rate=RATE,
29 output=True,
30 frames_per_buffer = CHUNK
31 )
32 while True:
33 while len(data) < payload_size:
34 data += conn.recv(81920)
35 packed_size = data[:payload_size]
36 data = data[payload_size:]
37 msg_size = struct.unpack("L", packed_size)[0]
38 while len(data) < msg_size:
39 data += conn.recv(81920)
40 frame_data = data[:msg_size]
41 data = data[msg_size:]
42 frames = pickle.loads(frame_data)
43 for frame in frames:
44 self.stream.write(frame, CHUNK)
45
46 class Audio_Client(threading.Thread):
47 def __init__(self ,ip, port, version):
48 threading.Thread.__init__(self)
49 self.setDaemon(True)
50 self.ADDR = (ip, port)
51 if version == 4:
52 self.sock = socket(AF_INET, SOCK_STREAM)
53 else:
54 self.sock = socket(AF_INET6, SOCK_STREAM)
55 self.p = pyaudio.PyAudio()
56 self.stream = None
57 def __del__(self) :
58 self.sock.close()
59 if self.stream is not None:
60 self.stream.stop_stream()
61 self.stream.close()
62 self.p.terminate()
63 def run(self):
64 print("AUDIO client starts...")
65 while True:
66 try:
67 self.sock.connect(self.ADDR)
68 break
69 except:
70 time.sleep(3)
71 continue
72 print("AUDIO client connected...")
73 self.stream = self.p.open(format=FORMAT,
74 channels=CHANNELS,
75 rate=RATE,
76 input=True,
77 frames_per_buffer=CHUNK)
78 while self.stream.is_active():
79 frames = []
80 for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
81 data = self.stream.read(CHUNK)
82 frames.append(data)
83 senddata = pickle.dumps(frames)
84 try:
85 self.sock.sendall(struct.pack("L", len(senddata)) + senddata)
86 except:
87 break
複製代碼
至此咱們完成了 vchat.py 的編寫。
4.6 編寫程序入口 main.py
爲了提供用戶參數解析，代碼使用了argparse。你可能對此前幾個類中初始化方法的self.setDaemon(True)有疑惑。
這個方法的調用使每一個線程在主線程結束以後自動退出，保證程序不會出現崩潰且沒法銷燬的狀況。在main.py中，
咱們經過每隔1s作一次線程的保活檢查，若是視頻/音頻中出現阻塞/故障，主線程會終止。
複製代碼
1 import sys
2 import time
3 import argparse
4 from vchat import Video_Server, Video_Client
5 from achat import Audio_Server, Audio_Client
6
7 parser = argparse.ArgumentParser()
8
9 parser.add_argument('--host', type=str, default='127.0.0.1')
10 parser.add_argument('--port', type=int, default=10087)
11 parser.add_argument('--level', type=int, default=1)
12 parser.add_argument('-v', '--version', type=int, default=4)
13
14 args = parser.parse_args()
15
16 IP = args.host
17 PORT = args.port
18 VERSION = args.version
19 LEVEL = args.level
20
21 if __name__ == '__main__':
22 vclient = Video_Client(IP, PORT, LEVEL, VERSION)
23 vserver = Video_Server(PORT, VERSION)
24 aclient = Audio_Client(IP, PORT+1, VERSION)
25 aserver = Audio_Server(PORT+1, VERSION)
26 vclient.start()
27 aclient.start()
28 time.sleep(1) # make delay to start server
29 vserver.start()
30 aserver.start()
31 while True:
32 time.sleep(1)
33 if not vserver.isAlive() or not vclient.isAlive():
34 print("Video connection lost...")
35 sys.exit(0)
36 if not aserver.isAlive() or not aclient.isAlive():
37 print("Audio connection lost...")
38 sys.exit(0)
複製代碼
最終能夠實現和本身視頻聊天
最終代碼：三個文件
代碼一：achat.py
1 from socket import *
2 import threading
3 import cv2
4 import re
5 import sys
6 import os
7 import time
8 import pyaudio
9 import struct
10 import pickle
11 import zlib
12 import wave
13
14
15 CHUNK = 1024
16 FORMAT = pyaudio.paInt16
17 CHANNELS = 2
18 RATE = 44100
19 RECORD_SECONDS = 5
20 WAVE_OUTPUT_FILENAME = "output.wav"
21
22
23
24 class Audio_Server(threading.Thread):
25 def __init__(self, port, version) :
26 threading.Thread.__init__(self)
27 self.setDaemon(True)
28 self.ADDR = ('', port)
29 if version == 4:
30 self.sock = socket(AF_INET ,SOCK_STREAM)
31 else:
32 self.sock = socket(AF_INET6 ,SOCK_STREAM)
33 self.p = pyaudio.PyAudio()
34 self.stream = None
35 def __del__(self):
36 self.sock.close()
37 if self.stream is not None:
38 self.stream.stop_stream()
39 self.stream.close()
40 self.p.terminate()
41
42 def run(self):
43 print("AUDIO server starts...")
44 self.sock.bind(self.ADDR)
45 self.sock.listen(1)
46 conn, addr = self.sock.accept()
47 print("remote AUDIO client success connected...")
48 data = "".encode("utf-8")
49 payload_size = struct.calcsize("L")
50 self.stream = self.p.open(format=FORMAT,
51 channels=CHANNELS,
52 rate=RATE,
53 output=True,
54 frames_per_buffer = CHUNK
55 )
56
57 while True:
58 while len(data) < payload_size:
59 data += conn.recv(81920)
60 packed_size = data[:payload_size]
61 data = data[payload_size:]
62 msg_size = struct.unpack("L", packed_size)[0]
63 while len(data) < msg_size:
64 data += conn.recv(81920)
65 frame_data = data[:msg_size]
66 data = data[msg_size:]
67 frames = pickle.loads(frame_data)
68 for frame in frames:
69 self.stream.write(frame, CHUNK)
70
71 class Audio_Client(threading.Thread):
72 def __init__(self ,ip, port, version):
73 threading.Thread.__init__(self)
74 self.setDaemon(True)
75 self.ADDR = (ip, port)
76 if version == 4:
77 self.sock = socket(AF_INET, SOCK_STREAM)
78 else:
79 self.sock = socket(AF_INET6, SOCK_STREAM)
80 self.p = pyaudio.PyAudio()
81 self.stream = None
82 def __del__(self) :
83 self.sock.close()
84 if self.stream is not None:
85 self.stream.stop_stream()
86 self.stream.close()
87 self.p.terminate()
88 def run(self):
89 print("AUDIO client starts...")
90 while True:
91 try:
92 self.sock.connect(self.ADDR)
93 break
94 except:
95 time.sleep(3)
96 continue
97 print("AUDIO client connected...")
98 self.stream = self.p.open(format=FORMAT,
99 channels=CHANNELS,
100 rate=RATE,
101 input=True,
102 frames_per_buffer=CHUNK)
103 while self.stream.is_active():
104 frames = []
105 for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
106 data = self.stream.read(CHUNK)
107 frames.append(data)
108 senddata = pickle.dumps(frames)
109 try:
110 self.sock.sendall(struct.pack("L", len(senddata)) + senddata)
111 except:
112 break
代碼二：vchat.py
from socket import *
import threading
import cv2
import re
import time
import sys
import os
import struct
import pickle
import zlib
import wave
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:
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("VEDIO client starts...")
while True:
try:
self.sock.connect(self.ADDR)
break
except:
time.sleep(3)
continue
print("VEDIO 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()
# 服務器端最終代碼以下，增長了對接收到數據的解壓縮處理。
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("VEDIO server starts...")
self.sock.bind(self.ADDR)
self.sock.listen(1)
conn, addr = self.sock.accept()
print("remote VEDIO client success connected...")
data = "".encode("utf-8")
payload_size = struct.calcsize("L")
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
代碼三：main.py
import sys
import time
import argparse
import cv2
import re
import pyaudio
import pickle
import os
import struct
import zlib
import wave
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)
新人學習python，多有不足，謝謝你們觀看。