MQTT adafruit.io test

test_adafruit_io.py

#!/usr/bin/env python3

import paho.mqtt.client as mqtt
# import ssl
import time

# some const
# mqtt account
MQTT_HOST = 'io.adafruit.com'
MQTT_USER = '<USERNAME>'
MQTT_PWD = '<AIO KEY>'
# mqtt topics
TOPIC_PATH = '%s/feeds/' % MQTT_USER


class MqttFeed:
    def __init__(self, client, key_name, init_value=None):
        # public
        self.key_name = key_name
        # private
        self._c = client
        self._value = init_value
        if self._value is not None:
            self.pub()

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, value):
        self._value = value
        self.pub()

    def pub(self):
        self._c.publish(TOPIC_PATH + self.key_name, self._value)


def on_button1(client, userdata, message):
    if message.payload.decode('utf8') == '1':
        # reset feed1
        feed1.value = 0
        # console message
        print('click on button1 !')


def on_connect(client, obj, flags, rc):
    # log connect to broker
    print("connect ok [rc: " + str(rc) + "]")
    # subscribes on connect and reconnect
    m.subscribe(TOPIC_PATH + '#')


def on_message(client, userdata, message):
    print('rx msg: %s = %s' % (message.topic, message.payload.decode('utf8')))


def on_publish(client, obj, mid):
    print('publish send')

# init mqtt client
m = mqtt.Client()
m.username_pw_set(MQTT_USER, MQTT_PWD)
# set callbacks
m.message_callback_add(TOPIC_PATH + 'button1', on_button1)
m.on_connect = on_connect
# m.on_publish = on_publish
# m.on_message = on_message
# connect
# m.tls_set(ca_certs='io.adafruit.pem')
# m.tls_insecure_set(True)
m.connect(MQTT_HOST, port=1883, keepalive=60)
# launch mqtt loop
m.loop_start()

# set mqtt var
feed1 = MqttFeed(m, 'feed1', 0)

# main loop
while True:
    feed1.value += 1
    print('feed1 %s' % feed1.value)
    time.sleep(30.0)