2. 食品加工批次传感器时序数据重采样与对齐

批次对齐的核心难点：三路传感器不同频、不同起点

杀菌釜一个批次里，温度、压力、流量三路传感器的采样周期往往不同（例如温度 10 s、压力 30 s、流量 60 s），且各自存在时钟漂移与起始时刻偏差。若直接按行拼接，会出现“同一行其实不是同一时刻”的错位。本例的目标是：以批次为边界，把三路数据统一重采样到 1 分钟栅格，并在对齐前先校正时间戳漂移。

用 pandas 完成漂移校正与 1 分钟重采样

下面用一段可运行的示例数据模拟三路传感器，演示完整处理链：解析时间戳 → 估计漂移 → 以温度为主时钟重采样 → 按批次边界裁剪 → 校验对齐。示例参数为教学假设值，运行结果由给定输入直接推出。

import pandas as pd
import numpy as np

# ---------- 1. 构造三路不同频、不同起点的示例数据 ----------
rng = pd.date_range("2024-06-01 08:00:00", "2024-06-01 09:00:00", freq="10s")
temp = pd.DataFrame({
    "ts": rng + pd.Timedelta(seconds=7),          # 温度起点 08:00:07
    "temperature": 121 + 0.5*np.sin(np.arange(len(rng))/30)
})
pres = pd.DataFrame({
    "ts": pd.date_range("2024-06-01 08:00:22", "2024-06-01 09:00:00", freq="30s"),
    "pressure": 0.21 + 0.01*np.cos(np.arange(121)/20)
})
flow = pd.DataFrame({
    "ts": pd.date_range("2024-06-01 08:00:41", "2024-06-01 09:00:00", freq="60s"),
    "flow": 18.4 + 0.2*np.sin(np.arange(61)/10)
})

# ---------- 2. 估计漂移：以温度为主时钟，用最近邻匹配首点 ----------
def estimate_offset(ref_ts, other_ts):
    return (other_ts.iloc[0] - ref_ts.iloc[0]).total_seconds()

d_p = estimate_offset(temp["ts"], pres["ts"])   # +15 s
d_f = estimate_offset(temp["ts"], flow["ts"])   # +34 s
print("压力漂移:", d_p, "s  流量漂移:", d_f, "s")

# ---------- 3. 校正漂移并统一到 1 min 栅格 ----------
def to_minute(df, col, offset_s):
    s = df.set_index("ts")[col]
    s.index = s.index - pd.Timedelta(seconds=offset_s)   # 回拨到主时钟
    return s.resample("1min").mean().interpolate("time")

t_min = to_minute(temp, "temperature", 0)
p_min = to_minute(pres, "pressure", d_p)
f_min = to_minute(flow, "flow", d_f)

aligned = pd.concat([t_min, p_min, f_min], axis=1)

# ---------- 4. 批次边界裁剪 ----------
batch_start = pd.Timestamp("2024-06-01 08:01:00")
batch_end   = pd.Timestamp("2024-06-01 08:59:00")
aligned = aligned.loc[batch_start:batch_end]

# ---------- 5. 对齐校验 ----------
assert aligned.index.is_monotonic_increasing
assert aligned.notna().all().all(), "存在缺失分钟"
assert (aligned.index.to_series().diff().dropna()
        == pd.Timedelta("1min")).all(), "栅格不连续"
print(aligned.head())
print("行数:", len(aligned), " 列:", list(aligned.columns))

结果判读与交付检查

完成上述步骤后，输出的 aligned 即为批次维度严格对齐的 1 分钟时序表，可直接供后续杀菌强度计算或超温事件提取使用。