MQTT client for Python

python3 publish.py -t topic -f fileName -q qos -c clientid -s hostname:port -u username -p password

Options may be present in any order, and if absent, default values will be used.

python3 publish.py -t testTopic -f data.json

Connecting to MQTT broker on host "127.0.0.1", port 1883, sending your file (data.json) to topic "testTopic", and then disconnecting.

After publish.single() call.  Goodbye!

# -*- coding: Latin-1 -*-
#
# publish.py
#
# This Python script connects to the specified FairCom EdgeMQ MQTT broker and publishes one MQTT message to the specified topic.  The contents of that message is read from disk.
#

import getopt
import os
import sys

try:
  import paho.mqtt.publish as publish
except ImportError:
  print('Error!  This script requires the Paho MQTT module, which does not appear to be installed.  Please install it, using a command such as "pip3 install paho-mqtt".')
  sys.exit(2)


def usage():
  print('This script connects to the specified MQTT broker, publishes one MQTT message (from the specified file), and then disconnects from the broker.\nCommand line options:\n  -h, --help : display help\n  -s MQTT server information - hostName or IPAddress : portNumber.  Default="127.0.0.1:1883".\n  -u user name.  No Default.\n  -p password.  No Default.  Ignored if user name is not specified.\n  -t MQTT topic name.  Default="testTopic".\n  -c MQTT Client ID.  Default="testPubClientID".\n  -q publish QoS.  Default=1.\n-f file that contains the message which will be published.  Required.')
  print("Example: python publish.py -s 127.0.0.1:1883 -u admin -p ADMIN -t testTopic -c testPubClientID -q 1 -f data.json\n")



# Default values for the command-line arguments
server_data = '127.0.0.1:1883'
user_name = None
user_pass = None
topic = 'testTopic'
client_id = 'testPubClientID'
publish_qos = 1
file_name = None  # Required argument - File to publish to the topic

# Read the command-line arguments
try:
  opts, args = getopt.getopt(sys.argv[1:], 'hs:u:p:t:c:f:q:', longopts=['help'])
except getopt.GetoptError as e:
  print(str(e))
  usage()
  sys.exit(2)
for opt, arg in opts:
  if opt in ('-h', '--help'):
    usage()
    sys.exit(1)
  if opt == '-s':
    server_data = arg
  elif opt == '-u':
    user_name = arg
  elif opt == '-p':
    user_pass = arg
  elif opt == '-t':
    topic = arg
  elif opt == '-c':
    client_id = arg
  elif opt == '-f':
    file_name = arg
  elif opt == '-q':
    # Make sure the publish QoS is valid
    if not arg.isdigit():
      print('Error!  The "-q" (publish QoS) argument must be a number from 0 to 2 inclusive.')
      usage()
      sys.exit(2)
    publish_qos = int(arg)
    if publish_qos<0 or publish_qos>2:
      print('Error!  The "-q" (publish QoS) argument must be a number from 0 to 2 inclusive.')
      usage()
      sys.exit(2)
  else:
    print('Error!  Unhandled option: "%s"' % opt)

# Make sure the required argument is present
if file_name is None:
  print('Error!  The "-f" (file) argument is required.')
  usage()
  sys.exit(2)

# Separate the host name and port from the 'server_data' variable, because Paho needs them individually.
host = '127.0.0.1'
port = 1883
parts = server_data.split(":")
if len(parts) == 1:  # No colon.  Assume the entire string is the host name / IP Address
  host = server_data
elif len(parts) == 2:  # One colon.  Assume hostname:port or IPAddress:port
  host = parts[0]
  port = int(parts[1])
else:
  print('Error!  I do not understand the -s (server) parameter you specified.  Please use one of the following:\n  hostname:port\n  IPAddress:port.')
  sys.exit(2)

# Create the authentication data dictionary which will be passed to the Paho MQTT client
if (user_name is None) or (user_pass is None):
  auth_data = None
else:
  auth_data = {'username': user_name, 'password': user_pass}

# Verify that the specified file actually exists
if not os.path.isfile(file_name):
  print('Error!  The specified file "%s" does not exist.' % file_name)
  sys.exit(2)

# Read in the file, all at once
with open(file_name, "rb") as inFile:
  message_payload = inFile.read()

# This is a blocking call unless there is an exception while connecting to the MQTT broker.
# This connects to the MQTT broker and sends the message contained in the file to the specified topic.
# https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php#id2
print('Connecting to MQTT broker on host "%s", port %d, sending your file (%s) to topic "%s", and then disconnecting.' % (host, port, file_name, topic))
try:
  publish.single(topic, payload=message_payload, qos=publish_qos, retain=False, hostname=host, port=port, client_id=client_id, keepalive=60, will=None, auth=auth_data, tls=None, protocol=publish.paho.MQTTv311, transport="tcp")
except ConnectionRefusedError as e:
  print("Connection Refused error.  Is your FairCom EDGE server running?  %s" % str(e))
except KeyboardInterrupt:
  print('User pressed <ctrl-c> to exit script.')
except Exception as e:
  print("MQTT exception while connecting to broker %s." % str(e))
  # try to print a little more information about the problem
  if len(e.args) > 0:
    print( "[%s]" % publish.paho.connack_string( e.args[0] ) )


print('After publish.single() call.  Goodbye!')

python3 subscribe.py -t topic -d directory -q QoS -c clientid -s hostname:port -u username -p password

Options may be present in any order, and if absent, default values will be used.

python3 subscribe.py -t testTopic -d testDir

Connecting to MQTT broker on host "127.0.0.1", port 1883 and subscribing to topic "testTopic".  Messages that arrive will be printed to the screen.

Msg 1: {

        "property":    "value",

        "string":      "string",

        "number":      1,

        "boolean":     true,

        "null...

When a directory is specified, the utility writes the payload of each received message to a file. The name of the file is the name of the topic you specified followed by an underscore plus a sequential number. Topic names are case-sensitive. The file has no extension because an MQTT payload can be anything.

# -*- coding: Latin-1 -*-
#
# subscribe.py
#
# This Python script connects to the specified FairCom EdgeMQ MQTT broker and subscribes to the specified topic.
# It then prints each message which arrives from the broker to the console and (optionally) writes the message to a numbered file in the specified directory.
#

import sys
import getopt
import os

try:
  import paho.mqtt.subscribe as subscribe
except ImportError:
  print('Error!  This script requires the Paho MQTT module, which does not appear to be installed.  Please install it, using a command such as "pip3 install paho-mqtt".')
  sys.exit(2)


def usage():
  print('This script connects to the specified MQTT broker and subscribes to the specified topic.  It then prints each message that arrives to the console and (if desired) writes it to a numbered file in the specified directory.  Warning!  Existing files are overwritten!\nCommand line options:\n  -h, --help : display help\n  -s MQTT server information - hostName or IPAddress : portNumber.  Default="127.0.0.1:1883".\n  -u user name.  No Default.\n  -p password.  No Default.  Ignored if user name is not specified.\n  -t MQTT topic name.  Default="testTopic".\n  -c MQTT Client ID.  Default="testSubClientID".\n  -q subscribe QoS.  Default=1.\n  -d directory to store files in, if this is desired.  Defaults to not writing files.')
  print("Example: python subscribe.py -s 127.0.0.1:1883 -u admin -p ADMIN -t testTopic -c testSubClientID -q 1 -d messages\n")


message_number = 0


# This function gets invoked each time a message arrives from the MQTT broker.
def on_message(client, userdata, msg):
  global message_number, directory, safe_file_name
  message_number += 1
  # Write the message to a new file in the specified directory, if a directory was specified.
  if not directory is None:
    with open(os.path.join(directory, "%s_%06d" % (safe_file_name, message_number)), "wb") as f:
      f.write(msg.payload)
  # Now try to decode the message as ASCII / UTF-8 Unicode for display.
  # If that fails, just show a hex dump of the message.
  try:
    payload=msg.payload.decode('utf8') # convert the bytes to a string.
  except UnicodeDecodeError:
    payload=msg.payload.hex()

  # Print the first 100 characters of the string-version of the payload.
  if len(payload)>100:
    print("Msg %d: %s..." % (message_number, payload[:100]))
  else:
    print("Msg %d: %s" % (message_number, payload))


# Default values for the command-line arguments
server_data = '127.0.0.1:1883'
user_name = None
user_pass = None
topic = 'testTopic'
client_id = 'testSubClientID'
subscribe_qos = 1
directory = None  # Required argument - Folder to store message payloads in

# Read the command-line arguments
try:
  opts, args = getopt.getopt(sys.argv[1:], 'hs:u:p:t:c:d:q:', longopts=['help'])
except getopt.GetoptError as e:
  print(str(e))
  usage()
  sys.exit(2)
for opt, arg in opts:
  if opt in ('-h', '--help'):
    usage()
    sys.exit(1)
  if opt == '-s':
    server_data = arg
  elif opt == '-u':
    user_name = arg
  elif opt == '-p':
    user_pass = arg
  elif opt == '-t':
    topic = arg
  elif opt == '-c':
    client_id = arg
  elif opt == '-d':
    directory = arg
  elif opt == '-q':
    # Make sure the subscribe QoS is valid
    if not arg.isdigit():
      print('Error!  The "-q" (subscribe QoS) argument must be a number from 0 to 2 inclusive.')
      usage()
      sys.exit(2)
    subscribe_qos = int(arg)
    if subscribe_qos<0 or subscribe_qos>2:
      print('Error!  The "-q" (subscribe QoS) argument must be a number from 0 to 2 inclusive.')
      usage()
      sys.exit(2)
  else:
    print('Error!  Unhandled option: "%s"' % opt)

# Separate the host name and port from the 'server_data' variable, because Paho needs them individually.
host = '127.0.0.1'
port = 1883
parts = server_data.split(":")
if len(parts) == 1:  # No colon.  Assume the entire string is the host name / IP Address
  host = server_data
elif len(parts) == 2:  # One colon.  Assume hostname:port or IPAddress:port
  host = parts[0]
  port = int(parts[1])
else:
  print('Error!  I do not understand the -s (server) parameter you specified.  Please use one of the following:\n  hostname:port\n  IPAddress:port.')
  sys.exit(2)

# Create the authentication data dictionary which will be passed to the Paho MQTT client
if (user_name is None) or (user_pass is None):
  auth_data = None
else:
  auth_data = {'username': user_name, 'password': user_pass}

# If the user specified an output directory, create it if it does not already exist.
# Then come up with a safe name to use for the files I create.
if not directory is None:
  if not os.path.exists(directory):
    print('The specified directory "%s" does not exist.  Attempting to create it...' % directory)
    os.makedirs(directory)
  # Sanitize the topic name to replace characters that are not allowed in file names
  # with underscores.  I will be rather agressive with this...
  safe_file_name = topic
  illegal_chars = ['/', '\\', '*', '"', '?', ':', '>', '<', '|', ' ', '.', '&', '%', '#']
  for c in illegal_chars:
    safe_file_name = safe_file_name.replace(c, '_')
  # Now, to prevent path overflows, save the last 500 chars of the file name.
  safe_file_name = safe_file_name[-500:]

# This is a blocking call unless there is an exception while connecting to the MQTT broker.
# This connects to the MQTT broker and subscribes to the specified topic.
# Each time an MQTT packet arrives, the on_message() callback will be invoked.
# This automatically handles re-connects to the broker.
# https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php#id2
print('Connecting to MQTT broker on host "%s", port %d and subscribing to topic "%s".  Messages that arrive will be printed to the screen' %  (host, port, topic), end='')
if directory is None:
  print(".")
else:
  print('and stored in the "%s" directory.' % (directory))

try:
  subscribe.callback(on_message, topic, qos=subscribe_qos, userdata=None, hostname=host, port=port, client_id=client_id, keepalive=60, will=None, auth=auth_data, tls=None, protocol=subscribe.paho.MQTTv311)
except ConnectionRefusedError as e:
  print("Connection Refused error.  Is your FairCom EDGE server running?  %s" % str(e))
except KeyboardInterrupt:
  print('User pressed <ctrl-c> to exit script.')
except Exception as e:
  print("MQTT exception while connecting to broker %s." % str(e))
  # try to print a little more information about the problem
  if len(e.args) > 0:
    print("[%s]" % subscribe.paho.connack_string(e.args[0]))


print('After subscribe.callback() call.  Goodbye!')