MqttClient.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. import json
  2. import logging
  3. import socket
  4. from threading import Thread
  5. import paho.mqtt.client as mqtt
  6. from kalliope.core.SynapseLauncher import SynapseLauncher
  7. logging.basicConfig()
  8. logger = logging.getLogger("kalliope")
  9. class MqttClient(Thread):
  10. def __init__(self, broker=None, brain=None):
  11. """
  12. Class used to instantiate mqtt client
  13. Thread used to be non blocking when called from parent class
  14. :param broker: broker object
  15. :type broker: Broker
  16. """
  17. super(MqttClient, self).__init__()
  18. self.broker = broker
  19. self.brain = brain
  20. self.client = mqtt.Client(client_id=self.broker.client_id, protocol=self.broker.protocol)
  21. self.client.on_connect = self.on_connect
  22. self.client.on_message = self.on_message
  23. self.client.on_subscribe = self.on_subscribe
  24. if self.broker.username is not None and self.broker.password is not None:
  25. logger.debug("[MqttClient] Username and password are set")
  26. self.client.username_pw_set(self.broker.username, self.broker.password)
  27. if self.broker.ca_cert is not None and self.broker.certfile is not None and self.broker.keyfile is not None:
  28. logger.debug("[MqttClient] Active TLS with client certificate authentication")
  29. self.client.tls_set(ca_certs=self.broker.ca_cert,
  30. certfile=self.broker.certfile,
  31. keyfile=self.broker.keyfile)
  32. self.client.tls_insecure_set(self.broker.tls_insecure)
  33. elif self.broker.ca_cert is not None:
  34. logger.debug("[MqttClient] Active TLS with server CA certificate only")
  35. self.client.tls_set(ca_certs=self.broker.ca_cert)
  36. self.client.tls_insecure_set(self.broker.tls_insecure)
  37. def run(self):
  38. logger.debug("[MqttClient] Try to connect to broker: %s, port: %s, "
  39. "keepalive: %s, protocol: %s" % (self.broker.broker_ip,
  40. self.broker.port,
  41. self.broker.keepalive,
  42. self.broker.protocol))
  43. try:
  44. self.client.connect(self.broker.broker_ip, self.broker.port, self.broker.keepalive)
  45. self.client.loop_forever()
  46. except socket.error:
  47. logger.debug("[MqttClient] Unable to connect to broker %s" % self.broker.broker_ip)
  48. def on_connect(self, client, userdata, flags, rc):
  49. """
  50. The callback for when the client receives a CONNACK response from the server.
  51. """
  52. logger.debug("[MqttClient] Broker %s connection result code %s" % (self.broker.broker_ip, str(rc)))
  53. if rc == 0: # success connection
  54. logger.debug("[MqttClient] Successfully connected to broker %s" % self.broker.broker_ip)
  55. # Subscribing in on_connect() means that if we lose the connection and
  56. # reconnect then subscriptions will be renewed.
  57. for topic in self.broker.topics:
  58. logger.debug("[MqttClient] Trying to subscribe to topic %s" % topic.name)
  59. client.subscribe(topic.name)
  60. else:
  61. logger.debug("[MqttClient] Broker %s connection failled. Disconnect" % self.broker.broker_ip)
  62. self.client.disconnect()
  63. def on_message(self, client, userdata, msg):
  64. """
  65. The callback for when a PUBLISH message is received from the server
  66. """
  67. logger.debug("[MqttClient] " + msg.topic + ": " + str(msg.payload))
  68. self.call_concerned_synapses(msg.topic, msg.payload)
  69. def on_subscribe(self, mqttc, obj, mid, granted_qos):
  70. """
  71. The callback for when the client successfully subscribe to a topic on the server
  72. """
  73. logger.debug("[MqttClient] Successfully subscribed to topic")
  74. def call_concerned_synapses(self, topic_name, message):
  75. """
  76. Call synapse launcher class for each synapse concerned by the subscribed topic
  77. convert the message to json if needed before.
  78. The synapse is loaded with a parameter called "mqtt_subscriber_message" that can be used in neurons
  79. :param topic_name: name of the topic that received a message from the broker
  80. :param message: string message received from the broker
  81. """
  82. # find concerned topic by name
  83. target_topic = next(topic for topic in self.broker.topics if topic.name == topic_name)
  84. # convert payload to a dict if necessary
  85. if target_topic.is_json:
  86. message = json.loads(message)
  87. logger.debug("[MqttClient] Payload message converted to JSON dict: %s" % message)
  88. else:
  89. logger.debug("[MqttClient] Payload message is plain text: %s" % message)
  90. # run each synapse
  91. for synapse in target_topic.synapses:
  92. logger.debug("[MqttClient] start synapse name %s" % synapse.name)
  93. overriding_parameter_dict = dict()
  94. overriding_parameter_dict["mqtt_subscriber_message"] = message
  95. SynapseLauncher.start_synapse_by_name(synapse.name,
  96. brain=self.brain,
  97. overriding_parameter_dict=overriding_parameter_dict)