LoRa-WindPower-Control/Source/lora_mqtt.py

#!/usr/bin/env python3

"""
MQTT Script to communicate with The Things Network and control digital output.

Author: Tobias Müller
Licence: GPL v3
Date: 23-12-06

Changelog:
    23-12-06 - create initial program
    23-12-07 - change output of payload
    23-12-12 - finish publish mqtt code
"""

from paho.mqtt import client as mqtt
import pigpio
from time import sleep
from sys import exit
import json
import socket
from base64 import b64encode

USER=""                                     # https://eu1.cloud.thethings.network/console/ -> Applications -> Integrations -> MQTT -> Username
PASSWORD=""                                 # https://eu1.cloud.thethings.network/console/ -> Applications -> Integrations -> MQTT -> Password
PUPLIC_MQTT_TLS_ADDRESS = "eu1.cloud.thethings.network"
PUPLIC_MQTT_TLS_PORT = 8883
DEVICE_IDS = ["eui-70b3d5e75e01131f"]       # list of device IDs in an application
DEVICE_PORT = 125                           # standard port for watteco devices
DEVICE_OUTPUT = 1                           # value of 1 to 4
PULISH_PRIORITY = "NORMAL"                  # priority to puplish messages: "LOWEST", "LOW", "BELOW_NORMAL", "NORMAL", "ABOVE_NORMAL", "HIGH", "HIGHEST"
MIN_REPORT_INTERVAL = 60                    # 0-32767; in production, value must be >= 60min/3600s to comply with the legal duty cycle in the EU at SF12
MAX_REPORT_INTERVAL = 120                   # 0-32767; in production, value must be >= 60min/3600s to comply with the legal duty cycle in the EU at SF12
INTERVAL_UNIT = "sec"                       # "sec" for seconds or "min" for minutes
GPIO = 5                                    # GPIO to detect shutdown
QOS = 0                                     # Quality of Service for MQTT
DEBUG = False

# Note: On The Things Network’s public community network a Fair Use
# Policy applies which limits the uplink airtime to 30 seconds per
# day (24 hours) per node and the downlink messages to 10
# messages per day (24 hours) per node. Higher SF (Spreading Factor)
# results in higher airtime usage and vice versa. Report intervals must
# be inside those limits.
# https://www.thethingsnetwork.org/forum/t/fair-use-policy-explained/1300

def debounce(gpio_client):
    active_low = 0
    active_high = 0
    max_cycle = 20
    delay = 0.01
    while active_low < max_cycle and active_high < max_cycle:
        if gpio_client.read(GPIO):
            active_high += 1
            active_low = 0
        else:
            active_high = 0
            active_low += 1
        sleep(delay)
    return 0 if active_high == max_cycle else 1


def set_payload(data):
    frame_setting = {"Fctrl": [{1: 0x11, 2: 0x31, 3: 0x51, 4:0x71}, 2],
                     "CmdID": [0x50, 2],
                     "ClusterID": [0x06, 4],
                     "Data": [2]}

    payload = ""
    for key, val in frame_setting.items():
        if key == "Fctrl":
            payload += F"{val[0].get(DEVICE_OUTPUT):0{val[-1]}X}"
        elif key == "CmdID" or key == "ClusterID":
            payload += F"{val[0]:0{val[-1]}X}"
        elif key == "Data":
            payload += F"{data:0{val[-1]}X}"

    return b64encode(bytes.fromhex(payload)).decode()


def report_payload():
    frame_setting = {"Fctrl": [{1: 0x11, 2: 0x31, 3: 0x51, 4:0x71}, 2],
                     "CmdID": [0x06, 2],
                     "ClusterID": [0x06, 4],
                     "AttributeID": [0x00, 4],
                     "Attribute_type": [0x10, 4],
                     "MinReport": [{"sec": 0x0000, "min": 0x8000}, 4],
                     "MaxReport": [{"sec": 0x0000, "min": 0x8000}, 4],
                     "ReportChange": [0x01, 2]}

    payload = ""
    for key, val in frame_setting.items():
        if key == "Fctrl":
            payload += F"{val[0].get(DEVICE_OUTPUT):0{val[-1]}X}"
        elif (key == "CmdID" or key == "ClusterID" or key == "AttributeID" or
              key == "Attribute_type" or key == "ReportChange"):
            payload += F"{val[0]:0{val[-1]}X}"
        elif key == "MinReport":
            payload += F"{val[0].get(INTERVAL_UNIT)|MIN_REPORT_INTERVAL:0{val[-1]}X}"
        elif key == "MaxReport":
            payload += F"{val[0].get(INTERVAL_UNIT)|MAX_REPORT_INTERVAL:0{val[-1]}X}"

    return b64encode(bytes.fromhex(payload)).decode()


def mqtt_send(mqtt_client, payload):
    if DEVICE_IDS:
        for id in DEVICE_IDS:
            topic = "v3/" + USER + "/devices/" + id + "/down/push"
            message = '{"downlinks":[{"f_port":' + str(DEVICE_PORT) + ',"frm_payload":"' + payload + '","priority":"' + PULISH_PRIORITY + '"}]}'
            result = mqtt_client.publish(topic, message, QOS)
            if result[0] == 0:
                print(F"Publish to topic: \"{topic}\", QoS: {QOS}, message ID: {result[1]}")
            else:
                print(F"Failed to publish to topic: \"{topic}\", QoS: {QOS}")
    else:
        print("There are no device IDs configured for MQTT-Subscription!")
        raise ValueError


def mqtt_on_connect_callback(client, userdata, flags, rc):
    if rc == 0:
        print(F"Connected successfully to MQTT broker \"{PUPLIC_MQTT_TLS_ADDRESS}:{PUPLIC_MQTT_TLS_PORT}\"")
    else:
        raise ConnectionError(F"Could not connect to MQTT-Broker \"{PUPLIC_MQTT_TLS_ADDRESS}:{PUPLIC_MQTT_TLS_PORT}\"" +
                              F"\nReturn-Code: {rc}")


def mqtt_on_message_callback(client, userdata, message):
    print("\nMessage received on topic '" + message.topic + "' with QoS = " + str(message.qos))
    parsed_json = json.loads(message.payload)
    if DEBUG:
        print(f"Full Message (json):\n{json.dumps(parsed_json, indent=4)}")
    else:
        print("Message:")
        for key, val in parsed_json["end_device_ids"].items():
            if key == "device_id":
                print(F"\tDevice ID: {val}")
            if key == "application_ids":
                print(F"\tApplication ID: {val['application_id']}")
        for key, val in parsed_json["uplink_message"].items():
            if key == "f_port":
                print(F"\tPort: {val}")
            if key == "frm_payload":
                print(F"\tPayload: {val}")
            if key == "decoded_payload":
                print(F"\tDecoded Payload:")
                try:
                    print(F"\t\tDate: {val['data'][0]['date']}")
                    print(F"\t\tLabel: {val['data'][0]['label']}")
                    print(F"\t\tValue: {val['data'][0]['value']}")
                except KeyError:
                    print("\t\tCould not decode payload.")
            if key == "settings":
                print(F"\tFrequency: {val['frequency']}")
                print(F"\tBandwidth: {val['data_rate']['lora']['bandwidth']}")
                print(F"\tSpreading Factor: {val['data_rate']['lora']['spreading_factor']}")
            if key == "consumed_airtime":
                print(F"\tAirtime: {val}")

def mqtt_on_subscribe_callback(client, userdata, mid, granted_qos):
    print(F"Subscription succesfull on message id: {mid}, and QoS: {granted_qos[0]}")


def mqtt_on_disconnect_callback(client, userdata, rc):
    print(F"Disconnect from MQTT with result code: {rc}")


def mqtt_on_publish_callback(client, userdata, mid):
    print(F"Publish succesfull on message ID: {mid}")


def init_mqtt():
    mqtt_client = mqtt.Client(f"axxeo_monitoring-client")
    mqtt_client.on_connect = mqtt_on_connect_callback
    mqtt_client.on_message = mqtt_on_message_callback
    mqtt_client.on_subscribe = mqtt_on_subscribe_callback
    mqtt_client.on_disconnect = mqtt_on_disconnect_callback
    mqtt_client.on_publish = mqtt_on_publish_callback
    mqtt_client.username_pw_set(USER, PASSWORD)
    mqtt_client.tls_set()
    mqtt_client.connect(PUPLIC_MQTT_TLS_ADDRESS, PUPLIC_MQTT_TLS_PORT)
    if DEVICE_IDS:
        for id in DEVICE_IDS:
            topic = "v3/" + USER + "/devices/" + id + "/up"
            result = mqtt_client.subscribe(topic, QOS)
            if result[0] == 0:
                print(F"Subscribe to topic: \"{topic}\", QoS: {QOS}, message ID: {result[1]}")
            else:
                print(F"Failed to subsribe to topic: \"{topic}\", QoS: {QOS}")
    else:
        print("There are no device IDs configured for MQTT-Subscription!")
        raise ValueError
    payload = report_payload()
    mqtt_send(mqtt_client, payload)
    return mqtt_client


def init_gpio():
    gpio_client = pigpio.pi()
    if not gpio_client.connected:
        print("The pigpio daemon is not running!")
        raise ConnectionError
    gpio_client.set_mode(GPIO, pigpio.INPUT)
    return gpio_client


def main():
    try:
        print("Initialize GPIO")
        gpio_client = init_gpio()
        print("Initialize MQTT")
        mqtt_client = init_mqtt()
        print("\nStart Communication:")
        old_val = debounce(gpio_client)
        payload = set_payload(old_val)
        mqtt_send(mqtt_client, payload)
        while True:
            new_val = debounce(gpio_client)
            if old_val != new_val:
                old_val = new_val
                if old_val:
                    print("\nSwitch on windpower plant!")
                else:
                    print("\nSwitch off windpower plant!")
                payload = set_payload(new_val)
                mqtt_send(mqtt_client, payload)
            mqtt_client.loop(10)
    except KeyboardInterrupt:
        exit(0)
    except ConnectionError:
        exit(1)
    except socket.gaierror:
        print(F"\nNo address associated with hostname \"{PUPLIC_MQTT_TLS_ADDRESS}\"")
        exit(1)
    except ValueError:
        exit(1)
    finally:
        if "mqtt_client" in locals() and mqtt_client.is_connected():
            mqtt_client.disconnect()
        if "gpio_client" in locals() and gpio_client.connected:
            gpio_client.stop()
        print("\nExit program!")


if __name__ == "__main__":
    main()