**Contents**


* `Prerequisites <#prerequisites-for-mqtt-publishing>`__

  * `Configure and start MQTT broker <#configure-and-start-mqtt-broker>`__
  * `Start MQTT Subscriber <#start-mqtt-subscriber>`__

* `Configuration <#configuration>`__

  * `Configuration options <#configuration-options>`__
  * `Metadata filtering <#metadata-filtering>`__

* `Error handling <#error-handling>`__
* `Secure Pubishing <#secure-publishing>`__

Prerequisites for MQTT publishing
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

*(Broker configuration, certificates generation are for development and testing purposes only)*

Prior to EVAM publishing, MQTT broker and subscriber needs to be configured and started.

Configure and start MQTT broker
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

MQTT broker should be configured to accept connections. Start the broker from [WORK_DIR]/IEdgeInsights/EdgeVideoAnalyticsMicroservice directory.
For example, start `eclipse mosquitto <https://mosquitto.org/>`_ MQTT broker using configuration in mosquitto folder(\ ``[WORK_DIR]/IEdgeInsights/EdgeVideoAnalyticsMicroservice/mosquitto``\ ) as below. Make sure ``echo $PWD`` shows the root of EVAM repository.

.. code-block:: sh

     docker run -d --name=mqtt_broker -p 1883:1883 -v $PWD/mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf eclipse-mosquitto

With the above configuration, the broker listens on port 1883.

Start MQTT Subscriber
~~~~~~~~~~~~~~~~~~~~~

Once the mqtt broker is configured and up, connect to the mqtt broker and subscriber to topics in order to receive messages that will be published.


* 
  For example, run the below command to subscribe using `mosquitto_sub <https://mosquitto.org/man/mosquitto_sub-1.html>`_ client,

  .. code-block:: sh

     sudo apt install mosquitto-clients

     mosquitto_sub --topic <topic name> -p 1883 -h <mqtt broker address>

* 
  Alternatively, `Eclipse Paho MQTT Python Client <https://github.com/eclipse/paho.mqtt.python>`_ can be used for subscribing to broker and receiving messages.

  Please make sure to update the ``<topic_name>`` and ``<mqtt broker address>`` in the script before running.

  Make sure to install the python packages:

  .. code-block:: sh

     pip install paho-mqtt opencv-python numpy

  .. code-block:: python

     import paho.mqtt.subscribe as subscribe

     def on_message(client, userdata, message):
         print("%s %s" % (message.topic, message.payload))

     subscribe.callback(on_message, "<topic name>", hostname="<mqtt broker address>")

  Here is a sample subscriber script for receiving both frame (JPEG encoded) and metadata, printing the metadata and saving the frame as jpg image.

  .. code-block:: python

     import paho.mqtt.subscribe as subscribe
     import json
     import base64
     import numpy as np
     import cv2

     def on_message(client, userdata, message):
         print("Receiving frame and metadata")
         msg = json.loads(message.payload)

         metadata = msg[0]
         print("Metadata:", metadata)

         output_file_name = "output_" + str(metadata['frame_id']) + ".jpg"
         print("Saving frame to {}".format(output_file_name))
         img = base64.b64decode(msg[1].encode("utf-8"))
         img = np.frombuffer(img, dtype="uint8")
         img = cv2.imdecode(img, cv2.IMREAD_UNCHANGED)
         cv2.imwrite(output_file_name, img)

     subscribe.callback(on_message, "<topic_name>", hostname="<mqtt broker address>")

  More details on the subscribe helper functions can be found `here <https://github.com/eclipse/paho.mqtt.python#subscribe-1>`_

Configuration
^^^^^^^^^^^^^

Configuration options
~~~~~~~~~~~~~~~~~~~~~

Add below configuration in config.json(\ ``[WORK_DIR]/IEdgeInsights/EdgeVideoAnalyticsMicroservice/eii/docs/config.json``\ ) to enable publishing to the mqtt broker.

.. code-block:: json

       "mqtt_publisher": {
         "host": "<mqtt broker address>",
         "port": 1883
       }


* ``host`` mqtt broker hostname or IP address
* ``port`` port to connect to the broker
* ``topic`` topic to which message will be published. Defaults to ``edge_video_analytics_results`` *(optional)*
* 
  ``publish_frame`` whether to publish only metadata or both frames and metadata can be published to the mqtt broker. Defaults to ``false`` *(optional)*


  * When ``publish_frame`` is false, only metadata will be published.
  * When ``publish_frame`` is true, both metadata and frame will be published.
      |  Payload   |   Type         |
      |  :-------------------   |  :-------------------------                                  |
      |  Only metadata          |   JSON (metadata)
      |  Both frame and metadata|   JSON (metadata, frame) where frames are Base64 encoded UTF-8 string |

* 
  ``qos`` quality of service level to use which defaults to 0. Values can be 0, 1, 2. *(optional)*
  More details on the QoS levels can be found `here <https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels>`_

* ``protocol`` protocol version to use which defaults to 4 i.e. MQTTv311. Values can be 3, 4, 5 based on the versions MQTTv3, MQTTv311, MQTTv5 respectively *(optional)*

Metadata filtering
~~~~~~~~~~~~~~~~~~

Below configuration can be used to optionally filter out messages sent to mqtt broker for classification and detection usecases.


* 
  For detection:

  .. code-block:: json

       "mqtt_publisher": {
         ...
         "filter": {
               "type": "detection",
               "label_score": {"Person": 0.4}
         }
         ...
       }

* 
  For classification:

  .. code-block:: json

     "mqtt_publisher": {
         ...
         "filter": {
             "type": "classification",
               "label_score": {
                   "anomalous": 0.5
               }
         }
         ...
       }


  * ``type`` to specify type of filter config ``classification`` or ``detection``
  * 
    ``label score`` to specify key-value pair ``class label``\ : ``threshold``. Any detections < threshold will be skipped.

  * 
    Note:


    * For detection, metadata is expected to have, for example,
      ...'predictions': {'labels_to_revisit_fullscene': None, 'kind': 'prediction', 'id': None, 'maps': [], 'media_identifier': None, 'modified': None,'annotations': [{'labels_to_revisit': None,'shape': {'type': 'RECTANGLE', 'x': 196, 'height': 328, 'y': 567, 'width': 272}, 'id': None 'labels': [{'id': None, 'probability': 0.527821958065033, 'source': None, 'color': '#25a18eff', 'name': 'Person'}], 'modified': None}... or
      ...'annotations': {'objects': [{'label': 'Person', 'score': 0.6827021241188049, 'bbox': [873, 484, 1045, 702], 'attributes': {'rotation': 0, 'occluded': 0}}],...
    * For classification, metadata is expected to have, for example, ...{'label': 'Person', 'score': 0.5}...

Error Handling
~~~~~~~~~~~~~~


#. If connection to MQTT broker is successful, messages are published to the broker.
#. If there are connection issues with MQTT broker, messages will not be published to the broker.
#. Reconnection is automatically attempted when connection is lost. The time between successive reconnect attempts starts with 1s and doubles for every attempt until a max of 30s is reached after which it will always be 30s.
#. If connection is re-established, subsequent messages will be published to the broker.
#. Publishing to EII Message bus remains unimpacted.

Secure Publishing
~~~~~~~~~~~~~~~~~

MQTT publishing to broker could be over a secure communication channel providing encryption and authentication over TLS.

Follow the below steps to establish a secure connection with MQTT broker,


#. 
   Generate Certificates

   CA (Certificate Authority), client and server certificates need to be generated which will be used for configuring MQTT broker and EVAM.

   Below script can be used for generating certificates using openssl. (Command reference: https://mosquitto.org/man/mosquitto-tls-7.html).


   * Make sure to edit the ``<IP address of broker>`` in the script. This will be the broker address.
   * 
     Executing the below script with ask for a password for ca.key and the same password to be used again when prompter during signing step.

     .. code-block:: bash

        echo "Creating CA Key and Certificate"
        openssl req -new -x509 -days 365 -extensions v3_ca -keyout ca.key -out ca.crt -subj "/CN=example.com"

        echo "Creating Server key"
        openssl genrsa -out server.key 2048

        echo "Creating Server Certificate signing request"
        openssl req -subj "/CN=<IP address of broker>" -out server.csr -key server.key -new

        echo "Signing Server certificate"
        openssl x509 -req -in server.csr -extfile <(echo "subjectAltName=IP:<IP address of broker>") -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 365

        echo "Creating Client key"
        openssl genrsa -out client.key 2048

        echo "Creating Client Certificate signing request"
        openssl req -subj "/" -out client.csr -key client.key -new

        echo "Signing Client certificate"
        openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out client.crt -days 365

        mkdir -p client
        mv client.* client/

        mkdir -p server
        mv server.* server/

     Once the certificates are generated, make sure to move the certificates to right location. Make sure ``echo $PWD`` shows the root of EVAM repository.


     * 
       Move ca.crt, server/server.crt, server/server.key to ``$PWD/mosquitto/certificates``

       .. code-block:: sh

          mkdir $PWD/mosquitto/certificates
          mkdir $PWD/mosquitto/certificates/server
          cp <path to ca.crt> $PWD/mosquitto/certificates
          cp <path to server/server.crt> $PWD/mosquitto/certificates/server
          cp <path to server/server.key> $PWD/mosquitto/certificates/server

     * 
       Move ca.crt, client/client.crt, client/client.key to ``$PWD/certificates``

       .. code-block:: sh

          mkdir $PWD/certificates
          mkdir $PWD/certificates/client
          cp <path to ca.crt> $PWD/certificates
          cp <path to client/client.crt> $PWD/certificates/client
          cp <path to client/client.key> $PWD/certificates/client

     * 
       Change permission for client.key in ``$PWD/certificates/client``

       .. code-block:: sh

          sudo chmod 644 $PWD/certificates/client/client.key

#. 
   Configure and start MQTT broker


   * 
     Modify the mosquitto.config(\ ``[WORK_DIR]/IEdgeInsights/EdgeVideoAnalyticsMicroservice/mosquitto/mosquitto.conf``\ ) file as below.

     .. code-block:: json

         allow_anonymous true

         listener 8883

         cafile /mosquitto/config/certificates/ca.crt
         keyfile /mosquitto/config/certificates/server/server.key
         certfile /mosquitto/config/certificates/server/server.crt

         require_certificate true

     With the above configuration, the broker listens on port 8883 and is set up for mutual authentication.


     * 
       Start MQTT broker

       .. code-block:: sh

          docker run -d --name=mqtt_broker -p 8883:8883 -v $PWD/mosquitto:/mosquitto/config eclipse-mosquitto

#. 
   Configure and start MQTT subscriber

    As broker is configured for secure connection, CA certificate and client certificate/key needs to be provided when starting the subscriber.


   * 
     For example, run the below command to subscribe using `mosquitto_sub <https://mosquitto.org/man/mosquitto_sub-1.html>`_ client,

     .. code-block:: sh

          sudo apt install mosquitto-clients

          mosquitto_sub --topic <topic name> -p 8883 -h <mqtt broker address> --cafile $PWD/certificates/ca.crt --cert $PWD/certificates/client/client.crt --key $PWD/certificates/client/client.key

   * 
     Alternatively, `Eclipse Paho MQTT Python Client <https://github.com/eclipse/paho.mqtt.python>`_ can be used for subscribing to broker and receiving messages.

     Please make sure to update the ``<topic_name>``\ , ``<mqtt broker address>``\ , ``<path to ca.rt>``\ , ``<path to client.crt>``\ , ``<path to client.key``\ > in the script before running.

     Make sure to install the python packages:

     .. code-block:: sh

        pip install paho-mqtt opencv-python numpy

     .. code-block:: python

        import paho.mqtt.subscribe as subscribe

        def on_message(client, userdata, message):
            print("%s %s" % (message.topic, message.payload))

        subscribe.callback(on_message, "<topic name>", hostname="<mqtt broker address>", port=8883, tls={'ca_certs': "<path to ca.crt>", 'certfile': "<path to client.crt>", 'keyfile': "<path to client.key>"})

     Here is a sample subscriber script for receiving both frame (JPEG encoded) and metadata, printing the metadata and saving the frame as jpg image.

     .. code-block:: python

          import paho.mqtt.subscribe as subscribe
          import json
          import base64
          import numpy as np
          import cv2

          def on_message(client, userdata, message):
              print("Receiving frame and metadata")
              msg = json.loads(message.payload)

              metadata = msg[0]
              print("Metadata:", metadata)

              output_file_name = "output_" + str(metadata['frame_id']) + ".jpg"
              print("Saving frame to {}".format(output_file_name))
              img = base64.b64decode(msg[1].encode("utf-8"))
              img = np.frombuffer(img, dtype="uint8")
              img = cv2.imdecode(img, cv2.IMREAD_UNCHANGED)
              cv2.imwrite(output_file_name, img)

          subscribe.callback(on_message, "<topic_name>", hostname="<mqtt broker address>", port=8883, tls={'ca_certs': "<path to ca.crt>", 'certfile': "<path to client.crt>", 'keyfile': "<path to client.key>"})

#. 
   Configure EVAM for establishing secure connection with MQTT broker

   Modify config.json(\ ``[WORK_DIR]/IEdgeInsights/EdgeVideoAnalyticsMicroservice/eii/config.json``\ )

   .. code-block:: json

         "mqtt_publisher": {
           "host": "<mqtt broker address>",
           "port": 8883,
           "tls": {
               "ca_cert": "/MqttCerts/ca.crt",
               "client_key": "/MqttCerts/client/client.key",
               "client_cert": "/MqttCerts/client/client.crt"
           }
         }