# -*- coding: utf-8 -*- #!/usr/bin/python import paho.mqtt.client as mqtt import json import time import threading import traceback 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 # 订阅服务器(中兴无人机的消息服务器)地址 #mqtt server URL：mqtt://7.65.210.186:1883 username:sanss password:123456 topic:rid/uav host = '10.7.145.30' port = 1884 username = 'sanss' password = '123456' arr_topic = ['rid/uav'] # 发送服务器(服务器62)配置 # 172.16.2.76 1883 endpoint 321123 send_host = '172.16.2.76' send_port = 1883 send_username = 'endpoint' send_password = '321123' send_topic = ['uavAddObj','uavAddTrack','uavRemoveObj'] #全局数据定义： object_dict = {} objs = LinkedList() data_queue = Queue() db_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]]) shapeUtil_dy = ShapelyUtil([[121.342248,30.814109],[121.335855,30.843758],[121.374132,30.849655],[121.38213,30.819001]]) # MQTT连接成功回调 def on_connect(client, userdata, flags, respons_code): print(f"连接状态码: {respons_code}") if respons_code == 0: print('Connection Succeed!') for topic in arr_topic: client.subscribe(topic) else: print(f"Connect Error status {respons_code}") # MQTT断开连接回调 def on_disconnect(client, userdata, rc): print(f"断开连接，状态码: {rc}") if rc != 0: print("意外断开，尝试重连...") # MQTT收到消息回调，仅入队，逻辑后移 def on_message(client, userdata, msg): try: jsondata = json.loads(msg.payload) data_queue.put(jsondata) except Exception as e: print("on_message 解析错误:", e) traceback.print_exc() # 独立线程：处理数据逻辑 def data_processor_(data_queue: Queue, client, db_data_queue): while True: jsondata = data_queue.get() if 'remoteId' not in jsondata or not jsondata['remoteId']: return None # 无 remoteId 则忽略 try: remote = jsondata['remoteId'] position = remote['position'] timestamp = remote['timestamp'] / 1000 # 微秒转毫秒 recv_time = int(time.time()) # \"position\":{\"x\":121.475845,\"y\":31.06572,\"z\":16.0},\"speedHorizontal\":5.0,\"speedVertical\":0.0,\"timestamp\":1745911833687349,\"uasId\":\"1581F5FHD238R00D8LJ6\" obj = { 'object_id': remote['uasId'], 'lng': float(position['x']), 'lat': float(position['y']), 'height': float(position['z']), 'speed': float(remote['speedHorizontal']), # 水平速度 'velocityH': float(remote['speedHorizontal']), 'velocityV': float(remote['speedVertical']), 'headH': float(remote['direction']), 'direction': float(remote['direction']), 'recvtime': recv_time, 'second': int(timestamp), 'isInside': 0, 'region': 0, 'time': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(recv_time)), 'is_track': 0 } # obj['lng'] = 121.475845 # obj['lat'] = 31.06572 # obj['height'] = 16.0 # obj['velocityH'] = 5.0 # 区域判断 # if shapeUtil.isInPolygon(obj['lng'], obj['lat']): # obj['isInside'] = 1 # obj['region'] = 1 # elif shapeUtil_df.isInPolygon(obj['lng'], obj['lat']): # obj['isInside'] = 1 # obj['region'] = 2 # elif shapeUtil_dy.isInPolygon(obj['lng'], obj['lat']): # obj['isInside'] = 1 # obj['region'] = 3 node = objs.find_by_value(obj['object_id']) if node is None: objs.insert_value_to_head(obj['object_id'], [obj]) node = objs.find_by_value(obj['object_id']) #db_data_queue.put(obj) else : obj0 = node.track[0] obj['is_track'] = 1 obj['direction'] = shapeUtil.angle(obj0['lat'], obj0['lng'], obj['lat'], obj['lng']) node.track.append(obj) #if obj['isInside'] == 1: client.publish(send_topic[1] if len(node.track)>1 else send_topic[0], json.dumps(obj)) except Exception as e: print("解析无人机数据异常:", e) traceback.print_exc() return None def monitor_list_(linked_list: LinkedList, client: mqtt.Client, db_data_queue): 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],json.dumps(track[-1])) track[-1]['isInside'] = -1 linked_list.delete_by_value(item, track) #db_data_queue.put(track) countBoat = countBoat+1 if track[-1]['isInside'] == 1 else countBoat except Exception as e: print(f"监控异常: {e}") def queueListen_(db_data_queue): while True: data = db_data_queue.get() if data is None: continue #db_operate.bt_cy_save(data) def messageListen(data_queue): f1 = open("uav.txt", 'w', encoding='UTF-8') while True: data = data_queue.get() if data is None: continue f1.writelines(data) f1.close() def main(): # 初始化 MQTT client = mqtt.Client() client.username_pw_set(username, password) client.on_connect = on_connect client.on_message = on_message client.on_disconnect = on_disconnect client.connect(host, port, keepalive=60) client.loop_start() # 初始化 发送 MQTT 客户端 send_client = mqtt.Client() send_client.username_pw_set(send_username, send_password) send_client.connect(send_host, send_port, keepalive=60) send_client.loop_start() # 启动处理线程 threading.Thread(target=data_processor_, args=(data_queue, send_client, db_data_queue), daemon=True).start() threading.Thread(target=monitor_list_, args=(objs, send_client, db_data_queue), daemon=True).start() #threading.Thread(target=queueListen_, args=(db_data_queue,), daemon=True).start() # 主线程挂起 while True: time.sleep(1) if __name__ == '__main__': #print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(1712900732))) main()