Send single MQTT notification

lib/paho/mqtt/__init__.py

@@ -0,0 +1 @@
+__version__ = "1.1"

lib/paho/mqtt/publish.py

@@ -0,0 +1,217 @@
+# Copyright (c) 2014 Roger Light <roger@atchoo.org>
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Public License v1.0
+# and Eclipse Distribution License v1.0 which accompany this distribution.
+#
+# The Eclipse Public License is available at
+#    http://www.eclipse.org/legal/epl-v10.html
+# and the Eclipse Distribution License is available at
+#   http://www.eclipse.org/org/documents/edl-v10.php.
+#
+# Contributors:
+#    Roger Light - initial API and implementation
+
+"""
+This module provides some helper functions to allow straightforward publishing
+of messages in a one-shot manner. In other words, they are useful for the
+situation where you have a single/multiple messages you want to publish to a
+broker, then disconnect and nothing else is required.
+"""
+
+import paho.mqtt.client as mqtt
+
+
+def _do_publish(c):
+    """Internal function"""
+    m = c._userdata[0]
+    c._userdata = c._userdata[1:]
+    if type(m) is dict:
+        topic = m['topic']
+        try:
+            payload = m['payload']
+        except KeyError:
+            payload = None
+        try:
+            qos = m['qos']
+        except KeyError:
+            qos = 0
+        try:
+            retain = m['retain']
+        except KeyError:
+            retain = False
+    elif type(m) is tuple:
+        (topic, payload, qos, retain) = m
+    else:
+        raise ValueError('message must be a dict or a tuple')
+
+    c.publish(topic, payload, qos, retain)
+
+
+def _on_connect(c, userdata, flags, rc):
+    """Internal callback"""
+    _do_publish(c)
+
+
+def _on_publish(c, userdata, mid):
+    """Internal callback"""
+    if len(userdata) == 0:
+        c.disconnect()
+    else:
+        _do_publish(c)
+
+
+def multiple(msgs, hostname="localhost", port=1883, client_id="", keepalive=60,
+             will=None, auth=None, tls=None, protocol=mqtt.MQTTv31):
+    """Publish multiple messages to a broker, then disconnect cleanly.
+
+    This function creates an MQTT client, connects to a broker and publishes a
+    list of messages. Once the messages have been delivered, it disconnects
+    cleanly from the broker.
+
+    msgs : a list of messages to publish. Each message is either a dict or a
+           tuple.
+
+           If a dict, only the topic must be present. Default values will be
+           used for any missing arguments. The dict must be of the form:
+
+           msg = {'topic':"<topic>", 'payload':"<payload>", 'qos':<qos>,
+           'retain':<retain>}
+           topic must be present and may not be empty.
+           If payload is "", None or not present then a zero length payload
+           will be published.
+           If qos is not present, the default of 0 is used.
+           If retain is not present, the default of False is used.
+
+           If a tuple, then it must be of the form:
+           ("<topic>", "<payload>", qos, retain)
+    hostname : a string containing the address of the broker to connect to.
+               Defaults to localhost.
+    port : the port to connect to the broker on. Defaults to 1883.
+    client_id : the MQTT client id to use. If "" or None, the Paho library will
+                generate a client id automatically.
+    keepalive : the keepalive timeout value for the client. Defaults to 60
+                seconds.
+    will : a dict containing will parameters for the client: will = {'topic':
+           "<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
+           Topic is required, all other parameters are optional and will
+           default to None, 0 and False respectively.
+           Defaults to None, which indicates no will should be used.
+    auth : a dict containing authentication parameters for the client:
+           auth = {'username':"<username>", 'password':"<password>"}
+           Username is required, password is optional and will default to None
+           if not provided.
+           Defaults to None, which indicates no authentication is to be used.
+    tls : a dict containing TLS configuration parameters for the client:
+          dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
+          'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
+          'ciphers':"<ciphers">}
+          ca_certs is required, all other parameters are optional and will
+          default to None if not provided, which results in the client using
+          the default behaviour - see the paho.mqtt.client documentation.
+          Defaults to None, which indicates that TLS should not be used.
+    """
+
+    if type(msgs) is not list:
+        raise ValueError('msgs must be a list')
+
+    client = mqtt.Client(client_id=client_id,
+                         userdata=msgs, protocol=protocol)
+    client.on_publish = _on_publish
+    client.on_connect = _on_connect
+
+    if auth is not None:
+        username = auth['username']
+        try:
+            password = auth['password']
+        except KeyError:
+            password = None
+        client.username_pw_set(username, password)
+
+    if will is not None:
+        will_topic = will['topic']
+        try:
+            will_payload = will['payload']
+        except KeyError:
+            will_payload = None
+        try:
+            will_qos = will['qos']
+        except KeyError:
+            will_qos = 0
+        try:
+            will_retain = will['retain']
+        except KeyError:
+            will_retain = False
+
+        client.will_set(will_topic, will_payload, will_qos, will_retain)
+
+    if tls is not None:
+        ca_certs = tls['ca_certs']
+        try:
+            certfile = tls['certfile']
+        except KeyError:
+            certfile = None
+        try:
+            keyfile = tls['keyfile']
+        except KeyError:
+            keyfile = None
+        try:
+            tls_version = tls['tls_version']
+        except KeyError:
+            tls_version = None
+        try:
+            ciphers = tls['ciphers']
+        except KeyError:
+            ciphers = None
+        client.tls_set(ca_certs, certfile, keyfile, tls_version=tls_version,
+                       ciphers=ciphers)
+
+    client.connect(hostname, port, keepalive)
+    client.loop_forever()
+
+
+def single(topic, payload=None, qos=0, retain=False, hostname="localhost",
+           port=1883, client_id="", keepalive=60, will=None, auth=None,
+           tls=None, protocol=mqtt.MQTTv31):
+    """Publish a single message to a broker, then disconnect cleanly.
+
+    This function creates an MQTT client, connects to a broker and publishes a
+    single message. Once the message has been delivered, it disconnects cleanly
+    from the broker.
+
+    topic : the only required argument must be the topic string to which the
+            payload will be published.
+    payload : the payload to be published. If "" or None, a zero length payload
+              will be published.
+    qos : the qos to use when publishing,  default to 0.
+    retain : set the message to be retained (True) or not (False).
+    hostname : a string containing the address of the broker to connect to.
+               Defaults to localhost.
+    port : the port to connect to the broker on. Defaults to 1883.
+    client_id : the MQTT client id to use. If "" or None, the Paho library will
+                generate a client id automatically.
+    keepalive : the keepalive timeout value for the client. Defaults to 60
+                seconds.
+    will : a dict containing will parameters for the client: will = {'topic':
+           "<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
+           Topic is required, all other parameters are optional and will
+           default to None, 0 and False respectively.
+           Defaults to None, which indicates no will should be used.
+    auth : a dict containing authentication parameters for the client:
+           auth = {'username':"<username>", 'password':"<password>"}
+           Username is required, password is optional and will default to None
+           if not provided.
+           Defaults to None, which indicates no authentication is to be used.
+    tls : a dict containing TLS configuration parameters for the client:
+          dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
+          'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
+          'ciphers':"<ciphers">}
+          ca_certs is required, all other parameters are optional and will
+          default to None if not provided, which results in the client using
+          the default behaviour - see the paho.mqtt.client documentation.
+          Defaults to None, which indicates that TLS should not be used.
+    """
+
+    msg = {'topic':topic, 'payload':payload, 'qos':qos, 'retain':retain}
+    multiple([msg], hostname, port, client_id, keepalive, will, auth, tls, protocol)
+

plexpy/notifiers.py

@@ -21,7 +21,7 @@ from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 import email.utils
 from httplib import HTTPSConnection
-import paho.mqtt.client as mqtt
+from paho.mqtt.publish import single
 import os
 import re
 import requests
@@ -1917,6 +1917,119 @@ class JOIN(Notifier):
         return config_option
 
 
+class MQTT(Notifier):
+    """
+    MQTT notifications
+    """
+    _DEFAULT_CONFIG = {'broker': '',
+                       'port': 1883,
+                       'protocol': 'MQTTv311',
+                       'username': '',
+                       'password': '',
+                       'client_id': 'plexpy',
+                       'topic': '',
+                       'qos': 1,
+                       'retain': 0,
+                       'keep_alive': 60
+                       }
+
+    def notify(self, subject='', body='', action='', **kwargs):
+        if not subject or not body:
+            return
+
+        if not self.config['topic']:
+            logger.error(u"PlexPy Notifiers :: MQTT topic not specified.")
+            return
+
+        data = {'subject': subject.encode("utf-8"),
+                'body': body.encode("utf-8"),
+                'topic': self.config['topic'].encode("utf-8")}
+
+        auth = {}
+        if self.config['username']:
+            auth['username'] = self.config['username']
+        if self.config['password']:
+            auth['password'] = self.config['password']
+
+        single(self.config['topic'], payload=json.dumps(data), qos=self.config['qos'], retain=bool(self.config['retain']),
+               hostname=self.config['broker'], port=self.config['port'], client_id=self.config['client_id'],
+               keepalive=self.config['keep_alive'], auth=auth or None, protocol=self.config['protocol'])
+
+        return True
+
+    def return_config_options(self):
+        config_option = [{'label': 'Broker',
+                          'value': self.config['broker'],
+                          'name': 'mqtt_broker',
+                          'description': 'The hostname or IP address of the MQTT broker.',
+                          'input_type': 'text'
+                          },
+                         {'label': 'Port',
+                          'value': self.config['port'],
+                          'name': 'mqtt_port',
+                          'description': 'The network port for connecting to the MQTT broker.',
+                          'input_type': 'number'
+                          },
+                         {'label': 'Protocol',
+                          'value': self.config['protocol'],
+                          'name': 'mqtt_protocol',
+                          'description': 'The MQTT protocol version.',
+                          'input_type': 'select',
+                          'select_options': {'MQTTv31': '3.1',
+                                             'MQTTv311': '3.1.1'
+                                             }
+                          },
+                         {'label': 'Client ID',
+                          'value': self.config['client_id'],
+                          'name': 'mqtt_client_id',
+                          'description': 'The client ID for connecting to the MQTT broker.',
+                          'input_type': 'text'
+                          },
+                         {'label': 'Username',
+                          'value': self.config['username'],
+                          'name': 'mqtt_username',
+                          'description': 'The username to authenticate with the MQTT broker.',
+                          'input_type': 'text'
+                          },
+                         {'label': 'Password',
+                          'value': self.config['password'],
+                          'name': 'mqtt_password',
+                          'description': 'The password to authenticate with the MQTT broker.',
+                          'input_type': 'password'
+                          },
+                         {'label': 'Topic',
+                          'value': self.config['topic'],
+                          'name': 'mqtt_topic',
+                          'description': 'The topic to publish notifications to.',
+                          'input_type': 'text'
+                          },
+                         {'label': 'Quality of Service',
+                          'value': self.config['qos'],
+                          'name': 'mqtt_qos',
+                          'description': 'The quality of service level to use when publishing the notification.',
+                          'input_type': 'select',
+                          'select_options': {0: 0,
+                                             1: 1,
+                                             2: 2
+                                             }
+                          },
+                         {'label': 'Retain Message',
+                          'value': self.config['retain'],
+                          'name': 'mqtt_retain',
+                          'description': 'Set the message to be retained on the MQTT broker.',
+                          'input_type': 'checkbox'
+                          },
+                         {'label': 'Keep-Alive',
+                          'value': self.config['keep_alive'],
+                          'name': 'mqtt_keep_alive',
+                          'description': 'Maximum period in seconds before timing out the connection with the broker.',
+                          'input_type': 'number'
+                          }
+                         ]
+
+        return config_option
+
+
 class NMA(Notifier):
     """
     Notify My Android notifications
@@ -3175,155 +3288,6 @@ class XBMC(Notifier):
         return config_option
 
 
-class MQTT(Notifier):
-    """
-    MQTT notifications
-    """
-    _DEFAULT_CONFIG = {'broker': '',
-                       'port': '1883',
-                       'keep_alive': '60',
-                       'bind_address': '',
-                       'protocol': 'MQTTv311',
-                       'username': '',
-                       'password': '',
-                       'topic': ''
-                       }
-
-    def __init__(self, config=None):
-        self.set_config(config)
-        self.data = ''
-        self.loop_flag = 1
-        self.counter = 0
-        self.success = False
-        self.qos = 1
-        self.retain = False
-        self.client_id = 'plexpy'
-        logger.info(u"PlexPy Notifiers :: MQTT init.")
-        self.mqtt_client = mqtt.Client(protocol=self.config['protocol'])
-        self.mqtt_client.username_pw_set(self.config['username'], self.config['password'])
-        self.mqtt_client.on_connect = self.on_connect
-        self.mqtt_client.on_publish = self.on_publish
-        logger.info(u"PlexPy Notifiers :: MQTT initialized.")
-
-    def on_connect(self, client, userdata, flags, rc):
-        logger.info(u"PlexPy Notifiers :: MQTT notification: connect")
-        if rc == 0:
-            status = 'Connection successful'
-        elif rc == 1:
-            status = 'Connection refused - incorrect protocol version'
-        elif rc == 2:
-            status = 'Connection refused - invalid client identifier'
-        elif rc == 3:
-            status = 'Connection refused - server unavailable'
-        elif rc == 4:
-            status = 'Connection refused - bad username or password'
-        elif rc == 5:
-            status = 'Connection refused - not authorised'
-        else:
-            status = 'Connection refused - unknown error'
-
-        if rc == 0:
-            logger.info(u"PlexPy Notifiers :: MQTT connection: %s.", status)
-            logger.info(u"PlexPy Notifiers :: MQTT notification: Publishing message.")
-            (rc, mid) = self.mqtt_client.publish(self.config['topic'], json.dumps(self.data), self.qos, self.retain)
-        else:
-            logger.warn(u"PlexPy Notifiers :: MQTT connection: %s.", status)
-
-    def on_publish(self, client, userdata, mid):
-        logger.info(u"PlexPy Notifiers :: MQTT notification: publish")
-        self.mqtt_client.disconnect()
-        self.mqtt_client.loop_stop()
-        self.loop_flag = 0
-        if mid == 1:
-            self.success = True
-            logger.info(u"PlexPy Notifiers :: MQTT notification: Publishing succeeded.")
-            if self.qos == 0:
-                logger.info(u"PlexPy Notifiers :: MQTT notification: Note that because QoS is set to %s we can't be sure everything was delivered, only that the message was sent.", self.qos)
-        else:
-            logger.warn(u"PlexPy Notifiers :: MQTT notification: Publishing failed.")
-
-    def notify(self, subject='', body='', action='', **kwargs):
-        logger.info(u"PlexPy Notifiers :: MQTT notification: notify.")
-        if not subject or not body:
-            return
-        logger.info(u"PlexPy Notifiers :: MQTT notification: subject/body okay.")
-        self.data = {'title': subject.encode("utf-8"),
-                     'body': body.encode("utf-8"),
-                     'topic': self.config['topic'].encode("utf-8")}
-        logger.info(u"PlexPy Notifiers :: MQTT notification: data okay.")
-
-        self.mqtt_client.connect(self.config['broker'], port=self.config['port'], keepalive=self.config['keep_alive'], bind_address=self.config['bind_address'])
-        self.mqtt_client.loop_start()
-        logger.info(u"PlexPy Notifiers :: MQTT notification: connect ran.")
-        logger.info(u"PlexPy Notifiers :: MQTT notification: loop started.")
-
-        while self.loop_flag == 1 and self.counter < 1000:
-            logger.info(u"PlexPy Notifiers :: MQTT notification: waiting for on_publish to complete %s", self.counter)
-            time.sleep(.01)
-            self.counter+=1
-
-        self.mqtt_client.disconnect()
-        self.mqtt_client.loop_stop()
-        return self.success
-
-    def return_config_options(self):
-        config_option = [{'label': 'Broker',
-                          'value': self.config['broker'],
-                          'name': 'mqtt_broker',
-                          'description': 'The hostname or IP address of the remote broker.',
-                          'input_type': 'text'
-                          },
-                         {'label': 'Port',
-                          'value': self.config['port'],
-                          'name': 'mqtt_port',
-                          'description': 'The network port of the server host to connect to.',
-                          'input_type': 'number'
-                          },
-                         {'label': 'Keep-alive',
-                          'value': self.config['keep_alive'],
-                          'name': 'mqtt_keep_alive',
-                          'description': 'Maximum period in seconds allowed between communications with the broker.',
-                          'input_type': 'number'
-                          },
-                         {'label': 'Bind Address',
-                          'value': self.config['bind_address'],
-                          'name': 'mqtt_bind_address',
-                          'description': 'The IP address of a local network interface to bind this client to, assuming multiple interfaces exist. Typically left empty.',
-                          'input_type': 'text'
-                          },
-                         {'label': 'Protocol',
-                          'value': self.config['protocol'],
-                          'name': 'mqtt_protocol',
-                          'description': 'The version of the MQTT protocol to use.',
-                          'input_type': 'select',
-                          'select_options': {'': '',
-                                             'MQTTv31': '3.1',
-                                             'MQTTv311': '3.1.1'
-                                             }
-                          },
-                         {'label': 'Username',
-                          'value': self.config['username'],
-                          'name': 'mqtt_username',
-                          'description': 'The username to authenticate to the mqtt broker.',
-                          'input_type': 'text'
-                          },
-                         {'label': 'Password',
-                          'value': self.config['password'],
-                          'name': 'mqtt_password',
-                          'description': 'The password to authenticate to the mqtt broker.',
-                          'input_type': 'password'
-                          },
-                         {'label': 'Topic',
-                          'value': self.config['topic'],
-                          'name': 'mqtt_topic',
-                          'description': 'The topic to publish notifications to.',
-                          'input_type': 'text'
-                          }
-                         ]
-
-        return config_option
-
-
 def upgrade_config_to_db():
     logger.info(u"PlexPy Notifiers :: Upgrading to new notification system...")