使用 FastAPI 和 OpenCV 的视频流应用程序

2025-02-14 09:50:00
admin
原创
42
摘要:问题描述:我正在尝试渲染一个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())
相关推荐
  政府信创国产化的10大政策解读一、信创国产化的背景与意义信创国产化,即信息技术应用创新国产化,是当前中国信息技术领域的一个重要发展方向。其核心在于通过自主研发和创新,实现信息技术应用的自主可控,减少对外部技术的依赖,并规避潜在的技术制裁和风险。随着全球信息技术竞争的加剧,以及某些国家对中国在科技领域的打压,信创国产化显...
工程项目管理   1579  
  为什么项目管理通常仍然耗时且低效?您是否还在反复更新电子表格、淹没在便利贴中并参加每周更新会议?这确实是耗费时间和精力。借助软件工具的帮助,您可以一目了然地全面了解您的项目。如今,国内外有足够多优秀的项目管理软件可以帮助您掌控每个项目。什么是项目管理软件?项目管理软件是广泛行业用于项目规划、资源分配和调度的软件。它使项...
项目管理软件   1355  
  信创产品在政府采购中的占比分析随着信息技术的飞速发展以及国家对信息安全重视程度的不断提高,信创产业应运而生并迅速崛起。信创,即信息技术应用创新,旨在实现信息技术领域的自主可控,减少对国外技术的依赖,保障国家信息安全。政府采购作为推动信创产业发展的重要力量,其对信创产品的采购占比情况备受关注。这不仅关系到信创产业的发展前...
信创和国产化的区别   8  
  信创,即信息技术应用创新产业,旨在实现信息技术领域的自主可控,摆脱对国外技术的依赖。近年来,国货国用信创发展势头迅猛,在诸多领域取得了显著成果。这一发展趋势对科技创新产生了深远的推动作用,不仅提升了我国在信息技术领域的自主创新能力,还为经济社会的数字化转型提供了坚实支撑。信创推动核心技术突破信创产业的发展促使企业和科研...
信创工作   9  
  信创技术,即信息技术应用创新产业,旨在实现信息技术领域的自主可控与安全可靠。近年来,信创技术发展迅猛,对中小企业产生了深远的影响,带来了诸多不可忽视的价值。在数字化转型的浪潮中,中小企业面临着激烈的市场竞争和复杂多变的环境,信创技术的出现为它们提供了新的发展机遇和支撑。信创技术对中小企业的影响技术架构变革信创技术促使中...
信创国产化   8  
热门文章
项目管理软件有哪些?
云禅道AD
禅道项目管理软件

云端的项目管理软件

尊享禅道项目软件收费版功能

无需维护,随时随地协同办公

内置subversion和git源码管理

每天备份,随时转为私有部署

免费试用