ysk-hack-log
戻る

MQTT連携でIoTダッシュボードを作る｜複数デバイスの値を集約（Raspberry Pi）【連載⑦】

「MQTT連携でIoTダッシュボードを作る｜複数デバイスの値を集約（Raspberry Pi）【連載⑦】」の記事アイキャッチ

前回（第6回） まででPi単体のモニタリングが完成しました。今回は MQTT を使い、他のデバイスの値も集約して、複数センサーを1画面で見る IoTダッシュボードにします。しきい値を超えたらLEDで警告もします。

Table of contents

Open Table of contents

MQTTとは

MQTT は、IoT向けの軽量なpublish/subscribe（発行/購読）型メッセージング方式です。

送信側と受信側は直接つながらず、必ずブローカーを経由します。「トピック名で投函・受け取りする郵便局」とイメージすると分かりやすいです。

flowchart LR
    P1["送信側（publisher）"]
    P2["送信側（publisher）"]
    B["ブローカー（Mosquitto）"]
    S1["受信側（subscriber）"]
    S2["受信側（subscriber）"]

    P1 --> B
    P2 --> B
    B -->|topicで振り分け| S1
    B -->|topicで振り分け| S2

ブローカーがあることで、送信側は受信側のIPや存在を知らなくてよく（疎結合）、デバイスが増えても「ブローカーに繋ぐだけ」で多対多の配信が成立します。

このプロジェクトでの送信側・受信側

役割誰かトピック
ブローカー（中継）Pi上の Mosquitto（localhost:1883）—
送信側① Piアプリ自身（PublishAsync）sensors/pi/temperature
② 他デバイス／テスト送信（mosquitto_pub）sensors/kitchen/... など
受信側Piアプリの MqttService（購読）→ ダッシュボード・LED警告sensors/+/temperature

ポイントは、Piアプリは「送信側」でもあり「受信側」でもあること。自分の温度を発行しつつ、sensors/+/temperature（+＝任意の1階層）で全デバイスの値を集約します。mosquitto_pub は「別デバイスの送信側」を手動で演じるものです。

一般的にはどれが担う？

実世界では、それぞれ次のようなものが担います。

役割一般的な担い手
ブローカーMosquitto／EMQX／HiveMQ。クラウドなら AWS IoT Core／Azure IoT Hub（Event Grid の MQTT）など
送信側センサー・マイコン（ESP32／Arduino／M5Stack／Raspberry Pi）
受信側ダッシュボード（Grafana／Node-RED／Home Assistant／自作Web）、DB保存、通知、別デバイスの制御

多くのデバイスは送信も受信も兼ねます（状態を発行しつつ、制御コマンドを購読）。

クラウド上のアプリ（例：Azure）は、テレメトリを受けて処理・可視化するなら受信側、デバイスへ制御コマンドを送るなら送信側（両方兼ねることも多い）。Azure IoT Hub などのマネージドサービスはブローカーの役割です。

ブローカー（Mosquitto）をPiに入れる

sudo apt install -y mosquitto mosquitto-clients

他デバイスからも接続できるよう、/etc/mosquitto/conf.d/local.conf を作成します（家庭内LAN前提の簡易設定）。

listener 1883 0.0.0.0
allow_anonymous true
sudo systemctl restart mosquitto

⚠️ allow_anonymous true は家庭内の検証用です。実運用では認証（ユーザー/パスワード）やTLSを設定してください。

動作確認（別ターミナルで送信）:

mosquitto_pub -h localhost -t sensors/livingroom/temperature -m 28.5

MQTTnet を追加

dotnet add package MQTTnet

MQTTサービス（購読・集約・LED警告）

バックグラウンドでブローカーに接続し、sensors/+/temperature（+ は1階層のワイルドカード）を購読してデバイスごとの最新値を集約します。しきい値超えで第3回の GpioService を使ってLED点灯します。

// Services/MqttService.cs
using System.Collections.Concurrent;
using MQTTnet;

public sealed class MqttService : BackgroundService
{
    private const double Threshold = 30.0;
    private readonly GpioService _gpio;
    private readonly ConcurrentDictionary<string, double> _latest = new();
    private IMqttClient? _client;

    public IReadOnlyDictionary<string, double> Latest => _latest;
    public event Action? Changed;

    public MqttService(GpioService gpio) => _gpio = gpio;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var factory = new MqttClientFactory();      // ← v5。旧v4は MqttFactory
        _client = factory.CreateMqttClient();

        _client.ApplicationMessageReceivedAsync += e =>
        {
            // topic: sensors/{deviceId}/temperature
            var parts = e.ApplicationMessage.Topic.Split('/');
            var deviceId = parts.Length >= 2 ? parts[1] : "unknown";
            var payload = e.ApplicationMessage.ConvertPayloadToString();

            if (double.TryParse(payload, out var value))
            {
                _latest[deviceId] = value;
                _gpio.SetLed(_latest.Values.Any(v => v >= Threshold)); // しきい値でLED
                Changed?.Invoke();
            }
            return Task.CompletedTask;
        };

        var options = new MqttClientOptionsBuilder()
            .WithTcpServer("localhost", 1883)
            .WithClientId("pi-dashboard")
            .Build();

        var subOptions = factory.CreateSubscribeOptionsBuilder()
            .WithTopicFilter("sensors/+/temperature")
            .Build();

        // 接続確立時（再接続時も）に購読し直す
        _client.ConnectedAsync += async _ =>
            await _client.SubscribeAsync(subOptions, stoppingToken);

        // ⚠️ ブローカー未起動でも落とさない：例外を握って5秒ごとに再接続
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                if (!_client.IsConnected)
                    await _client.ConnectAsync(options, stoppingToken);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"[MQTT] 接続待機中… ({ex.Message})");
            }
            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
        }
    }

    // Pi自身の温度を発行（ダッシュボードにも自分が並ぶ）
    public async Task PublishAsync(double celsius)
    {
        if (_client is null || !_client.IsConnected) return; // 未接続なら何もしない
        var msg = new MqttApplicationMessageBuilder()
            .WithTopic("sensors/pi/temperature")
            .WithPayload(celsius.ToString("F1"))
            .Build();
        await _client.PublishAsync(msg);
    }
}

Program.cs に登録します。

builder.Services.AddSingleton<MqttService>();
builder.Services.AddHostedService(sp => sp.GetRequiredService<MqttService>());

Pi自身の温度も並べたいときは、第5回の TemperatureMonitor から MqttService.PublishAsync(Current) を呼べば、sensors/pi/temperature として集約に加わります。

ダッシュボードで一覧表示

@* Components/Pages/Dashboard.razor *@
@page "/dashboard"
@rendermode InteractiveServer
@implements IDisposable
@inject MqttService Mqtt

<h1>IoTダッシュボード</h1>
<ul>
    @foreach (var kv in Mqtt.Latest.OrderBy(k => k.Key))
    {
        <li>
            <b>@kv.Key</b>：@kv.Value.ToString("F1") ℃
            @if (kv.Value >= 30.0) { <span style="color:red"> ⚠️ 高温</span> }
        </li>
    }
</ul>

@code {
    protected override void OnInitialized() => Mqtt.Changed += OnChanged;
    private void OnChanged() => InvokeAsync(StateHasChanged);
    public void Dispose() => Mqtt.Changed -= OnChanged;
}

実機で起動し /dashboard を開いた状態で、別ターミナルから

mosquitto_pub -h localhost -t sensors/kitchen/temperature -m 31.2

を送ると、画面に kitchen が追加され、31.2℃なのでLEDが点灯すれば成功です。

ブラウザ画面

つまずきポイント

まとめ

次回（第8回）は、ここまでのアプリを systemdで常駐化し、電源を入れれば自動で立ち上がる24時間モニタリング機にします。

参考：MQTTnet 公式サンプル（GitHub） 前の記事：連載⑥ 温度ログをSQLite＋EF Coreで蓄積し時系列グラフで振り返る

この記事をシェア:
この記事をWhatsappでシェアこの記事をFacebookでシェアこの記事をXでシェアこの記事をTelegramでシェアこの記事をPinterestでシェア
前の記事
温度ログをSQLite＋EF Coreで蓄積し時系列グラフで振り返る（Raspberry Pi）【連載⑥】
次の記事
BlazorアプリをsystemdでRaspberry Piに常駐させる｜自動起動と自動再起動【連載⑧】