#!/usr/bin/python
|
# -*- coding: utf-8 -*-
|
|
import json
|
import time
|
import threading
|
from queue import Queue
|
|
import paho.mqtt.client as mqtt
|
from flask import Flask, request, jsonify, render_template
|
|
import timeutil
|
import db_operate
|
from linkListUtil import Node, LinkedList
|
|
# === 配置项 ===
|
host = '192.168.0.79'
|
port = 1883
|
username = 'device'
|
password = '321!@3'
|
|
arr_topic = [
|
'nr_message', 'lte_message', 'device_status_message',
|
'80211beacon_message', 'bluetooth_message', 'cdma_message', 'gsm_message'
|
]
|
|
# === 状态变量 ===
|
objs = LinkedList()
|
data_queue = Queue()
|
|
action_status = False
|
check_line = "0"
|
|
# === Flask Web 控制服务 ===
|
app = Flask(__name__)
|
|
@app.route('/')
|
def index():
|
return render_template('control.html')
|
|
@app.route('/start', methods=['GET'])
|
def start_collection():
|
global action_status, check_line
|
check_line = request.args.get("check_line", "0")
|
action_status = True
|
print(f"[WEB] ✅ 接收已开启, check_line={check_line}")
|
return jsonify({"status": "started", "check_line": check_line})
|
|
@app.route('/stop', methods=['GET'])
|
def stop_collection():
|
global action_status
|
action_status = False
|
print("[WEB] 🛑 接收已停止")
|
return jsonify({"status": "stopped"})
|
|
@app.route('/status', methods=['GET'])
|
def get_status():
|
return jsonify({
|
"status": "active" if action_status else "inactive",
|
"check_line": check_line
|
})
|
|
# === MQTT 回调逻辑 ===
|
def on_connect(client, userdata, flags, respons_code):
|
if respons_code == 0:
|
print('✅ MQTT连接成功')
|
for topic in arr_topic:
|
client.subscribe(topic)
|
print('✅ 已订阅所有数据主题')
|
else:
|
print(f'❌ MQTT连接失败，错误码：{respons_code}')
|
|
def on_message(client, userdata, msg):
|
global action_status, check_line
|
|
topic = msg.topic
|
payload = msg.payload
|
|
if not action_status:
|
return
|
|
if topic not in arr_topic:
|
return
|
|
try:
|
jsondata = json.loads(payload)
|
if isinstance(jsondata, dict) and 'data' in jsondata:
|
jsondata['check_line'] = check_line
|
data_queue.put(jsondata)
|
except Exception as e:
|
print(f"[ERROR] 处理主题 {topic} 的消息时失败: {e}")
|
|
# === 消息处理线程 ===
|
def messageListen(data_queue):
|
while True:
|
data = data_queue.get()
|
if not data:
|
continue
|
try:
|
data_ = data['data']
|
data_['messageType'] = data['messageType']
|
data_['ts'] = timeutil.timestr_to_us(data_['deviceTime'])
|
data_['check_line'] = data.get('check_line', "0")
|
process_deviceMessage(data['messageType'], data_)
|
except Exception as e:
|
print(f"[ERROR] 数据处理失败: {e}")
|
|
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:
|
action()
|
else:
|
print(f"[WARN] 未处理的消息类型: {option}")
|
|
# === 主函数 ===
|
def start_mqtt():
|
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)
|
|
threading.Thread(target=messageListen, args=(data_queue,), daemon=True).start()
|
client.loop_forever()
|
|
def main():
|
threading.Thread(target=start_mqtt, daemon=True).start()
|
app.run(host="0.0.0.0", port=5000)
|
|
if __name__ == '__main__':
|
main()
|
