物联网数据采集与可视化：基于 Python 和 MQTT 的实时监控系统

引言

在物联网项目中，数据只是原材料。如果无法实时处理、分析和展示这些数据，我们就很难做出及时的决策。今天我们来搭建一个完整的物联网数据采集与可视化系统，利用 Python、MQTT 协议和 WebSocket 技术，实现从传感器到前端页面的实时数据监控。

我们将模拟一个温度传感器系统，涵盖以下流程：

模拟传感器数据生成
使用 MQTT 协议发布数据
通过 Python 服务订阅并处理数据
将处理后的数据通过 WebSocket 推送到前端
使用 HTML/CSS/JavaScript 实现可视化展示

系统架构

整个链路非常清晰：

[传感器] --> [MQTT Broker] --> [Python 服务] --> [WebSocket Server] --> [Web 浏览器]

环境准备

我们需要安装以下依赖：

Python 3.x
paho-mqtt: MQTT 客户端库
websockets: WebSocket 服务端库
asyncio: 异步 I/O 库（Python 内置）
flask: Web 框架（用于提供静态文件）
matplotlib: 数据可视化库
numpy: 数值计算库

安装命令如下：

pip install paho-mqtt websockets flask matplotlib numpy

项目结构

为了保持代码整洁，我们采用如下目录结构：

iot_monitoring/
├── app.py                 # 主应用逻辑
├── mqtt_publisher.py      # MQTT 发布者（模拟传感器）
├── requirements.txt       # 依赖包列表
├── templates/
│   └── index.html         # 前端页面
└── static/
    └── script.js          # 前端 JavaScript

1. MQTT 发布者（模拟传感器）

首先，我们需要一个'设备'来产生数据。创建 mqtt_publisher.py 文件：

# mqtt_publisher.py
import random
import time
import json
import paho.mqtt.client as mqtt

# MQTT 配置
MQTT_BROKER = 
MQTT_PORT = 
TOPIC = 


SENSOR_ID = 

 ():
     rc == :
        ()
    :
        ()

 ():
    client = mqtt.Client()
    client.on_connect = on_connect
    
    
    client.connect(MQTT_BROKER, MQTT_PORT, )
    
    
    :
         :
            
            temperature = (random.uniform(, ), )
            
            humidity = (random.uniform(, ), )
            
            payload = {
                : SENSOR_ID,
                : temperature,
                : humidity,
                : time.time()
            }
            
            
            client.publish(TOPIC, json.dumps(payload))
            ()
            
            
            time.sleep()
     KeyboardInterrupt:
        ()
        client.disconnect()

 __name__ == :
    simulate_sensor_data()

# app.py import asyncio import websockets import json import threading import time import queue import paho.mqtt.client as mqtt from flask import Flask, render_template import logging # 设置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 全局变量 latest_data = {} data_queue = queue.Queue() # 用于线程间传递数据 connected_clients = set() lock = threading.Lock() # 保护共享资源 # MQTT 配置 MQTT_BROKER = "localhost" MQTT_PORT = 1883 MQTT_TOPIC = "iot/sensors/temperature" # Flask 应用 app = Flask(__name__) @app.route('/') def index(): return render_template('index.html') # WebSocket 服务器 async def handle_websocket(websocket, path): logger.info(f"WebSocket client connected from {websocket.remote_address}") with lock: connected_clients.add(websocket) try: async for message in websocket: pass # 预留处理客户端消息的逻辑 except websockets.exceptions.ConnectionClosed: logger.info(f"WebSocket client disconnected from {websocket.remote_address}") finally: with lock: connected_clients.discard(websocket) async def broadcast_data(data): """向所有连接的 WebSocket 客户端广播数据""" if connected_clients: data_to_send = json.dumps(data) tasks = [] with lock: clients_copy = list(connected_clients) for client in clients_copy: tasks.append(client.send(data_to_send)) await asyncio.gather(*tasks, return_exceptions=True) def on_message(client, userdata, msg): """处理从 MQTT 接收到的消息""" try: data = json.loads(msg.payload.decode()) global latest_data latest_data = data logger.info(f"Received MQTT message: {data}") # 将数据放入队列，由主循环处理 data_queue.put(data) except Exception as e: logger.error(f"Error processing MQTT message: {e}") def start_mqtt_client(): """启动 MQTT 客户端""" client = mqtt.Client() client.on_message = on_message client.connect(MQTT_BROKER, MQTT_PORT, 60) client.subscribe(MQTT_TOPIC) logger.info("MQTT Client started and subscribed to topic.") client.loop_forever() async def main(): """主异步函数""" # 启动 Flask 应用 (在单独线程中) def run_flask(): app.run(host='0.0.0.0', port=5000, debug=False) flask_thread = threading.Thread(target=run_flask) flask_thread.daemon = True flask_thread.start() # 启动 MQTT 客户端 (在单独线程中) mqtt_thread = threading.Thread(target=start_mqtt_client) mqtt_thread.daemon = True mqtt_thread.start() # 启动 WebSocket 服务器 server = await websockets.serve(handle_websocket, "0.0.0.0", 8765) logger.info("WebSocket server started on ws://0.0.0.0:8765") # 处理队列中的 MQTT 数据 async def process_queue(): while True: try: data = data_queue.get_nowait() await broadcast_data(data) except queue.Empty: await asyncio.sleep(0.1) # 启动队列处理任务 asyncio.create_task(process_queue()) await server.wait_closed() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("Shutting down...")

// static/script.js const wsUri = "ws://localhost:8765"; let websocket; let chart; document.addEventListener('DOMContentLoaded', function() { const ctx = document.getElementById('sensorChart').getContext('2d'); chart = new Chart(ctx, { type: 'line', data: { labels: [], datasets: [ { label: 'Temperature (°C)', data: [], borderColor: 'rgb(255, 99, 132)', backgroundColor: 'rgba(255, 99, 132, 0.2)', tension: 0.1, fill: false }, { label: 'Humidity (%)', data: [], borderColor: 'rgb(54, 162, 235)', backgroundColor: 'rgba(54, 162, 235, 0.2)', tension: 0.1, fill: false } ] }, options: { responsive: true, maintainAspectRatio: false, scales: { x: { title: { display: true, text: 'Time' } }, y: { beginAtZero: false } }, plugins: { legend: { position: 'top' }, title: { display: true, text: 'Real-time Sensor Data' } } } }); connectWebSocket(); }); function connectWebSocket() { console.log("Attempting to connect to WebSocket server..."); websocket = new WebSocket(wsUri); websocket.onopen = function(event) { console.log("WebSocket connection opened."); document.getElementById('status').textContent = "Connected"; document.getElementById('status').className = "status connected"; }; websocket.onclose = function(event) { console.log("WebSocket connection closed."); document.getElementById('status').textContent = "Disconnected"; document.getElementById('status').className = "status disconnected"; setTimeout(connectWebSocket, 5000); }; websocket.onerror = function(error) { console.error("WebSocket error:", error); document.getElementById('status').textContent = "Connection Error"; document.getElementById('status').className = "status disconnected"; }; websocket.onmessage = function(event) { try { const data = JSON.parse(event.data); console.log("Received data:", data); // 更新 DOM 元素 document.getElementById('temperature').textContent = data.temperature.toFixed(2) + ' °C'; document.getElementById('humidity').textContent = data.humidity.toFixed(2) + ' %'; // 显示人类可读的时间 const date = new Date(data.timestamp * 1000); document.getElementById('timestamp').textContent = date.toLocaleString(); updateChart(data); } catch (e) { console.error("Error parsing JSON:", e); } }; } function updateChart(data) { const timestamp = new Date(data.timestamp * 1000).toLocaleTimeString(); const temp = data.temperature; const humidity = data.humidity; chart.data.labels.push(timestamp); chart.data.datasets[0].data.push(temp); chart.data.datasets[1].data.push(humidity); // 限制图表上的数据点数量（例如，保留最近的 20 个） if (chart.data.labels.length > 20) { chart.data.labels.shift(); chart.data.datasets[0].data.shift(); chart.data.datasets[1].data.shift(); } chart.update(); }

物联网数据采集与可视化：基于 Python 和 MQTT 的实时监控系统