Flask-SocketIO で BME280 センサーデータを配信

📝 はじめに

Flask + Socket.IO を使って、IoT センサー（BME280）から収集したデータをブラウザにリアルタイム配信する方法を解説します。

対象読者

• Web 経由で IoT データを可視化したい人
• REST API のポーリングから WebSocket に移行したい人

前提条件

• Python 3.10 以上、Flask の基本知識
• 前回の記事
  Raspberry Pi + Docker で BME280 センサーデータ収集 #Python - Qiita
  が完了していること

---

🎯 背景・動機

IoT センサーデータを Web で表示する場合、REST API を定期的にポーリングする方法が一般的ですが、遅延や無駄な通信が発生します。WebSocket を使うことでサーバーからのプッシュ型配信が可能になり、これらの問題を解決できます。

---

システム構成

┌─────────────────┐     WebSocket      ┌─────────────────┐
│  SensorClient   │ ────────────────→ │  Flask Server   │
│  (BME280)       │   sense_data       │  (SocketIO)     │
└─────────────────┘                    └────────┬────────┘
                                                │ broadcast
                                                ↓
                                       ┌─────────────────┐
                                       │  Web Browser    │
                                       └─────────────────┘

---

🛠️ 手順

1. 必要なパッケージ

[tool.poetry.dependencies]
flask = "^3.0.3"
flask-socketio = "^5.3.6"
python-socketio = "^5.11.3"
eventlet = "^0.36.1"

2. Flask-SocketIO サーバーの実装

from flask import Flask, send_from_directory
from flask_socketio import SocketIO, emit
from datetime import datetime

app = Flask(__name__, static_folder="dist", static_url_path="")
socketio = SocketIO(app, async_mode='eventlet')

@app.route('/')
def index():
    return send_from_directory(app.static_folder, "index.html")

@socketio.on('connect')
def ws_connect():
    app.logger.info('クライアントが接続しました')

@socketio.on('sense_data')
def ws_sense_data(sense_data):
    """センサーデータ受信時の処理"""
    sense_data['timestamp'] = str(datetime.now())
    app.logger.debug(f"{sense_data['host']}: {sense_data['temperature']}°C")

socketio.run(app, host="0.0.0.0", port=5010)

eventlet を選んだ理由: 軽量で Flask との相性が良く、monkey patching で既存コード対応が容易。

3. センサークライアントの実装

from socketio import Client
import time
import os

class SensorClient:
    def __init__(self, server_url='http://localhost:5010', interval=10):
        self.server_url = server_url
        self.interval = interval
        self.client = None
        self.hostname = os.getenv('HOSTNAME', 'unknown-host')

    def connect_to_server(self):
        try:
            self.client = Client()
            self.client.connect(self.server_url, retry=True)
            return True
        except Exception as e:
            print(f'接続失敗: {e}')
            return False

    def send_sensor_data(self, data):
        if self.client and self.client.connected:
            data['host'] = self.hostname
            self.client.emit('sense_data', data)
            return True
        return False

    def run(self):
        if not self.connect_to_server():
            return
        try:
            while True:
                data = self.read_sensor_data()
                if data:
                    self.send_sensor_data(data)
                time.sleep(self.interval)
        except KeyboardInterrupt:
            pass
        finally:
            if self.client:
                self.client.disconnect()

---

🚨 つまずきポイント

接続が切れる問題

長時間接続が維持されない場合は、ping/pong の設定を確認します。

socketio = SocketIO(app, async_mode='eventlet', ping_timeout=60, ping_interval=25)

CORS 設定

開発時にフロントエンドとバックエンドが別ポートで動作する場合:

socketio = SocketIO(app, cors_allowed_origins="*")

---

📈 まとめ

• リアルタイム通信で UX が大幅に向上: ポーリング不要で即座にデータ反映
• Flask-SocketIO の使いやすさ: デコレータベースで直感的に実装可能
• 双方向通信の可能性: センサー→ブラウザだけでなく、ブラウザ→センサーの制御も可能