参考链接：
读取多个(海康大华)网络摄像头的视频流 (使用opencv-python)，解决实时读取延迟问题：https://zhuanlan.zhihu.com/p/38136322
websockets：
https://github.com/aaugustin/websockets
---------------------------------------------------------------------------------
流程： 使用opencv采取多进程实时读取网络摄像头， 经过算法之后采用websocket视频流传给摄像头。
关键代码：具体参考第一篇博客。
import multiprocessing as mp
...
img_queues = [mp.Queue(maxsize=2) for _ in camera_ip_l] # queue
...
q.put(frame) if is_opened else None # 线程A不仅将图片放入队列
q.get() if q.qsize() > 1 else time.sleep(0.01) # 线程A还负责移除队列中的旧图
服务端代码：
import time
import multiprocessing as mp
import threading
from queue import Queue
import cv2
import numpy as np
import asyncio
import websockets
from websockets import ConnectionClosed
frame = None
def websocket_process(img_dict):
# 服务器端主逻辑
async def main_logic(websocket, path):
await recv_msg(websocket, img_dict)
# new_loop = asyncio.new_event_loop()
# asyncio.set_event_loop(new_loop)
start_server = websockets.serve(main_logic, '0.0.0.0', 5678)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
async def recv_msg(websocket, img_dict):
recv_text = await websocket.recv()
if recv_text == 'begin':
while True:
frame = img_dict['img']
if isinstance(frame, np.ndarray):
enconde_data = cv2.imencode('.png', frame)[1]
enconde_str = enconde_data.tostring()
try:
await websocket.send(enconde_str)
except Exception as e:
print(e)
return True
def image_put(q, user, pwd, ip):
rtsp_url = 'rtsp://{0}:{1}@{2}:554/h265/ch1/main/av_stream'.format(user, pwd, ip)
cap = cv2.VideoCapture(rtsp_url)
i = 0
while True:
ret, frame = cap.read()
if ret:
frame = cv2.resize(frame, (500, 500))
q.put(frame)
q.get() if q.qsize() > 1 else time.sleep(0.01)
def image_get(q, img_dict):
while True:
frame = q.get()
if isinstance(frame, np.ndarray):
img_dict['img'] = frame
def run_single_camera(user_name, user_pwd, camera_ip):
mp.set_start_method(method='spawn') # init
# queue = Queue(maxsize=3)
# threads = [threading.Thread(target=image_put, args=(queue, user_name, user_pwd, camera_ip)),
# threading.Thread(target=image_get, args=(queue, )),
# threading.Thread(target=websocket_process)]
#
# [thread.start() for thread in threads]
# [thread.join() for thread in threads]
queue = mp.Queue(maxsize=3)
m = mp.Manager()
img_dict = m.dict()
Processes = [mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)),
mp.Process(target=image_get, args=(queue, img_dict)),
mp.Process(target=websocket_process, args=(img_dict, ))]
[process.start() for process in Processes]
[process.join() for process in Processes]
def run():
run_single_camera('admin', 'admin', '192.168.101.65')
if __name__ == '__main__':
run()
前端代码：
<!DOCTYPE HTML>
<html lang="zh">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0">
<script src="js/jquery-2.1.0.js" type="text/javascript" charset="utf-8"></script>
<title>websocket通信客户端</title>
<script type="text/javascript">
function WebSocketTest()
{
if ("WebSocket" in window)
{
// 打开一个 web socket
var ws = new WebSocket("ws://192.168.101.95:5678");
// 连接建立后的回调函数
ws.onopen = function()
{
// Web Socket 已连接上，使用 send() 方法发送数据
ws.send("begin");
};
// 接收到服务器消息后的回调函数
ws.onmessage = function (evt)
{
var received_msg = evt.data;
// alert("收到消息："+received_msg);
console.log("要输出的内容" + received_msg);
blobToDataURI(received_msg,function(result){ //blob格式再转换为base64格式
document.getElementById('img').src = result;
})
var imgUrl=tobase(received_msg)
// $('#img').html('<img src="'+imgUrl+'" />')
};
// 连接关闭后的回调函数
ws.onclose = function()
{
// 关闭 websocket
alert("连接已关闭...");
};
}
else
{
// 浏览器不支持 WebSocket
alert("您的浏览器不支持 WebSocket!");
}
}
function blobToDataURI(blob, callback) {
var reader = new FileReader();
reader.readAsDataURL(blob);
reader.onload = function (e) {
callback(e.target.result);
}
}
</script>
</head>
<body onload="WebSocketTest()">
<div><img src="" id="img" alt=""></div>
</body>
</html>