概要

ImageFlux Live StreamingのWebRTC to WebRTC配信は、遅延0.5秒以下の超低遅延配信が可能である。また、株式会社時雨堂が開発運用しているSora互換で動作する。
WebRTCは基本的にブラウザから配信できることを強みとしたプロトコルであるが、今回は敢えて、Raspberry Pi 4からコマンドで配信してみた。同じく株式会社時雨堂が開発しているWebRTCネイティブクライアント、MomoがRaspberry Pi 4のハードウェアエンコーダにも対応しているため、気軽に実装することが可能である。「単純な配信」項でコマンドを解説している。
また、今回は遠隔でRaspberry Pi 4からの配信を開始・終了できるようにもしてみた。こちらはロジックが複雑になるため、大まかな動かし方だけ解説する。
なにはともあれ、まずは単純にRaspberry Pi 4からMomoを使って配信してみよう。

単純な配信

環境

Raspberry Pi 4
- Raspberry Pi OS
- メモリ4GB
Raspberry Pi Camera Module 3

コマンド

Momoの動かし方についてはGitHub上に記載がある。

明瞭かつ簡潔な文書なのでこれを読むだけで初学者でも配信まではできるだろう。
まずはRaspberry Pi 4上でMomoを取得・展開する。

wget https://github.com/shiguredo/momo/releases/download/2025.1.3/momo-2025.1.3_raspberry-pi-os_armv8.tar.gz
tar -zxvf momo-2025.1.3_raspberry-pi-os_armv8.tar.gz

展開したフォルダ内にあるmomoがバイナリなので、基本はそれを呼び出すだけで良い。
今回はRaspberry Pi Camera Module 3を使うので、ガイドに記載がある設定も反映する。

sudo modprobe bcm2835-v4l2 max_video_width=2592 max_video_height=1944

あとはmomoに実行権限を付与して動かせばよい。ImageFlux Live StreamingのImageFlux_20200316.CreateMultistreamChannel APIを呼び出し、取得したチャンネルIDとSora URLを転記するだけだ。

curl -H "Content-Type: application/json" -H "X-Sora-Target: ImageFlux_20200316.CreateMultistreamChannel" -H "Authorization: Bearer [ImageFlux Live Streamingのアクセストークン]" -d "{}" https://live-api.imageflux.jp/
# 応答値：{"channel_id":"チャンネルID","sora_url":"Sora URL"}
./momo --no-audio-device --force-i420 --use-libcamera --resolution FHD --framerate 30 --log-level 1 sora --signaling-urls [取得したSora URL] --channel-id [取得したチャンネルID] --audio false --video-codec-type H264 --video-bit-rate 4000 --role sendonly

私のRaspberry Pi 4では、特にパッケージのインストールはしなくても動作した。最新のRaspberry Pi OSであれば大丈夫なのかもしれない。

遠隔配信

Raspberry Pi 4を気軽に操作できる環境であれば上記のようにコマンドで配信すれば良いが、そうもいかないのが世の常だろう。
特に複数台を監視用途で用いたい場合は遠隔で制御できる方が便利である。
そこで今回は、IoTでよく使われるMQTTプロトコルを用いて遠隔で配信開始・終了を制御できるようにした。また、せっかくなのでIoTらしく、BME280というセンサーモジュールから温度・湿度・気圧を取得して表示、さらには配信時には赤色LEDが点灯するような仕掛けも作ってみた。
部材は秋月電子通商で調達した。

アーキテクチャ

さくらのクラウドで構築した。基本的には、中継サーバをかますだけである。

動作機序は大まかに以下の通り：

Raspberry Pi 4は中継サーバから配信制御用MQTTトピックをSubscribeする。また、別トピックでセンサー情報をPublishする。
Webブラウザから配信開始要求を受け取った中継サーバは、ImageFlux Live Streamingのチャンネルを作成し、Raspberry Pi 4に配信開始を命じるMQTT・Publishを行う。同時に、WebブラウザにチャンネルIDとSora URLを返却する。
Raspberry Pi 4は、受け取ったチャンネルにMomoを用いて映像をIngestする。
Webブラウザは受け取った情報を元に配信の視聴を開始する。センサー情報も定期取得・更新する。

というもの。配信終了時は視聴終了し、中継サーバを通じてチャンネル削除をするだけだ。
サーバ上では、視聴用の簡易Webページのホスト、APIサーバのコンテナ、MQTTブローカーのMosquittoコンテナの合計３つを動作させている。
記事用に動かしているだけなので、認証・認可は特に実装していない。本番運用する場合は、ネットワーク制御とID管理は必須だろうし、状況に応じて、認証Webhookも使うことを推奨する。

プログラム

中継サーバのロジックは省略し、Raspberry Pi 4で動作させているPythonを掲載する。

from gpiozero import LED
import paho.mqtt.client as mqtt
import json
import signal
import subprocess
import threading
from datetime import datetime, timezone

import smbus2
import bme280

# MQTTブローカーの接続設定
BROKER = "中継サーバのドメイン"
PORT = 1883
TOPIC = "rpi/command"
TOPIC_TELEMETRY_BME280 = "rpi/telemetry/bme280"
USERNAME = ""
PASSWORD = ""

# momoの実行ファイルパス
MOMO_PATH = "./momo"

# LED接続GPIOピン番号
GPIO_GREEN = 17
GPIO_RED = 27

# BME280のI2C設定
I2C_PORT = 1
I2C_ADDRESS = 0x76
TELEMETRY_INTERVAL_SEC = 0.1
DEVICE_ID = "Raspberry Pi 4"

momo_proc = None
green_led = None
red_led = None
mqtt_client = None
sensor_bus = None
calibration_params = None
stop_event = threading.Event()

def shutdown_handler(signum, frame):
    """終了シグナル受信時にストリームとLEDを停止するハンドラ。"""
    _ = signum
    _ = frame
    stop_stream()
    stop_event.set()
    cleanup_led()
    cleanup_sensor()
    raise SystemExit(0)

def init_sensor() -> None:
    """BME280センサーを初期化する。"""
    global sensor_bus, calibration_params

    try:
        sensor_bus = smbus2.SMBus(I2C_PORT)
        calibration_params = bme280.load_calibration_params(sensor_bus, I2C_ADDRESS)
        print("BME280を初期化しました。")
    except Exception as exc:
        sensor_bus = None
        calibration_params = None
        print(f"BME280の初期化に失敗しました：{exc}")

def cleanup_sensor() -> None:
    """BME280センサー用I2Cバスをクローズする。"""
    global sensor_bus

    if sensor_bus is not None:
        try:
            sensor_bus.close()
        except Exception:
            pass
        sensor_bus = None

def read_sensor_payload() -> dict | None:
    """BME280から1回分の値を読み取り、MQTT送信用ペイロードを返す。"""
    if sensor_bus is None or calibration_params is None:
        return None

    try:
        data = bme280.sample(sensor_bus, I2C_ADDRESS, calibration_params)
        return {
            "device_id": DEVICE_ID,
            "temperature_c": round(float(data.temperature), 2),
            "humidity_pct": round(float(data.humidity), 2),
            "pressure_hpa": round(float(data.pressure), 2),
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
    except Exception as exc:
        print(f"BME280の読み取りに失敗しました：{exc}")
        return None

def publish_telemetry() -> None:
    """最新のBME280値をMQTTへ送信する。"""
    if mqtt_client is None or not mqtt_client.is_connected():
        return

    payload = read_sensor_payload()
    if payload is None:
        return

    mqtt_client.publish(
        TOPIC_TELEMETRY_BME280,
        json.dumps(payload),
        qos=1,
        retain=True,
    )

def telemetry_loop() -> None:
    """一定間隔でBME280テレメトリを送信する。"""
    while not stop_event.is_set():
        publish_telemetry()
        stop_event.wait(TELEMETRY_INTERVAL_SEC)

def init_led() -> None:
    """LEDを初期化し、待機状態の表示にする。"""
    global green_led, red_led

    green_led = LED(GPIO_GREEN)
    red_led = LED(GPIO_RED)
    set_idle_led()

def set_idle_led() -> None:
    """待機状態を示す緑色LEDを点灯する。"""
    if green_led is None or red_led is None:
        return
    green_led.on()
    red_led.off()

def set_streaming_led() -> None:
    """配信中状態を示す赤色LEDを点灯する。"""
    if green_led is None or red_led is None:
        return
    green_led.off()
    red_led.on()

def cleanup_led() -> None:
    """LEDを消灯し、GPIOリソースを解放する。"""
    global green_led, red_led

    if green_led is None or red_led is None:
        return
    green_led.off()
    red_led.off()
    green_led.close()
    red_led.close()
    green_led = None
    red_led = None

def start_stream(payload: dict) -> None:
    """受信したコマンドに基づいてmomoを起動し、配信を開始する。"""
    global momo_proc

    # すでに配信中の場合は何もしない
    if momo_proc is not None and momo_proc.poll() is None:
        return

    # ペイロードから配信パラメータを取得
    channel_id = payload["channel_id"]
    sora_url = payload["sora_ingest_url"]
    resolution = payload.get("resolution", "FHD")
    framerate = str(payload.get("frame_rate", 30))
    bitrate = str(payload.get("bit_rate", 4000))

    # momoのコマンドライン引数を組み立てる
    cmd = [
        MOMO_PATH,
        "--no-audio-device",
        "--force-i420",
        "--use-libcamera",
        "--resolution", resolution,
        "--framerate", framerate,
        "--log-level", "1",
        "sora",
        "--signaling-urls", sora_url,
        "--channel-id", channel_id,
        "--audio", "false",
        "--video-codec-type", "H264",
        "--video-bit-rate", bitrate,
        "--role", "sendonly",
    ]

    momo_proc = subprocess.Popen(cmd)
    set_streaming_led()

def stop_stream() -> None:
    """配信中のmomoプロセスを終了する。"""
    global momo_proc

    if momo_proc is None:
        return

    if momo_proc.poll() is None:
        # プロセスがまだ終了していない場合は終了シグナルを送信する。TERMINATEシグナルを送信しても終了しない場合は、強制終了する。
        momo_proc.terminate()
        try:
            momo_proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            momo_proc.kill()

    momo_proc = None
    set_idle_led()

def on_connect(client, userdata, flags, rc, properties=None):
    """MQTTブローカー接続時にコマンドトピックを購読する。"""
    client.subscribe(TOPIC)

def on_message(client, userdata, msg):
    """MQTTメッセージ受信時に配信開始・終了処理を行う。"""
    payload = json.loads(msg.payload.decode("utf-8"))
    command = payload.get("command")

    if command == "start":
        start_stream(payload)
    elif command == "stop":
        stop_stream()

def main() -> None:
    """主処理。LED、シグナルハンドラ、MQTTクライアントを初期化する。"""
    global mqtt_client

    init_led()
    init_sensor()
    signal.signal(signal.SIGTERM, shutdown_handler)
    signal.signal(signal.SIGINT, shutdown_handler)

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    mqtt_client = client

    # MQTTユーザー名とパスワードを設定する（必要な場合）
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)

    client.on_connect = on_connect
    client.on_message = on_message

    client.connect(BROKER, PORT)

    telemetry_thread = threading.Thread(target=telemetry_loop, daemon=True)
    telemetry_thread.start()

    client.loop_forever()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        stop_event.set()
        stop_stream()
        cleanup_led()
        cleanup_sensor()
        raise SystemExit(0)

やっていることは単純で、MQTTを受け取ったら、MQTT内のパラメータに応じてMomoで配信を開始しているだけである。
Pythonの仮想環境で

pip install paho-mqtt gpiozero smbus2 RPi.bme280

を叩いた後、実行すれば動作する。

配信

SPA側にアクセスすると、センサーデータが表示される。センサーデータは、プロセスを動かしている時点で、制御用とは別のMQTTトピックを通じて常時送出されているからである（もちろん配信開始後に送出するように組むことも可能だ）。

配信開始を命令すると、

無事配信が開始された。私が愛してやまない名馬、ガイアフォースのぬいぐるみの頭が映っている。
センサーデータも0.1秒おきにちゃんと更新された。実際は1秒おきで十分だろう。

LEDも無事点灯している。

まとめ

今回は、ImageFlux Live Streamingとさくらのクラウドを活用して、Raspberry Pi 4からの遠隔配信を実現してみた。
これを応用することで、安価な超低遅延監視システムを構築することが可能である。
ぜひ試してみて欲しい。