Explorar el Código

add mqtt_subscriber signal

nico hace 7 años
padre
commit
d76b4e6fef

+ 1 - 0
install/files/python_requirements.txt

@@ -24,3 +24,4 @@ SoundFile>=0.9.0
 pyalsaaudio>=0.8.4
 RPi.GPIO>=0.6.3
 sox>=1.3.0
+paho-mqtt>=1.3.0

+ 34 - 0
kalliope/brain.yml

@@ -26,3 +26,37 @@
             - "Veuillez renouveller votre ordre"
             - "Veuillez reformuller s'il vous plait"
             - "Je n'ai pas saisi cet ordre"
+
+# https://pypi.python.org/pypi/paho-mqtt/1.3.0#constructor-reinitialise
+  - name: "test-mqtt-1"
+    signals:
+      - mqtt_subscriber:
+          broker_ip: "127.0.0.1"
+          topic: "topic1"
+      - mqtt_subscriber:
+          broker_ip: "192.168.0.1"
+          topic: "topic2"
+    neurons:
+      - say:
+          message:
+            - "Bonjour monsieur"
+
+  - name: "test-mqtt-2"
+    signals:
+      - mqtt_subscriber:
+          broker_ip: "127.0.0.1"
+          topic: "topic3"
+    neurons:
+      - say:
+          message:
+            - "Bonjour monsieur"
+
+  - name: "test-mqtt-3"
+    signals:
+      - mqtt_subscriber:
+          broker_ip: "127.0.0.1"
+          topic: "topic1"
+    neurons:
+      - say:
+          message:
+            - "Bonjour monsieur"

+ 99 - 0
kalliope/signals/mqtt_subscriber/MqttClient.py

@@ -0,0 +1,99 @@
+import logging
+import socket
+from threading import Thread
+
+import paho.mqtt.client as mqtt
+from kalliope.core.SynapseLauncher import SynapseLauncher
+
+logging.basicConfig()
+logger = logging.getLogger("kalliope")
+
+
+class MqttClient(Thread):
+
+    def __init__(self, broker=None, brain=None):
+        """
+        Class used to instantiate mqtt client
+        Thread used to be non blocking when called from parent class
+        :param broker: broker object
+        :type broker: Broker
+        """
+        super(MqttClient, self).__init__()
+        self.broker = broker
+        self.brain = brain
+
+        self.client = mqtt.Client(client_id=self.broker.client_id, protocol=self.broker.protocol)
+        self.client.on_connect = self.on_connect
+        self.client.on_message = self.on_message
+        self.client.on_subscribe = self.on_subscribe
+
+        if self.broker.username is not None and self.broker.password is not None:
+            logger.debug("[MqttClient] Username and password are set")
+            self.client.username_pw_set(self.broker.username, self.broker.password)
+
+    def run(self):
+        logger.debug("[MqttClient] Try to connect to broker: %s, port: %s, "
+                     "keepalive: %s, protocol: %s" % (self.broker.broker_ip,
+                                                      self.broker.port,
+                                                      self.broker.keepalive,
+                                                      self.broker.protocol))
+        try:
+            self.client.connect(self.broker.broker_ip, self.broker.port, self.broker.keepalive)
+            self.client.loop_forever()
+        except socket.error:
+            logger.debug("[MqttClient] Unable to connect to broker %s" % self.broker.broker_ip)
+
+    def on_connect(self, client, userdata, flags, rc):
+        """
+        The callback for when the client receives a CONNACK response from the server.
+        """
+        logger.debug("[MqttClient] Broker %s connection result code %s" % (self.broker.broker_ip, str(rc)))
+
+        if rc == 0:  # success connection
+            logger.debug("[MqttClient] Successfully connected to broker %s" % self.broker.broker_ip)
+            # Subscribing in on_connect() means that if we lose the connection and
+            # reconnect then subscriptions will be renewed.
+            for topic in self.broker.topics:
+                logger.debug("[MqttClient] Trying to subscribe to topic %s" % topic.name)
+                client.subscribe(topic.name)
+        else:
+            logger.debug("[MqttClient] Broker %s connection failled. Disconnect" % self.broker.broker_ip)
+            self.client.disconnect()
+
+    def on_message(self, client, userdata, msg):
+        """
+        The callback for when a PUBLISH message is received from the server
+        """
+        print(msg.topic + ": " + str(msg.payload))
+
+        # obj = json.loads(msg.payload)
+
+        self.call_concerned_synapses(msg.topic, msg.payload)
+
+    def on_subscribe(self, mqttc, obj, mid, granted_qos):
+        """
+        The callback for when the client successfully subscribe to a topic on the server
+        """
+        logger.debug("[MqttClient] Successfully subscribed to topic")
+
+    def call_concerned_synapses(self, topic_name, message):
+        """
+        Call synapse launcher class for each synapse concerned by the subscribed topic
+        convert the message to json if needed before.
+        The synapse is loaded with a parameter called "mqtt_subscriber_message" that can be used in neurons
+        :param topic_name: name of the topic that received a message from the broker
+        :param message: string message received from the broker
+        """
+        # find concerned topic by name
+        target_topic = next(topic for topic in self.broker.topics if topic.name == topic_name)
+        # run each synapse
+        for synapse in target_topic.synapses:
+            logger.debug("[MqttClient] start synapse name %s" % synapse.name)
+            overriding_parameter_dict = dict()
+            overriding_parameter_dict["mqtt_subscriber_message"] = message
+            SynapseLauncher.start_synapse_by_name(synapse.name,
+                                                  brain=self.brain,
+                                                  overriding_parameter_dict=overriding_parameter_dict)
+
+
+

+ 46 - 0
kalliope/signals/mqtt_subscriber/README.md

@@ -0,0 +1,46 @@
+# MQTT Subscriber
+
+## Install rabbitmq
+
+```
+sudo apt-get install rabbitmq-server amqp-tools
+```
+
+Enable mqtt plugin
+```
+sudo rabbitmq-plugins enable rabbitmq_mqtt
+sudo systemctl restart rabbitmq-server
+```
+
+Active web ui
+```bash
+sudo rabbitmq-plugins enable rabbitmq_management
+```
+
+Get the cli and make it available to use
+```
+wget http://127.0.0.1:15672/cli/rabbitmqadmin
+sudo mv rabbitmqadmin /etc/rabbitmqadmin
+sudo chmod 755 /etc/rabbitmqadmin
+```
+
+Create admin account
+```bash
+sudo rabbitmqctl add_user admin p@ssw0rd
+sudo rabbitmqctl set_user_tags admin administrator
+sudo rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"
+```
+
+Publish a message with amqp-tools in the default rabbitmq exchange with the topic key "test.light"
+```
+amqp-publish -e "amq.topic" -r "test.light" -b "your message"
+```
+
+
+Test publish json
+```
+amqp-publish -e "amq.topic" -r "test.light" -b '{"test" : "message"}'
+```
+
+
+amqp-publish -e "amq.topic" -r "topic1" -b "your message"

+ 1 - 0
kalliope/signals/mqtt_subscriber/__init__.py

@@ -0,0 +1 @@
+from .mqtt_subscriber import Mqtt_subscriber

+ 94 - 0
kalliope/signals/mqtt_subscriber/models.py

@@ -0,0 +1,94 @@
+import logging
+
+logging.basicConfig()
+logger = logging.getLogger("kalliope")
+
+
+class Topic(object):
+    def __init__(self, name=None, synapses=None):
+        self.name = name
+        self.synapses = synapses
+
+    def __eq__(self, other):
+        """
+        This is used to compare 2 objects
+        :param other:
+        :return:
+        """
+        return self.__dict__ == other.__dict__
+
+
+class Broker(object):
+    def __init__(self, broker_ip=None, topics=None, port=None, client_id=None, keepalive=None,
+                 username=None, password=None, protocol=None):
+        self.broker_ip = broker_ip
+        self.topics = topics
+        if self.topics is None:
+            self.topics = list()
+
+        # optional value
+        self.port = port
+        self.client_id = client_id
+        self.keepalive = keepalive
+        self.username = username
+        self.password = password
+        self.protocol = protocol
+
+    def build_from_signal_dict(self, dict_parameters):
+        """
+        Build the Broker object from a received dict of parameters
+        :param dict_parameters: dict of parameters used to build the Broker object
+        """
+        logger.debug("[Broker] Build broker object from received parameters: %s" % dict_parameters)
+
+        self.broker_ip = dict_parameters["broker_ip"]
+
+        if "broker_port" in dict_parameters:
+            self.port = dict_parameters["broker_port"]
+            # keep alive must be an integer
+            if not isinstance(self.keepalive, int):
+                try:
+                    self.port = int(self.port)
+                except ValueError:
+                    logger.debug("[Broker] Invalid port value, fallback to 1883")
+                    self.port = 1883
+        else:
+            # set default port
+            self.port = 1883
+
+        if "username" in dict_parameters:
+            self.username = dict_parameters["username"]
+
+        if "password" in dict_parameters:
+            self.password = dict_parameters["password"]
+
+        if "keepalive" in dict_parameters:
+            self.keepalive = dict_parameters["keepalive"]
+            # keep alive must be an integer
+            if not isinstance(self.keepalive, int):
+                try:
+                    self.keepalive = int(self.keepalive)
+                except ValueError:
+                    logger.debug("[Broker] Invalid keepalive value, fallback to 60")
+                    self.keepalive = 60
+        else:
+            # set default value
+            self.keepalive = 60
+
+        if "protocol" in dict_parameters:
+            if dict_parameters["protocol"] not in ["MQTTv31", "MQTTv311"]:
+                logger.debug("[Broker] Invalid protocol value, fallback to MQTTv311")
+                self.protocol = "MQTTv311"
+            else:
+                self.protocol = dict_parameters["protocol"]
+        else:
+            self.protocol = "MQTTv311"
+
+
+    def __eq__(self, other):
+        """
+        This is used to compare 2 objects
+        :param other:
+        :return:
+        """
+        return self.__dict__ == other.__dict__

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

@@ -0,0 +1,133 @@
+import logging
+from threading import Thread
+
+from kalliope.core.ConfigurationManager import BrainLoader
+from kalliope.signals.mqtt_subscriber.MqttClient import MqttClient
+from kalliope.signals.mqtt_subscriber.models import Broker, Topic
+
+CLIENT_ID = "kalliope"
+
+logging.basicConfig()
+logger = logging.getLogger("kalliope")
+
+
+class Mqtt_subscriber(Thread):
+
+    def __init__(self, brain=None):
+        super(Mqtt_subscriber, self).__init__()
+        logger.debug("[Mqtt_subscriber] Mqtt_subscriber class created")
+        # variables
+        self.broker_ip = None
+        self.topic = None
+        self.json_message = False
+
+        self.brain = brain
+        if self.brain is None:
+            self.brain = BrainLoader().get_brain()
+
+    def run(self):
+        logger.debug("[Mqtt_subscriber] Starting Mqtt_subscriber")
+        # get the list of synapse that use Mqtt_subscriber as signal
+        list_synapse_with_mqtt_subscriber = self.get_list_synapse_with_mqtt_subscriber(brain=self.brain)
+
+        # we need to sort broker URL by ip, then for each broker, we sort by topic and attach synapses name to run to it
+        list_broker_to_instantiate = self.get_list_broker_to_instantiate(list_synapse_with_mqtt_subscriber)
+
+        # now instantiate a MQTT client for each broker object
+        self.instantiate_mqtt_client(list_broker_to_instantiate)
+
+    def get_list_synapse_with_mqtt_subscriber(self, brain):
+        """
+        return the list of synapse that use Mqtt_subscriber as signal in the provided brain
+        :param brain: Brain object that contain all synapses loaded
+        :type brain: Brain
+        :return: list of synapse that use Mqtt_subscriber as signal
+        """
+        for synapse in brain.synapses:
+            for signal in synapse.signals:
+                # if the signal is an event we add it to the task list
+                if signal.name == "mqtt_subscriber":
+                    if self.check_mqtt_dict(signal.parameters):
+                        yield synapse
+
+    @staticmethod
+    def check_mqtt_dict(mqtt_signal_parameters):
+        """
+        receive a dict of parameter from a mqtt_subscriber signal and them
+        :param mqtt_signal_parameters: dict of parameters
+        :return: True if parameters are valid
+        """
+        # check mandatory parameters
+        mandatory_parameters = ["broker_ip", "topic"]
+        if not all(key in mqtt_signal_parameters for key in mandatory_parameters):
+            return False
+
+        return True
+
+    @staticmethod
+    def get_list_broker_to_instantiate(list_synapse_with_mqtt_subscriber):
+        """
+        return a list of Broker object from the given list of synapse
+        :param list_synapse_with_mqtt_subscriber: list of Synapse object
+        :return: list of Broker
+        """
+        returned_list_of_broker = list()
+
+        for synapse in list_synapse_with_mqtt_subscriber:
+            for signal in synapse.signals:
+                # check if the broker exist in the list
+                if not any(x.broker_ip == signal.parameters["broker_ip"] for x in returned_list_of_broker):
+                    logger.debug("[Mqtt_subscriber] Create new broker: %s" % signal.parameters["broker_ip"])
+                    # create a new broker object
+                    new_broker = Broker()
+                    new_broker.build_from_signal_dict(signal.parameters)
+                    # add the current topic
+                    logger.debug("[Mqtt_subscriber] Add new topic to broker %s: %s" % (new_broker.broker_ip,
+                                                                                       signal.parameters["topic"]))
+                    new_topic = Topic()
+                    new_topic.name = signal.parameters["topic"]
+                    # add the current synapse to the topic
+                    new_topic.synapses = list()
+                    new_topic.synapses.append(synapse)
+                    new_broker.topics.append(new_topic)
+
+                    logger.debug("[Mqtt_subscriber] Add new synapse to topic %s :%s" % (new_topic.name, synapse.name))
+                    returned_list_of_broker.append(new_broker)
+                else:
+                    # the broker exist. get it from the list of broker
+                    broker_to_edit = next((broker for broker in returned_list_of_broker
+                                           if signal.parameters["broker_ip"] == broker.broker_ip))
+                    # check if the topic already exist
+                    if not any(topic.name == signal.parameters["topic"] for topic in broker_to_edit.topics):
+                        new_topic = Topic()
+                        new_topic.name = signal.parameters["topic"]
+                        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
+                        logger.debug("[Mqtt_subscriber] Add new synapse "
+                                     "to topic %s :%s" % (new_topic.name, synapse.name))
+                        new_topic.synapses = list()
+                        new_topic.synapses.append(synapse)
+                        # add the topic to the broker
+                        broker_to_edit.topics.append(new_topic)
+                    else:
+                        # the topic already exist, get it from the list
+                        topic_to_edit = next((topic for topic in broker_to_edit.topics
+                                              if topic.name == signal.parameters["topic"]))
+                        # add the synapse
+                        logger.debug("[Mqtt_subscriber] Add synapse %s to existing topic %s "
+                                     "in existing broker %s" % (synapse.name,
+                                                                topic_to_edit.name,
+                                                                broker_to_edit.broker_ip))
+                        topic_to_edit.synapses.append(synapse)
+
+        return returned_list_of_broker
+
+    def instantiate_mqtt_client(self, list_broker_to_instantiate):
+        """
+        Instantiate a MqttClient thread for each broker
+        :param list_broker_to_instantiate: list of broker to run
+        """
+        for broker in list_broker_to_instantiate:
+            mqtt_client = MqttClient(broker=broker, brain=self.brain)
+            mqtt_client.start()

+ 205 - 0
kalliope/signals/mqtt_subscriber/test_mqtt_subscriber.py

@@ -0,0 +1,205 @@
+import unittest
+
+from kalliope.core.Models import Neuron, Signal, Synapse, Brain
+from kalliope.signals.mqtt_subscriber import Mqtt_subscriber
+from kalliope.signals.mqtt_subscriber.models import Broker, Topic
+
+
+class TestMqtt_subscriber(unittest.TestCase):
+
+    def test_check_mqtt_dict(self):
+
+        valid_dict_of_parameters = {
+            "topic": "my_topic",
+            "broker_ip": "192.168.0.1"
+        }
+
+        invalid_dict_of_parameters = {
+            "topic": "my_topic"
+        }
+
+        self.assertTrue(Mqtt_subscriber.check_mqtt_dict(valid_dict_of_parameters))
+        self.assertFalse(Mqtt_subscriber.check_mqtt_dict(invalid_dict_of_parameters))
+
+    def test_get_list_synapse_with_mqtt_subscriber(self):
+
+        # test with one signal mqtt
+        neuron = Neuron(name='say', parameters={'message': ['test message']})
+        signal1 = Signal(name="mqtt_subscriber", parameters={"topic": "test", "broker_ip": "192.168.0.1"})
+        synapse1 = Synapse(name="synapse1", neurons=[neuron], signals=[signal1])
+        synapses = [synapse1]
+        brain = Brain()
+        brain.synapses = synapses
+
+        expected_result = synapses
+
+        mq = Mqtt_subscriber(brain=brain)
+
+        generator = mq.get_list_synapse_with_mqtt_subscriber(brain)
+
+        self.assertEqual(expected_result, list(generator))
+
+        # test with two synapse
+        neuron = Neuron(name='say', parameters={'message': ['test message']})
+        signal1 = Signal(name="order", parameters="test_order")
+        signal2 = Signal(name="mqtt_subscriber", parameters={"topic": "test", "broker_ip": "192.168.0.1"})
+        synapse1 = Synapse(name="synapse1", neurons=[neuron], signals=[signal1])
+        synapse2 = Synapse(name="synapse2", neurons=[neuron], signals=[signal1, signal2])
+
+        synapses = [synapse1, synapse2]
+        brain = Brain()
+        brain.synapses = synapses
+
+        expected_result = [synapse2]
+
+        mq = Mqtt_subscriber(brain=brain)
+        generator = mq.get_list_synapse_with_mqtt_subscriber(brain)
+
+        self.assertEqual(expected_result, list(generator))
+
+    def test_get_list_broker_to_instantiate(self):
+        # ----------------
+        # only one synapse
+        # ----------------
+        neuron = Neuron(name='say', parameters={'message': ['test message']})
+        signal1 = Signal(name="mqtt_subscriber", parameters={"topic": "topic1", "broker_ip": "192.168.0.1"})
+        synapse1 = Synapse(name="synapse1", neurons=[neuron], signals=[signal1])
+        brain = Brain()
+        brain.synapses = [synapse1]
+
+        list_synapse_with_mqtt_subscriber = [synapse1]
+
+        expected_broker = Broker()
+        expected_broker.broker_ip = "192.168.0.1"
+        expected_broker.topics = list()
+        expected_topic = Topic()
+        expected_topic.name = "topic1"
+        # add the current synapse to the topic
+        expected_topic.synapses = list()
+        expected_topic.synapses.append(synapse1)
+        expected_broker.topics.append(expected_topic)
+
+        expected_retuned_list = [expected_broker]
+
+        mq = Mqtt_subscriber(brain=brain)
+
+        self.assertEqual(expected_retuned_list, mq.get_list_broker_to_instantiate(list_synapse_with_mqtt_subscriber))
+
+        # ----------------
+        #  one synapse, two different broker
+        # ----------------
+        neuron = Neuron(name='say', parameters={'message': ['test message']})
+        signal1 = Signal(name="mqtt_subscriber", parameters={"topic": "topic1", "broker_ip": "192.168.0.1"})
+        signal2 = Signal(name="mqtt_subscriber", parameters={"topic": "topic2", "broker_ip": "172.16.0.1"})
+        synapse1 = Synapse(name="synapse1", neurons=[neuron], signals=[signal1, signal2])
+        brain = Brain()
+        brain.synapses = [synapse1]
+
+        list_synapse_with_mqtt_subscriber = [synapse1]
+
+        expected_broker1 = Broker()
+        expected_broker1.broker_ip = "192.168.0.1"
+        expected_broker1.topics = list()
+        expected_topic = Topic()
+        expected_topic.name = "topic1"
+        # add the current synapse to the topic
+        expected_topic.synapses = list()
+        expected_topic.synapses.append(synapse1)
+        expected_broker1.topics.append(expected_topic)
+
+        expected_broker2 = Broker()
+        expected_broker2.broker_ip = "172.16.0.1"
+        expected_broker2.topics = list()
+        expected_topic = Topic()
+        expected_topic.name = "topic2"
+        # add the current synapse to the topic
+        expected_topic.synapses = list()
+        expected_topic.synapses.append(synapse1)
+        expected_broker2.topics.append(expected_topic)
+
+        expected_retuned_list = [expected_broker1, expected_broker2]
+
+        mq = Mqtt_subscriber(brain=brain)
+
+        self.assertEqual(expected_retuned_list, mq.get_list_broker_to_instantiate(list_synapse_with_mqtt_subscriber))
+
+        # ----------------
+        #  two synapse, same broker, different topics
+        # ----------------
+        # synapse 1
+        neuron1 = Neuron(name='say', parameters={'message': ['test message']})
+        signal1 = Signal(name="mqtt_subscriber", parameters={"topic": "topic1", "broker_ip": "192.168.0.1"})
+        synapse1 = Synapse(name="synapse1", neurons=[neuron1], signals=[signal1])
+
+        # synapse 2
+        neuron2 = Neuron(name='say', parameters={'message': ['test message']})
+        signal2 = Signal(name="mqtt_subscriber", parameters={"topic": "topic2", "broker_ip": "192.168.0.1"})
+        synapse2 = Synapse(name="synapse2", neurons=[neuron2], signals=[signal2])
+
+        brain = Brain()
+        brain.synapses = [synapse1, synapse2]
+
+        list_synapse_with_mqtt_subscriber = [synapse1, synapse2]
+
+        expected_broker1 = Broker()
+        expected_broker1.broker_ip = "192.168.0.1"
+        expected_broker1.topics = list()
+        expected_topic1 = Topic()
+        expected_topic1.name = "topic1"
+        expected_topic2 = Topic()
+        expected_topic2.name = "topic2"
+        # add the current synapse to the topic
+        expected_topic1.synapses = [synapse1]
+        expected_topic2.synapses = [synapse2]
+        # add both topic to the broker
+        expected_broker1.topics.append(expected_topic1)
+        expected_broker1.topics.append(expected_topic2)
+
+        expected_retuned_list = [expected_broker1]
+
+        mq = Mqtt_subscriber(brain=brain)
+
+        self.assertEqual(expected_retuned_list, mq.get_list_broker_to_instantiate(list_synapse_with_mqtt_subscriber))
+
+        # ----------------
+        #  two synapse, same broker, same topic
+        # ----------------
+        # synapse 1
+        neuron1 = Neuron(name='say', parameters={'message': ['test message']})
+        signal1 = Signal(name="mqtt_subscriber", parameters={"topic": "topic1", "broker_ip": "192.168.0.1"})
+        synapse1 = Synapse(name="synapse1", neurons=[neuron1], signals=[signal1])
+
+        # synapse 2
+        neuron2 = Neuron(name='say', parameters={'message': ['test message']})
+        signal2 = Signal(name="mqtt_subscriber", parameters={"topic": "topic1", "broker_ip": "192.168.0.1"})
+        synapse2 = Synapse(name="synapse2", neurons=[neuron2], signals=[signal2])
+
+        brain = Brain()
+        brain.synapses = [synapse1, synapse2]
+
+        list_synapse_with_mqtt_subscriber = [synapse1, synapse2]
+
+        expected_broker1 = Broker()
+        expected_broker1.broker_ip = "192.168.0.1"
+        expected_broker1.topics = list()
+        expected_topic1 = Topic()
+        expected_topic1.name = "topic1"
+        # add both synapses to the topic
+        expected_topic1.synapses = [synapse1, synapse2]
+        # add the topic to the broker
+        expected_broker1.topics.append(expected_topic1)
+
+        expected_retuned_list = [expected_broker1]
+
+        mq = Mqtt_subscriber(brain=brain)
+
+        self.assertEqual(expected_retuned_list, mq.get_list_broker_to_instantiate(list_synapse_with_mqtt_subscriber))
+
+
+if __name__ == '__main__':
+    unittest.main()
+
+    # suite = unittest.TestSuite()
+    # suite.addTest(TestMqtt_subscriber("test_get_list_broker_to_instantiate"))
+    # runner = unittest.TextTestRunner()
+    # runner.run(suite)

+ 2 - 1
setup.py

@@ -91,7 +91,8 @@ setup(
         'SoundFile>=0.9.0',
         'pyalsaaudio>=0.8.4',
         'RPi.GPIO>=0.6.3',
-        'sox>=1.3.0'
+        'sox>=1.3.0',
+        'paho-mqtt>=1.3.0'
     ],