通常のPythonでセンサーゲートウェイを構築。

1. 通信方式を選ぶ

HTTPはPython標準ライブラリだけで動作し、定期実行に最適です。MQTTはTLS常時接続、QoS 1送信、last will、再接続待ちを利用します。

2. SDK不要のHTTP

デバイスを作成し、Raspberry Pi / PythonでHTTPを選びます。read_sensor()をGPIO、I2C、シリアル、USBセンサーの読み取りへ置き換えます。

#!/usr/bin/env python3
"""Devicecamp Raspberry Pi / Linux HTTP starter."""

import json
import time
import urllib.error
import urllib.request

DEVICECAMP_URL = "https://api.devicecamp.app/v1/devices/DEVICE_ID/telemetry"
DEVICECAMP_SECRET = "DEVICE_SECRET"
PUBLISH_INTERVAL_SECONDS = 60


def read_sensor() -> dict[str, float]:
    # Replace these values with reads from your GPIO, I2C, serial, or USB sensor.
    return {"temperature": 22.4, "humidity": 51.0}


def publish(reading: dict[str, float]) -> None:
    request = urllib.request.Request(
        DEVICECAMP_URL,
        data=json.dumps(reading).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {DEVICECAMP_SECRET}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=15) as response:
        if response.status not in (200, 202):
            raise RuntimeError(f"Devicecamp returned HTTP {response.status}")


while True:
    try:
        reading = read_sensor()
        publish(reading)
        print(f"Published metric keys: {', '.join(reading)}")
    except (OSError, RuntimeError, urllib.error.HTTPError) as error:
        # Do not print credentials or the full telemetry payload.
        print(f"Publish failed: {error}")
    time.sleep(PUBLISH_INTERVAL_SECONDS)

3. ゲートウェイ向けMQTT

MQTTを選び、一度だけ表示される認証情報を生成します。ダウンロードしたスクリプトを実行する前にpython3 -m pip install "paho-mqtt>=2,<3"でPahoを導入します。

#!/usr/bin/env python3
"""Devicecamp Raspberry Pi / Linux MQTT/TLS starter.

Install the only dependency with:
    python3 -m pip install "paho-mqtt>=2,<3"
"""

import json
import ssl
import time

import paho.mqtt.client as mqtt

MQTT_HOST = "mqtt.devicecamp.app"
MQTT_PORT = 8883
MQTT_CLIENT_ID = "MQTT_CLIENT_ID"
MQTT_USERNAME = "MQTT_USERNAME"
MQTT_PASSWORD = "MQTT_PASSWORD"
TELEMETRY_TOPIC = "v1/devices/DEVICE_ID/telemetry"
STATUS_TOPIC = "v1/devices/DEVICE_ID/status"
PUBLISH_INTERVAL_SECONDS = 60


def read_sensor() -> dict[str, float]:
    # Replace these values with reads from your GPIO, I2C, serial, or USB sensor.
    return {"temperature": 22.4, "humidity": 51.0}


client = mqtt.Client(
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    client_id=MQTT_CLIENT_ID,
    protocol=mqtt.MQTTv311,
)
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
client.tls_set(tls_version=ssl.PROTOCOL_TLS_CLIENT)
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.will_set(STATUS_TOPIC, '{"status":"offline"}', qos=1, retain=True)
client.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
client.loop_start()
client.publish(STATUS_TOPIC, '{"status":"online"}', qos=1, retain=True)

try:
    while True:
        reading = read_sensor()
        result = client.publish(
            TELEMETRY_TOPIC,
            json.dumps(reading, separators=(",", ":")),
            qos=1,
        )
        result.wait_for_publish(timeout=15)
        if result.is_published():
            print(f"Published metric keys: {', '.join(reading)}")
        time.sleep(PUBLISH_INTERVAL_SECONDS)
finally:
    client.publish(STATUS_TOPIC, '{"status":"offline"}', qos=1, retain=True)
    client.disconnect()
    client.loop_stop()

4. 常時運転

手動確認後は専用ユーザーとsystemdサービスで実行します。生成ファイルには実際の認証情報が含まれるため、そのユーザーだけが読める権限にしてください。

[Unit]
Description=Devicecamp sensor gateway
Wants=network-online.target
After=network-online.target

[Service]
Type=simple
User=devicecamp
Group=devicecamp
WorkingDirectory=/opt/devicecamp
Environment=PYTHONDONTWRITEBYTECODE=1
ExecStart=/usr/bin/python3 /opt/devicecamp/devicecamp_sensor.py
Restart=on-failure
RestartSec=10
NoNewPrivileges=true
PrivateTmp=true
ProtectHome=true
ProtectSystem=strict

[Install]
WantedBy=multi-user.target

5. サービスをインストール

上のunitをdevicecamp-sensor.serviceとしてダウンロード済みスクリプトと同じ場所へ保存します。コマンドはログイン不可のユーザーを作成し、認証ファイルを保護してログを表示します。

sudo useradd --system --home /opt/devicecamp \
  --shell /usr/sbin/nologin devicecamp
sudo install -d -o devicecamp -g devicecamp -m 750 /opt/devicecamp
sudo install -o devicecamp -g devicecamp -m 600 \
  devicecamp_raspberry_pi_http.py /opt/devicecamp/devicecamp_sensor.py
sudo install -m 644 devicecamp-sensor.service \
  /etc/systemd/system/devicecamp-sensor.service
sudo systemctl daemon-reload
sudo systemctl enable --now devicecamp-sensor.service
sudo journalctl -u devicecamp-sensor.service -f