4. 温室环境数据采集与预处理脚本实现

任务拆解：从多路原始读数到一份可用的清洗数据集

本节要交付的是一个可重复运行的数据预处理脚本。它面对的是温室里真实存在的三类麻烦：不同传感器采样周期不一致（温度 30 s、湿度 30 s、CO₂ 60 s、光照 15 s）、偶发跳变与丢数（通信重试、探头结露、遮阳网遮挡瞬间）、以及单位混杂（温度有 ℃ 与 0.1 ℃ 定点整数、光照有 lux 与 klx、CO₂ 有 ppm 与 %VOL）。脚本的输出必须是一张统一时间栅格、统一单位、异常已标记的宽表，供后续控制算法直接读取。

原始数据长什么样

假设现场网关每 10 分钟导出一个 CSV，字段为 ts, sensor_id, raw_value, unit，同一时刻不同传感器各占一行（长表）。温度通道上报的是 0.1 ℃ 定点整数（如 214 表示 21.4 ℃），CO₂ 通道部分时段上报 %VOL，光照通道部分时段上报 klx。这种长表结构必须先转成宽表才能做时间对齐。

ts,sensor_id,raw_value,unit
2024-06-12 08:00:00,T_air,214,0.1C
2024-06-12 08:00:00,RH,68.2,%RH
2024-06-12 08:00:00,CO2,612,ppm
2024-06-12 08:00:15,PAR,18.4,klx
2024-06-12 08:00:30,T_air,216,0.1C
2024-06-12 08:00:30,RH,67.9,%RH
2024-06-12 08:01:00,CO2,0.0608,%VOL
2024-06-12 08:01:00,T_air,9999,0.1C   ← 通信错误码，需剔除

脚本实现（Python + pandas）

下面脚本自包含，可直接对上述 CSV 运行。核心设计是先对齐、再判异常、最后换算：若先换算再判异常，量程阈值会随单位变化而失效；若先判异常再对齐，则无法利用对齐后的连续序列做滑动统计。

import pandas as pd
import numpy as np

GRID = "60s"                 # 统一时间栅格
MAX_HOLD = pd.Timedelta("180s")   # 前向保持最大窗
MAD_WIN = 15                 # 滑动中位数窗口（点数）
MAD_K = 3.5                  # 离群阈值倍数

# 物理量程（换算到 SI 单位后）
RANGE = {
    "T_air": (-10.0, 60.0),
    "RH":    (0.0, 100.0),
    "CO2":   (300.0, 2000.0),
    "PAR":   (0.0, 120000.0),
}

def to_si(sensor_id, value, unit):
    """把原始读数换算到统一 SI 单位。"""
    if sensor_id == "T_air" and unit == "0.1C":
        return value / 10.0
    if sensor_id == "CO2" and unit == "%VOL":
        return value * 10000.0
    if sensor_id == "PAR" and unit == "klx":
        return value * 1000.0
    return value

def load_raw(path):
    df = pd.read_csv(path, parse_dates=["ts"])
    df["si_value"] = [to_si(s, v, u) for s, v, u in
                      zip(df["sensor_id"], df["raw_value"], df["unit"])]
    return df

def pivot_wide(df):
    """长表转宽表：每个传感器一列，索引为原始时间戳。"""
    return df.pivot_table(index="ts", columns="sensor_id",
                          values="si_value", aggfunc="last")

def align_to_grid(wide):
    """重采样到统一栅格，前向保持但限制最大保持窗。"""
    grid_idx = pd.date_range(wide.index.min().floor(GRID),
                             wide.index.max().ceil(GRID), freq=GRID)
    out = wide.reindex(wide.index.union(grid_idx)).sort_index()
    out = out.ffill(limit=None)
    # 计算每个值距其原始采样点的时间差，超窗则置 NaN
    last_valid = {}
    for col in out.columns:
        src = wide[col].dropna()
        if src.empty:
            continue
        # 对每个栅格点找最近的、不晚于它的原始采样时刻
        pos = src.index.searchsorted(out.index, side="right") - 1
        age = np.where(pos >= 0,
                       (out.index - src.index[np.clip(pos, 0, len(src)-1)]),
                       pd.Timedelta.max)
        out[col] = np.where(pd.to_timedelta(age) <= MAX_HOLD,
                            out[col].values, np.nan)
    return out.loc[grid_idx]

def flag_anomalies(wide):
    """返回 qc 标记表：0 正常 / 1 量程越界 / 2 MAD 离群 / 3 保持超窗。"""
    qc = pd.DataFrame(0, index=wide.index, columns=wide.columns, dtype=int)
    for col in wide.columns:
        lo, hi = RANGE[col]
        s = wide[col]
        qc.loc[s.isna(), col] = 3
        qc.loc[s.notna() & ((s < lo) | (s > hi)), col] = 1
        med = s.rolling(MAD_WIN, min_periods=5, center=True).median()
        mad = (s - med).abs().rolling(MAD_WIN, min_periods=5, center=True).median()
        scale = 1.4826 * mad
        outlier = (scale > 0) & ((s - med).abs() > MAD_K * scale)
        qc.loc[outlier & (qc[col] == 0), col] = 2
    return qc

def build_clean_dataset(raw_path, out_path):
    raw = load_raw(raw_path)
    wide = pivot_wide(raw)
    aligned = align_to_grid(wide)
    qc = flag_anomalies(aligned)
    qc.columns = [f"qc_{c}" for c in qc.columns]
    clean = aligned.join(qc)
    clean.index.name = "ts"
    clean.to_csv(out_path, float_format="%.2f")
    return clean

if __name__ == "__main__":
    ds = build_clean_dataset("raw_20240612.csv", "clean_20240612.csv")
    print(ds.head())
    print("各通道异常计数：")
    print(ds.filter(like="qc_").apply(pd.Series.value_counts).fillna(0).astype(int))

运行与结果判读

脚本对示例 CSV 运行后，clean_20240612.csv 的索引为 60 s 等间隔时间戳，列包含四个 SI 单位通道与四个 qc_* 标记列。判读要点如下：

• 08:01 的 T_air：原始值 9999 换算后为 999.9 ℃，超出 −10~60 ℃ 量程，qc_T_air=1，该值保留但被标记，控制算法应跳过。
• 08:01 的 CO₂：原始 0.0608 %VOL 换算为 608 ppm，与前后 612、605 ppm 连续，qc_CO2=0，说明单位换算正确。
• PAR 通道：15 s 采样在 60 s 栅格上每格有 4 个原始点，pivot_table 的 aggfunc="last" 取最后一个，若需均值应改为 "mean" 并在文档中注明。
• 保持超窗：若某传感器断线超过 180 s，对应栅格点 qc=3 且值为 NaN，而不是被旧值无限填充。