上文提到,我已经实现了将 OpenCV 获取到的视频流通过 Flask 推送到Web浏览器上,本文在此基础上,增加了哥 OpenCV 上的人脸追踪功能,使用了官方提供的人脸数据集(如果自己机器学习的话,需要花费大量的人力物力财力,请各位量力而行!!


一、用到的技术

  1. Python 3.10
  2. OpenCV 4.4.5+
  3. Flask模块

二、demo设想

近些年由于互联网的发展,市面上出现了Ip网络监控摄像机,这样大家在任何地方都能通过摄像机厂家提供的系统来进行实时查看监控的画面,但有些时候需要定制画面,这些系统多少会有点局限性,这样,利用OpenCV、Flask制作一个可将自己的IP摄像机画面推流到web浏览器上显示,将会很方便(设想)

三、实现方法

1. 使用OpenCV获取摄像头实时视频流并呈现在创建的窗口中

此时,主要使用 cap = cv2.VideoCapture(0) 创建一个 VideoCapture 获取摄像头的视频流,调用本地摄像头括号内可以是0、1(一般情况下设备最多两个视频采集设备),若调用准备好的视频文件,则按照以下写法 cap = cv2.VideoCapture(‘视频文件的路径’)

设置好摄像头的分辨率(一般默认摄像头的物理分辨率即可):

# 调节摄像头分辨率
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)

在结束所有服务并关闭窗口后,需要释放申请的资源:

# 当一切结束后,释放VideoCapture对象
cap.release()
cv2.destroyAllWindows()

整体的实现代码:

# 使用 Python3.10 版本

import cv2

# 创建一个窗口 名字叫做CameraLive
cv2.namedWindow('CameraLive', flags=cv2.WINDOW_NORMAL | cv2.WINDOW_KEEPRATIO | cv2.WINDOW_GUI_EXPANDED)

# 创建一个VideoCapture
cap = cv2.VideoCapture(0)

print('摄像头是否开启: {}'.format(cap.isOpened()))

# 显示缓存数
print(cap.get(cv2.CAP_PROP_BUFFERSIZE))
# 设置缓存区的大小
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)

# 调节摄像头分辨率
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)

print(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
print(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

# 设置FPS
print('setfps', cap.set(cv2.CAP_PROP_FPS, 25))
print(cap.get(cv2.CAP_PROP_FPS))

while (True):
    # 逐帧捕获
    ret, frame = cap.read()  # 第一个参数返回一个布尔值(True/False),代表有没有读取到图片;第二个参数表示截取到一帧的图片
    # gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    cv2.imshow('CameraLive', frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

# 当一切结束后,释放VideoCapture对象
cap.release()
cv2.destroyAllWindows()

2. 使用Flask模块将通过OpenCV获取到的视频流推流到Web浏览器

Flask是一个用Python便携的Web应用程序框架(微架构),其基于 Werkzeug WSGI 工具包和 Jinja2 模版引擎,属于Pocco项目。

要想使用 Flask 模块,则需在项目中引用,“Flask类的一个对象是我们的 WSGI 应用程序,其构造函数使用 当前模块(name)的名称作为参数,其 router() 函数是一个装饰器,它告诉应用程序哪个 URL 应该调用相关的函数”,如:

from flask import Flask
app = Flask(__name__)

@app.route('/')
def hello_world():
   return 'Hello World'

if __name__ == '__main__':
   app.run()

将 Flask 的微架构与 OpenCV 调取摄像头视频流进行结合,可以创建一个 **视频流微服务器 **并将视频推流到Web上,局域网内的所有设备均可以访问查看这个被调用的视频流,这个视频流此时是被共享的,程序运行时访问 http://127.0.0.1:9000/video_feed 即可访问被共享的视频流。相关代码:

# Python3.10 版本

from flask import Flask, render_template, Response
import cv2


class VideoCamera(object):
    def __init__(self):
        # 通过opencv获取实时视频流
        self.video = cv2.VideoCapture(0)

    def __del__(self):
        self.video.release()

    def get_frame(self):
        success, image = self.video.read()
        # 因为opencv读取的图片并非jpeg格式,因此要用motion JPEG模式需要先将图片转码成jpg格式图片
        ret, jpeg = cv2.imencode('.jpg', image)
        return jpeg.tobytes()


app = Flask(__name__)


@app.route('/')  # 主页
def index():
    # jinja2模板,具体格式保存在index.html文件中
    return render_template('index.html')


def gen(camera):
    while True:
        frame = camera.get_frame()
        # 使用generator函数输出视频流, 每次请求输出的content类型是image/jpeg
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n')


@app.route('/video_feed')  # 这个地址返回视频流响应
def video_feed():
    return Response(gen(VideoCamera()),
                    mimetype='multipart/x-mixed-replace; boundary=frame')


if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=True, port=9000)

3. 在此基础上,进行人脸追踪的实现

已经有了调用并共享本地摄像头视频流,且推流到Web浏览器上的基础,此时要想实现人脸追踪,主要需要实现的是对人脸进行识别并通过线给圈画起来,为了节省时间,这里选择使用 OpenCV 官方提供的人脸数据集 haarcascade_frontalface_default.xml (只提供个大致的脸部数据,满足大部分人脸特征,若需提高识别的准确度,则需要自己创建算法使用大量的人脸图片进行机器学习),这个数据集一般情况下在安装Python环境的时候已经被安装了,你只需要找到并复制到你的项目文件夹中即可,大致路径是 /python3.10/site-packages/cv2/data/haarcascade_frontalface_default.xml

因为opencv读取的图片并非jpeg格式,因此要用motion JPEG模式需要先将图片转码成jpg格式图片:

success, jpeg = cv2.imencode('.jpg', image)
return jpeg.tobytes()

主要代码部分:

def get_frame(self):
    while True:
        # 调用模型库文件
        face_cascade = cv2.CascadeClassifier('haarcascade_frontalface_default.xml')
        # 读取视频帧
        success, image = self.video.read()
        # 图像灰度处理
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        # 设定人脸识别参数
        faces = face_cascade.detectMultiScale(gray, scaleFactor=1.3, minNeighbors=3)
        faceNum = len(faces)
        print("人脸数量: %s" % faceNum)

        if len(faces) > 0:
            for faceRect in faces:
                x, y, w, h = faceRect
                # -------- 在人脸周围绘制矩形
                cv2.rectangle(image, (x, y), (x + w, y + h), (255, 255, 0), 2)
        # 因为opencv读取的图片并非jpeg格式,因此要用motion JPEG模式需要先将图片转码成jpg格式图片
        success, jpeg = cv2.imencode('.jpg', image)
        return jpeg.tobytes()
    # 释放资源
    cv2.destroyAllWindows()
    cap.release()
  
 # 创建一个gen()函数,使用generator函数输出视频流,完成视频流的格式的输出转换
def gen(camera):
    while True:
        frame = camera.get_frame()
        # 使用generator函数输出视频流, 每次请求输出的content类型是image/jpeg
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n')

 # 创建一个live()函数,并在router中将其绑定到‘/live’规则
@app.route('/live')  # 这个地址返回视频流响应
def live():
    return Response(gen(VideoCamera()),
                    mimetype='multipart/x-mixed-replace; boundary=frame')

一个完整的代码实现:

from flask import Flask, render_template, Response
import cv2


class VideoCamera(object):
    def __init__(self):
        # 通过opencv获取实时视频流
        self.video = cv2.VideoCapture(0)
        self.faceNum = 0

    def __del__(self):
        self.video.release()

    def get_frame(self):
        while True:
            # 调用模型库文件
            face_cascade = cv2.CascadeClassifier('haarcascade_frontalface_default.xml')
            # 读取视频帧
            success, image = self.video.read()
            # 图像灰度处理
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
            # 设定人脸识别参数
            faces = face_cascade.detectMultiScale(gray, scaleFactor=1.3, minNeighbors=3)
            faceNum = len(faces)
            print("人脸数量: %s" % faceNum)

            if len(faces) > 0:
                for faceRect in faces:
                    x, y, w, h = faceRect
                    # -------- 在人脸周围绘制矩形
                    cv2.rectangle(image, (x, y), (x + w, y + h), (255, 255, 0), 2)
            # 因为opencv读取的图片并非jpeg格式,因此要用motion JPEG模式需要先将图片转码成jpg格式图片
            success, jpeg = cv2.imencode('.jpg', image)
            return jpeg.tobytes()
        # 释放资源
        cv2.destroyAllWindows()
        cap.release()


app = Flask(__name__)


@app.route('/')  # 主页
def index():
    return render_template('index.html')


def gen(camera):
    while True:
        frame = camera.get_frame()
        # 使用generator函数输出视频流, 每次请求输出的content类型是image/jpeg
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n')


@app.route('/live')  # 这个地址返回视频流响应
def live():
    return Response(gen(VideoCamera()),
                    mimetype='multipart/x-mixed-replace; boundary=frame')


if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False, port=9900)

四、总结

除了上述的方法外,应该还会有更简单的代码,目前粗学了下 Flask 和 OpenCV ,应该还有使用不妥当的地方。在 OpenCV 读取视频流时,实际上是进行了图片的读取,而 Flask 模块进行视频流推流也是主要对图片进行操作,图片的高速实时刷新完成了视频的对流,这里对图片格式的转换确认十分重要!!