93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
import paho.mqtt.client as mqtt
|
|
import sys
|
|
import json
|
|
import threading
|
|
from datetime import datetime
|
|
from queue import Queue, Empty
|
|
from flask import Flask, render_template, request
|
|
from flask_socketio import SocketIO, emit
|
|
|
|
app = Flask(__name__)
|
|
socketio = SocketIO(app)
|
|
|
|
# MQTT Broker 配置
|
|
mqtt_server = {
|
|
"remote": "192.168.1.106",
|
|
"port": 1883,
|
|
"user_name": "admin",
|
|
"password": "miler502"
|
|
}
|
|
topic = "measure/#"
|
|
|
|
# 消息队列
|
|
message_queue = Queue()
|
|
|
|
# 后端 API
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
# 有问题暂时未实现
|
|
@app.route('/history/<path:topic>/', methods=['POST'])
|
|
def clear_history(topic):
|
|
return 'History feature not implemented'
|
|
|
|
# 定时器线程函数
|
|
def process_messages():
|
|
while True:
|
|
try:
|
|
topic, alarm_id, alarm_value, car_position_msg = message_queue.get(timeout=3)
|
|
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
# 将消息通过 WebSocket 发送给前端
|
|
socketio.emit('new_message', {
|
|
'topic': topic,
|
|
'alarm_id': alarm_id,
|
|
'alarm_value': alarm_value,
|
|
'car_position_msg': car_position_msg,
|
|
'timestamp': current_time
|
|
})
|
|
except Empty:
|
|
pass
|
|
except Exception as e:
|
|
print(f"Failed to process message: {e}", file=sys.stderr)
|
|
|
|
# 启动定时器线程
|
|
threading.Thread(target=process_messages, daemon=True).start()
|
|
|
|
# MQTT 回调函数
|
|
def on_connect(client, userdata, flags, rc):
|
|
if rc == 0:
|
|
print("Connected successfully")
|
|
client.subscribe(topic)
|
|
else:
|
|
print(f"Connect failed with code {rc}")
|
|
|
|
def on_message(client, userdata, msg):
|
|
try:
|
|
payload = msg.payload.decode()
|
|
data = json.loads(payload)
|
|
alarm_id = data.get("AlarmId", "")
|
|
alarm_value = data.get("AlarmMessage", {}).get("value", "")
|
|
car_position_msg = data.get("CarPositionMsg", "")
|
|
message_queue.put((msg.topic, alarm_id, alarm_value, car_position_msg))
|
|
except Exception as e:
|
|
print(f"Failed to process message: {e}", file=sys.stderr)
|
|
|
|
# 设置 MQTT 客户端
|
|
client = mqtt.Client()
|
|
client.username_pw_set(mqtt_server["user_name"], mqtt_server["password"])
|
|
client.on_connect = on_connect
|
|
client.on_message = on_message
|
|
|
|
# 启动 Flask 应用和 MQTT 客户端
|
|
if __name__ == '__main__':
|
|
try:
|
|
mqtt_thread = threading.Thread(target=lambda: client.connect(mqtt_server["remote"], mqtt_server["port"], 60))
|
|
mqtt_thread.start()
|
|
client.loop_start()
|
|
socketio.run(app, host='0.0.0.0', port=5000)
|
|
except Exception as e:
|
|
print(f"Failed to start server: {e}", file=sys.stderr)
|
|
finally:
|
|
client.loop_stop()
|