# -*- 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]])
|
|
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'])
|
x['velocityH'] = float(0)
|
x['headH'] = float(0)
|
x['direction'] = x['headH']
|
x['isInside'] = 1 if shapeUtil.isInPolygon(x['lng'],x['lat']) else 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['direction'] = shapeUtil.angle(obj0['lat'], obj0['lng'], x['lat'], x['lng'])
|
if x['isInside'] == 1:
|
client.publish(send_topic[1], json.dumps(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)
|
|
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()
|
