数据采集基础：传感器数据采集、MQTT协议基础、数据预处理与清洗

各位同学，欢迎来到第二章。这一章我们聊聊IoT系统的“口粮”——数据采集。你想想看，一个异常检测系统，如果连数据都拿不准、传不好、洗不干净，后面再牛的算法也是白搭。我在早期做项目时，就吃过这个亏，所以这一章我会把实战中踩过的坑、总结的经验，都掰开揉碎了讲给你听。

2.1 传感器数据采集：从物理世界到数字信号

传感器，说白了就是把物理世界的温度、湿度、振动、电流这些模拟量，变成单片机或网关能识别的数字信号。我个人习惯把采集过程分成三步：感知、转换、传输。

2.1.1 常见传感器类型与接口

不同的物理量，对应不同的传感器。我整理了一个常用清单，你在选型时可以直接参考：

2.1.2 采集代码实战：以DHT22为例

我们拿最常用的温湿度传感器DHT22来演示。它用单总线协议，一根数据线就能搞定。下面是我在树莓派上采集数据的代码片段：

import Adafruit_DHT
import time

# 设置传感器类型和GPIO引脚
SENSOR = Adafruit_DHT.DHT22
PIN = 4

def read_sensor():
    humidity, temperature = Adafruit_DHT.read_retry(SENSOR, PIN)
    if humidity is not None and temperature is not None:
        # 保留一位小数，方便后续处理
        return {
            "temperature": round(temperature, 1),
            "humidity": round(humidity, 1),
            "timestamp": time.time()
        }
    else:
        # 我曾经遇到过连续读取失败的情况，后来发现是线没插紧
        print("读取失败，检查接线或传感器是否损坏")
        return None

# 主循环：每5秒采集一次
if __name__ == "__main__":
    while True:
        data = read_sensor()
        if data:
            print(f"温度: {data['temperature']}°C, 湿度: {data['humidity']}%")
        time.sleep(5)

2.2 MQTT协议基础：IoT数据的“快递员”

数据采集上来之后，怎么传到服务器？Wi-Fi、蓝牙、LoRa……方案很多。但在IoT领域，MQTT几乎是事实标准。为什么？因为它轻量、支持发布/订阅模式、还有三个QoS等级保证消息可靠性。说白了，它就是为低带宽、高延迟、不稳定的网络环境设计的。

2.2.1 MQTT的核心概念

MQTT有三个角色：发布者、代理（Broker）、订阅者。发布者把消息发到某个“主题”，订阅者订阅这个主题，就能收到消息。代理负责中转。我画了一张图，帮你理解这个流程：

2.2.2 实战：用Python实现MQTT通信

我们用paho-mqtt库，写一个简单的发布者和订阅者。先安装库：

pip install paho-mqtt

发布者代码（传感器端）：

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

# 连接回调
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("连接成功")
    else:
        print(f"连接失败，返回码: {rc}")

client = mqtt.Client()
client.on_connect = on_connect

# 连接本地Broker（假设你在本机运行了Mosquitto）
client.connect("localhost", 1883, 60)

# 模拟传感器数据发布
while True:
    payload = json.dumps({
        "temperature": 25.3,
        "humidity": 60.1,
        "timestamp": time.time()
    })
    # 发布到 sensor/temp 主题，QoS=1
    client.publish("sensor/temp", payload, qos=1)
    print(f"已发布: {payload}")
    time.sleep(5)

订阅者代码（服务器端）：

import paho.mqtt.client as mqtt
import json

def on_message(client, userdata, msg):
    # 收到消息后解析JSON
    data = json.loads(msg.payload.decode())
    print(f"收到主题 {msg.topic}: 温度={data['temperature']}°C, 湿度={data['humidity']}%")
    # 这里可以调用后续的异常检测逻辑

client = mqtt.Client()
client.on_message = on_message

client.connect("localhost", 1883, 60)
# 订阅 sensor/temp 主题
client.subscribe("sensor/temp", qos=1)

print("等待消息...")
client.loop_forever()

2.3 数据预处理与清洗：把脏数据变成“干净粮”

数据到了服务器，你以为就能直接用了？太天真了。我在一个智能楼宇项目里，采集到的温度数据里，有-127°C的“幽灵值”，还有连续10分钟不变的“死值”。如果不处理，异常检测模型会把这些当成正常行为。所以，预处理是必须的。

2.3.1 常见的数据问题

2.3.2 清洗流程实战

我一般按这个流程处理：去重 → 过滤异常值 → 填充缺失值 → 时间戳对齐。下面是一个完整的清洗函数：

import pandas as pd
import numpy as np

def clean_sensor_data(df):
    """
    清洗传感器数据
    df: DataFrame，必须包含 'timestamp', 'temperature', 'humidity' 列
    """
    # 1. 去重：按时间戳去重，保留第一条
    df = df.drop_duplicates(subset=['timestamp'], keep='first')
    
    # 2. 过滤异常值：温度合理范围 -20~60°C，湿度 0~100%
    df = df[(df['temperature'] >= -20) & (df['temperature'] <= 60)]
    df = df[(df['humidity'] >= 0) & (df['humidity'] <= 100)]
    
    # 3. 填充缺失值：用前向填充法（ffill）
    # 我曾经试过用均值填充，但会导致数据平滑过度，掩盖异常
    df = df.fillna(method='ffill')
    
    # 4. 时间戳对齐：按1分钟间隔重采样，取均值
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
    df = df.set_index('timestamp').resample('1T').mean().reset_index()
    
    return df

# 示例用法
raw_data = pd.DataFrame({
    'timestamp': [1700000000, 1700000005, 1700000010, 1700000015],
    'temperature': [25.3, -127.0, 25.4, 25.5],  # 第二个是异常值
    'humidity': [60.1, 60.2, 60.1, 60.3]
})

cleaned_data = clean_sensor_data(raw_data)
print(cleaned_data)

2.4 本章小结：数据采集的“铁三角”

这一章我们走了完整的数据采集链路：从传感器读取原始信号，到MQTT协议传输，再到服务器端的清洗预处理。这三步环环相扣，任何一个环节出问题，都会影响后续的异常检测效果。

我个人觉得，数据采集阶段花的时间，往往比模型调参还多。但这是值得的——干净、可靠的数据，是异常检测系统成功的基石。下一章，我们会基于这些清洗后的数据，开始构建特征工程。但那是后话了，先把今天的内容消化掉。