devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

笔记（三）mqtt客户端代码分析（python）_python mqtt消息解析

作者：我家小花儿 | 2024-03-04 20:30:11

赞

踩

python mqtt消息解析

一、逻辑分析

逻辑一： 

逻辑二： 

 注：因为在我们这个课设项目中服务器向客户端传输图片时将一个图片分包传输，所以在客户端需要将图片按序重新组装，然后再进行人脸识别 

二、重要模块代码

1.通过影子设备文件获取userid

  1. a = []#获取userId
  2. with open("/home/simon/mqtt/"+"shadow.ini","r") as ud: 
  3.     mes = ud.read()
  4.     mes_parsed = json.loads(mes)
  5.     meslist = mes_parsed['payload']['state']['reported']['faceList']
  6.     for i in meslist:
  7.         #print(str(i))
  8.         userid = i['userId']
  9.         a.append(int(userid))
  10.         #print(userid)
  11.     #print(a)

2.对文件夹中的图片进行循环扫描

  1. def file_circle(dirs5,houzhui,i,picture):
  2.     global picture2
  3.     for dirs4 in dirs5:
  4.         #print("dirs4:"+dirs4)
  5.         d2 = os.path.basename(dirs4)#图片名加后缀
  6.         num2 = os.path.splitext(d2)[0]#图片名
  7.         basename2 = os.path.splitext(d2)[1] #后缀名
  8.         if(int(num2)==i):
  9.             with open("/home/simon/mqtt/picture_recieve/"+str(num2)+basename2,"rb") as p1:
  10.                 picture2 = picture + str(base64.b64encode(p1.read()),encoding="utf-8")#bytes
  11.             break
  12.     return(picture2)

3.当topic为control且type为imageUpdate时需现将存有图片的文件夹中的对应userid的图片删除

  1.         if(msg_type=="imageUpdate"):
  2.             rootdir = "/home/simon/mqtt/picture_recieve/"
  3.             file_delete = os.listdir(rootdir)
  4.             for file in file_delete:
  5.                 if houzhui in file:
  6.                     del_file = rootdir + file#当代码和要删除的文件不在同一个文件夹时，必须使用绝对路径
  7.                     os.remove(del_file)#删除文件

4.判断接收的到一个userid的图片是否全部收到

  1.         key = len(u)-1
  2.         if(key<0):
  3.             u.append(useridnum)
  4.             v[key]=1
  5.         else:
  6.             while(key>=0):
  7.                 if(uesridnum==u[key]):
  8.                     v[key]+=1                   
  9.                 else:
  10.                     u.append(useridnum)
  11.                     v[key]=1
  12.                 if(v[key]==packettotal):
  13.                     break
  14.                 key = key-1

5.一张图片接收完以后进行组装

  1.         #一张图片全部接收后进行组装
  2.         if(v[key]==packettotal):
  3.             i=1
  4.             #all = r'/home/simon/mqtt/picture_recieve'
  5.             bool_flag2 = False#大循环跳出标志
  6.             dirs2 = []
  7.             dirs0 = []
  8.             for dirs1 in glob.glob('/home/simon/mqtt/picture_recieve'+"/*"+houzhui):
  9.                 dirs2.append(dirs1)
  10.                 dirs0.append(dirs1)
  11.             #print(dirs2)
  12.             for dirs3 in dirs2:
  13.                 #print("dirs3:"+dirs3)
  14.                 d1 = os.path.basename(dirs3)#图片名加后缀
  15.                 num1 = os.path.splitext(d1)[0]#图片名
  16.                 basename1 = os.path.splitext(d1)[1] #后缀名
  17.                 if(int(num1)==1):
  18.                     with open("/home/simon/mqtt/picture_recieve/"+str(num1)+basename1,"rb") as p1:
  19.                         picture1 = str(base64.b64encode(p1.read()),encoding="utf-8")#bytes
  20.                     i=i+1
  21.                     bool_flag1 = False#跳出小循环标志
  22.                     while True:
  23.                         if(i!=packettotal):
  24.                             picture1 = file_circle(dirs0,houzhui,i,picture1)
  25.                             i=i+1
  26.                         if(i==packettotal):
  27.                             picture1 = file_circle(dirs0,houzhui,i,picture1)
  28.                             i=1
  29.                             bool_flag1 = True
  30.                             break
  31.                     if(bool_flag1):
  32.                         bool_flag2 = True
  33.                         temp = base64.b64decode(picture1)#bytes
  34.                         with open("/home/simon/mqtt/picture/"+str(recieve_parsed['payload']['user'])+str(recieve_parsed['payload']['userName'])+".jpg","wb") as o:
  35.                             o.write(temp)
  36.                 if(bool_flag2):
  37.                     break

6.如果get到的影子设备文件内desired有内容，即表示客户有需要我做的事情，所以完成desired的事情后，将其放置到reported内发送delta，报告给客户表明我做完了这件事

  1.         if(shadow_desired!={}):
  2.             if('state' in shadow_desired):
  3.                 state = shadow_desired['state']
  4.             else:state = shadow_reported['state']
  5.             if('faceList' in shadow_desired):
  6.                 s_d_facelist = shadow_desired['faceList']#数组
  7.                 s_d_f = ",".join('%s'%id for id in s_d_facelist)#字符串
  8.             else:
  9.                 s_r_facelist = shadow_reported['faceList']#数组
  10.                 s_d_f = ",".join('%s'%id for id in s_r_facelist)#字符串
  11.             data_delta = {
  12.                 "type":"delta",
  13.                 "state":{
  14.                     "reported":{
  15.                         "state":state,
  16.                         "faceList":[s_d_f]
  17.                     }
  18.                 },
  19.                 "version":shadow_version,
  20.                 "clientToken":str(clientToken)
  21.             }
  22.             data_delta_param = json.dumps(data_delta)
  23.             client.publish("$shadow/operation/414ce45bde0a4bc1949fec75a7d2f699", payload=data_delta_param,qos=2,retain=False)#delta

7.循环扫描存有要发送信息的文件，有信息则发布，然后删除发送的信息，继续循环扫描

  1.     while True:
  2.         with open('/home/simon/mqtt/'+'FaceAndTime.json','r',encoding="utf-8") as old_f:
  3.             #读取所有行，每行会是一个字符串
  4.             for j in old_f.readlines():
  5.                 if(j):
  6.                     print(j)
  7.                     face = json.loads(j,object_hook=dict2Data)
  8.                     data_face = {
  9.                                 "type":"history",
  10.                                 "payload":{
  11.                                     "history":[
  12.                                     {
  13.                                         "timestamp":face.timestamp,
  14.                                         "history":{"name":face.name}
  15.                                         }]
  16.                                     },
  17.                                 "clientToken":str(clientToken)
  18.                     }
  19.                     client.publish("414ce45bde0a4bc1949fec75a7d2f699/event", payload=str(data_face), qos=0,retain=False)#发送人脸数据
  20.                     with open('/home/simon/mqtt/'+'FaceAndTime.json','w+',encoding="utf-8") as new_f:
  21.                         seek_point = old_f.tell()#光标在被删除行的行首，记录该位置
  22.                         new_f.seek(seek_point,0)#设置光标位置
  23.                         old_f.readline()#读需要删除的行，光标移动到下一行行首
  24.                         next_line=old_f.readline()#被删除行的下一行读给next_line
  25.                         while next_line:#连续覆盖剩余行，后面所有行上移一行
  26.                             new_f.write(next_line)
  27.                             next_line=old_f.readline()
  28.                             new_f.truncate()

三、完整代码

逻辑一： 

  1. import paho.mqtt.client as mqtt
  2. import json
  3. import base64
  4. import uuid
  5. import time
  6. import sys
  7. import multiprocessing
  8. import re
  9. import os
  10. from face_recognition.face_recognition_cli import image_files_in_folder
  11. import glob
  12.  
  13. HOST = "api.yumik.top" 
  14. PORT = 1883
  15. client_id="414ce45bde0a4bc1949fec75a7d2f699"
  16. clientToken=uuid.uuid4()
  17.  
  18. class Data(object):
  19.     def __init__(self,name,timestamp):
  20.         self.timestamp = timestamp
  21.         self.name = name
  22. def dict2Data(d):
  23.     return Data(d['name'],d['timestamp'])
  24.  
  25. a = []#获取userId
  26. with open("/home/simon/mqtt/"+"shadow.ini","r") as ud: 
  27.     mes = ud.read()
  28.     mes_parsed = json.loads(mes)
  29.     meslist = mes_parsed['payload']['state']['reported']['faceList']
  30.     for i in meslist:
  31.         #print(str(i))
  32.         userid = i['userId']
  33.         a.append(int(userid))
  34.         #print(userid)
  35.     #print(a)
  36.  
  37. #publish event
  38. data_in = {
  39.      "type":"face",
  40.      "payload":{"userList":a,},
  41.      #"payload":{"userList":[2],},
  42.      "clientToken":str(clientToken)
  43.      }
  44. param_in = json.dumps(data_in)
  45. #publish shadow/operation
  46. data1 = {
  47.      "type":"get",
  48.      "clientToken":str(clientToken)
  49.      }
  50. param1 = json.dumps(data1)
  51.  
  52. def on_connect(client, userdata, flags, rc):
  53.     print("Connected with result code "+str(rc))        
  54.  
  55. def file_circle(dirs5,houzhui,i,picture):
  56.     global picture2
  57.     for dirs4 in dirs5:
  58.         #print("dirs4:"+dirs4)
  59.         d2 = os.path.basename(dirs4)#图片名加后缀
  60.         num2 = os.path.splitext(d2)[0]#图片名
  61.         basename2 = os.path.splitext(d2)[1] #后缀名
  62.         if(int(num2)==i):
  63.             with open("/home/simon/mqtt/picture_recieve/"+str(num2)+basename2,"rb") as p1:
  64.                 picture2 = picture + str(base64.b64encode(p1.read()),encoding="utf-8")#bytes
  65.             break
  66.     return(picture2)
  67.  
  68. picture = ""
  69. houzhui = ""
  70. u = []#存用户的id
  71. v = []#存每个用户的图片数量，递增
  72. def on_message(client, userdata, msg):
  73.     global picture
  74.     global i
  75.     global houzhui
  76.     recieve = msg.payload.decode("utf-8")
  77.     recieve_parsed = json.loads(recieve)
  78.     msg_type = recieve_parsed['type']
  79.     if(msg.topic=="414ce45bde0a4bc1949fec75a7d2f699/control" and (msg_type=="face" or msg_type=="imageUpdate")):
  80.         print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  81.         print("control achieve!")
  82.         packetnum = recieve_parsed['payload']['packet']
  83.         packettotal = recieve_parsed['payload']['totalPacket']
  84.         picture = recieve_parsed['payload']['userImageBase64']
  85.         temp = base64.b64decode(picture)#bytes
  86.         useridnum = str(recieve_parsed['payload']['user'])
  87.         houzhui = "."+str(recieve_parsed['payload']['user'])
  88.         if(msg_type=="imageUpdate"):
  89.             rootdir = "/home/simon/mqtt/picture_recieve/"
  90.             file_delete = os.listdir(rootdir)
  91.             for file in file_delete:
  92.                 if houzhui in file:
  93.                     del_file = rootdir + file#当代码和要删除的文件不在同一个文件夹时，必须使用绝对路径
  94.                     os.remove(del_file)#删除文件
  95.         with open("/home/simon/mqtt/picture_recieve/"+str(packetnum)+"."+str(recieve_parsed['payload']['user']),"wb") as f:
  96.             f.write(temp)#int
  97.         #判断一张图片是否全部接收
  98.         key = len(u)-1
  99.         if(key<0):
  100.             u.append(useridnum)
  101.             v[key]=1
  102.         else:
  103.             while(key>=0):
  104.                 if(uesridnum==u[key]):
  105.                     v[key]+=1                   
  106.                 else:
  107.                     u.append(useridnum)
  108.                     v[key]=1
  109.                 if(v[key]==packettotal):
  110.                     break
  111.                 key = key-1
  112.         #一张图片全部接收后进行组装
  113.         if(v[key]==packettotal):
  114.             i=1
  115.             #all = r'/home/simon/mqtt/picture_recieve'
  116.             bool_flag2 = False#大循环跳出标志
  117.             dirs2 = []
  118.             dirs0 = []
  119.             for dirs1 in glob.glob('/home/simon/mqtt/picture_recieve'+"/*"+houzhui):
  120.                 dirs2.append(dirs1)
  121.                 dirs0.append(dirs1)
  122.             #print(dirs2)
  123.             for dirs3 in dirs2:
  124.                 #print("dirs3:"+dirs3)
  125.                 d1 = os.path.basename(dirs3)#图片名加后缀
  126.                 num1 = os.path.splitext(d1)[0]#图片名
  127.                 basename1 = os.path.splitext(d1)[1] #后缀名
  128.                 if(int(num1)==1):
  129.                     with open("/home/simon/mqtt/picture_recieve/"+str(num1)+basename1,"rb") as p1:
  130.                         picture1 = str(base64.b64encode(p1.read()),encoding="utf-8")#bytes
  131.                     i=i+1
  132.                     bool_flag1 = False#跳出小循环标志
  133.                     while True:
  134.                         if(i!=packettotal):
  135.                             picture1 = file_circle(dirs0,houzhui,i,picture1)
  136.                             i=i+1
  137.                         if(i==packettotal):
  138.                             picture1 = file_circle(dirs0,houzhui,i,picture1)
  139.                             i=1
  140.                             bool_flag1 = True
  141.                             break
  142.                     if(bool_flag1):
  143.                         bool_flag2 = True
  144.                         temp = base64.b64decode(picture1)#bytes
  145.                         with open("/home/simon/mqtt/picture/"+str(recieve_parsed['payload']['user'])+str(recieve_parsed['payload']['userName'])+".jpg","wb") as o:
  146.                             o.write(temp)
  147.                 if(bool_flag2):
  148.                     break
  149.     if(msg.topic=="414ce45bde0a4bc1949fec75a7d2f699/control" and msg_type=="history"):
  150.         print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  151.     if(msg.topic=="$shadow/operation/result/414ce45bde0a4bc1949fec75a7d2f699"):
  152.         #print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  153.         print("shadow achieve!")
  154.         #本地的shadow
  155.         with open("/home/simon/mqtt/"+"shadow.ini","r") as sd: 
  156.             vsn = sd.read()
  157.         vsn_parsed = json.loads(vsn)
  158.         in_version = vsn_parsed['payload']['version']
  159.         #get到的shadow
  160.         shadow = msg.payload.decode('utf-8')
  161.         shadow_get = json.loads(shadow)
  162.         shadow_desired = shadow_get['payload']['state']['desired']
  163.         shadow_reported = shadow_get['payload']['state']['reported']
  164.         print(shadow_get)
  165.         shadow_version = shadow_get['payload']['version']#字符串
  166.         if(shadow_desired!={}):
  167.             if('state' in shadow_desired):
  168.                 state = shadow_desired['state']
  169.             else:state = shadow_reported['state']
  170.             if('faceList' in shadow_desired):
  171.                 s_d_facelist = shadow_desired['faceList']#数组
  172.                 s_d_f = ",".join('%s'%id for id in s_d_facelist)#字符串
  173.             else:
  174.                 s_r_facelist = shadow_reported['faceList']#数组
  175.                 s_d_f = ",".join('%s'%id for id in s_r_facelist)#字符串
  176.             data_delta = {
  177.                 "type":"delta",
  178.                 "state":{
  179.                     "reported":{
  180.                         "state":state,
  181.                         "faceList":[s_d_f]
  182.                     }
  183.                 },
  184.                 "version":shadow_version,
  185.                 "clientToken":str(clientToken)
  186.             }
  187.             data_delta_param = json.dumps(data_delta)
  188.             client.publish("$shadow/operation/414ce45bde0a4bc1949fec75a7d2f699", payload=data_delta_param,qos=2,retain=False)#delta
  189.         elif(shadow_version!=in_version):
  190.             client.publish("$shadow/operation/414ce45bde0a4bc1949fec75a7d2f699", payload=param1,qos=2,retain=False)#get
  191.         with open("/home/simon/mqtt/"+"shadow.ini","w") as sd: 
  192.             sd.write(msg.payload.decode('utf-8'))
  193.  
  194. def on_publish(client,userdata,mid):
  195.     print("On onPublish: qos = %d" % mid)
  196.  
  197. def on_subscribe(client, userdata, mid, granted_qos):
  198.     print("On Subscribed: qos = %d" % granted_qos)
  199.  
  200. def on_disconnect(client, userdata, rc):
  201.     if rc != 0:
  202.         print("Unexpected disconnection %s" % rc)
  203.  
  204. def client_loop():
  205.     while True:
  206.         with open('/home/simon/mqtt/'+'FaceAndTime.json','r',encoding="utf-8") as old_f:
  207.             #读取所有行，每行会是一个字符串
  208.             for j in old_f.readlines():
  209.                 if(j):
  210.                     print(j)
  211.                     face = json.loads(j,object_hook=dict2Data)
  212.                     data_face = {
  213.                                 "type":"history",
  214.                                 "payload":{
  215.                                     "history":[
  216.                                     {
  217.                                         "timestamp":face.timestamp,
  218.                                         "history":{"name":face.name}
  219.                                         }]
  220.                                     },
  221.                                 "clientToken":str(clientToken)
  222.                     }
  223.                     client.publish("414ce45bde0a4bc1949fec75a7d2f699/event", payload=str(data_face), qos=0,retain=False)#发送人脸数据
  224.                     with open('/home/simon/mqtt/'+'FaceAndTime.json','w+',encoding="utf-8") as new_f:
  225.                         seek_point = old_f.tell()#光标在被删除行的行首，记录该位置
  226.                         new_f.seek(seek_point,0)#设置光标位置
  227.                         old_f.readline()#读需要删除的行，光标移动到下一行行首
  228.                         next_line=old_f.readline()#被删除行的下一行读给next_line
  229.                         while next_line:#连续覆盖剩余行，后面所有行上移一行
  230.                             new_f.write(next_line)
  231.                             next_line=old_f.readline()
  232.                             new_f.truncate()
  233.  
  234. if __name__ == '__main__':
  235.     client = mqtt.Client(client_id)
  236.     client.on_connect = on_connect
  237.     client.connect(HOST, PORT, 60)
  238.     client.on_subscribe = on_subscribe
  239.     client.subscribe("414ce45bde0a4bc1949fec75a7d2f699/control",qos=1)        
  240.     client.subscribe("$shadow/operation/result/414ce45bde0a4bc1949fec75a7d2f699",qos=1)
  241.     client.on_publish = on_publish
  242.     client.publish("414ce45bde0a4bc1949fec75a7d2f699/event", payload=param_in,qos=2,retain=False)#获得图片
  243.     client.publish("$shadow/operation/414ce45bde0a4bc1949fec75a7d2f699", payload=param1,qos=2,retain=False)#get
  244.     client.on_message = on_message
  245.     client.loop_start()
  246.     client_loop()

逻辑二： 

  1. import paho.mqtt.client as mqtt
  2. import json
  3. import base64
  4. import uuid
  5. import time
  6. import sys
  7. import multiprocessing
  8. import re
  9. import os
  10. from face_recognition.face_recognition_cli import image_files_in_folder
  11. import glob
  12.  
  13. HOST = "api.yumik.top" 
  14. PORT = 1883
  15. client_id="414ce45bde0a4bc1949fec75a7d2f699"
  16. clientToken=uuid.uuid4()
  17.  
  18. class Data(object):
  19.     def __init__(self,name,timestamp):
  20.         self.timestamp = timestamp
  21.         self.name = name
  22. def dict2Data(d):
  23.     return Data(d['name'],d['timestamp'])
  24.  
  25. #publish shadow/operation
  26. data1 = {
  27.      "type":"get",
  28.      "clientToken":str(clientToken)
  29.      }
  30. param1 = json.dumps(data1)
  31.  
  32. def on_connect(client, userdata, flags, rc):
  33.     print("Connected with result code "+str(rc))        
  34.  
  35. def file_circle(dirs5,houzhui,i,picture):
  36.     global picture2
  37.     for dirs4 in dirs5:
  38.         d2 = os.path.basename(dirs4)#图片名加后缀
  39.         num2 = os.path.splitext(d2)[0]#图片名
  40.         basename2 = os.path.splitext(d2)[1] #后缀名
  41.         if(int(num2)==i):
  42.             with open("/home/simon/mqtt/picture_recieve/"+str(num2)+basename2,"rb") as p1:
  43.                 picture2 = picture + str(base64.b64encode(p1.read()),encoding="utf-8")#bytes
  44.             break
  45.     return(picture2)
  46.  
  47. picture = ""
  48. houzhui = ""
  49. u = []#存用户的id
  50. v = []#存每个用户的图片数量，递增
  51. a = []#获取userId
  52. def on_message(client, userdata, msg):
  53.     global picture
  54.     global i
  55.     global houzhui
  56.     recieve = msg.payload.decode("utf-8")
  57.     recieve_parsed = json.loads(recieve)
  58.     msg_type = recieve_parsed['type']
  59.     if(msg.topic=="414ce45bde0a4bc1949fec75a7d2f699/control" and msg_type=="history"):
  60.         print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  61.     if(msg.topic=="$shadow/operation/result/414ce45bde0a4bc1949fec75a7d2f699"):
  62.         if(msg_type=="get"):
  63.             print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  64.             print("shadow achieve!")
  65.             #get到的shadow
  66.             #shadow = msg.payload.decode('utf-8')
  67.             #shadow_get = json.loads(shadow)
  68.             shadow_desired = recieve_parsed['payload']['state']['desired']
  69.             shadow_reported = recieve_parsed['payload']['state']['reported']
  70.             shadow_version = recieve_parsed['payload']['version']#字符串
  71.             if(shadow_desired!={}):
  72.                 if('faceList' in shadow_desired):
  73.                     s_d_facelist = shadow_desired['faceList']#数组
  74.                     if('state' in shadow_desired):
  75.                         state = shadow_desired['state']
  76.                         data_delta = {
  77.                             "type":"delta",
  78.                             "state":{
  79.                                 "reported":{
  80.                                     "state":state,
  81.                                     "faceList":s_d_facelist
  82.                                 }
  83.                             },
  84.                             "version":shadow_version,
  85.                             "clientToken":str(clientToken)
  86.                         }
  87.                     else:
  88.                         data_delta = {
  89.                             "type":"delta",
  90.                             "state":{
  91.                                 "reported":{
  92.                                     "faceList":s_d_facelist
  93.                                 }
  94.                             },
  95.                             "version":shadow_version,
  96.                             "clientToken":str(clientToken)
  97.                         }
  98.                 else:
  99.                     state = shadow_desired['state']
  100.                     data_delta = {
  101.                         "type":"delta",
  102.                         "state":{
  103.                             "reported":{
  104.                                 "state":state
  105.                             }
  106.                         },
  107.                         "version":shadow_version,
  108.                         "clientToken":str(clientToken)
  109.                     }
  110.                 data_delta_param = json.dumps(data_delta)
  111.                 client.publish("$shadow/operation/414ce45bde0a4bc1949fec75a7d2f699", payload=data_delta_param,qos=2,retain=False)#delta
  112.             else:#desired内没有内容需要完成时
  113.                 with open("/home/simon/mqtt/"+"shadow.ini","r") as ud: 
  114.                     mes = ud.read()
  115.                     mes_parsed = json.loads(mes)
  116.                     meslist = mes_parsed['payload']['state']['reported']['faceList']
  117.                     for i in meslist:
  118.                         userid = i['userId']
  119.                         a.append(int(userid))
  120.                 #publish event
  121.                 data_in = {
  122.                      "type":"face",
  123.                      "payload":{"userList":a,},
  124.                      #"payload":{"userList":[2,3,4,5],},
  125.                      "clientToken":str(clientToken)
  126.                      }
  127.                 param_in = json.dumps(data_in)
  128.                 client.publish("414ce45bde0a4bc1949fec75a7d2f699/event", payload=param_in,qos=2,retain=False)#获得图片            
  129.         if(msg_type=="delta"):
  130.             print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  131.             print("shadow achieve!")
  132.             with open("/home/simon/mqtt/"+"shadow.ini","w") as sd: 
  133.                 sd.write(msg.payload.decode('utf-8'))
  134.             with open("/home/simon/mqtt/"+"shadow.ini","r") as ud: 
  135.                 mes = ud.read()
  136.                 mes_parsed = json.loads(mes)
  137.                 meslist = mes_parsed['payload']['state']['reported']['faceList']
  138.                 for i in meslist:
  139.                     userid = i['userId']
  140.                     a.append(int(userid))
  141.             #publish event
  142.             data_in = {
  143.                  "type":"face",
  144.                  "payload":{"userList":a,},
  145.                  #"payload":{"userList":[2,3,4,5],},
  146.                  "clientToken":str(clientToken)
  147.                  }
  148.             param_in = json.dumps(data_in)
  149.             client.publish("414ce45bde0a4bc1949fec75a7d2f699/event", payload=param_in,qos=2,retain=False)#获得图片
  150.     if(msg.topic=="414ce45bde0a4bc1949fec75a7d2f699/control" and (msg_type=="face" or msg_type=="imageUpdate")):
  151.         #print("topic:"+msg.topic+" message:"+str(msg.payload.decode('utf-8')))
  152.         print("control achieve!")
  153.         packetnum = recieve_parsed['payload']['packet']
  154.         packettotal = recieve_parsed['payload']['totalPacket']
  155.         picture = recieve_parsed['payload']['userImageBase64']
  156.         temp = base64.b64decode(picture)#bytes
  157.         useridnum = str(recieve_parsed['payload']['user'])
  158.         houzhui = "."+str(recieve_parsed['payload']['user'])
  159.         if(msg_type=="imageUpdate"):
  160.             rootdir = "/home/simon/mqtt/picture_recieve/"
  161.             file_delete = os.listdir(rootdir)
  162.             for file in file_delete:
  163.                 if houzhui in file:
  164.                     del_file = rootdir + file#当代码和要删除的文件不在同一个文件夹时，必须使用绝对路径
  165.                     os.remove(del_file)#删除文件
  166.         with open("/home/simon/mqtt/picture_recieve/"+str(packetnum)+"."+str(recieve_parsed['payload']['user']),"wb") as f:
  167.             f.write(temp)#int
  168.         #判断一张图片是否全部接收
  169.         key = len(u)-1
  170.         if(key<0):#接收第一张图片
  171.             u.append(useridnum)
  172.             v.append(1)
  173.         else:
  174.             while(key>=0):
  175.                 if(useridnum==u[key]):
  176.                     v[key]+=1                   
  177.                 else:
  178.                     u.append(useridnum)
  179.                     v.append(1)
  180.                 if(v[key]==packettotal):
  181.                     break
  182.                 key = key-1
  183.         #一张图片全部接收后进行组装
  184.         if(v[key]==packettotal):
  185.             i=1
  186.             bool_flag2 = False#大循环跳出标志
  187.             dirs2 = []#大循环的数组
  188.             dirs0 = []#小循环的数组
  189.             for dirs1 in glob.glob('/home/simon/mqtt/picture_recieve'+"/*"+houzhui):
  190.                 dirs2.append(dirs1)
  191.                 dirs0.append(dirs1)
  192.             for dirs3 in dirs2:
  193.                 d1 = os.path.basename(dirs3)#图片名加后缀
  194.                 num1 = os.path.splitext(d1)[0]#图片名
  195.                 basename1 = os.path.splitext(d1)[1] #后缀名
  196.                 if(int(num1)==1):#找到序号为1的照片部分
  197.                     with open("/home/simon/mqtt/picture_recieve/"+str(num1)+basename1,"rb") as p1:
  198.                         picture1 = str(base64.b64encode(p1.read()),encoding="utf-8")#bytes
  199.                     i=i+1
  200.                     bool_flag1 = False#跳出小循环标志
  201.                     while True:
  202.                         if(i!=packettotal):#不是最后一部分就一直组装照片
  203.                             picture1 = file_circle(dirs0,houzhui,i,picture1)
  204.                             i=i+1
  205.                         if(i==packettotal):#找到照片的最后一部分
  206.                             picture1 = file_circle(dirs0,houzhui,i,picture1)
  207.                             i=1
  208.                             bool_flag1 = True
  209.                             break
  210.                     if(bool_flag1):
  211.                         bool_flag2 = True
  212.                         temp = base64.b64decode(picture1)#bytes
  213.                         with open("/home/simon/mqtt/picture/"+str(recieve_parsed['payload']['user'])+str(recieve_parsed['payload']['userName'])+".jpg","wb") as o:
  214.                             o.write(temp)
  215.                 if(bool_flag2):
  216.                     break
  217.  
  218. def on_publish(client,userdata,mid):
  219.     print("On onPublish: qos = %d" % mid)
  220.  
  221. def on_subscribe(client, userdata, mid, granted_qos):
  222.     print("On Subscribed: qos = %d" % granted_qos)
  223.  
  224. def on_disconnect(client, userdata, rc):
  225.     if rc != 0:
  226.         print("Unexpected disconnection %s" % rc)
  227.  
  228. def client_loop():
  229.     while True:
  230.         with open('/home/simon/mqtt/'+'FaceAndTime.json','r',encoding="utf-8") as old_f:
  231.             #读取所有行，每行会是一个字符串
  232.             for j in old_f.readlines():
  233.                 if(j):
  234.                     print(j)
  235.                     face = json.loads(j,object_hook=dict2Data)
  236.                     data_face = {
  237.                                 "type":"history",
  238.                                 "payload":{
  239.                                     "history":[
  240.                                     {
  241.                                         "timestamp":face.timestamp,
  242.                                         "history":{"name":face.name}
  243.                                         }]
  244.                                     },
  245.                                 "clientToken":str(clientToken)
  246.                     }
  247.                     client.publish("414ce45bde0a4bc1949fec75a7d2f699/event", payload=str(data_face), qos=0,retain=False)#发送人脸数据
  248.                     with open('/home/simon/mqtt/'+'FaceAndTime.json','w+',encoding="utf-8") as new_f:
  249.                         seek_point = old_f.tell()#光标在被删除行的行首，记录该位置
  250.                         new_f.seek(seek_point,0)#设置光标位置
  251.                         old_f.readline()#读需要删除的行，光标移动到下一行行首
  252.                         next_line=old_f.readline()#被删除行的下一行读给next_line
  253.                         while next_line:#连续覆盖剩余行，后面所有行上移一行
  254.                             new_f.write(next_line)
  255.                             next_line=old_f.readline()
  256.                             new_f.truncate()
  257.  
  258. if __name__ == '__main__':
  259.     client = mqtt.Client(client_id)
  260.     client.username_pw_set("device", "15f56228e74f2c978b0c376df35d3df8166da22f858f016f6906430d0c4b6070")
  261.     client.on_connect = on_connect
  262.     client.connect(HOST, PORT, 60)
  263.     client.on_subscribe = on_subscribe
  264.     client.subscribe("414ce45bde0a4bc1949fec75a7d2f699/control",qos=1)        
  265.     client.subscribe("$shadow/operation/result/414ce45bde0a4bc1949fec75a7d2f699",qos=1)
  266.     client.on_publish = on_publish
  267.     client.publish("$shadow/operation/414ce45bde0a4bc1949fec75a7d2f699", payload=param1,qos=2,retain=False)#get
  268.     client.on_message = on_message
  269.     client.loop_start()
  270.     client_loop()

 此篇博客仅作为课设的笔记，如有错误欢迎指正~

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/188558
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号