使用 FastAPI 和 OpenCV 的视频流应用程序-IT科技

摘要：问题描述：我正在尝试渲染一个HTML显示网络摄像头视频流的页面。但是，我遇到了以下错误：500 Server Error TypeError: TemplateResponse() missing 1 required positional argument: 'context' 我的 FastAPI 应用：...

问题描述：

我正在尝试渲染一个HTML显示网络摄像头视频流的页面。但是，我遇到了以下错误：

500 Server Error TypeError: TemplateResponse() missing 1 required positional argument: 'context'

我的 FastAPI 应用：

from fastapi import FastAPI
import uvicorn
from fastapi import Depends, FastAPI
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
import cv2

app = FastAPI(debug=True)
templates = Jinja2Templates(directory="templates")

@app.get("/")
async def index():
    return templates.TemplateResponse("index.html")


async def gen_frames(camera_id):
    cap=  cv2.VideoCapture(0)

    while True:
        # for cap in caps:
        # # Capture frame-by-frame
        success, frame = cap.read()  # read the camera frame
        if not success:
            break
        else:
            ret, buffer = cv2.imencode('.jpg', frame)
            frame = buffer.tobytes()
            yield (b'--frame
'b'Content-Type: image/jpeg

' + frame + b'
')  
    
   if __name__ == '__main__':
    uvicorn.run(app,  host="127.0.0.1",port=8000)

我的 HTML 页面 (index.html)：

<!doctype html>
<html lang="en">
<head>
    <!-- Required meta tags -->
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">

    <!-- Bootstrap CSS -->
    <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.1.3/css/bootstrap.min.css"
          integrity="sha384-MCw98/SFnGE8fJT3GXwEOngsV7Zt27NXFoaoApmYm81iuXoPkFOJwJ8ERdknLPMO" crossorigin="anonymous">

    <title>Multiple Live Streaming</title>
</head>
<body>
<div class="container">
    <div class="row">
        <div class="col-lg-7">
            <h3 class="mt-5">Multiple Live Streaming</h3>
            <img src="{{ url_for('video_feed', id='0') }}" width="100%">
        </div>
    </div>
</div>
</body>
</html>

追溯：

错误捕获

解决方案 1：

使用模板时您需要传递请求。

from fastapi import Request

@app.get("/")
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

因此，您也可以使用StreamingResponse将视频作为另一条路径提供

from fastapi.responses import StreamingResponse


@app.get("/serve/{camera_id}", include_in_schema=False)
async def serve_video(camera_id: int):
    return StreamingResponse(gen_frames(camera_id))

然后使用 Ajax 或 Axios 等获取该响应。

解决方案 2：

关于你问题中提到的错误，即：

 TemplateResponse() missing 1 required positional argument: 'context'

使用时Templates，需要使用作为键传递字典Request中的对象，如下所示（更新：如上面的文档链接所述，FastAPI/Starlette 最近做了一些更改，包括不再将对象作为的一部分传递，以及模板的不再是的第一个参数（用代替它）。但是，如相关源代码所示，Starlette 似乎仍然支持（但不确定会持续多久）两种创建的方式，即使从源代码中可以看出最初的一种现在已经弃用了。有关更多详细信息，请参阅上面的链接）：context`TemplateResponserequestrequestcontextnameTemplateResponserequest`TemplateResponse

from fastapi import Request

@app.get('/')
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

下面给出了两个使用 FastAPI 和流式传输（实时）视频的选项（带有完整的代码示例）OpenCV。选项 1演示了一种基于您的问题使用HTTP协议和 FastAPI/Starlette 的方法StreamingResponse。选项 2使用协议，它可以轻松处理高清视频流，并得到 FastAPI/Starlette 的支持（文档可以在这里和这里WebSocket找到）

我还建议看一下这个答案，以便更好地理解Python 中的async/ await，特别是 FastAPI ，以及何时使用async def或 normal在 FastAPI 中定义端点def。例如，cv2库的函数（例如camera.read()和cv2.imencode()）是同步阻塞（IO 绑定或 CPU 绑定）函数，这意味着，当调用它们时，它们会阻塞事件循环，直到它们完成。这就是选项 1 中的StreamingResponse生成器（即gen_frames()函数）和/video_feed端点都使用 normal 定义的原因def，因为这会导致对该端点的任何请求都在与随后将被编辑的外部线程池不同的线程中运行await（因此，FastAPI 仍将异步工作）。但需要注意的是，在的情况下StreamingResponse，即使使用定义端点/video_feed并使用async def定义StreamingResponse生成器def（因为它包含阻塞操作），FastAPI/Starlette 为了防止事件循环被阻塞，也会使用在单独的线程中iterate_in_threadpool()运行生成StreamingResponse器，然后将其await编辑（类似于 FastAPI 对def端点的操作）——有关更多详细信息和相关源代码，请参阅此答案。因此，任何对该async def端点的请求都不会因为某些阻塞StreamingResponse生成器而阻塞事件循环。请注意，如果生成器包含阻塞操作，则应避免使用定义生成器async def，因为 FastAPI/Starlette 会直接在事件循环中运行它，而不是在单独的线程内。只有async def当您确定内部没有发生会阻塞事件循环的同步await操作时，和/或需要用于协程/async函数时，才使用定义它。

在选项 2 中，/ws必须使用定义端点async def，因为 FastAPI/Starlette 的websockets函数是async函数，需要进行await编辑。在这种情况下，cv2阻塞函数将在每次调用时阻塞事件循环，直到它们完成（考虑到完成这些操作所需的时间以及项目的要求，这可能是也可能不是微不足道的；例如，应用程序是否需要同时为多个请求/用户提供服务）。但是，仍然可以在端点cv2内部运行阻塞操作（例如的函数），而不会阻塞事件循环，并且可以同时运行应用程序的多个实例（工作程序/进程）。请查看上面链接的答案以获取有关此主题的更多详细信息和解决方案。async def

选项 1 - 使用`HTTP`协议

您可以通过http://127.0.0.1:8000/访问直播。

应用程序

import cv2
import time
import uvicorn
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import StreamingResponse

app = FastAPI()
camera = cv2.VideoCapture(0, cv2.CAP_DSHOW)
templates = Jinja2Templates(directory="templates")


def gen_frames():
    while True:
        success, frame = camera.read()
        if not success:
            break
        else:
            ret, buffer = cv2.imencode('.jpg', frame)
            frame = buffer.tobytes()
            yield (b'--frame
'
                   b'Content-Type: image/jpeg

' + frame + b'
')
        time.sleep(0.03)


@app.get('/')
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})


@app.get('/video_feed')
def video_feed():
    return StreamingResponse(gen_frames(), media_type='multipart/x-mixed-replace; boundary=frame')


if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000, debug=True)

模板/index.html

<!DOCTYPE html>
<html>
    <body>
        <div class="container">
            <h3> Live Streaming </h3>
            <img src="{{ url_for('video_feed') }}" width="50%">
        </div>
    </body>
</html>

选项 2 - 使用`WebSocket`协议

您可以通过http://127.0.0.1:8000/访问直播。使用该协议的相关答案可以在这里、这里和这里WebSocket找到。

应用程序

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from websockets.exceptions import ConnectionClosed
from fastapi.templating import Jinja2Templates
import uvicorn
import asyncio
import cv2

app = FastAPI()
camera = cv2.VideoCapture(0,cv2.CAP_DSHOW)
templates = Jinja2Templates(directory="templates")


@app.get('/')
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})


@app.websocket("/ws")
async def get_stream(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            success, frame = camera.read()
            if not success:
                break
            else:
                ret, buffer = cv2.imencode('.jpg', frame)
                await websocket.send_bytes(buffer.tobytes()) 
            await asyncio.sleep(0.03)
    except (WebSocketDisconnect, ConnectionClosed):
        print("Client disconnected")   

 
if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000)

下面是HTML建立WebSocket连接、接收图像字节和创建BlobURL（在图像加载后释放，以便对象随后被垃圾收集，而不是不必要地保存在内存中）的模板，如下所示，以在浏览器中显示视频帧。

模板/index.html

<!DOCTYPE html>
<html>
    <head>
        <title>Live Streaming</title>
    </head>
    <body>
        <img id="frame" src="">
        <script>
            let ws = new WebSocket("ws://localhost:8000/ws");
            let image = document.getElementById("frame");
            image.onload = function(){
                URL.revokeObjectURL(this.src); // release the blob URL once the image is loaded
            } 
            ws.onmessage = function(event) {
                image.src = URL.createObjectURL(event.data);
            };
        </script>
    </body>
</html>

下面也是一个基于websockets库和的Python 客户端OpenCV，您可以使用它连接到服务器，以便在 Python 应用程序中接收和显示视频帧。

客户端.py

from websockets.exceptions import ConnectionClosed
import websockets
import numpy as np
import asyncio
import cv2


async def main():
    url = 'ws://127.0.0.1:8000/ws'
    
    async for websocket in websockets.connect(url):
        try:
             #count = 1
             while True:
                contents = await websocket.recv()
                arr = np.frombuffer(contents, np.uint8)
                frame = cv2.imdecode(arr, cv2.IMREAD_UNCHANGED)
                cv2.imshow('frame', frame)
                cv2.waitKey(1)
                
                #cv2.imwrite("frame%d.jpg" % count, frame)
                #count += 1
        except ConnectionClosed:
            continue  # attempt reconnecting to the server (otherwise, call `break` instead)


asyncio.run(main())

问题描述：

解决方案 1：

解决方案 2：

选项 1 - 使用HTTP协议

选项 2 - 使用WebSocket协议

云端的项目管理软件

选项 1 - 使用`HTTP`协议

选项 2 - 使用`WebSocket`协议