背景

前回shapelyを使って矩形の中に居るかどうかを判別しました。
判別状態を処理を行うのとは別の計算機で結果を確認したかったので、結果の画像をストリーム配信してみました。
備忘録として行った内容を記事に残します。

使ったもの

ストリーム配信用計算機
Raspberry Pi 3B + Raspberry Pi OS liteで動かしました。
ストリーム受信用計算機
ブラウザを使えるならスマートフォンでも良いです。
表示確認時にも言及しますが、なぜかfirefoxでは表示がちらついてしまい見づらいのでchromeを使うのが良いです。
配信するLAN
配信機も受信機も同一LAN（WiFiや有線LAN）に繋ぎます。

サーバーライブラリflaskをインストール

pythonのサーバーライブラリflaskを使ってサーバーを立ち上げるので、ストリーミングを行う計算機にインストールします。
pipでインストールできます。

pip install flask --break-system-packages

動作確認したflaskのバージョンは3.1.0です。

プログラム解説

全体を共有後に要所を解説します。

全体像

下記のプログラムを実行すると3030ポートで計算機外部からも接続可能なサーバーが起動します。

#!/usr/bin/python
import time
from flask import Flask, Response
from shapely import Polygon, Point
import matplotlib.pyplot as plt
import io
import threading

arrPoligon = [
    Polygon([(36.17615365206663, 139.29274439262977),
             (36.176069213304274, 139.29292611226978),
             (36.17615744098333, 139.29299383803968),
             (36.17625487020672, 139.2928342466215)]),
    Polygon([(36.17610006593953, 139.29270885336436),
             (36.17600209524877, 139.29289124355657),
             (36.17592306907745, 139.29284095214328),
             (36.17603565428372, 139.29261698771606)]),
]

arrPoint = [
    Point(36.17614534102644, 139.29286477985536),
    Point(36.17608958979661, 139.2928252172769),
    Point(36.17601110498835, 139.29275883261136),
]
image_bytes = b''

def thread_update_image_bytes():
    global image_bytes
    while True:
        for point in arrPoint:
            for polygon in arrPoligon:
                plt.fill(*polygon.exterior.xy, color="green")
            isPointInSomePolygon = False
            for polygon in arrPoligon:
                if polygon.contains(point):
                    # print("%s contains %s" % (polygon, point))
                    isPointInSomePolygon = True
                    break
            plt.scatter(*point.xy, color= "orange" if isPointInSomePolygon else "black")

            plt.plot()
            buf = io.BytesIO()
            plt.savefig(buf, format='png')
            buf.seek(0)
            plt.gca().clear()
            new_bytes = buf.read()
            image_bytes = new_bytes
            print("update image %d" % time.time())
            time.sleep(1)

app = Flask(__name__)

@app.route("/")
def app_route():
    return "<p>Hello, world</p><a href=\"/stream_image\">see image stream</a>"

def watch_image():
    global image_bytes
    while True:
        if image_bytes is not None:
            yield(b'--frame\r\n'
                  b'Content-Type: image/png\r\n\r\n'+ image_bytes + b'\r\n')
        time.sleep(0.1) # my Firefox needs some time to display image / Chrome displays image without it

@app.route("/stream_image")
def stream_image():
    return Response(watch_image(), mimetype="multipart/x-mixed-replace; boundary=frame")

def thread_server():
    app.run(port=3030, host="0.0.0.0")

for thread_fn in [
    thread_update_image_bytes,
    thread_server,
]:
    t = threading.Thread(target=thread_fn)
    t.start()

画像データはbytes形式で受け渡す

shapelyのポリゴンと点をmatplotlibで描画したデータはio.BytesIOに対して保存を行うことでメモリに保有したままbytes形式のデータに変換できます。

import matplotlib.pyplot as plt
import io

def thread_update_image_bytes():
    global image_bytes
    while True:
        for point in arrPoint:
            # plot data process
            buf = io.BytesIO()
            plt.savefig(buf, format='png')
            buf.seek(0)
            plt.gca().clear()
            new_bytes = buf.read()
            image_bytes = new_bytes
            time.sleep(1)

参考
matplotlibでRGBヒストグラムを描画し、画像に埋め込んでPythonによりストリーミング配信する

ブラウザなどから接続されたら実行し続ける関数で画像を配信

この記事のコードでは/stream_imageのパスで画像をストリーム配信します。

def watch_image():
    global image_bytes
    while True:
        if image_bytes is not None:
            yield(b'--frame\r\n'
                  b'Content-Type: image/png\r\n\r\n'+ image_bytes + b'\r\n')
        time.sleep(0.1) # my Firefox needs some time to display image / Chrome displays image without it

@app.route("/stream_image")
def stream_image():
    return Response(watch_image(), mimetype="multipart/x-mixed-replace; boundary=frame")

参考
Image streaming with OpenCV and flask - Why imencode is needed?

画像更新とサーバーの動作は別スレッドで実行

サーバーの処理と画像の更新処理は別に行う必要があるので、python環境に標準で組み込まれているthreadingを利用して並列処理しました。

import threading

def thread_update_image_bytes():
    global image_bytes
    while True:
        for point in arrPoint:
            # ...
            time.sleep(1)

def thread_server():
    app.run(port=3030, host="0.0.0.0")

for thread_fn in [
    thread_update_image_bytes,
    thread_server,
]:
    t = threading.Thread(target=thread_fn)
    t.start()

動作確認

先程のプログラムをストリーム配信したい計算機で動かし、その計算機のLANのIPアドレスの3030ポートをブラウザで開くとHello worldページが表示されます。

IPアドレスはlinuxなら下記のコマンドで確認できます。

ip a | grep 192

「see image stream」のリンクをクリックすると/stream_image ページに移り、動く画像を見れます。

なぜかfirefoxではガタつく

firefoxで今回作成した画像ストリームページを見ると、真っ白な画像が時々表示されてガタつきます。
ubuntuでもandroidでも同様でした。
不具合が解消する方法をご存知でしたら、コメントなどで教えていただけると嬉しいです。

おわり

shapelyで作った形状をmatplotlibで描画し、その画像をflaskでストリーム配信できました。
モニタを繋いでいない計算機の処理結果を別の計算機のブラウザで表示できて便利です。

参考

matplotlibでRGBヒストグラムを描画し、画像に埋め込んでPythonによりストリーミング配信する
Image streaming with OpenCV and flask - Why imencode is needed?

試行錯誤な日々

ページ

2025年1月26日日曜日

shapelyとmatplotlibで作る画像をストリーム配信

背景