e08_ngsi_v2_iota_paho_mqtt.py

"""
# This example shows how to provision a virtual iot device in a FIWARE-based
# IoT Platform using FiLiP and PahoMQTT
"""

# ## Import packages
import json
import logging
import random
import time
import paho.mqtt.client as mqtt

from urllib.parse import urlparse
from filip.config import settings
from filip.models import FiwareHeader
from filip.models.ngsi_v2.iot import (
    Device,
    DeviceAttribute,
    DeviceCommand,
    ServiceGroup,
    StaticDeviceAttribute,
)
from filip.models.ngsi_v2.context import NamedCommand
from filip.clients.ngsi_v2 import HttpClient, HttpClientConfig

# ## Parameters
#
# To run this example you need a working Fiware v2 setup with a
# context-broker, an IoT-Agent and mqtt-broker. Here you can 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)

# Here you can 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 Device
DEVICE_APIKEY = "filip-example-device"
# 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 FiwareHeader
    # Since we want to use the multi-tenancy concept of fiware we always start
    # with creating a fiware header
    fiware_header = FiwareHeader(service=SERVICE, service_path=SERVICE_PATH)

    # # 2 Setup
    #
    # ## 2.1 Device creation
    #
    # First we create our device configuration using the models provided for
    # filip.models.ngsi_v2.iot

    # Creating an attribute for incoming measurements from e.g. a sensor. We
    # add the metadata for units using the unit models. You will later
    # notice that the library automatically augments the provided
    # information about units.
    device_attr1 = DeviceAttribute(
        name="temperature",
        object_id="t",
        type="Number",
        metadata={"unit": {"type": "Unit", "value": {"name": "degree Celsius"}}},
    )

    # creating a static attribute that holds additional information
    static_device_attr = StaticDeviceAttribute(
        name="info", type="Text", value="Filip example for " "virtual IoT device"
    )
    # creating a command that the IoT device will liston to
    device_command = DeviceCommand(name="heater")

    # NOTE: You need to know that if you define an apikey for a single device it
    # will be only used for outgoing traffic. This is not clearly defined
    # in the official documentation.
    # https://fiware-iotagent-json.readthedocs.io/en/latest/usermanual/index.html
    device = Device(
        device_id="urn:ngsi-ld:device:001",
        entity_name="urn:ngsi-ld:device:001",
        entity_type="Thing",
        protocol="IoTA-JSON",
        transport="MQTT",
        apikey=DEVICE_APIKEY,
        attributes=[device_attr1],
        static_attributes=[static_device_attr],
        commands=[device_command],
    )

    # you can also add additional attributes via the Device API
    device_attr2 = DeviceAttribute(
        name="humidity",
        object_id="h",
        type="Number",
        metadata={"unitText": {"value": "percent", "type": "Text"}},
    )

    device.add_attribute(attribute=device_attr2)

    # this will print our configuration that we will send
    logging.info(
        "This is our device configuration: \n" + device.model_dump_json(indent=2)
    )

    # ## 2.2 Device Provision
    #
    # Send the device configuration to FIWARE via the IoT-Agent. We use the general
    # ngsiv2 httpClient.
    # This will automatically create a data entity in the context broker and
    # make the device with it. The id of the entity in the context broker will
    # be our entity_name of the device in this case.
    # For more complex configuration you need to set the entity_name
    # and entity_type in the previous Device-Model.

    # In order to change the apikey of our devices for incoming data we need to
    # create a service group that our device will be we attached to.
    # NOTE: This is important in order to adjust the apikey for incoming traffic.
    service_group = ServiceGroup(
        service=fiware_header.service,
        subservice=fiware_header.service_path,
        apikey=SERVICE_GROUP_APIKEY,
        resource="/iot/json",
    )

    # Create the Http client node that once sent, the device can't be posted
    # again, and you need to use the update command.
    config = HttpClientConfig(cb_url=CB_URL, iota_url=IOTA_URL)
    client = HttpClient(fiware_header=fiware_header, config=config)
    client.iota.post_group(service_group=service_group, update=True)
    client.iota.post_device(device=device, update=True)

    # ## 2.3 Check for correctness
    # Check if the device is correctly configured. You will notice that
    # unfortunately the iot API does not return all the metadata. However,
    # it will still appear in the context-entity.
    device = client.iota.get_device(device_id=device.device_id)
    logging.info(f"{device.model_dump_json(indent=2)}")

    # check if the data entity is created in the context broker
    entity = client.cb.get_entity(
        entity_type=device.entity_type, entity_id=device.device_id
    )
    logging.info(
        "This is our data entity in the context broker belonging to our device: \n"
        + entity.model_dump_json(indent=2)
    )

    # # 3 MQTT Client
    #
    # Create a mqtt client that we use as representation of an IoT device
    # following the official documentation of Paho-MQTT.
    # https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php

    # We use the implementation of MQTTv5 which is slightly different from former
    # versions. Especially the arguments of the well-known function have
    # changed. It's now more verbose than it used to be. Furthermore,
    # you have to handle the 'properties' argument.

    # The callback for when the mqtt client receives a CONNACK response from the
    # server. All callbacks need to have this specific arguments, Otherwise the
    # client will not be able to execute them.
    def on_connect(client, userdata, flags, reason_code, properties=None):
        if reason_code != 0:
            logger.error(
                f"MQTT Client failed to connect with the error code: '{reason_code}'"
            )
            raise ConnectionError
        else:
            logger.info(
                f"MQTT Client successfully connected with the reason code: {reason_code}"
            )

        # Subscribing in on_connect() means that if we lose the connection and
        # reconnect then subscriptions will be renewed.
        # We do subscribe to the topic that the platform will publish our
        # commands on
        client.subscribe(f"/{device.apikey}/{device.device_id}/cmd")

    # Callback when the command topic is successfully subscribed
    def on_subscribe(client, userdata, mid, granted_qos, properties=None):
        logger.info(f"MQTT Client successfully subscribed: {granted_qos[0]}")

    # The callback for when the device receives a PUBLISH message like a
    # command from the server. Here, the received command will be printed and a
    # command-acknowledge will be sent to the platform.

    # NOTE: We need to use the apikey of the service-group to send the message
    # to the platform
    def on_message(client, userdata, msg):
        logger.info(msg.topic + " " + str(msg.payload))
        data = json.loads(msg.payload)
        res = {k: v for k, v in data.items()}
        print(f"MQTT Client on_message payload: {msg.payload}")
        client.publish(
            topic=f"/json/{service_group.apikey}" f"/{device.device_id}/cmdexe",
            payload=json.dumps(res),
        )

    def on_disconnect(client, userdata, flags, reasonCode, properties):
        logger.info("MQTT Client disconnected" + str(reasonCode))

    mqtt_client = mqtt.Client(
        client_id="filip-iot-example",
        userdata=None,
        protocol=mqtt.MQTTv5,
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        transport="tcp",
    )
    # bind callbacks to the client
    mqtt_client.on_connect = on_connect
    mqtt_client.on_subscribe = on_subscribe
    mqtt_client.on_message = on_message
    mqtt_client.on_disconnect = on_disconnect
    # connect to the server
    mqtt_url = urlparse(MQTT_BROKER_URL)
    mqtt_client.connect(
        host=mqtt_url.hostname,
        port=mqtt_url.port,
        keepalive=60,
        bind_address="",
        bind_port=0,
        clean_start=mqtt.MQTT_CLEAN_START_FIRST_ONLY,
        properties=None,
    )
    # create a non-blocking thread for mqtt communication
    mqtt_client.loop_start()

    for attr in device.attributes:
        random_number = random.randint(0, 9)
        payload = json.dumps({attr.object_id: random_number})
        logger.info("Send data to platform:" + payload)
        mqtt_client.publish(
            topic=f"/json/{service_group.apikey}/{device.device_id}/attrs",
            payload=payload,
        )

    time.sleep(1)
    entity = client.cb.get_entity(
        entity_id=device.device_id, entity_type=device.entity_type
    )
    logger.info(
        "This is updated entity status after measurements are "
        "received: \n" + entity.model_dump_json(indent=2)
    )

    # create and send a command via the context broker
    for i in range(10):
        if i % 2 == 1:
            value = True
        else:
            value = False

        context_command = NamedCommand(name=device_command.name, value=value)
        client.cb.post_command(
            entity_id=entity.id, entity_type=entity.type, command=context_command
        )

        time.sleep(1)
        # check the entity the command attribute should now show the PENDING
        entity = client.cb.get_entity(
            entity_id=device.device_id, entity_type=device.entity_type
        )
        logger.info(
            "This is updated entity status after the command was sent "
            "and the acknowledge message was received: "
            "\n" + entity.model_dump_json(indent=2)
        )

    # close the mqtt listening thread
    mqtt_client.loop_stop()
    # disconnect the mqtt device
    mqtt_client.disconnect()

    # # 4 Cleanup the server and delete everything
    #
    client.iota.delete_device(device_id=device.device_id)
    client.iota.delete_group(
        resource=service_group.resource, apikey=service_group.apikey
    )
    client.cb.delete_entity(entity_id=entity.id, entity_type=entity.type)