由于 Flask 集成的帧到字节的 OpenCV 问题

Question

我正在使用以下代码：def gen_frames():  # generate frame by frame from camera    while True:        # Capture frame-by-frame        success, frame = camera.read()  # read the camera frame        if not success:            break        else:            ret, buffer = cv2.imencode('.jpg', frame)            frame = buffer.tobytes()            yield (b'--frame\r\n'                   b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')  # concat frame one by one and show result            #frame_resized = cv2.resize(frame,None,fx=0.75,fy=0.75)            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)我收到一个错误：gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)TypeError: Expected Ptr<cv::UMat> for argument 'src'这是正常的，因为帧数组被转换为字节。但这对于 Flask 中的视频输入来说是必需的：def video_feed():    """Video streaming route. Put this in the src attribute of an img tag."""    return Response(gen_frames(),    mimetype='multipart/x-mixed-replace; boundary=frame')我应该如何使其发挥作用？

Answer 1

已修复，需要移动一些代码。现在的解决方案是：

from flask import Flask, render_template, Response

import numpy as np

import cv2

import pickle

import rtsp

import PIL as Image

import threading

import time

import queue

app = Flask(__name__)

url = 'rtsp://user:[e-mail]:YYYY/stream0/mobotix.mjpeg'

#camera = cv2.VideoCapture(url)  # use 0 for web camera

#  for cctv camera use rtsp://username:password@ip_address:554/user=username_password='password'_channel=channel_number_stream=0.sdp' instead of camera

face_cascade = cv2.CascadeClassifier('cascades\data\haarcascade_frontalface_alt2.xml')

recognizer = cv2.face.LBPHFaceRecognizer_create()

recognizer.read("trainner.yml")

labels = {"person_name": 1}

with open("labels.pickle", 'rb') as f:

    og_labels = pickle.load(f)

    labels = {v:k for k,v in og_labels.items()}

class VideoCapture:

  def __init__(self, name):

    self.cap = cv2.VideoCapture(name)

    self.q = queue.Queue()

    t = threading.Thread(target=self._reader)

    t.daemon = True

    t.start()

  # read frames as soon as they are available, keeping only most recent one

  def _reader(self):

    while True:

      ret, frame = self.cap.read()

      if not ret:

        break

      if not self.q.empty():

        try:

          self.q.get_nowait()   # discard previous (unprocessed) frame

        except queue.Empty:

          pass

      self.q.put(frame)

  def read(self):

    return self.q.get()

def gen_frames():

    cap = VideoCapture(url)

    while True:

        time.sleep(.5)   # simulate time between events

        frame = cap.read()

        frame_detection(frame)

        ret, buffer = cv2.imencode('.jpg', frame)

        frame = buffer.tobytes()

        yield (b'--frame\r\n'

               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')  # concat frame one by one and show result

def frame_detection(frame):

    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    faces = face_cascade.detectMultiScale(gray, scaleFactor=1.3, minNeighbors=5)

    for (x,y,w,h) in faces:

      print(x,y,w,h)

      roi_gray = gray[y:y+h, x:x+w]

      roi_color = frame[y:y+h, x:x+w]

            

      #recognize?

      id_, conf = recognizer.predict(roi_gray)

      if conf>=45: # and conf <=85:

        print(id_)

        print(labels[id_])

        font = cv2.FONT_HERSHEY_SIMPLEX

        name = labels[id_]

        color = (0,0,255)

        stroke = 2

        cv2.putText(frame, name, (x,y), font, 1, color, stroke, cv2.LINE_AA)

        #img_item = "my-image.png"

        #cv2.imwrite(img_item, roi_gray)

        color = (0, 0, 255)

        stroke = 2

        end_cord_x = x + w

        end_cord_y = y + h

        cv2.rectangle(frame, (x,y), (end_cord_x, end_cord_y), color, stroke)

        return frame

@app.route('/video_feed')

def video_feed():

    """Video streaming route. Put this in the src attribute of an img tag."""

    return Response(gen_frames(),

                    mimetype='multipart/x-mixed-replace; boundary=frame')

@app.route('/')

def index():

    """Video streaming home page."""

    return render_template('index.html')

if __name__ == '__main__':

    app.run(host='0.0.0.0')