3. 时序数据采集与标签关联实战

在设备时序标签工程中，采集只是起点，真正决定数据可用性的是“时间对齐”和“标签关联”两个环节。PLC、SCADA、IoT网关三类数据源在采样频率、时间基准、字段命名上往往互不统一：PLC通常以毫秒级扫描周期产生布尔量或寄存器值，SCADA以秒级轮询汇总工艺参数，IoT网关则可能以分钟级上报振动、温度等状态量。若直接拼接，会出现同一时刻多条记录、时间戳漂移、设备编码不一致等问题，导致后续建模或查询时无法准确回答“某台设备在某段时间内处于什么状态、是否发生过维修”。本节以一个可运行的Python原型为主线，说明如何把三类采集流统一到同一时间轴，并与设备台账、工单、维修记录完成标签关联。

一、采集对象与数据形态

假设某工厂有三类设备：注塑机（PLC控制）、空压机（SCADA监控）、风机（IoT网关接入）。采集侧约定如下示例参数：PLC侧每200 ms采集一次运行状态与合模压力，字段为device_code、ts、run_state、pressure；SCADA侧每5 s采集一次排气温度与电流，字段为equip_id、collect_time、temp、current；IoT网关每60 s上报一次振动均方根值，字段为dev_sn、report_ts、vib_rms。三类数据的时间字段名称不同，设备标识也不同，这是后续对齐与关联必须解决的第一层问题。

设备台账记录设备编码、设备名称、所属产线、设备类型；工单记录工单号、设备编码、计划开始与结束时间、工单类型；维修记录记录维修单号、设备编码、维修开始与结束时间、故障代码。台账是静态维表，工单和维修记录是带时间区间的动态事件表。标签关联的本质，是把每条时序采样点映射到“设备身份 + 当时所处工单/维修状态”上。

二、时间戳对齐与频率统一

时间对齐的第一步是统一时间字段。三类数据分别使用ts、collect_time、report_ts，在Python中统一重命名为ts，并全部转换为带时区的datetime64[ns, Asia/Shanghai]类型。若原始时间戳为字符串，使用pd.to_datetime(..., utc=True).dt.tz_convert("Asia/Shanghai")，避免夏令时或跨时区采集造成偏移。第二步是统一频率。以1秒为基准栅格，对PLC的200 ms数据按秒取均值或取末值，对SCADA的5 s数据按秒前向填充，对IoT的60 s数据按秒前向填充。前向填充的语义是“在两次上报之间，认为状态保持不变”，这符合温度、振动等缓变量的物理特性；但对run_state这类开关量，应使用“最近一次有效值”而非插值。

重采样时要注意：不能简单用resample('1S').mean()处理所有列，否则布尔量会被转成浮点。正确做法是按列类型分组处理，数值型用均值或前向填充，状态型用ffill()。重采样后会产生大量NaN，需根据业务决定是丢弃还是填充。示例中，PLC的pressure在1秒内取均值，run_state取该秒内最后一个有效值；SCADA的temp和current用前向填充；IoT的vib_rms用前向填充。最终每条记录都落在整秒时间点上，形成统一时间轴。

三、设备编码归一化与标签关联

三类采集流的设备标识不同：PLC用device_code，SCADA用equip_id，IoT用dev_sn。设备台账中主键为device_code，因此需要建立映射表，把equip_id和dev_sn映射到统一的device_code。映射关系可以来自台账中的别名列，也可以单独维护一张编码对照表。归一化后，所有时序记录都带有统一的device_code，才能与工单、维修记录关联。

标签关联采用区间匹配：对每条时序记录，查找该设备在ts时刻所处的工单和维修区间。工单表字段为work_order_id、device_code、plan_start、plan_end、order_type；维修表字段为repair_id、device_code、repair_start、repair_end、fault_code。匹配条件为device_code相等且ts落在区间内。若同一时刻同时命中工单和维修，按“维修优先”原则取维修标签，因为维修期间设备状态通常异常，工单计划时间可能被覆盖。实现时可用pd.merge_asof按时间就近匹配，或对每条记录做区间判断。数据量不大时，直接对工单和维修表做笛卡尔积后过滤更直观。

四、Python原型实现

下面给出一个可运行的原型骨架，使用pandas完成读取、对齐、关联和输出。示例数据为构造的模拟数据，用于说明流程，不代表真实采集结果。

import pandas as pd

# 1. 读取三类采集流（示例为模拟数据）
plc = pd.read_csv("plc_stream.csv")      # device_code, ts, run_state, pressure
scada = pd.read_csv("scada_stream.csv")  # equip_id, collect_time, temp, current
iot = pd.read_csv("iot_stream.csv")      # dev_sn, report_ts, vib_rms

# 2. 统一时间字段与设备编码
plc = plc.rename(columns={"ts": "ts"})
scada = scada.rename(columns={"collect_time": "ts", "equip_id": "device_code"})
iot = iot.rename(columns={"report_ts": "ts", "dev_sn": "device_code"})

for df in [plc, scada, iot]:
    df["ts"] = pd.to_datetime(df["ts"], utc=True).dt.tz_convert("Asia/Shanghai")

# 3. 重采样到1秒栅格
plc = plc.set_index("ts").groupby("device_code").resample("1S").agg(
    {"run_state": "last", "pressure": "mean"}).reset_index()
scada = scada.set_index("ts").groupby("device_code").resample("1S").agg(
    {"temp": "ffill", "current": "ffill"}).reset_index()
iot = iot.set_index("ts").groupby("device_code").resample("1S").agg(
    {"vib_rms": "ffill"}).reset_index()

# 4. 合并三类流
ts_all = plc.merge(scada, on=["device_code", "ts"], how="outer") \
             .merge(iot, on=["device_code", "ts"], how="outer") \
             .sort_values(["device_code", "ts"])

# 5. 关联设备台账
asset = pd.read_csv("asset.csv")  # device_code, device_name, line, device_type
ts_all = ts_all.merge(asset, on="device_code", how="left")

# 6. 关联工单与维修记录（区间匹配）
work_order = pd.read_csv("work_order.csv")
repair = pd.read_csv("repair.csv")

def match_interval(row, table, start_col, end_col, label_col):
    hit = table[(table["device_code"] == row["device_code"]) &
                (table[start_col] <= row["ts"]) &
                (table[end_col] >= row["ts"])]
    return hit[label_col].iloc[0] if len(hit) > 0 else None

ts_all["work_order_id"] = ts_all.apply(
    lambda r: match_interval(r, work_order, "plan_start", "plan_end", "work_order_id"), axis=1)
ts_all["repair_id"] = ts_all.apply(
    lambda r: match_interval(r, repair, "repair_start", "repair_end", "repair_id"), axis=1)
ts_all["fault_code"] = ts_all.apply(
    lambda r: match_interval(r, repair, "repair_start", "repair_end", "fault_code"), axis=1)

# 7. 输出统一时序标签表
ts_all.to_parquet("device_ts_labeled.parquet", index=False)
print(ts_all.head())

运行后，ts_all的每一行代表某设备在某一秒的采集值，并带有设备名称、产线、工单号、维修单号和故障代码。判断结果是否合理，可以抽查几个时间点：例如某注塑机在2024-06-01 10:00:00的run_state为1、pressure为82.3，同时work_order_id为WO-20240601-001，说明该秒处于计划工单内；若同一时刻repair_id非空，则说明维修优先，工单标签被覆盖。若发现大量记录work_order_id为空，应检查工单时间区间是否与采集时间重叠、设备编码是否归一化成功。

五、结果校验与常见问题

原型跑通后，至少做三项校验。第一，时间连续性：按设备分组检查ts是否以1秒为间隔递增，若出现跳变，说明重采样或合并环节有遗漏。第二，标签覆盖率：统计work_order_id和repair_id的非空比例，若工单覆盖率过低，优先排查工单区间是否包含采集时段。第三，语义一致性：随机抽取若干秒，人工核对run_state、temp、vib_rms与原始采集流是否一致，尤其检查前向填充是否把异常值扩散到了后续时间点。常见问题包括：时间戳未带时区导致合并错位、设备编码大小写不一致导致关联失败、工单区间使用字符串比较导致边界判断错误。解决方式分别是统一时区、统一编码格式、将区间字段转为datetime后再比较。

完成上述步骤后，采集流就不再是孤立的时序点，而是带有设备身份、工单上下文和维修状态的标签化数据。这份统一时序标签表可以直接用于后续的故障分析、工单效率统计或设备健康度建模，也为下一阶段的标签目录设计提供了可落地的数据基础。