Browse Source

support json message

nico 7 years ago
parent
commit
d4b683ee4d

+ 4 - 3
kalliope/brain.yml

@@ -33,17 +33,18 @@
       - mqtt_subscriber:
           broker_ip: "127.0.0.1"
           broker_port: 8883
+          is_json: True
           topic: "topic1"
           username: "guest"
           password: "guest"
           ca_cert: "/home/nico/Desktop/PKI_rabbit/ca.cert"
-          certfile: "/home/nico/Desktop/PKI_rabbit/client.crt"
-          keyfile: "/home/nico/Desktop/PKI_rabbit/client.key"
+#          certfile: "/home/nico/Desktop/PKI_rabbit/client.crt"
+#          keyfile: "/home/nico/Desktop/PKI_rabbit/client.key"
           tls_insecure: True
     neurons:
       - say:
           message:
-            - "Bonjour monsieur"
+            - "Bonjour monsieur {{ mqtt_subscriber_message['test'] }}"
 
 #  - name: "test-mqtt-2"
 #    signals:

+ 14 - 6
kalliope/signals/mqtt_subscriber/MqttClient.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import socket
 from threading import Thread
@@ -32,13 +33,16 @@ class MqttClient(Thread):
             self.client.username_pw_set(self.broker.username, self.broker.password)
 
         if self.broker.ca_cert is not None and self.broker.certfile is not None and self.broker.keyfile is not None:
-            logger.debug("[MqttClient] Active TLS")
+            logger.debug("[MqttClient] Active TLS with client certificate authentication")
             self.client.tls_set(ca_certs=self.broker.ca_cert,
                                 certfile=self.broker.certfile,
                                 keyfile=self.broker.keyfile)
             self.client.tls_insecure_set(self.broker.tls_insecure)
-        else:
-            print("miss")
+
+        elif self.broker.ca_cert is not None:
+            logger.debug("[MqttClient] Active TLS with server CA certificate only")
+            self.client.tls_set(ca_certs=self.broker.ca_cert)
+            self.client.tls_insecure_set(self.broker.tls_insecure)
 
     def run(self):
         logger.debug("[MqttClient] Try to connect to broker: %s, port: %s, "
@@ -73,9 +77,7 @@ class MqttClient(Thread):
         """
         The callback for when a PUBLISH message is received from the server
         """
-        print(msg.topic + ": " + str(msg.payload))
-
-        # obj = json.loads(msg.payload)
+        logger.debug("[MqttClient] " + msg.topic + ": " + str(msg.payload))
 
         self.call_concerned_synapses(msg.topic, msg.payload)
 
@@ -95,6 +97,11 @@ class MqttClient(Thread):
         """
         # find concerned topic by name
         target_topic = next(topic for topic in self.broker.topics if topic.name == topic_name)
+        # convert payload to a dict if necessary
+        if target_topic.is_json:
+            message = json.loads(message)
+            logger.debug("[MqttClient] Payload converted to JSON dict: %s" % message)
+
         # run each synapse
         for synapse in target_topic.synapses:
             logger.debug("[MqttClient] start synapse name %s" % synapse.name)
@@ -106,3 +113,4 @@ class MqttClient(Thread):
 
 
 
+

+ 7 - 1
kalliope/signals/mqtt_subscriber/models.py

@@ -5,9 +5,10 @@ logger = logging.getLogger("kalliope")
 
 
 class Topic(object):
-    def __init__(self, name=None, synapses=None):
+    def __init__(self, name=None, synapses=None, is_json=None):
         self.name = name
         self.synapses = synapses
+        self.is_json = is_json
 
     def __eq__(self, other):
         """
@@ -61,6 +62,11 @@ class Broker(object):
             # set default port
             self.port = 1883
 
+        if "client_id" in dict_parameters:
+            self.client_id = dict_parameters["client_id"]
+        else:
+            self.client_id = "kalliope"
+
         if "username" in dict_parameters:
             self.username = dict_parameters["username"]
 

+ 12 - 0
kalliope/signals/mqtt_subscriber/mqtt_subscriber.py

@@ -86,6 +86,12 @@ class Mqtt_subscriber(Thread):
                                                                                        signal.parameters["topic"]))
                     new_topic = Topic()
                     new_topic.name = signal.parameters["topic"]
+                    if "is_json" in signal.parameters:
+                        logger.debug("[Mqtt_subscriber] Message for the topic %s will be json converted"
+                                     % new_topic.name)
+                        new_topic.is_json = bool(signal.parameters["is_json"])
+                    else:
+                        new_topic.is_json = False
                     # add the current synapse to the topic
                     new_topic.synapses = list()
                     new_topic.synapses.append(synapse)
@@ -101,6 +107,12 @@ class Mqtt_subscriber(Thread):
                     if not any(topic.name == signal.parameters["topic"] for topic in broker_to_edit.topics):
                         new_topic = Topic()
                         new_topic.name = signal.parameters["topic"]
+                        if "is_json" in signal.parameters:
+                            logger.debug("[Mqtt_subscriber] Message for the topic %s will be json converted"
+                                         % new_topic.name)
+                            new_topic.is_json = bool(signal.parameters["is_json"])
+                        else:
+                            new_topic.is_json = False
                         logger.debug("[Mqtt_subscriber] Add new topic to existing broker "
                                      "%s: %s" % (broker_to_edit.broker_ip, signal.parameters["topic"]))
                         # add the current synapse to the topic