FastAPIを使ってWebカメラ映像をストリーミング配信するAPI

はじめに
必要な環境
APIの実装
サーバー全体
クライアントの実装
まとめ
リポジトリ
1. 関連

はじめに

近年、IoTデバイスの普及に伴い、リアルタイムでのビデオストリーミングが様々な用途で利用されるようになりました。本記事では、Python製のWebフレームワークであるFastAPIを使って、Webカメラからの映像を複数の端末にブロードキャストするAPIの作成方法を解説します。

こちらの記事もおすすめ

Faster WhisperをDockerでワンパン起動させてみた (CUDA12)

Faster Whisperを使用して、手早くかつ効率的に音声認識を行う方法について、Dockerを使った手順を初心者にもわかりやすく解説します。Dockerを用いることで、環境依存を減らし、どのようなマシンでも同じ条件でFaster Wh...

【claude3】YlvaVisionAPI(Webカメラからリアルタイムの画像を取得+ブロードキャストするためのAPI)

必要な環境

Python 3.7以上
FastAPI
OpenCV
Uvicorn

APIの実装

ステップ1: 必要なモジュールのインポート

まず、必要なモジュールをインポートします。FastAPIとOpenCVを使用するため、以下のようにインポートします。

from fastapi import FastAPI, Response
import cv2
import numpy as np
from fastapi.responses import StreamingResponse
import asyncio

ステップ2: FastAPIアプリケーションの作成

FastAPIアプリケーションを作成します。

app = FastAPI()

ステップ3: カメラの設定

使用するWebカメラの設定を行います。カメラのID、解像度、フォーマットを指定します。

camera_id = 0
cap = cv2.VideoCapture(camera_id)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc('M','J','P','G'))

ステップ4: 最新のフレームを保持する変数の定義

最新のフレームを保持するグローバル変数を定義します。

latest_frame = None

ステップ5: カメラからフレームを取得する非同期関数の定義

カメラから連続的にフレームを取得する非同期関数を定義します。取得したフレームはJPEG形式にエンコードされ、latest_frame変数に格納されます。

async def fetch_camera_frame():
    global latest_frame
    while True:
        ret, frame = cap.read()
        if ret:
            _, img_encoded = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
            latest_frame = img_encoded.tobytes()
        await asyncio.sleep(0.01)  # カメラからのフレーム取得間隔

ステップ6: フレームをクライアントに送信する非同期ジェネレータ関数の定義

latest_frame変数に格納されたフレームをクライアントに送信する非同期ジェネレータ関数を定義します。この関数はマルチパートレスポンスを生成し、フレームをJPEG形式で送信します。

async def gen_frames():
    while True:
        if latest_frame is not None:
            yield (b'--frame\r\n'
                   b'Content-Type: image/jpeg\r\n\r\n' + latest_frame + b'\r\n')
        await asyncio.sleep(0.01)  # クライアントへの送信間隔

ステップ7: アプリケーション起動時のイベントハンドラの定義

アプリケーション起動時に、fetch_camera_frame関数をバックグラウンドで実行するためのイベントハンドラを定義します。

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(fetch_camera_frame())

ステップ8: カメラストリームを提供するエンドポイントの定義

/cameraエンドポイントを定義し、gen_frames関数を使用してカメラストリームを提供します。

@app.get("/camera")
async def get_camera_stream():
    return StreamingResponse(gen_frames(), media_type="multipart/x-mixed-replace;boundary=frame")

ステップ9: アプリケーション終了時のイベントハンドラの定義

アプリケーション終了時に、カメラリソースを解放するためのイベントハンドラを定義します。

@app.on_event("shutdown")
async def shutdown_event():
    cap.release()

ステップ10: アプリケーションの実行

Uvicornを使用してアプリケーションを実行します。

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

サーバー全体


from fastapi import FastAPI, Response
import cv2
import numpy as np
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

# カメラ設定
camera_id = 0
cap = cv2.VideoCapture(camera_id)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc('M','J','P','G'))

# 最新のフレームを保持する変数
latest_frame = None

async def fetch_camera_frame():
    global latest_frame
    while True:
        ret, frame = cap.read()
        if ret:
            _, img_encoded = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
            latest_frame = img_encoded.tobytes()
        await asyncio.sleep(0.01)  # カメラからのフレーム取得間隔

async def gen_frames():
    while True:
        if latest_frame is not None:
            yield (b'--frame\r\n'
                   b'Content-Type: image/jpeg\r\n\r\n' + latest_frame + b'\r\n')
        await asyncio.sleep(0.01)  # クライアントへの送信間隔

@app.on_event("startup")
async def startup_event():
    # カメラフレームのフェッチをバックグラウンドで開始
    asyncio.create_task(fetch_camera_frame())

@app.get("/camera")
async def get_camera_stream():
    return StreamingResponse(gen_frames(), media_type="multipart/x-mixed-replace;boundary=frame")

@app.on_event("shutdown")
async def shutdown_event():
    cap.release()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8002)

クライアントの実装

APIを使用してWebカメラ映像を受信し、表示するクライアントの実装例を以下に示します。

import cv2
import requests
import numpy as np

def stream_video(url):
    response = requests.get(url, stream=True)
    byte_buffer = bytes()

    for chunk in response.iter_content(chunk_size=1024):
        byte_buffer += chunk
        a = byte_buffer.find(b'\xff\xd8')
        b = byte_buffer.find(b'\xff\xd9')
        if a != -1 and b != -1:
            jpg = byte_buffer[a:b+2]
            byte_buffer = byte_buffer[b+2:]
            frame = cv2.imdecode(np.frombuffer(jpg, dtype=np.uint8), cv2.IMREAD_COLOR)
            cv2.imshow('Video Stream', frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    cv2.destroyAllWindows()

stream_url = 'http://localhost:8001/camera'
stream_video(stream_url)

このクライアントは、指定されたURLからビデオストリームを受信し、OpenCVを使用して表示します。