OpenCV利用Flask+Python+OpenCV实现摄像头读取图像帧网页视频流
Posted 小风_
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了OpenCV利用Flask+Python+OpenCV实现摄像头读取图像帧网页视频流相关的知识，希望对你有一定的参考价值。
利用Flask+Python+OpenCV实现摄像头读取图像帧网页视频流
1.环境准备
python3.x
cv2
flask
2.代码
文件树：
----文件名\\
|----templates\\
| |----index.html
|----app.py
|----base_camera.py
|----camera_opencv.py
2.1 index.html
<html>
<head>
<title>视频流演示</title>
</head>
<body>
<h1>视频流演示</h1>
<img src="{{ url_for('video_feed') }}">
</body>
</html>
说明：该代码块输入在templates文件夹下的index.html
- 其中，图片的路径，对应flask中的
url_for函数，这个函数用于给指定函数（video_feed函数）构造url链接，下面看一下app.py中的相关函数
2.2 app.py
from importlib import import_module
import os
from flask import Flask, render_template, Response
from camera_opencv import Camera
app = Flask(__name__)
@app.route('/')
def index():
"""主页"""
return render_template('index.html')
def gen(camera):
"""视频流生成"""
while True:
frame = camera.get_frame()
yield (b'--frame\\r\\n'
b'Content-Type: image/jpeg\\r\\n\\r\\n' + frame + b'\\r\\n')
@app.route('/video_feed')
def video_feed():
"""视频流的路线。将其放在img标记的src属性中。"""
return Response(gen(Camera()),
mimetype='multipart/x-mixed-replace; boundary=frame')
if __name__ == '__main__':
app.run(host='127.0.0.1', port=9011, threaded=True)
说明：主函数中，定义了服务器的host和port，127.0.0.1代表本地host，9011可以改为其他没有被占用的端口
- 函数一开始，执行
@app.route('/')下的函数，即index()，对应index.html - 在
index.html中，图片的链接对应flask中的url_for()，在该函数中的video_feed()中构造了图片的链接 - 在
video_feed中，使用Response()函数返回数据，而由gen()函数流式地yeild出对应的数据 Camera()中主要利用到opencv中的read函数，读取图像数据
2.3 camera_opencv.py
import os
import cv2
from base_camera import BaseCamera
class Camera(BaseCamera):
video_source = 0
def __init__(self):
if os.environ.get('OPENCV_CAMERA_SOURCE'):
Camera.set_video_source(int(os.environ['OPENCV_CAMERA_SOURCE']))
super(Camera, self).__init__()
@staticmethod
def set_video_source(source):
Camera.video_source = source
@staticmethod
def frames():
camera = cv2.VideoCapture(Camera.video_source)
if not camera.isOpened():
raise RuntimeError('无法打开摄像头')
while True:
_, img = camera.read()
# encode 成为 jpeg image 然后 return it
yield cv2.imencode('.jpg', img)[1].tobytes()
- 本段代码中，利用
cv2.VideoCapture(0)和read()函数不断读取摄像头中的图像数据 - 然后利用
cv2.imencode函数将其编码成jpeg图像并返回
2.4 base_camera.py
import time
import threading
try:
from greenlet import getcurrent as get_ident
except ImportError:
try:
from thread import get_ident
except ImportError:
from _thread import get_ident
class CameraEvent(object):
"""An Event-like class that signals all active clients when a new frame is
available.
"""
def __init__(self):
self.events = {}
def wait(self):
"""Invoked from each client's thread to wait for the next frame."""
ident = get_ident()
if ident not in self.events:
# this is a new client
# add an entry for it in the self.events dict
# each entry has two elements, a threading.Event() and a timestamp
self.events[ident] = [threading.Event(), time.time()]
return self.events[ident][0].wait()
def set(self):
"""Invoked by the camera thread when a new frame is available."""
now = time.time()
remove = None
for ident, event in self.events.items():
if not event[0].isSet():
# if this client's event is not set, then set it
# also update the last set timestamp to now
event[0].set()
event[1] = now
else:
# if the client's event is already set, it means the client
# did not process a previous frame
# if the event stays set for more than 5 seconds, then assume
# the client is gone and remove it
if now - event[1] > 5:
remove = ident
if remove:
del self.events[remove]
def clear(self):
"""Invoked from each client's thread after a frame was processed."""
self.events[get_ident()][0].clear()
class BaseCamera(object):
thread = None # background thread that reads frames from camera
frame = None # current frame is stored here by background thread
last_access = 0 # time of last client access to the camera
event = CameraEvent()
def __init__(self):
"""Start the background camera thread if it isn't running yet."""
if BaseCamera.thread is None:
BaseCamera.last_access = time.time()
# start background frame thread
BaseCamera.thread = threading.Thread(target=self._thread)
BaseCamera.thread.start()
# wait until frames are available
while self.get_frame() is None:
time.sleep(0)
def get_frame(self):
"""Return the current camera frame."""
BaseCamera.last_access = time.time()
# wait for a signal from the camera thread
BaseCamera.event.wait()
BaseCamera.event.clear()
return BaseCamera.frame
@staticmethod
def frames():
""""Generator that returns frames from the camera."""
raise RuntimeError('Must be implemented by subclasses.')
@classmethod
def _thread(cls):
"""Camera background thread."""
print('Starting camera thread.')
frames_iterator = cls.frames()
for frame in frames_iterator:
BaseCamera.frame = frame
BaseCamera.event.set() # send signal to clients
time.sleep(0)
# if there hasn't been any clients asking for frames in
# the last 10 seconds then stop the thread
if time.time() - BaseCamera.last_access > 10:
frames_iterator.close()
print('Stopping camera thread due to inactivity.')
break
BaseCamera.thread = None
3.效果
（打码处理）
4.参考
https://blog.miguelgrinberg.com/post/video-streaming-with-flask
以上是关于OpenCV利用Flask+Python+OpenCV实现摄像头读取图像帧网页视频流的主要内容，如果未能解决你的问题，请参考以下文章
在Azure上使用OpenCV部署Python Flask应用的最佳方法