当前位置：首页 > news >正文

智能城市管理系统设计思路详解：集成InfluxDB、Grafana和MQTTx协议（代码示例）

news 2025/11/7 12:25:29

引言

随着城市化进程的加快，城市管理面临越来越多的挑战。智能城市管理系统的出现，为城市的基础设施管理、资源优化和数据分析提供了现代化的解决方案。本文将详细介绍一个基于开源技术的智能城市管理系统，涵盖系统功能、技术实现、环境搭建、代码示例及使用说明等内容。

系统功能

智能城市管理系统主要包括以下功能：

监控城市基础设施：
- 实时监控路灯、交通灯、垃圾桶等基础设施的状态。
- 提供故障报警和维护建议。
优化城市资源使用：
- 根据实际需求动态调整路灯亮度。
- 在高峰时段优化交通信号灯的切换策略。
数据展示与历史分析：
- 实时展示城市管理数据。
- 提供历史数据的分析功能，帮助决策者优化管理策略。

技术实现

本智能城市管理系统的核心技术包括：

Node-RED：用于管理城市基础设施的控制逻辑。
Mosquitto：实现基础设施设备之间的通信。
InfluxDB：用于存储城市管理数据。
Grafana：用于展示和分析城市管理数据。
Python：进行数据处理和优化策略计算。

环境搭建

在开始实现智能城市管理系统之前，需要搭建相应的开发环境。以下是环境搭建的步骤：

安装 Node-RED：
```
npm install -g node-red
```
安装 Mosquitto：
- 在 Ubuntu 上，使用以下命令安装 Mosquitto：
```
sudo apt install mosquitto mosquitto-clients
```
安装 InfluxDB：
- 下载并安装 InfluxDB，具体安装步骤请参考 InfluxDB 官方文档.
安装 Grafana：
- 下载并安装 Grafana，具体安装步骤请参考 Grafana 官方文档.

安装 Python 及相关库：

pip install paho-mqtt influxdb-client matplotlib numpy

系统架构

代码示例说明

1. Node-RED 流程

在上面提供的 Node-RED 流程示例中，我们创建了一个简单的流程来接收和处理来自基础设施设备的 MQTT 消息。下面是各个节点的详细说明：

[{"id": "mqtt_in","type": "mqtt in","z": "flow_id","name": "设备状态","topic": "city/infrastructure/status","qos": "2","datatype": "auto","broker": "mqtt_broker","x": 100,"y": 100,"wires": [["function_process"]]},{"id": "function_process","type": "function","z": "flow_id","name": "处理状态","func": "msg.payload = JSON.parse(msg.payload);\nreturn msg;","outputs": 1,"noerr": 0,"x": 300,"y": 100,"wires": [["influxdb_out"]]},{"id": "influxdb_out","type": "influxdb out","z": "flow_id","influxdb": "influxdb_config","name": "存储到 InfluxDB","measurement": "infrastructure_status","x": 500,"y": 100,"wires": [[]]}
]

节点配置说明

MQTT 输入节点 (mqtt_in)：
- 功能：此节点用于接收来自城市基础设施设备的状态消息。
- 配置：
  - 主题：city/infrastructure/status，表示我们订阅的是基础设施设备的状态更新。
  - QoS：设置为 2，表示“至多一次传递”，确保消息的可靠性。
- 输出：接收到的消息将传递到下一个节点（function_process）。
函数节点 (function_process)：
- 功能：此节点用于处理接收到的消息，将 JSON 格式的消息解析为 JavaScript 对象，以便进行后续处理。
- 代码：
```
msg.payload = JSON.parse(msg.payload);
return msg;
```
  - 这段代码将消息的负载（msg.payload）从字符串格式转换为 JavaScript 对象，以便更方便地访问其属性。
- 输出：处理后的消息将发送到下一个节点（influxdb_out）。
InfluxDB 输出节点 (influxdb_out)：
- 功能：此节点用于将处理后的数据存储到 InfluxDB 数据库中。
- 配置：
  - 测量名称：infrastructure_status，表示我们将数据存储在名为 infrastructure_status 的测量中。
- 输出：此节点没有后续连接，主要用于数据存储。

2. Python 数据处理示例

接下来，我们使用 Python 脚本进行数据处理和优化策略计算。以下是示例代码：

import paho.mqtt.client as mqtt
import json
from influxdb import InfluxDBClient# InfluxDB 配置
influxdb_host = 'localhost'
influxdb_port = 8086
influxdb_database = 'city_management'# 创建 InfluxDB 客户端
client = InfluxDBClient(host=influxdb_host, port=influxdb_port, database=influxdb_database)# MQTT 回调函数
def on_message(client, userdata, message):# 解析消息data = json.loads(message.payload.decode())print("Received data:", data)# 数据存储到 InfluxDBinflux_data = [{"measurement": "infrastructure_status","tags": {"device": data["device"],"type": data["type"]},"fields": {"status": data["status"],"brightness": data.get("brightness", 0)  # 可选字段}}]client.write_points(influx_data)print("Data written to InfluxDB")# MQTT 客户端配置
mqtt_client = mqtt.Client()
mqtt_client.on_message = on_message
mqtt_client.connect("localhost", 1883, 60)
mqtt_client.subscribe("city/infrastructure/status")# 启动 MQTT 循环
mqtt_client.loop_forever()

代码说明

InfluxDB 配置：
- 定义了 InfluxDB 的连接参数，包括主机名、端口号和数据库名称。
- 创建了 InfluxDB 客户端，用于与数据库进行交互。
MQTT 回调函数 (on_message)：
- 此函数在接收到 MQTT 消息时被调用。
- 解析消息：使用 json.loads 将 JSON 数据解析为 Python 字典。
- 数据存储：
  - 创建一个包含测量名称、标签和字段的字典，准备将数据写入 InfluxDB。
  - 标签包括设备名称和类型，字段包括状态和亮度（如果存在）。
  - 使用 client.write_points(influx_data) 将数据写入 InfluxDB。
MQTT 客户端配置：
- 创建一个 MQTT 客户端实例，并设置消息回调函数。
- 连接到 MQTT 代理（Mosquitto），并订阅 city/infrastructure/status 主题。
- 启动 MQTT 循环，保持客户端在线并接收消息。

3. Grafana 数据展示

Grafana 是一个开源的数据可视化和监控工具，可以与 InfluxDB 等多种数据源集成，轻松创建动态仪表板和可视化图表。通过 Grafana，我们可以实时监控城市基础设施的状态和历史数据。

3.1 安装和配置 Grafana

安装 Grafana：
- 如果您尚未安装 Grafana，可以按照以下命令在 Ubuntu 上安装：
```
sudo apt-get install -y software-properties-common
sudo add-apt-repository -y ppa:grafana/stable
sudo apt-get update
sudo apt-get install grafana
```
- 也可以访问 Grafana 官方网站下载适合您操作系统的安装包。
启动 Grafana：
```
sudo systemctl start grafana-server
sudo systemctl enable grafana-server
```
- 启动后，Grafana 默认运行在 http://localhost:3000。
登录 Grafana：
- 打开浏览器，访问 http://localhost:3000。
- 默认用户名和密码均为 admin，首次登录后会要求您更改密码。

3.2 添加 InfluxDB 数据源

添加数据源：
- 在左侧菜单中，点击 Configuration (配置)，然后选择 Data Sources (数据源)。
- 点击 Add data source (添加数据源)，选择 InfluxDB。
配置 InfluxDB 数据源：
- 在数据源配置页面，填写以下信息：
  - HTTP URL：http://localhost:8086
  - Database：city_management
  - User 和 Password：如果您在 InfluxDB 中设置了用户名和密码，请填写相应信息。
- 点击 Save & Test (保存并测试)，确保 Grafana 能成功连接到 InfluxDB。

3.3 创建仪表板

创建新仪表板：
- 在左侧菜单中，点击 +，然后选择 Dashboard (仪表板)。
- 点击 Add new panel (添加新面板)。
配置图表：
- 在面板编辑器中，选择 Query (查询) 选项卡。
- 在 Data source (数据源) 下拉框中，选择之前添加的 InfluxDB 数据源。
- 在 Query 输入框中，可以使用类似以下的查询：
```
SELECT mean("status") FROM "infrastructure_status" WHERE $timeFilter GROUP BY time($__interval) fill(null)
```
  - 该查询从 infrastructure_status 测量中获取状态的平均值，并根据时间进行分组。
设置图表类型：
- 在右侧的 Panel 设置中，选择图表类型（例如，折线图、柱状图等）。
- 可以自定义图表的标题、轴标签、颜色等。
保存仪表板：
- 完成图表配置后，点击右上角的 Save dashboard (保存仪表板)，为仪表板命名并保存。

4. 使用说明

4.1 启动系统

启动 Mosquitto 代理：
```
sudo systemctl start mosquitto
```
启动 Node-RED：
```
node-red
```
启动 InfluxDB：
```
influxd
```
启动 Grafana：
```
sudo systemctl start grafana-server
```

4.2 模拟设备数据

为了测试整个系统，您可以使用一个简单的 Python 脚本来模拟城市基础设施设备发送状态消息。以下是一个简单的示例：

4.2 模拟设备数据

为了测试整个系统，您可以使用一个简单的 Python 脚本来模拟城市基础设施设备发送状态消息。以下是一个完整的示例代码：

import paho.mqtt.publish as publish
import json
import random
import time# MQTT 配置
mqtt_broker = "localhost"
topic = "city/infrastructure/status"while True:# 随机生成设备状态device_data = {"device": f"light_{random.randint(1, 10)}",  # 生成设备名称，如 light_1, light_2 等"type": "light",  # 设备类型"status": random.choice(["on", "off"]),  # 随机选择设备状态"brightness": random.randint(0, 100)  # 随机生成亮度值}# 将设备数据转换为 JSON 格式payload = json.dumps(device_data)# 发布 MQTT 消息publish.single(topic, payload, hostname=mqtt_broker)print(f"Published: {payload} to topic: {topic}")# 每隔 5 秒发送一次消息time.sleep(5)

代码说明

导入库：
- paho.mqtt.publish：用于发布 MQTT 消息。
- json：用于将数据转换为 JSON 格式。
- random：用于生成随机数。
- time：用于设置延时。
MQTT 配置：
- mqtt_broker：设置 MQTT 代理的地址，这里为 localhost。
- topic：消息发布的主题为 city/infrastructure/status。
无限循环：
- 使用 while True 创建一个无限循环，以便定期发布消息。
随机生成设备数据：
- device：生成设备名称，格式为 light_x，其中 x 是随机生成的数字（1 到 10）。
- type：设备类型，这里固定为 light。
- status：随机选择设备状态，可以是 on 或 off。
- brightness：随机生成亮度值，范围为 0 到 100。
发布消息：
- 将生成的设备数据转换为 JSON 格式，并使用 publish.single() 方法将其发布到指定主题。
- 每次发布后打印出发布的内容，方便调试。
延时：
- 使用 time.sleep(5) 控制每隔 5 秒发送一次消息。这可以通过修改该值来调节发送频率。

4.3 观察数据流

运行模拟设备数据脚本：
- 在终端中运行上述模拟设备数据的 Python 脚本。您将看到脚本定期发布设备状态消息到 MQTT 代理。
```
python simulate_device.py
```
观察 Node-RED 流程：
- 打开 Node-RED 的浏览器界面（通常为 http://localhost:1880）。
- 查看 Node-RED 流程的调试窗口，您将看到接收到的设备状态消息。
查看 InfluxDB 数据：
- 您可以使用 InfluxDB CLI 或者其他工具（如 Chronograf）查看存储在数据库中的数据，确保数据正确写入。
```
influx
USE city_management
SELECT * FROM infrastructure_status
```
在 Grafana 中查看数据可视化：
- 打开 Grafana 的浏览器界面（通常为 http://localhost:3000），查看您创建的仪表板。
- 您应该能够看到实时更新的图表，展示城市基础设施的状态变化。

5. 总结

通过本教程，我们搭建了一个简单的智能城市管理系统，使用了以下技术栈：

Node-RED：用于管理和处理城市基础设施的控制逻辑。
Mosquitto：作为 MQTT 代理，负责设备之间的通信。
InfluxDB：用于存储城市管理数据。
Grafana：用于实时展示和分析城市管理数据。
Python：用于模拟设备数据和进行数据处理。

引言

系统功能

技术实现

环境搭建

系统架构

代码示例说明

1. Node-RED 流程

节点配置说明

2. Python 数据处理示例

代码说明

3. Grafana 数据展示

3.1 安装和配置 Grafana

3.2 添加 InfluxDB 数据源

3.3 创建仪表板

4. 使用说明

4.1 启动系统

4.2 模拟设备数据

4.2 模拟设备数据

代码说明

4.3 观察数据流

5. 总结

相关文章：