参考链接：
读取多个(海康大华)网络摄像头的视频流 (使用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>