使用 FastAPI 和 OpenCV 的视频流应用程序
- 2025-02-14 09:50:00
- admin 原创
- 42
问题描述:
我正在尝试渲染一个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
`TemplateResponserequest
requestcontext
nameTemplateResponse
request`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
连接、接收图像字节和创建Blob
URL(在图像加载后释放,以便对象随后被垃圾收集,而不是不必要地保存在内存中)的模板,如下所示,以在浏览器中显示视频帧。
模板/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())