# -*- 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
|
from linkListUtil import Node, LinkedList
|
from datetime import datetime
|
from shapelyUtil import ShapelyUtil
|
|
|
def gettime():
|
time1=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())
|
return time1
|
|
# 服务器地址
|
host = '7.65.0.207' #'7.65.0.207'
|
# 通信端口 默认端口1883
|
port = 1883
|
|
|
username = 'eadpoint'
|
password = '321123'
|
|
merged_list = {}
|
objs = LinkedList()
|
currobjs = LinkedList()
|
currentObjs = []
|
shapeUtil = ShapelyUtil()
|
|
arr_topic = ['RadarRecord']
|
send_topic = ['dfAddTrack','dfRemoveObj','dfAddObj']
|
|
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 send_file(client, topic, filename):
|
try:
|
with open(filename, 'r') as file:
|
for line in file:
|
radarData = json.loads(line)
|
#client.publish(topic, payload=json.dumps(radarData), qos=2, retain=False)
|
client.publish(topic, line)
|
print(f"Sent: {radarData}")
|
except Exception as e:
|
print(f"发生了异常: {e}")
|
|
# 接收到数据后事件
|
def on_message(client, userdata, msg):
|
global dddd
|
# 打印订阅消息主题
|
#print(msg.payload)
|
try:
|
jsondata=json.loads(msg.payload)
|
currentObjs.clear()
|
if jsondata['object'] is None:
|
print('has no object data')
|
else :
|
index_ = 0
|
for x in jsondata['object']:
|
x['time'] = jsondata['time']
|
x['minOfYear'] = jsondata['minOfYear']
|
x['second'] = jsondata['second']
|
x['direction'] = 0
|
x['fusDevID'] = jsondata['fusDevID']
|
x['fps'] = jsondata['fps']
|
x['cellid'] = jsondata['cellid']
|
x['msgCnt'] = jsondata['msgCnt']
|
x['nofobjects'] = jsondata['nofobjects']
|
x['dataSource'] = jsondata['dataSource']
|
x['refPos'] = jsondata['refPosList'][0]
|
x['isInside'] = 1 if shapeUtil.isInPolygon_(x['lng'],x['lat']) else 0
|
if x['isInside'] == 0:
|
client.publish(send_topic[1], x['object_id'])
|
node = objs.find_by_value(x['object_id'])
|
currentObjs.append(x['object_id'])
|
if node is None:
|
objs.insert_value_to_head(x['object_id'], [x])
|
client.publish(send_topic[1], json.dumps(x))
|
else :
|
obj0 = node.track[-1]
|
x['direction'] = shapeUtil.angle(obj0['lat'], obj0['lng'], x['lat'], x['lng'])
|
client.publish(send_topic[0], json.dumps(x))
|
node.track.append(x)
|
index_ = index_+1
|
p = objs.head
|
while p:
|
#print(p.track)
|
savejsondata(p.item, jsondata, p.track)
|
p = p.next
|
|
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(object_id, msgdata, track):
|
# SQL 插入语句
|
try:
|
obj = track[-1]
|
#sql2 = 'insert into dt_radar_message (object_id,lng,lat,height,minOfYear,second,direction,serverTime,deviceTime,fusDevID,fps,cellid,msgCnt,nofobjects,dataSource,msgdata,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,now()) on duplicate key update msgdata=values(msgdata), lon=values(lon), lat=values(lat), height=values(height), update_time=now()'
|
sql2 = 'insert into dt_radar_message (object_id,lng,lat,height,minOfYear,second,direction,deviceTime,fusDevID,fps,cellid,msgCnt,nofobjects,dataSource,msgdata,isInside,serverTime,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,now(),now()) on duplicate key update msgdata=values(msgdata), lng=values(lng), lat=values(lat), direction=values(direction), height=values(height), isInside=values(isInside), update_time=now()'
|
ret = db.insertone(sql2, (object_id,"{:.8f}".format(obj['lng']),"{:.8f}".format(obj['lat']),"{:.2f}".format(obj['height']),obj['minOfYear'],obj['second'],"{:.2f}".format(obj['direction']),obj['time'],obj['fusDevID'],obj['fps'],obj['cellid'],obj['msgCnt'],obj['nofobjects'],obj['dataSource'],json.dumps(track),obj['isInside']))
|
print(f"数据库保存成功 {ret}:！")
|
except Exception as e:
|
print(e)
|
pass
|
|
if __name__ == '__main__':
|
main()
|
|
|
|
|
