# -*- coding: utf-8 -*-
|
|
#!/usr/bin/python
|
# -*- coding: utf-8 -*
|
|
import paho.mqtt.client as mqtt
|
import json
|
import time
|
import timeutil
|
import threading
|
import db_operate
|
from linkListUtil import Node, LinkedList
|
from queue import Queue
|
|
|
def gettime():
|
time1=time.strftime("%Y-%m-%d %H:%M:%S",time.localtime())
|
return time1
|
|
# 服务器地址
|
host = '192.168.0.79' #'7.65.0.207' 10.5.226.22
|
# 通信端口 默认端口1883
|
port = 1883
|
username = 'device'
|
password = '321!@3'
|
|
objs = LinkedList()
|
# 创建数据队列
|
data_queue = Queue()
|
action_status = True
|
'''
|
TODO:
|
1.增加一个接受开启和停止的功能,接受topic为action_command的消息,格式如：{"action":True, "check_line":"line1_2_3"}
|
a.当消息中包含action字段且值为start时,将变量action_status设置为True
|
b.当消息中包含action字段且值为stop时,将变量action_status设置为False
|
2.根据当前action_status的值
|
a.当为True时,开启接收并解析topic在数组中arr_topic中的消息,并将check_line的值作为属性添加到消息json格式中。
|
b.当为True时,停止接收并解析topic在数组中arr_topic中的消息。
|
'''
|
|
arr_topic = ['nr_message', 'lte_message', 'device_status_message', '80211beacon_message', 'bluetooth_message', 'cdma_message', 'gsm_message']
|
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 send_file(client, topic, filename):
|
try:
|
with open(filename, 'r') as file:
|
for line in file:
|
radarData = json.loads(line)
|
#client.publish(topic, payload=json.dumps(radarData), qos=2, retain=False)
|
client.publish(topic, line)
|
print(f"Sent: {radarData}")
|
except Exception as e:
|
print(f"发生了异常: {e}")
|
|
# 接收到数据后事件
|
def on_message(client, userdata, msg):
|
global dddd
|
# 打印订阅消息主题
|
#print(msg.payload)
|
try:
|
jsondata=json.loads(msg.payload)
|
print(jsondata)
|
data_queue.put(jsondata)
|
#print(jsondata)
|
except Exception as e:
|
print(f"发生了异常: {e}")
|
|
|
def messageListen(data_queue):
|
while True:
|
data = data_queue.get()
|
if data is None:
|
continue
|
data_ = data['data']
|
data_['messageType'] = data['messageType']
|
data_['ts'] = timeutil.timestr_to_us(data_['deviceTime'])
|
process_deviceMessage(data['messageType'] ,data_)
|
|
def process_deviceMessage(option, data):
|
actions = {
|
"LteRecord": lambda: db_operate.lte_message_save(data),
|
"NrRecord": lambda: db_operate.nr_message_save(data),
|
"DeviceStatus": lambda: db_operate.device_message_save(data),
|
"PhoneState": lambda: db_operate.device_message_save(data),
|
"WifiBeaconRecord": lambda: db_operate.device_wifi_save(data),
|
"BluetoothRecord": lambda: db_operate.device_bluetooth_save(data),
|
"GnssRecord": lambda: db_operate.device_gnss_save(data),
|
}
|
action = actions.get(option)
|
if action is None:
|
return
|
action()
|
|
|
def main():
|
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)
|
file_thread = threading.Thread(target=messageListen, args=(data_queue,))
|
file_thread.start()
|
# 守护连接状态
|
client.loop_forever()
|
|
|
if __name__ == '__main__':
|
#print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(1712900732)))
|
main()
|
# data_ = [{"message_title":"lte_message", "lng":121.2223122,"lat":31.231234,"msg": "lte msg"},{"message_title":"nr_message", "lng":121.2223122,"lat":31.231234,"msg":"nr msg"}]
|
# for data in data_:
|
# process_deviceMessage(data['message_title'], data)
|
