e09_ngsi_v2_iota_filip_mqtt.py

"""
# This example shows in more detail how to interact with a device over MQTT
# using FiLiP's IoTA-MQTT Client. This client comes along with a convenient
# API for handling MQTT communication with FIWARE's IoT-Agent
"""
# ## Import packages
import logging
import random
import time
import paho.mqtt.client as mqtt
from filip.config import settings
from urllib.parse import urlparse
from filip.clients.mqtt import IoTAMQTTClient
from filip.models import FiwareHeader
from filip.models.ngsi_v2.iot import \
    Device, \
    DeviceAttribute, \
    DeviceCommand, \
    ServiceGroup, \
    PayloadProtocol
from filip.models.ngsi_v2.context import NamedCommand

# ## Parameters
#
# To run this example you need a working Fiware v2 setup with a
# context-broker, an IoT-Agent and mqtt-broker. You can here set the
# addresses:
#
# Host address of Context Broker
CB_URL = settings.CB_URL
# Host address of IoT-Agent
IOTA_URL = settings.IOTA_URL
# Host address of the MQTT-Broker
MQTT_BROKER_URL = str(settings.MQTT_BROKER_URL)

# You can here also change the used Fiware service
# FIWARE-Service
SERVICE = 'filip'
# FIWARE-Service path
SERVICE_PATH = '/example'

# You may also change the ApiKey Information
# ApiKey of the ServiceGroup
SERVICE_GROUP_APIKEY = 'filip-example-service-group'


# Setting up logging
logging.basicConfig(
    level='INFO',
    format='%(asctime)s %(name)s %(levelname)s: %(message)s',
    datefmt='%d-%m-%Y %H:%M:%S')

logger = logging.getLogger(__name__)

if __name__ == '__main__':

    # # 1 Setup
    #
    # ## 1.1 FiwareHeader
    #
    # Since we want to use the multi-tenancy concept of fiware we always start
    # with create a fiware header
    fiware_header = FiwareHeader(service=SERVICE,
                                 service_path=SERVICE_PATH)

    # ## 1.2 Device configuration
    #
    service_group_json = ServiceGroup(
        apikey=SERVICE_PATH.strip('/'),
        resource="/iot/json")
    service_group_ul = ServiceGroup(
        apikey=SERVICE_PATH.strip('/'),
        resource="/iot/d")

    device_attr = DeviceAttribute(name='temperature',
                                  object_id='t',
                                  type="Number")
    device_command = DeviceCommand(name='heater', type="Boolean")

    device_json = Device(device_id='my_json_device',
                         entity_name='my_json_device',
                         entity_type='Thing',
                         protocol='IoTA-JSON',
                         transport='MQTT',
                         apikey=service_group_json.apikey,
                         attributes=[device_attr],
                         commands=[device_command])

    device_ul = Device(device_id='my_ul_device',
                       entity_name='my_ul_device',
                       entity_type='Thing',
                       protocol='PDI-IoTA-UltraLight',
                       transport='MQTT',
                       apikey=service_group_ul.apikey,
                       attributes=[device_attr],
                       commands=[device_command])

    # ## 1.3 IoTAMQTTClient
    #
    mqttc = IoTAMQTTClient()

    def on_connect(mqttc, obj, flags, rc, properties=None):
        mqttc.logger.info(f"on_connect callback function: Reason code: {rc}")

    def on_connect_fail(mqttc, obj):
        mqttc.logger.info("Connect failed")

    def on_publish(mqttc, obj, mid, rc, properties=None):
        mqttc.logger.info(f"on_publish callback function: Message identifier: {mid}")

    def on_subscribe(mqttc, obj, mid, granted_qos, properties=None):
        mqttc.logger.info(f"Subscribed: {granted_qos[0]}, message identifier: {mid}")

    def on_log(mqttc, obj, level, string):
        mqttc.logger.info(string)


    mqttc.on_connect = on_connect
    mqttc.on_connect_fail = on_connect_fail
    mqttc.on_publish = on_publish
    mqttc.on_subscribe = on_subscribe
    mqttc.on_log = on_log

    # # 2 Normal client behaviour
    #
    # this section demonstrates normal client behavior
    # For additional examples on how to use the client please check:
    # https://github.com/eclipse/paho.mqtt.python/tree/master/examples
    #
    first_topic = f"/filip/{SERVICE_PATH.strip('/')}/first"
    second_topic = f"/filip/{SERVICE_PATH.strip('/')}/second"
    first_payload = "filip_test_1"
    second_payload = "filip_test_2"


    def on_message_first(mqttc, obj, msg, properties=None):
        pass
        # do something

    def on_message_second(mqttc, obj, msg, properties=None):
        pass
        # do something


    mqttc.message_callback_add(sub=first_topic,
                               callback=on_message_first)
    mqttc.message_callback_add(sub=second_topic,
                               callback=on_message_second)
    mqtt_broker_url = urlparse(MQTT_BROKER_URL)

    mqttc.connect(host=mqtt_broker_url.hostname,
                  port=mqtt_broker_url.port,
                  keepalive=60,
                  bind_address="",
                  bind_port=0,
                  clean_start=mqtt.MQTT_CLEAN_START_FIRST_ONLY,
                  properties=None)
    mqttc.subscribe(topic=first_topic)

    # create a non-blocking loop
    mqttc.loop_start()
    mqttc.publish(topic=first_topic, payload="filip_test")

    # add additional subscription to the connection
    mqttc.subscribe(topic=second_topic)
    mqttc.publish(topic=second_topic, payload="filip_test")

    # remove subscriptions and callbacks
    mqttc.message_callback_remove(first_topic)
    mqttc.message_callback_remove(second_topic)
    mqttc.unsubscribe(first_topic)
    mqttc.unsubscribe(second_topic)

    # stop the network loop and disconnect cleanly
    # close the mqtt listening thread
    mqttc.loop_stop()
    # disconnect the mqtt device
    mqttc.disconnect()

    # # 3 Devices provisioning
    #
    # ## 3.1 Service groups
    #
    # ### 3.1.1 create service Groups
    #
    mqttc.add_service_group(service_group=service_group_json)

    # ### 3.1.2 Interact with service groups
    #
    mqttc.get_service_group(service_group_json.apikey)
    mqttc.update_service_group(service_group=service_group_json)

    # ### 3.1.2 Delete service groups
    #
    mqttc.delete_service_group(apikey=service_group_json.apikey)

    # ## 3.2 Devices
    #
    # ### 3.2.1 Create Device
    #
    mqttc.add_device(device=device_json)

    # ### 3.2.2 Interact with device
    #
    mqttc.get_device(device_json.device_id)
    mqttc.update_device(device=device_json)

    # ### 3.2.3 Delete device
    #
    mqttc.delete_device(device_id=device_json.device_id)

    # # 4 Commands
    #
    # This example is written for the JSON MQTT client, but it can be easily
    # adapted for Ultralight, by changing all JSON/json variable name parts
    # with the corresponding UL/ul parts
    #
    # ## 4.1 Setup MQTT client
    #
    # small clean up
    for group in mqttc.service_groups:
        mqttc.delete_service_group(group.apikey)

    for device in mqttc.devices:
        mqttc.delete_device(device.device_id)


    def on_command(client, obj, msg):
        apikey, device_id, payload = \
            client.get_encoder(PayloadProtocol.IOTA_JSON).decode_message(
                msg=msg)

        # acknowledge a command. Here command are usually single
        # messages. The first key is equal to the commands name.
        client.publish(device_id=device_id,
                       command_name=next(iter(payload)),
                       payload=payload)


    mqttc.add_service_group(service_group_json)
    mqttc.add_device(device_json)
    mqttc.add_command_callback(device_id=device_json.device_id,
                               callback=on_command)

    from filip.clients.ngsi_v2 import HttpClient, HttpClientConfig

    httpc_config = HttpClientConfig(cb_url=CB_URL,
                                    iota_url=IOTA_URL)
    httpc = HttpClient(fiware_header=fiware_header,
                       config=httpc_config)
    httpc.iota.post_group(service_group=service_group_json, update=True)
    httpc.iota.post_device(device=device_json, update=True)

    mqtt_broker_url = urlparse(MQTT_BROKER_URL)

    mqttc.connect(host=mqtt_broker_url.hostname,
                  port=mqtt_broker_url.port,
                  keepalive=60,
                  bind_address="",
                  bind_port=0,
                  clean_start=mqtt.MQTT_CLEAN_START_FIRST_ONLY,
                  properties=None)
    mqttc.subscribe()
    mqttc.loop_start()

    # ## 4.2 Command
    #
    entity = httpc.cb.get_entity(entity_id=device_json.device_id,
                                 entity_type=device_json.entity_type)
    context_command = NamedCommand(name=device_json.commands[0].name,
                                   value=False)

    httpc.cb.post_command(entity_id=entity.id,
                          entity_type=entity.type,
                          command=context_command)

    time.sleep(2)

    entity = httpc.cb.get_entity(entity_id=device_json.device_id,
                                 entity_type=device_json.entity_type)

    # The entity.heater_status.value should now have the status 'OK'
    print(f"Heater status value: {entity.heater_status.value}")

    # ## 4.3 Publish
    #
    payload = random.randint(0, 30)
    mqttc.publish(device_id=device_json.device_id,
                  payload={device_json.attributes[0].object_id: payload})
    time.sleep(1)
    entity = httpc.cb.get_entity(entity_id=device_json.device_id,
                                 entity_type=device_json.entity_type)

    # Set Temperature Value
    print(f"Entity temperature value before publishing: {entity.temperature.value}")

    payload = random.randint(0, 30)
    mqttc.publish(device_id=device_json.device_id,
                  attribute_name="temperature",
                  payload=payload)
    time.sleep(1)
    entity = httpc.cb.get_entity(entity_id=device_json.device_id,
                                 entity_type=device_json.entity_type)
    # Changed Temperature Value
    print(f"Entity temperature value after publishing: {entity.temperature.value}")

    # ## 4.4 Close Client

    # stop network loop and disconnect cleanly
    # close the mqtt listening thread
    mqttc.loop_stop()
    # disconnect the mqtt device
    mqttc.disconnect()