Mosquittoに、Python Paho MQTT libraryを使ってアクセスする

あらかじめ起動しておいたMosquittoにPython Paho MQTT library (pkgsrc/net/py-paho-mqtt 2.1.0)を使って、Mosquittoの受信したデータを、InfluxDB 1 (pkgsrc/databases/influxdb)に保存する。 なぜInfluxDBに保存するのかと言うと、最終的にMosquittoで受信したデータをGrafana (pkgsrc/www/grafana)で表示するためである。

乱数をMosquittoに送信し、それを購読して、InfluxDBに保存する。

#!/usr/pkg/bin/python3.14

#
# Subscribe mosquitto MQTT server and store data into InfluxDB 1
#

import time
import random
import ssl

from paho.mqtt.enums import MQTTProtocolVersion
import paho.mqtt.client as mqtt
from influxdb import InfluxDBClient

# Connection information for InfluxDB 1.x
INFLUXDB_HOSTNAME = 'localhost'
INFLUXDB_PORT = 8086
INFLUXDB_DATABASE = 'mqtt001'
INFLUXDB_USERID = 'admin'
INFLUXDB_PASSWORD = 'ADMIN_PASSWORD'

# Connection information of MQTT
MQTT_BROKER = 'MYHOSTNAME'
MQTT_PORT = 8883
MQTT_TOPIC = 'topic/0001'

# Instantiate InfluxDB
influx_client = InfluxDBClient (
    host=INFLUXDB_HOSTNAME,
    port=INFLUXDB_PORT,
    username=INFLUXDB_USERID,
    password=INFLUXDB_PASSWORD,
    database=INFLUXDB_DATABASE
)

# Create database at first at once
influx_client.create_database(INFLUXDB_DATABASE)

# Callback function for MQTT subscription
def on_message (client, userdata, message):
    payload = message.payload.decode('utf-8')
    timestamp = int(time.time() * 1000000000) # in ns
    # debug print
    print (f'Received: {payload}')

    json_body = [
        {
            "measurement": "sensor_data",
            "tags": {
                "device": "dev001_reg700"
            },
            "fields": {
                "value": float(payload)
            },
            "time": timestamp
         }
    ]

    # Write data to InfluxDB
    influx_client.write_points(json_body)

# Create MQTT client
mqtt_client = mqtt.Client(
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    protocol=MQTTProtocolVersion.MQTTv5
)

# Assign callback function on receiving
mqtt_client.on_message = on_message

# Setup TLS connection
tls_context = ssl.create_default_context()
tls_context.verify_mode = ssl.CERT_REQUIRED
mqtt_client.tls_set_context(tls_context)

# Use user ID and password auth
PUBSUB_ID = 'user001'
PUBSUB_PW = 'password001'
mqtt_client.username_pw_set(
    username=PUBSUB_ID,
    password=PUBSUB_PW
)

# Connect to broker
mqtt_client.connect(
    host=MQTT_BROKER,
    port=MQTT_PORT, 
    keepalive=60
)


# Start to subscribe
mqtt_client.subscribe(topic=MQTT_TOPIC)
## debug print
print (f'Subscribe topic: {MQTT_TOPIC}')

# main loop in another thread
mqtt_client.loop_start()


# Testing publishing
while True:
    value = round(random.uniform(10, 30), 2)
    mqtt_client.publish(MQTT_TOPIC, str(value))
    print(f'Published: {value}')
    time.sleep(1)