📚 课程目录第4章模拟数据生成器开发

4. 模拟数据生成器开发

在数字工地看板项目中，真实设备数据往往难以在开发阶段获取。因此，开发一个模拟数据生成器是启动数据管线的关键步骤。本章将使用 Python 构建一个轻量级模拟器，生成包含设备 ID、时间戳、温度、湿度、噪音、粉尘浓度的结构化数据，并通过 MQTT 协议发布到指定主题，为后续看板可视化提供稳定的数据源。

4.1 模拟数据模型设计

模拟数据需覆盖工地环境监测的核心指标。每个数据点包含以下字段：

字段名	类型	说明	模拟范围
device_id	string	设备唯一标识	“sensor-001” ~ “sensor-020”
timestamp	string (ISO 8601)	数据采集时间	当前时间 ± 随机偏移
temperature	float	环境温度 (°C)	15.0 ~ 45.0
humidity	float	相对湿度 (%)	30.0 ~ 90.0
noise	float	噪音分贝 (dB)	50.0 ~ 120.0
dust	float	粉尘浓度 (μg/m³)	10.0 ~ 500.0

设计要点： 模拟范围应贴近真实工地环境。例如，粉尘浓度在挖掘作业时可达 300 μg/m³ 以上，而夜间停工时段则降至 50 μg/m³ 以下。后续可通过概率分布模拟这种波动。

4.2 数据生成器核心逻辑

生成器采用 Python 标准库 实现，无需第三方依赖。核心步骤包括：

设备池初始化：预定义 5~20 个设备 ID，支持循环或随机选取。
时间戳生成：使用 datetime.utcnow() 生成当前时间，格式化为 ISO 8601 字符串。
传感器值模拟：使用 random.uniform() 在预设范围内生成浮点数，并保留两位小数。
数据封装：将字段组合为 Python 字典，再序列化为 JSON 字符串。
发布循环：通过 MQTT 客户端以固定间隔（如 2 秒）发布数据。

import random
import json
import time
from datetime import datetime

# 设备池
DEVICE_IDS = [f"sensor-{i:03d}" for i in range(1, 6)]

def generate_sensor_data():
    """生成单条工地传感器数据"""
    return {
        "device_id": random.choice(DEVICE_IDS),
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "temperature": round(random.uniform(15.0, 45.0), 2),
        "humidity": round(random.uniform(30.0, 90.0), 2),
        "noise": round(random.uniform(50.0, 120.0), 2),
        "dust": round(random.uniform(10.0, 500.0), 2)
    }

# 示例：生成一条数据并打印
sample = generate_sensor_data()
print(json.dumps(sample, indent=2))

输出示例：

{
  "device_id": "sensor-003",
  "timestamp": "2025-03-21T14:32:18.123456Z",
  "temperature": 32.45,
  "humidity": 67.89,
  "noise": 85.12,
  "dust": 234.56
}

4.3 MQTT 发布集成

MQTT 是物联网场景下最常用的轻量级消息协议。本课程使用 paho-mqtt 库实现发布端。安装命令：

pip install paho-mqtt

发布器核心流程如下：

连接 Broker：指定主机地址（如 localhost 或 broker.emqx.io）和端口（默认 1883）。
设置主题：采用分层主题结构，例如 construction/site/{device_id}/sensors。
循环发布：每 2 秒生成一条数据并调用 client.publish()。
异常处理：捕获连接断开或发布失败异常，自动重连。

import paho.mqtt.client as mqtt
import json
import time

BROKER = "localhost"
PORT = 1883
TOPIC_TEMPLATE = "construction/site/{}/sensors"

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")

client = mqtt.Client()
client.on_connect = on_connect
client.connect(BROKER, PORT, 60)
client.loop_start()

try:
    while True:
        data = generate_sensor_data()
        topic = TOPIC_TEMPLATE.format(data["device_id"])
        payload = json.dumps(data)
        client.publish(topic, payload, qos=1)
        print(f"Published to {topic}: {payload}")
        time.sleep(2)
except KeyboardInterrupt:
    print("Stopping publisher...")
    client.loop_stop()
    client.disconnect()

注意事项： 生产环境中建议使用 qos=1 确保至少一次送达；同时为每个设备设置独立的客户端 ID，避免冲突。若 Broker 需要认证，需在 connect() 前调用 client.username_pw_set()。

4.4 系统架构与数据流

下图展示了模拟数据生成器在整个数字工地看板系统中的位置与数据流向：

4.5 运行与验证

启动模拟器后，可通过 MQTT 客户端工具（如 MQTTX 或 mosquitto_sub）订阅主题验证数据是否正常到达：

# 使用 mosquitto_sub 订阅所有工地传感器数据
mosquitto_sub -h localhost -t "construction/site/+/sensors" -v

预期输出将每秒显示两条 JSON 格式的传感器数据，例如：

construction/site/sensor-001/sensors {"device_id":"sensor-001","timestamp":"2025-03-21T14:32:20.456789Z","temperature":28.73,"humidity":55.21,"noise":92.45,"dust":187.32}
construction/site/sensor-003/sensors {"device_id":"sensor-003","timestamp":"2025-03-21T14:32:22.123456Z","temperature":36.18,"humidity":72.64,"noise":78.90,"dust":312.45}

扩展建议： 可在生成器中加入 异常事件模拟（如温度骤升、粉尘超标），用于测试看板的告警逻辑。只需在 generate_sensor_data() 中按一定概率（如 5%）将某个字段值设为阈值以上即可。

4.6 本章小结

通过本章实战，你已掌握：

使用 Python 标准库 random 和 datetime 生成结构化工地传感器数据；
通过 paho-mqtt 库将数据发布到 MQTT 主题，实现设备与看板之间的实时通信；
理解 MQTT 发布/订阅模型在物联网数据管道中的核心作用。