e04_ngsi_v2_context_subscriptions_mqtt.py

"""
# Examples for subscriptions

# create new subscription following the API Walkthrough example:
# https://fiware-orion.readthedocs.io/en/master/user/walkthrough_apiv2.html#subscriptions
"""

# ## Import packages
import logging
import datetime
import time
from filip.config import settings
from filip.models.ngsi_v2.subscriptions import Subscription, Message
from filip.clients.ngsi_v2.cb import ContextBrokerClient
from filip.models.base import FiwareHeader
from filip.models.ngsi_v2.context import ContextEntity
from urllib.parse import urlparse

# ## Parameters
#
# To run this example you need a working Fiware v2 setup with a context-broker
# You can set the address:
#
# Host address of Context Broker
CB_URL = settings.CB_URL

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

# MQTT URL for eclipse mosquitto
MQTT_BROKER_URL_INTERNAL = "mqtt://mosquitto:1883"
MQTT_BROKER_URL_EXPOSED = str(settings.MQTT_BROKER_URL)

# MQTT topic that the subscription will send to
mqtt_topic = "".join([SERVICE, SERVICE_PATH])

# 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 Client setup
    #
    # create the client, for more details view the example: e01_http_clients.py
    fiware_header = FiwareHeader(service=SERVICE, service_path=SERVICE_PATH)
    cb_client = ContextBrokerClient(url=CB_URL, fiware_header=fiware_header)

    room_001 = {
        "id": "urn:ngsi-ld:Room:001",
        "type": "Room",
        "temperature": {"value": 11, "type": "Float"},
        "pressure": {"value": 111, "type": "Integer"},
    }
    room_entity = ContextEntity(**room_001)
    cb_client.post_entity(entity=room_entity, update=True)

    # # 2 Setup for a subscription and MQTT notifications
    #
    # Create the data for the subscription. Have a look at the condition and
    # the attribute section. Only the change of the temperature attribute will
    # trigger the subscription and only temperature data will be included
    # into the message.
    # Additionally, you should be aware of the throttling and expiration of a
    # subscription.
    #
    # For more details on subscriptions you might want to
    # check the Subscription model or the official tutorials.
    sub_example = {
        "description": "Subscription to receive MQTT-Notifications about "
        "urn:ngsi-ld:Room:001",
        "subject": {
            "entities": [{"id": "urn:ngsi-ld:Room:001", "type": "Room"}],
            "condition": {"attrs": ["temperature"]},
        },
        "notification": {
            "mqtt": {"url": MQTT_BROKER_URL_INTERNAL, "topic": mqtt_topic},
            "attrs": ["temperature"],
        },
        "expires": datetime.datetime.now() + datetime.timedelta(minutes=15),
        "throttling": 0,
    }
    # Generate Subscription object for validation
    sub = Subscription(**sub_example)

    # Posting an example subscription for Room1. Make sure that you store the
    # returned id because you might need it for later updates of the subscription.
    sub_id = cb_client.post_subscription(subscription=sub)

    # # 3 Setup for callbacks and the MQTT client
    #
    # Define callbacks for the mqtt client. They will be triggered by
    # different events. Do not change their signature!
    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: " f"{reason_code}"
            )
            raise ConnectionError
        else:
            logger.info(
                f"MQTT Client successfully connected with the reason code: {reason_code}"
            )
        client.subscribe(mqtt_topic)

    def on_subscribe(client, userdata, mid, granted_qos, properties=None):
        logger.info(f"MQTT Client successfully subscribed: {granted_qos[0]}")

    def on_message(client, userdata, msg):
        message = Message.model_validate_json(msg.payload)
        logger.info(
            "MQTT Client received this message:\n" + message.model_dump_json(indent=2)
        )

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

    # MQTT client
    import paho.mqtt.client as mqtt

    mqtt_client = mqtt.Client(
        userdata=None,
        protocol=mqtt.MQTTv5,
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        transport="tcp",
    )
    # add callbacks to the mqtt-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 mqtt-broker to receive the notifications message
    mqtt_url = urlparse(MQTT_BROKER_URL_EXPOSED)
    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,
    )

    # # 4 send new value via MQTT
    #
    # create a non-blocking thread for mqtt communication
    mqtt_client.loop_start()
    new_value = 55

    cb_client.update_attribute_value(
        entity_id="urn:ngsi-ld:Room:001",
        attr_name="temperature",
        value=new_value,
        entity_type="Room",
    )
    cb_client.update_attribute_value(
        entity_id="urn:ngsi-ld:Room:001",
        attr_name="pressure",
        value=new_value,
        entity_type="Room",
    )
    time.sleep(1)

    # # 5 Deleting the example entity and the subscription
    #
    cb_client.delete_subscription(sub_id)
    cb_client.delete_entity(entity_id=room_entity.id, entity_type=room_entity.type)
    # # 6 Clean up (Optional)
    #
    # Close clients
    mqtt_client.loop_stop()
    mqtt_client.disconnect()
    cb_client.close()