# -*- coding: utf-8 -*- # File Name:mqtt_chat_client.py # Python Version:3.5.1 import os import django import sys BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # 定位到你的django根目录 sys.path.append(os.path.abspath(os.path.join(BASE_DIR, os.pardir))) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "yfwlw_pro.settings") # project_name 项目名称 django.setup() print("<-----python_mqtt_client_bzy is run----->") import paho.mqtt.client as mqtt import json from apps.AppInfoManage.models import Equip, Equip_type, MyUser, Alarm_record, BZYdata, BZYstatus import re import datetime import time def mkdir(path): # 去除首位空格 path=path.strip() # 去除尾部 \ 符号 path=path.rstrip("\\") # 判断路径是否存在 # 存在 True # 不存在 False isExists=os.path.exists(path) # 判断结果 if not isExists: # 如果不存在则创建目录 # 创建目录操作函数 os.makedirs(path) print(path+' 创建成功') return True else: # 如果目录存在则不创建，并提示目录已存在 print(path+' 目录已存在') return False def variance_of_laplacian(image): # 拉普拉斯的方差 return cv2.Laplacian(image, cv2.CV_64F).var() class CJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(obj, date): return obj.strftime('%Y-%m-%d') else: return json.JSONEncoder.default(self, obj) # 连接后的操作： 0为成功 def on_connect(client, userdata, flags, rc): # print("Connected with result code "+str(rc)) client.subscribe("/yfkj/bzy/c2s/#") client.subscribe("/yfkj/bzy/offline/#") # *****成功发布****** def on_publish(msg, rc): if rc == 0: print("publish success,msg = "+msg) # 从服务器接受到消息后回调此函数 ： def on_message(client, userdata, msg): print('\r') print('=================================================') print('\r') print("<-----topic:\n" + msg.topic + ';\n') print("Message:\n" + str(msg.payload) + "----->\n") # 从主题中获取imei # imei = msg.topic[14:len(msg.topic)] reg = re.compile(r"(?<=/)\d+") imei = reg.search(msg.topic).group(0) # imei = re.sub("\D", "", msg.topic) # nowtime = datetime.datetime.now().strftime('%Y%m%d') # origin = sys.stdout # f = open('./logs/'+nowtime+'bzymqtt.txt','a+') # sys.stdout = f print("<-----imei:", imei, "----->") # 判断主题： if "c2s" in msg.topic: # 将json字符串解析： payload = json.loads(msg.payload.decode()) if payload.get("cmd") == "status": print("<-----uploading data!----->") extdata = payload.get("ext") # print("extdata:", extdata) bzy_exist = Equip.objects.filter(equip_id=imei) # 设备存在，进一步判断状态表是否存在： if bzy_exist.exists(): print("<-----this equip is existed!----->") try: e_id = Equip.objects.get(equip_id=imei) except: print("<-----this equip didn't exist!----->") try: # 设备数据表直接储存数据 BZYdata.objects.create(equip_id=e_id, bzy_data=extdata) print("<-----data update success!----->") except: print("<-----data update failed!----->") # 设备状态表存在、刷新状态表： if BZYstatus.objects.filter(equip_id=imei).exists(): print("<-----this equip's status is existed!----->") try: sta = BZYstatus.objects.get(equip_id=imei) sta.bzy_status=extdata sta.is_online = '1' sta.save() print("<-----status update success!----->") except: print("<-----status update failed!----->") else: # 设备状态表不存在、创建状态表： print("<-----this equip's status is not existed!----->") try: BZYstatus.objects.create(equip_id=e_id, bzy_status=extdata, is_online="1") print("<-----this equip's status table re-create successed!----->") except: print("<-----this equip's status table re-create failed!----->") else: # 设备不存在，在设备列表中创建： equip_t = Equip_type.objects.get(type_id=7) try: e_id = Equip.objects.create(equip_id=imei, equip_type=equip_t) print("<-----this imei add successed!----->") try: # 设备数据表直接储存数据 BZYdata.objects.create(equip_id=e_id, bzy_data=extdata) print("<-----data update success!----->") except: print("<-----data update failed!----->") try: BZYstatus.objects.create(equip_id=e_id, bzy_status=extdata, is_online="1") print("<-----this imei register successed!----->") except: print("<-----this imei register failed!----->") except: print("<-----this imei add failed!----->") elif payload.get("cmd") == "netconf": print("<-----uploading netconf!----->") mqttsta = payload.get("mqtt") ftpsta = payload.get("ftp") netsta = {"mqtt": mqttsta, "ftp": ftpsta} # print("extdata:", extdata) # 更新状态表中的参数配置信息； try: sta = BZYstatus.objects.get(equip_id=imei) sta.netconf = netsta sta.save() print("<-----status.netconf update success!----->") except: print("<-----bzystatus table is not exist,status.netconf upload failed!----->") # 离线消息： elif "offline" in msg.topic: try: # 将json字符串解析： payload = json.loads(msg.payload.decode()) if payload.get("cmd") == "offline": print("<-----离线消息!----->") print("%s is offline!" % imei) bzy_exist = Equip.objects.filter(equip_id=imei) # 设备存在，进一步判断状态表是否存在： if bzy_exist.exists(): try: e_id = Equip.objects.get(equip_id=imei) # now_time = json.dumps(datetime.datetime.now(), cls=CJSONEncoder) # 更新状态表中未离线状态： BZYstatus.objects.filter(equip_id=imei).update(is_online = '0',off_time = datetime.datetime.now()) # 创建预警记录： Alarm_record.objects.create(equip_id=e_id, alarm_desc="{'status':0,'type':'offline'}") print("update offline ok!") except: print("update offline failed!") else: print("this imei is not exist!") except: pass # sys.stdout = origin # f.close() if __name__ == '__main__': client = mqtt.Client( client_id="PY_MQTT_CLIENTC_BZY", clean_session=True, userdata=None, # protocol=MQTTv311,# 数据库版本 ) # 必须设置，否则会返回「Connected with result code 4」 client.username_pw_set("admin", "password") client.on_connect = on_connect client.on_message = on_message HOST = "127.0.0.1" client.connect(HOST, 1883, 60) client.loop_forever()