Pythonで画像処理プログラムを開発している際に、cv2.VideoCapture(x)で撮影した映像を、GUI無し(imshow無し)で手軽に表示したい、配信したいというユースケースは割と多いと感じます。そこで今回はPythonでWebサーバーを立てることができるFlaskを利用して、USBカメラにより撮影した画像をHTTPにより配信するサンプルプログラムを作成してみました。
実行結果
ハマりポイント
なお、本記事では撮影にRaspberry Piのカメラモジュールではなく、USBカメラを利用しています。理由はOpenCVからカメラモジュールを利用するのは手間がかかりそうなためです...
プログラムを実行する手順
今回作成したプログラムはGitHubにpushされています。ですので、下記の手順を辿ることにより、みなさまのお手元にてストリーミング配信機能を備えたPython製のWebサーバーをお試しいただけます。Pythonのスクリプトを「python app.py」として実行した後に、ブラウザでReaspberry Pi 5に割り当てられたIPアドレスのポート5000にアクセスして、/streams/stream を開くことにより、HTTPによるストリーミング配信を見ることができます。
#####
## 本手順ではパッケージ管理ツールuvを使います
#####
# パッケージ管理ツールのUVをインストールする
$ curl -LsSf https://astral.sh/uv/install.sh | sh
$ source ~/.bashrc
# ソースコードをGitHubより取得する
$ git clone https://github.com/xtrizeShino/opencv-web-sandbox.git
$ cd opencv-web-sandbox
# v0.1.0のタグのソースコードを取得
$ git checkout -b v0.1.0 refs/tags/v0.1.0
# 作業ディレクトリへ移動する
$ cd opencv-web-sandbox
# uvの仮想環境をセットアップする
$ uv sync
# 仮想環境に入る
$ . .venv/bin/activate
# app.pyを実行
(opencv-web-sandbox) $ python app.py
# * Serving Flask app 'web_root'
# * Debug mode : off
# * Running on akk addresses (0.0.0.0)
# * Running on http://127.0.0.1:5000
# * Running on http://192.168.10.140:5000
# Press CTRL+C to quit
############
###【終了するにはCtrl+c】
############
なお、本手順ではパッケージ管理ツールのuvを使っています。uvってなぁに？という方は下記の記事をご参照ください。
プログラムの概要
それでは、プログラムの内容を一緒に見ていきましょう。ファイルが若干多いですが、見るべきポイントは「 opencv-web-sandbox/opencv-web-sandbox/app.py 」と「 opencv-web-sandbox/opencv-web-sandbox/web_root/__init__.py 」そして「 opencv-web-sandbox/opencv-web-sandbox/web_root/views/streams.py 」です。
1. Webサーバーを起動する（app.py）
まず、起動スクリプト「app.py」にて「web_root」ディレクトリにある、Webサーバーを定義するFlaskクラスのインスタンス「app」を参照し、「run(host="0.0.0.0")」を実行し、Webコンテンツを公開します。runを実行すると、Flaskのデフォルトのポート番号(5000)でWebサーバが起動します。なお、hostを指定しない場合、127.0.0.1以外からのアクセスが禁止されますので注意してください。 ここで扱うappインスタンスは、最初にweb_rootが参照された際に実行される__init__.pyにより生成されています。
from web_root import app
# entry point to run Flask class
# -->> Flask class is generated in __init__.py
if __name__ == '__main__':
app.run(host="0.0.0.0")
Webサーバーのインスタンスを定義する（__init__.py）
__init__.pyは、Flaskのインスタンスappを作成します。作成後、最後のimport文にて指定したviewsとstreamsにより、web_root/viewsディレクトリ以下にある「views.py」と「streams.py」を呼び出し、配信するコンテンツを定義します。
from flask import Flask
app = Flask(__name__)
app.config.from_object('web_root.config')
from web_root.views import views, streams
2. ストリーミング配信のフレームを生成する（streams.py）
streams.pyにて定義するCameraクラスはストリーミング配信を構成するフレームを生成するクラスです。クラスのコンストラクタにてUSBカメラにcv2.VideoCapture(0)でアクセスします。 またCameraクラスはコンストラクタの引数として画像加工するコールバック関数を取得しており、ここで指定する関数を変更することにより撮影した映像を自由に加工することが可能です。 get_frameは、USBカメラをreadして画像を取得し、取得した画像をコールバック関数にて加工した後に、cv2.imencodeにより画像をJPEG形式へと変換し、ストリーミング配信を構成するフレームを出力します。
from flask import render_template, Response
from web_root import app
import cv2
class Camera:
def __init__(self, callback):
# Open Video device
self.video = cv2.VideoCapture(0)
self.imageproc = callback
def __del__(self):
# Close Video device
self.video.release()
def get_frame(self):
# fetch new frame from Video device
_, image = self.video.read()
image = self.imageproc(image)
if image is not None:
# Set Quarity for compress with jpg
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), app.config['WEBCAM_QUARITY']]
# Convert jpg on memory
_, frame = cv2.imencode('.jpg', image, encode_param)
else:
# Video devices are not found.
frame = None
# return Image as jpg-binary
return frame
### ... (続く) ...
3. ストリーミング配信をする(streams.py)
ストリーミング配信のコンテンツは、HTTPのレスポンスとして「multipart/x-mixed-replace」を返す必要があります。また、そのコンテンツにはフレームとして「Content-Type: image/jpeg」を含まなれけばなりません。そこで、下記の処理にて、前述のCameraクラスのget_frameにより取得したフレームをvideo_feedでストリーミング配信のコンテンツへと変換しています。フレームは関数を呼ばれるたびにyeildで毎回再生成されます。また、video_feedはResponseでHTTPの応答を返します。
### ... (上の続き) ...
# Generate Frame in Streaming Contents
def gen(camera):
while True:
# get Image as jpg-binary
frame = camera.get_frame()
if frame is not None:
# update frame in Multimedia
yield (b"--frame\r\n"
b"Content-Type: image/jpeg\r\n\r\n"
+ frame.tobytes() + b"\r\n")
### ... (省略) ...
# Streaming Contents
@app.route("/streams/video_feed")
def video_feed():
# return Newest Multimedia data
return Response(gen(Camera(imageproc_x2)),
mimetype="multipart/x-mixed-replace; boundary=frame")
4. ストリーミング配信を表示する(streams/stream.html)
ストリーミング配信されたデータはHTMLの「<img src="">」を利用して表示されます。このタグは「streams/stream.html」にて定義しています。{{ url_for('video_feed') }}はvideo_feed関数に対応するURLを生成するFlaskの機能です。
### ... (上の続き) ...
# HTML for put Streaming Contents
@app.route("/streams/stream")
def stream():
return render_template("streams/stream.html")
<!DOCTYPE html>
<head>
<title>Live View</title>
<link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='index.css') }}" />
</head>
<body>
<h1>Streaming</h1>
<img src="{{ url_for('video_feed') }}" />
</body>
5. 補助機能(streams.py)
前述しましたが、今回のプログラムでは、拡張性をもたせるためにWebカメラから取得した画像をJPEG形式へと変換する前に、一旦コールバック関数を呼び出すようになっています。本タグではサンプルとして現在の実装では画像サイズを2倍に引き伸ばす「imageproc_x2」を定義し、コールバック関数に指定しています。
# Callback function for Image Proc
def imageproc_x2(image):
h, w = image.shape[:2]
image = cv2.resize(image, (w*2, h*2))
return image
以上が、USBカメラの映像をPythonにストリーミング配信する方法です。本実装によりGUI無しでOpenCVに対応した画像データを加工し、処理結果を簡単にブラウザで確認することができます。PyTorch等と組み合わせ、ディープラーニングによる物体検出などと組み合わせても面白いかもしれませんね。AIKitが手元にあるので後日トライしてみます。是非ご期待ください！