4. 基于时间戳的数据对齐与融合

任务说明

本案例面向健康管理PHM系统中的多源异构传感器数据，解决因采样率不同、采集起始时间不同导致的数据难以联合分析的问题。目标是将振动（高频）、温度（低频）、压力（低频）三类数据基于时间戳对齐，并融合为统一格式的数据集，便于后续特征提取与故障预测。

数据源定义

为便于独立复现，本案例直接构造三类传感器在相同时间窗口内的原始采集数据，模拟实际采集场景：振动以1000Hz采样，温度和压力以10Hz采样，且三者起始时间存在偏移。原始数据格式为CSV，每行包含时间戳（Unix毫秒）与测量值。

实现方案

采用最近邻时间戳匹配策略：以低频数据（温度、压力）的时间戳为基准，为每个低频时间点查找最近的振动数据点，实现对齐。融合后的统一格式包含：时间戳、温度值、压力值、振动值、振动时间戳偏移量（用于质量评估）。代码使用Pandas实现，完整自洽，可直接运行。

代码实现

import pandas as pd
import numpy as np

# ========== 1. 构造模拟原始数据（模拟实际采集） ==========
def generate_raw_data():
    """生成三类传感器原始CSV数据（模拟），实际使用时替换为文件读取"""
    # 振动：1000Hz，起始0ms，持续2秒
    vib_t = np.arange(0, 2000, 1.0)  # 0~1999ms
    vib_val = 5.0 * np.sin(2 * np.pi * 50 * vib_t / 1000) + 0.5 * np.random.randn(len(vib_t))
    vib_df = pd.DataFrame({'timestamp_ms': vib_t.astype(np.int64), 'vibration_g': vib_val})
    
    # 温度：10Hz，起始150ms，持续2秒
    temp_t = np.arange(150, 2150, 100.0)  # 150~2050ms
    temp_val = 65.0 + 2.0 * np.sin(2 * np.pi * 0.2 * temp_t / 1000) + 0.1 * np.random.randn(len(temp_t))
    temp_df = pd.DataFrame({'timestamp_ms': temp_t.astype(np.int64), 'temperature_c': temp_val})
    
    # 压力：10Hz，起始300ms，持续2秒
    pres_t = np.arange(300, 2300, 100.0)  # 300~2200ms
    pres_val = 101.3 + 0.8 * np.sin(2 * np.pi * 0.1 * pres_t / 1000) + 0.05 * np.random.randn(len(pres_t))
    pres_df = pd.DataFrame({'timestamp_ms': pres_t.astype(np.int64), 'pressure_kpa': pres_val})
    
    return vib_df, temp_df, pres_df

# ========== 2. 时间戳对齐与融合 ==========
def align_and_fuse(vib_df, temp_df, pres_df):
    """
    以低频数据（温度、压力）时间戳为基准，最近邻匹配振动数据
    返回统一格式的融合DataFrame
    """
    # 将振动时间戳转为numpy数组用于最近邻搜索
    vib_times = vib_df['timestamp_ms'].values
    vib_values = vib_df['vibration_g'].values
    
    # 以温度时间戳为主基准（压力通过merge对齐）
    # 先对温度与压力做时间戳最近邻对齐（两者均为10Hz，但起始不同）
    # 此处采用：以温度时间戳为基准，压力最近邻匹配
    pres_times = pres_df['timestamp_ms'].values
    pres_values = pres_df['pressure_kpa'].values
    
    aligned_rows = []
    for temp_row in temp_df.itertuples():
        ts = temp_row.timestamp_ms
        
        # 查找最近的压力值
        pres_idx = np.argmin(np.abs(pres_times - ts))
        pres_val = pres_values[pres_idx]
        pres_offset = pres_times[pres_idx] - ts
        
        # 查找最近的振动值
        vib_idx = np.argmin(np.abs(vib_times - ts))
        vib_val = vib_values[vib_idx]
        vib_offset = vib_times[vib_idx] - ts
        
        aligned_rows.append({
            'timestamp_ms': ts,
            'temperature_c': round(temp_row.temperature_c, 3),
            'pressure_kpa': round(pres_val, 3),
            'vibration_g': round(vib_val, 4),
            'vib_time_offset_ms': int(vib_offset),
            'pres_time_offset_ms': int(pres_offset)
        })
    
    fused_df = pd.DataFrame(aligned_rows)
    return fused_df

# ========== 3. 主流程 ==========
if __name__ == "__main__":
    # 生成模拟数据（实际使用时替换为 pd.read_csv('vib.csv') 等）
    vib_df, temp_df, pres_df = generate_raw_data()
    
    # 执行对齐融合
    fused_df = align_and_fuse(vib_df, temp_df, pres_df)
    
    # 输出结果
    print("融合后数据集形状:", fused_df.shape)
    print("\n前5行数据:")
    print(fused_df.head())
    
    # 保存为统一格式CSV
    fused_df.to_csv('fused_dataset.csv', index=False)
    print("\n已保存至 fused_dataset.csv")
    
    # 质量评估：时间偏移统计
    print("\n时间偏移统计（毫秒）:")
    print(fused_df[['vib_time_offset_ms', 'pres_time_offset_ms']].describe())

运行与验证

运行方式：将上述代码保存为data_fusion.py，在安装有Pandas和NumPy的Python环境中执行python data_fusion.py。

验证方法：观察控制台输出，确认：

• 融合数据集行数与温度数据行数一致（约20行，对应2秒@10Hz）；
• 每行同时包含温度、压力、振动值，且时间戳为温度采集时刻；
• vib_time_offset_ms值不超过1ms（振动采样间隔为1ms，最近邻匹配最大偏移0.5ms）；
• pres_time_offset_ms值不超过50ms（压力采样间隔100ms，最近邻最大偏移50ms）；
• 生成fused_dataset.csv文件，可用Excel或文本编辑器打开检查列完整性。

判定标准：程序无异常退出，输出数据形状符合预期，时间偏移量在理论范围内，即视为对齐融合成功。若实际使用真实传感器数据，仅需将generate_raw_data()替换为对应的CSV读取代码，并确保时间戳列为毫秒Unix时间戳即可。