# -*- 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
|
|
# 服务器地址
|
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()
|
# 创建数据队列
|
msg_data_queue = Queue()
|
obj_data_queue = Queue()
|
track_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]])
|
|
arr_topic = ['RadarRecord']
|
send_topic = ['dfAddObj','dfAddTrack','dfRemoveObj']
|
uav_send_topic = ['uavAddObj','uavAddTrack','uavRemoveObj']
|
uav_height = 100
|
object_dict = {}
|
|
# 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)
|
msg_data_queue.put(jsondata)
|
track_data_queue.put(jsondata)
|
except Exception as e:
|
print("on_message 解析错误:", e)
|
traceback.print_exc()
|
|
# 独立线程：处理数据逻辑
|
def msg_data_processor_(msg_data_queue: Queue, client, obj_data_queue):
|
while True:
|
jsondata = msg_data_queue.get()
|
if not jsondata or 'participants' not in jsondata or jsondata['participants'] is None:
|
continue
|
try:
|
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'] = round(float(x['speed']) / 3.6, 2) # 速率转换: km/h-->m/s
|
x['velocityH'] = float(0)
|
x['headH'] = float(0)
|
x['direction'] = x['headH']
|
x['type'] = 'boat' if x['height'] < uav_height else 'uav'
|
x['isInside'] = 1 if shapeUtil.isInPolygon(x['lng'],x['lat']) else 0
|
x['region'] = 0
|
if shapeUtil.isInPolygon(x['lng'],x['lat']):
|
x['isInside'] = 1
|
x['region'] = 1
|
elif shapeUtil_df.isInPolygon(x['lng'],x['lat']):
|
x['isInside'] = 1
|
x['region'] = 2
|
elif shapeUtil_dy.isInPolygon(x['lng'],x['lat']):
|
x['isInside'] = 1
|
x['region'] = 3
|
else :
|
x['isInside'] = 0
|
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'])
|
obj_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'])
|
node.track.append(x)
|
if x['type']=='boat':
|
client.publish(uav_send_topic[1] if len(node.track)>1 else uav_send_topic[0], json.dumps(x))
|
elif x['isInside'] == 1:
|
client.publish(send_topic[1] if len(node.track)>1 else send_topic[0], json.dumps(x))
|
|
except Exception as e:
|
print("数据处理异常:", e)
|
traceback.print_exc()
|
|
# 独立线程：处理数据逻辑
|
def track_processor(track_data_queue: Queue):
|
region_buffers = {1: [], 2: [], 3: []}
|
buffer_size = 50
|
flush_interval = 1.0
|
last_flush_time = time.time()
|
|
while True:
|
try:
|
now = time.time()
|
while not track_data_queue.empty():
|
jsondata = track_data_queue.get()
|
if not jsondata or 'participants' not in jsondata:
|
continue
|
time_ = jsondata['time']
|
for x in jsondata['participants']:
|
x['date_time'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time_ / 1000)))
|
x['timestamp'] = int(time.time())
|
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'])
|
x['direction'] = 0
|
x['region'] = 0
|
# 区域判断
|
if x['type'] == 'boat':
|
if shapeUtil.isInPolygon(x['lng'], x['lat']):
|
x['region'] = 1
|
elif shapeUtil_df.isInPolygon(x['lng'], x['lat']):
|
x['region'] = 2
|
elif shapeUtil_dy.isInPolygon(x['lng'], x['lat']):
|
x['region'] = 3
|
region_buffers[x['region']].append(x)
|
|
now = time.time()
|
if now - last_flush_time > flush_interval:
|
for region, buffer in region_buffers.items():
|
if buffer:
|
db_operate.bt_tracks_batch_save(region, buffer)
|
region_buffers[region] = []
|
last_flush_time = now
|
time.sleep(0.1)
|
except Exception as e:
|
print("数据处理异常:", e)
|
traceback.print_exc()
|
|
|
def monitor_list_(linked_list: LinkedList, client: mqtt.Client, obj_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] if track[-1]['type']=='boat' else uav_send_topic[2], json.dumps(track[-1]))
|
track[-1]['isInside'] = -1
|
linked_list.delete_by_value(item, track)
|
obj_data_queue.put(track)
|
countBoat = countBoat+1 if track[-1]['isInside'] == 1 else countBoat
|
except Exception as e:
|
print(f"监控异常: {e}")
|
|
def obj_queueListen_(obj_data_queue):
|
while True:
|
data = obj_data_queue.get()
|
if data is None:
|
continue
|
db_operate.bt_cy_save(data)
|
|
def messageListen(msg_data_queue):
|
f1 = open("boatdata.txt", 'w', encoding='UTF-8')
|
while True:
|
data = msg_data_queue.get()
|
if data is None:
|
continue
|
f1.writelines(data)
|
f1.close()
|
|
|
def main():
|
db_operate.bt_old_cy_clean()
|
# 初始化 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()
|
|
# 启动处理线程
|
threading.Thread(target=msg_data_processor_, args=(msg_data_queue, client, obj_data_queue), daemon=True).start()
|
threading.Thread(target=track_processor, args=(track_data_queue, ), daemon=True).start()
|
threading.Thread(target=monitor_list_, args=(objs, client, obj_data_queue), daemon=True).start()
|
threading.Thread(target=obj_queueListen_, args=(obj_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()
|
