# -*- coding: utf-8 -*-
|
|
#!/usr/bin/python
|
# -*- coding: utf-8 -*
|
|
import paho.mqtt.client as mqtt
|
import json
|
import pymysql
|
import time
|
from queue import LifoQueue
|
from mysqlhelper import MySqLHelper
|
|
|
def gettime():
|
time1=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())
|
return time1
|
|
# 服务器地址
|
host = '7.65.0.207'
|
# 通信端口 默认端口1883
|
port = 1883
|
|
|
username = 'eadpoint'
|
password = '321123'
|
|
jsonLte,jsonDevice,jsonGnss,jsonBluetooth,jsonPhone,jsonNr = {}, {}, {}, {}, {}, {}
|
dictLte, dictDevice, dictGnss, dictPhone, dictNr = {}, {}, {}, {}, {}
|
merged_list = {}
|
#LIFO即Last in First Out,后进先出。与栈的类似，使用也很简单,maxsize用法同上
|
lqLte = LifoQueue(maxsize=0)
|
lqDevice = LifoQueue(maxsize=0)
|
lqGnss = LifoQueue(maxsize=0)
|
lqBluetooth = LifoQueue(maxsize=0)
|
lqPhone = LifoQueue(maxsize=0)
|
lqNr = LifoQueue(maxsize=0)
|
lqWifi = LifoQueue(maxsize=0)
|
|
# 订阅主题名
|
arr_topic = ['RadarRecord']
|
|
db = MySqLHelper()
|
|
# 连接后事件
|
def on_connect(client, userdata, flags, respons_code):
|
if respons_code == 0:
|
# 连接成功
|
print('Connection Succeed!')
|
else:
|
# 连接失败并显示错误代码
|
print('Connect Error status {0}'.format(respons_code))
|
# 订阅信息
|
#client.subscribe(topic)
|
for x in arr_topic:
|
client.subscribe(x)
|
|
# 接收到数据后事件
|
def on_message(client, userdata, msg):
|
global dddd
|
# 打印订阅消息主题
|
print(msg.payload)
|
try:
|
jsondata=json.loads(msg.payload)
|
savejsondata(jsondata)
|
except Exception as e:
|
print(f"发生了异常: {e}")
|
|
|
def main():
|
client = mqtt.Client()
|
# 注册事件
|
client.on_connect = on_connect
|
client.on_message = on_message
|
# 设置账号密码（如果需要的话）
|
client.username_pw_set(username, password=password)
|
# 连接到服务器
|
client.connect(host, port=port, keepalive=60)
|
# 守护连接状态
|
client.loop_forever()
|
|
#MySQL保存
|
def savejsondata(msgdata):
|
# SQL 插入语句
|
try:
|
sql2 = 'insert into dt_radar_message (msgCnt,minOfYear,second,deviceTime,fusDevID,cellid,nofobjects,fps,dataSource,msgdata) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'
|
ret = db.insertone(sql2, (msgdata['msgCnt'],msgdata['minOfYear'],msgdata['second'],msgdata['time'],msgdata['fusDevID'],msgdata['cellid'],msgdata['nofobjects'],msgdata['fps'],msgdata['dataSource'],json.dumps(msgdata)))
|
print(f"数据库保存成功 {ret}:！")
|
except :
|
pass
|
|
if __name__ == '__main__':
|
main()
|
|
|
|
|
