python使用Opencv實時讀取攝像頭進行處理之后傳到前端

參考鏈接：

讀取多個(海康\大華)網絡攝像頭的視頻流 (使用opencv-python)，解決實時讀取延遲問題：https://zhuanlan.zhihu.com/p/38136322

websockets：

https://github.com/aaugustin/websockets

---------------------------------------------------------------------------------

流程： 使用opencv采取多進程實時讀取網絡攝像頭， 經過算法之后采用websocket視頻流傳給攝像頭。

關鍵代碼：具體參考第一篇博客。

1.  
   import multiprocessing as mp
2.  
   ...
3.  
   img_queues = [mp.Queue(maxsize= 2) for _ in camera_ip_l] # queue
4.  
   ...
5.  
   q.put(frame) if is_opened else None # 線程A不僅將圖片放入隊列
6.  
   q.get() if q.qsize() > 1 else time.sleep(0.01) # 線程A還負責移除隊列中的舊圖

服務端代碼：

1.  
   import time
2.  
   import multiprocessing as mp
3.  
   import threading
4.  
   from queue import Queue
5.  
   import cv2
6.  
   import numpy as np
7.  
   import asyncio
8.  
   import websockets
9.  
   from websockets import ConnectionClosed
10.  
     
11.  
     
12.  
    frame = None
13.  
     
14.  
     
15.  
    def websocket_process(img_dict):
16.  
    # 服務器端主邏輯
17.  
    async def main_logic(websocket, path):
18.  
    await recv_msg(websocket, img_dict)
19.  
     
20.  
    # new_loop = asyncio.new_event_loop()
21.  
    # asyncio.set_event_loop(new_loop)
22.  
    start_server = websockets.serve(main_logic, '0.0.0.0', 5678)
23.  
    asyncio.get_event_loop().run_until_complete(start_server)
24.  
    asyncio.get_event_loop().run_forever()
25.  
     
26.  
     
27.  
     
28.  
    async def recv_msg(websocket, img_dict):
29.  
    recv_text = await websocket.recv()
30.  
    if recv_text == 'begin':
31.  
    while True:
32.  
    frame = img_dict[ 'img']
33.  
    if isinstance(frame, np.ndarray):
34.  
    enconde_data = cv2.imencode( '.png', frame)[1]
35.  
    enconde_str = enconde_data.tostring()
36.  
    try:
37.  
    await websocket.send(enconde_str)
38.  
    except Exception as e:
39.  
    print(e)
40.  
    return True
41.  
     
42.  
     
43.  
    def image_put(q, user, pwd, ip):
44.  
    rtsp_url = 'rtsp://{0}:{1}@{2}:554/h265/ch1/main/av_stream'.format(user, pwd, ip)
45.  
    cap = cv2.VideoCapture(rtsp_url)
46.  
    i = 0
47.  
    while True:
48.  
    ret, frame = cap.read()
49.  
    if ret:
50.  
    frame = cv2.resize(frame, ( 500, 500))
51.  
    q.put(frame)
52.  
    q.get() if q.qsize() > 1 else time.sleep(0.01)
53.  
     
54.  
     
55.  
    def image_get(q, img_dict):
56.  
    while True:
57.  
    frame = q.get()
58.  
    if isinstance(frame, np.ndarray):
59.  
    img_dict[ 'img'] = frame
60.  
     
61.  
     
62.  
     
63.  
     
64.  
     
65.  
    def run_single_camera(user_name, user_pwd, camera_ip):
66.  
    mp.set_start_method(method= 'spawn') # init
67.  
    # queue = Queue(maxsize=3)
68.  
    # threads = [threading.Thread(target=image_put, args=(queue, user_name, user_pwd, camera_ip)),
69.  
    # threading.Thread(target=image_get, args=(queue, )),
70.  
    # threading.Thread(target=websocket_process)]
71.  
    #
72.  
    # [thread.start() for thread in threads]
73.  
    # [thread.join() for thread in threads]
74.  
     
75.  
    queue = mp.Queue(maxsize= 3)
76.  
    m = mp.Manager()
77.  
    img_dict = m.dict()
78.  
    Processes = [mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)),
79.  
    mp.Process(target=image_get, args=(queue, img_dict)),
80.  
    mp.Process(target=websocket_process, args=(img_dict, ))]
81.  
     
82.  
    [process.start() for process in Processes]
83.  
    [process.join() for process in Processes]
84.  
     
85.  
     
86.  
    def run():
87.  
    run_single_camera( 'admin', 'admin', '192.168.101.65')
88.  
     
89.  
     
90.  
     
91.  
    if __name__ == '__main__':
92.  
    run()

前端代碼：

1.  
   <! DOCTYPE HTML>
2.  
   < html lang="zh">
3.  
   < head>
4.  
   < meta charset="utf-8">
5.  
   < meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0">
6.  
   < script src="js/jquery-2.1.0.js" type="text/javascript" charset="utf-8"></script>
7.  
   < title>websocket通信客戶端</title>
8.  
   < script type="text/javascript">
9.  
   function WebSocketTest()
10.  
    {
11.  
    if ("WebSocket" in window)
12.  
    {
13.  
    // 打開一個 web socket
14.  
    var ws = new WebSocket(" ws://192.168.101.95:5678");
15.  
     
16.  
    // 連接建立后的回調函數
17.  
    ws.onopen = function()
18.  
    {
19.  
    // Web Socket 已連接上，使用 send() 方法發送數據
20.  
    ws.send("begin");
21.  
    };
22.  
     
23.  
    // 接收到服務器消息后的回調函數
24.  
    ws.onmessage = function (evt)
25.  
    {
26.  
    var received_msg = evt.data;
27.  
     
28.  
    // alert("收到消息："+received_msg);
29.  
    console.log("要輸出的內容" + received_msg);
30.  
    blobToDataURI(received_msg,function(result){ //blob格式再轉換為base64格式
31.  
    document.getElementById('img').src = result;
32.  
    })
33.  
    var imgUrl=tobase(received_msg)
34.  
    // $('#img').html('<img src="'+imgUrl+'" />')
35.  
    };
36.  
     
37.  
    // 連接關閉后的回調函數
38.  
    ws.onclose = function()
39.  
    {
40.  
    // 關閉 websocket
41.  
    alert("連接已關閉...");
42.  
    };
43.  
    }
44.  
    else
45.  
    {
46.  
    // 瀏覽器不支持 WebSocket
47.  
    alert("您的瀏覽器不支持 WebSocket!");
48.  
    }
49.  
    }
50.  
     
51.  
    function blobToDataURI(blob, callback) {
52.  
    var reader = new FileReader();
53.  
    reader.readAsDataURL(blob);
54.  
    reader.onload = function (e) {
55.  
    callback(e.target.result);
56.  
    }
57.  
    }
58.  
     
59.  
     
60.  
     
61.  
    </script>
62.  
    </head>
63.  
     
64.  
    <body onload="WebSocketTest()">
65.  
    <div><img src="" id="img" alt=""></div>
66.  
    </body>
67.  
    </html>