# -*- 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()
|
