# -*- coding: utf-8 -*- #!/usr/bin/python # -*- coding: utf-8 -* import paho.mqtt.client as mqtt import json import time from queue import LifoQueue from mysqlhelper import MySqLHelper from linkListUtil import Node, LinkedList from shapelyUtil import ShapelyUtil import threading def gettime(): time1=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime()) return time1 # 服务器地址 host = '7.65.0.207' #'10.5.226.22 7.65.0.207' # 通信端口 默认端口1883 port = 1883 username = 'eadpoint' password = '321123' merged_list = {} objs = LinkedList() currentObjs = [] removeObjIds = [] removeObjs = [] #[[12146938.804545141,3106558.9023983314],[12147505.698887747,3106620.0104321464],[12147188.527927307,3108004.3420571982],[12146593.188072586,3107932.8914925163]] #经度:121.342248, 纬度:30.814109 经度:121.335855, 纬度:30.843758 经度:121.374132, 纬度:30.849655, 经度:121.38213, 纬度:30.819001 shapeUtil = ShapelyUtil([[12134224.8,3081410.9],[12133585.5,3084375.8],[12137413.2,3084965.5],[12138213,3081900.1]]) arr_topic = ['uav_zte_radar'] send_topic = ['uavAddObj','uavAddTrack','uavRemoveObj'] 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() removeObjIds.clear() print(f" cellid : {jsondata['cellid']}") if jsondata['object'] is None: print('has no object data') else : for x in jsondata['object']: x['time'] = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(x['ts'])) x['recvtime'] = int(time.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 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]) node = objs.head if x['isInside'] == 1: print(f"{send_topic[0]} : {x['object_id']}") client.publish(send_topic[0], json.dumps(x)) else : obj0 = node.track[0] x['direction'] = shapeUtil.angle(obj0['lat'], obj0['lng'], x['lat'], x['lng']) if x['isInside'] == 1: print(f"{send_topic[1]} : {x['object_id']}") client.publish(send_topic[1], json.dumps(x)) node.track.append(x) savejsondata(x['object_id'], node.track) except Exception as e: print(f"发生了异常: {e}") def monitor_list(linked_list, client): p = linked_list.head while True: if p is None: p = linked_list.head continue item = p.item track = p.track if abs(track[-1]['recvtime'] - int(time.time())) > 5 : print(f"{send_topic[2]} : {item}") client.publish(send_topic[2],item) track[-1]['isInside'] = -1 savejsondata(item, track) q = p.next linked_list.delete_by_value(p.item, p.track) p = q else : p = p.next def main(): cleanHistorydata() 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) list_thread = threading.Thread(target=monitor_list, args=(objs, client)) list_thread.start() # 守护连接状态 client.loop_forever() #MySQL保存 def savejsondata(object_id, 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_uav_zte_msg (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['speed'],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 #MySQL保存 def cleanHistorydata(): # SQL 插入语句 try: sql2 = 'update dt_radar_uav_zte_msg set isInside = -1 where isInside =1' ret = db.update(sql2) print(f"数据库保存成功 {ret}:！") except Exception as e: print(e) pass if __name__ == '__main__': #print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(1712900732))) main()