# -*- coding: utf-8 -*- #!/usr/bin/python # -*- coding: utf-8 -* import paho.mqtt.client as mqtt import json import time import threading import db_operate from linkListUtil import Node, LinkedList from shapelyUtil import ShapelyUtil from queue import Queue def gettime(): time1=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime()) return time1 # 服务器地址 host = '172.16.2.76' #'7.65.0.207' 10.5.226.22 172.16.2.76 # 通信端口 默认端口1883 port = 1883 username = 'endpoint' password = '321123' objs = LinkedList() # 创建数据队列 data_queue = Queue() # [[121.506064, 31.212161],[121.502294, 31.212988],[121.503238, 31.215669],[121.503828, 31.218579],[121.5024, 31.221577],[121.506519, 31.234141],[121.507887, 31.218478],[121.507595, 31.215546],[121.505041, 31.210298,]] #shapeUtil = ShapelyUtil([[121.497272, 31.208397],[121.506529, 31.205425],[121.510571, 31.224502],[121.495167, 31.221426]]) shapeUtil = ShapelyUtil([[121.501106,31.223485],[121.505473,31.223778],[121.507622,31.219178],[121.507838,31.217541],[121.507721,31.216351],[121.506934,31.214187],[121.505682,31.21227],[121.504602,31.210381],[121.504053,31.209363],[121.50158,31.211439],[121.50277,31.213707],[121.503714,31.21566],[121.5038,31.216795],[121.503908,31.217588],[121.503699,31.218523],[121.502979,31.220407],[121.502423,31.221324]]) shapeUtil_df = ShapelyUtil([[121.469388,31.065589],[121.47505698,31.06620010],[121.4718852,31.08004342],[121.4659318,31.07932891]]) arr_topic = ['RadarRecord'] send_topic = ['dfAddObj','dfAddTrack','dfRemoveObj'] object_dict = {} # 连接后事件 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,"ddd") try: jsondata=json.loads(msg.payload) print(jsondata) if jsondata['participants'] is None: print('has no object data') else : countBoat = 0 countBoat_ = 0 time_ = jsondata['time'] for x in jsondata['participants']: x['time'] = time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(int(time_/1000))) x['recvtime'] = int(time.time()) x['minOfYear'] = 0 x['second'] = int(time_) x['direction'] = 0 x['fusDevID'] = 0 x['fps'] = 0 x['cellid'] = 0 x['msgCnt'] = 0 x['nofobjects'] = 0 x['dataSource'] = 0 x['refPos'] = '' x['lng'] = float(x['location']['longitude']) x['lat'] = float(x['location']['latitude']) x['height'] = float(x['location']['height']) x['object_id'] = x['id'] x['speed'] = float(x['speed']) / 3.6 # 速率转换: km/h-->m/s x['velocityH'] = float(0) x['headH'] = float(0) x['direction'] = x['headH'] x['isInside'] = 1 if shapeUtil.isInPolygon(x['lng'],x['lat']) else 0 if shapeUtil.isInPolygon(x['lng'],x['lat']): x['isInside'] = 1 else: if shapeUtil_df.isInPolygon(x['lng'],x['lat']): x['isInside'] = 1 else : x['isInside'] = 0 countBoat = countBoat+1 if x['isInside'] == 1 else countBoat countBoat_ = countBoat_ +1 node = objs.find_by_value(x['object_id']) if node is None: objs.insert_value_to_head(x['object_id'], [x]) node = objs.find_by_value(x['object_id']) if x['isInside'] == 1: client.publish(send_topic[0], json.dumps(x)) data_queue.put(node.track) else : obj0 = node.track[0] x['is_track'] = 1 x['direction'] = shapeUtil.angle(obj0['lat'], obj0['lng'], x['lat'], x['lng']) if x['isInside'] == 1: client.publish(send_topic[1], json.dumps(x)) data_queue.put([x]) #else : #client.publish(send_topic[2],x['object_id']) #objs.delete_by_value(x['object_id'], node.track) #data_queue.put(node.track) node.track.append(x) print(f"上报当前区域内船舶数量：{countBoat},总数：{countBoat_} ") except Exception as e: print(f"发生了异常: {e}") def monitor_list(linked_list: LinkedList, client: mqtt.Client): p = linked_list.head countBoat = 0 while True: if p is None: #print(f"当前区域内船舶数量：{countBoat}: ") p = linked_list.head countBoat = 0 continue try: item = p.item track = p.track p = p.next if abs(int(track[-1]['recvtime']) - int(time.time())) > 5 : client.publish(send_topic[2],item) track[-1]['isInside'] = -1 linked_list.delete_by_value(item, track) data_queue.put(track) #else: #if track[-1]['isInside'] == 1 and len(track)%20 == 1: #print(f"更新状态:{item}") #data_queue.put(track) countBoat = countBoat+1 if track[-1]['isInside'] == 1 else countBoat except Exception as e: print(f"监控异常: {e}") def queueListen(data_queue): while True: data = data_queue.get() if data is None: continue #print(f"无人机： {data[-1]['object_id']} 状态： {data[-1]['isInside']}") db_operate.bt_cy_save(data) db_operate.bt_cy_tracks_save(data) def messageListen(data_queue): f1 = open("boatdata.txt", 'w', encoding='UTF-8') while True: data = data_queue.get() if data is None: continue f1.writelines(data) f1.close() def main(): db_operate.bt_old_cy_clean() #队列监听 db_thread = threading.Thread(target=queueListen, args=(data_queue,)) db_thread.start() 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() if __name__ == '__main__': #print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(1712900732))) main()