How to Send a File Using MQTT and Python

In this project guide we will look at how we can send a file or picture using MQTT.

The script is written in Python and the approach I used was to send the file as bytes.

So as to identify the start and end of file we use a header block and end block frame of 200 bytes.

The Send format is

• Send header frame
• Loop to Read data from disk in byte chunks and send. Also calculate a file hash.
• Send end block containing the file hash when done.

The Receive format is:

• Read in data and extract header block
• Look for end block while receiving data.
• Save file to disk and calculate a file hash
• When end detected compare the file hash in the received end data block with the one calculated at the receiver to verify success.

Notes:

1. I didn’t extract the file name from the header to create the received file with the same name as the sent file.
2. I used a QOS of 1 and waited for a PUBACK before sending the next data block to be nice to the MQTT broker and to help guarantee a successful send.
3. I don’t detect network failures in this script.

The Code Is shown below and Code notes follow the code.

#! python3.4
###demo code provided by Steve Cope at www.steves-internet-guide.com
##email steve@steves-internet-guide.com
###Free to use for any purpose
"""
Send File Using MQTT
"""
import time
import paho.mqtt.client as paho
import hashlib
broker="broker.hivemq.com"
broker="iot.eclipse.org"
broker="192.168.1.206"
#filename="DSCI0027.jpg"
filename="passwords.py" #file to send
topic="data/files"
qos=1
data_block_size=2000
fo=open(filename,"rb")
fout=open("1out.txt","wb") #use a different filename
# for outfile as I'm rnning sender and receiver together
def process_message(msg):
   """ This is the main receiver code
   """
   if len(msg)==200: #is header or end
      msg_in=msg.decode("utf-8")
      msg_in=msg_in.split(",,")
      if msg_in[0]=="end": #is it really last packet?
         in_hash_final=in_hash_md5.hexdigest()
         if in_hash_final==msg_in[2]:
            print("File copied OK -valid hash  ",in_hash_final)
         else:
            print("Bad file receive   ",in_hash_final)
         return False
      else:
         if msg_in[0]!="header":
            in_hash_md5.update(msg)
            return True
         else:
            return False
   else:
      in_hash_md5.update(msg)
      return True
#define callback
def on_message(client, userdata, message):
   time.sleep(1)
   #print("received message =",str(message.payload.decode("utf-8")))
   if process_message(message.payload):
      fout.write(message.payload)

def on_publish(client, userdata, mid):
    #logging.debug("pub ack "+ str(mid))
    client.mid_value=mid
    client.puback_flag=True  


def wait_for(client,msgType,period=0.25,wait_time=40,running_loop=False):
    client.running_loop=running_loop #if using external loop
    wcount=0  
    while True:
        #print("waiting"+ msgType)
        if msgType=="PUBACK":
            if client.on_publish:        
                if client.puback_flag:
                    return True
     
        if not client.running_loop:
            client.loop(.01)  #check for messages manually
        time.sleep(period)
        #print("loop flag ",client.running_loop)
        wcount+=1
        if wcount>wait_time:
            print("return from wait loop taken too long")
            return False
    return True 

def send_header(filename):
   header="header"+",,"+filename+",,"
   header=bytearray(header,"utf-8")
   header.extend(b','*(200-len(header)))
   print(header)
   c_publish(client,topic,header,qos)

def send_end(filename):
   end="end"+",,"+filename+",,"+out_hash_md5.hexdigest()
   end=bytearray(end,"utf-8")
   end.extend(b','*(200-len(end)))
   print(end)
   c_publish(client,topic,end,qos)

def c_publish(client,topic,out_message,qos):
   res,mid=client.publish(topic,out_message,qos)#publish
   if res==0: #published ok
      if wait_for(client,"PUBACK",running_loop=True):
         if mid==client.mid_value:
            print("match mid ",str(mid))
            client.puback_flag=False #reset flag
         else:
            raise SystemExit("not got correct puback mid so quitting")
         
      else:
         raise SystemExit("not got puback so quitting")
client= paho.Client("client-001")  #create client object client1.on_publish = on_publish                          #assign function to callback client1.connect(broker,port)                                 #establish connection client1.publish("data/files","on")  
######
client.on_message=on_message
client.on_publish=on_publish
client.puback_flag=False #use flag in publish ack
client.mid_value=None
#####
print("connecting to broker ",broker)
client.connect(broker)#connect
client.loop_start() #start loop to process received messages
print("subscribing ")
client.subscribe(topic)#subscribe
time.sleep(2)
start=time.time()
print("publishing ")
send_header(filename)
Run_flag=True
count=0
##hashes
out_hash_md5 = hashlib.md5()
in_hash_md5 = hashlib.md5()

while Run_flag:
   chunk=fo.read(data_block_size) # change if want smaller or larger data blcoks
   if chunk:
      out_hash_md5.update(chunk)
      out_message=chunk
      #print(" length =",type(out_message))
      c_publish(client,topic,out_message,qos)
         
   else:
      #send hash
      out_message=out_hash_md5.hexdigest()
      send_end(filename)
      #print("out Message ",out_message)
      res,mid=client.publish("data/files",out_message,qos=1)#publish
      Run_flag=False
time_taken=time.time()-start
print("took ",time_taken)
time.sleep(4)
client.disconnect() #disconnect
client.loop_stop() #stop loop
fout.close() #close files
fo.close()

When I run the script with a test file this is what I see:

Code Notes:

1. The process_message function handles the received messages and stores the data in a file.
2. I use a number of flags attached to the client instance like:
   client.puback_flag
   client.mid_value
3. The has is used to verify data integrity

Code Improvements

You can send file details in the header section as JSON encoded data the following snippets will send the file name and on the receive extract the file name and use it.

If you need to include a lot of information you can always increase the header size.

On the sender you need to change the send_header and send_end functions.

def send_header(filename):

   file_data={"filename":filename}
   file_data_json=json.dumps(file_data)
   header="header"+",,"+file_data_json + ",,"
   header=bytearray(header,"utf-8")
   header.extend(b'x'*(200-len(header)))
   print(header)
   c_publish(client,topic,header,qos)
def send_end(filename):
   end="end"+",,"+filename+",,"+out_hash_md5.hexdigest()+",,"
   end=bytearray(end,"utf-8")
   end.extend(b'x'*(200-len(end)))
   print(end)
   c_publish(client,topic,end,qos)

On the receiver you need to change the on_message and process_message functions

def on_message(client, userdata, message):
   global fout
   #time.sleep(1)
   #print("received message =",str(message.payload.decode("utf-8")))
   if process_message(message.payload):
      fout.write(message.payload)
############
def extract_file_data(file_data):
   data=json.loads(file_data)
   filename=data["filename"]
   return filename
def process_message(msg):
   """ This is the main receiver code
   """
   global fout
   print("received ")
   if len(msg)==200: #is header or end
      msg_in=msg.decode("utf-8","ignore")
      msg_in=msg_in.split(",,")
      if msg_in[0]=="header": #header
         filename=extract_file_data(msg_in[1])
         file_out="copy-"+filename
         fout=open(file_out,"wb") #use a different filename

      if msg_in[0]=="end": #is it really last packet?
         in_hash_final=in_hash_md5.hexdigest()
         if in_hash_final==msg_in[2]:
            print("File copied OK -valid hash  ",in_hash_final)
         else:
            print("Bad file receive   ",in_hash_final)
         return False
      else:
         if msg_in[0]!="header":
            in_hash_md5.update(msg)
            return True
         else:
            return False
   else:
      in_hash_md5.update(msg)
      #msg_in=msg.decode("utf-8","ignore")
      if len(msg) <100:
         print(msg)
      return True

Code Download

Send File and receive file split and single send/receive file

Sending the entire File

In the example above we split the file into chunks which is necessary when sending over the Internet using third party brokers as they will probably have message size restrictions.

If you control the broker then you don’t need to do this and you can send the entire file in one go.

All you need to do is read in the file and place it into a MQTT payload. I do this for backing up databases on my system with files around 1MB and it works perfectly.

Comments and Feedback

Was there enough detail in this guide for you to follow?

Please help me improve these guides by leaving your comments,rating them,asking questions.

Was This Useful?

Related tutorials and resources:

• Encrypting MQTT Payloads with Python – Example Code.
• Simple Python MQTT Data Logger
• Sending JSON Data over MQTT Using Python
• Logging Sensor Data to a SQL Database Using  Python
• Simple Controllable MQTT Sensor Simulator in Python
• Control Raspberry Pi GPIO Pins Using MQTT
• Republish HTML Data Over MQTT