第1章：数据读取与预处理——CSV/二进制文件解析、时间戳对齐、缺失值处理、异常值检测

大家好，欢迎来到《传感器数据采集与滤波算法精讲》。

我是你们这门课的主讲。做了十几年嵌入式，跟传感器数据打了太多交道。说实话，数据采集这步看着简单，但坑最多。我见过太多项目，算法选得挺好，最后栽在数据预处理上。

这一章，咱们就聊聊数据进来之后，第一件事该做什么。

1.1 数据读取：CSV与二进制文件解析

传感器数据存成什么格式？最常见的就是CSV和二进制。CSV直观，二进制省空间。但各有各的脾气。

1.1.1 CSV文件解析

CSV说白了就是逗号分隔的文本。Python里用pandas.read_csv()最顺手。但我建议你，别直接一把梭。

import pandas as pd

# 我个人习惯先预览前几行
df = pd.read_csv('sensor_data.csv', nrows=5)
print(df.head())

# 确认分隔符，有时候是制表符\t或者分号;
df = pd.read_csv('sensor_data.csv', sep=',', header=0)

# 指定数据类型，避免pandas猜错
dtype_dict = {
    'timestamp': 'str',
    'accel_x': 'float32',
    'accel_y': 'float32',
    'accel_z': 'float32'
}
df = pd.read_csv('sensor_data.csv', dtype=dtype_dict)

1.1.2 二进制文件解析

二进制文件，嗯，这里要注意。它没有自描述性，你必须知道协议。

我曾经接手过一个项目，前任工程师离职了，二进制文件格式没留文档。我对着十六进制看了三天，才把数据结构猜出来。你想想看，这多耽误事。

常见的二进制格式有：

• 固定长度记录：每条数据长度相同，直接按字节偏移读取
• 可变长度记录：每条数据前有个长度字段
• 带时间戳的原始帧：通常包含帧头、数据体、校验和

import struct
import numpy as np

def parse_binary_sensor(file_path):
    """
    解析固定长度的传感器二进制数据
    假设每条记录：4字节时间戳 + 4字节float x + 4字节float y + 4字节float z
    """
    records = []
    record_size = 16  # 4 + 4 + 4 + 4
    
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(record_size)
            if len(chunk) < record_size:
                break
            # 解包：I 无符号整型，3f 三个float
            timestamp, x, y, z = struct.unpack('<I3f', chunk)
            records.append((timestamp, x, y, z))
    
    return np.array(records)

1.2 时间戳对齐——多传感器融合的第一步

多传感器系统里，每个传感器有自己的时钟。GPS可能用UTC，IMU用本地晶振，摄像头用帧率。时间戳不对齐，后面所有算法都是白搭。

说白了，时间戳对齐就是让所有数据站在同一个时间轴上。

1.2.1 时间戳归一化

先把所有时间戳转成统一格式。我个人习惯用Unix时间戳（秒或毫秒）。

import pandas as pd

# 假设CSV里时间戳是字符串 '2024-01-15 10:30:25.123'
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 转成Unix时间戳（秒）
df['timestamp_unix'] = df['timestamp'].astype('int64') / 1e9

1.2.2 插值对齐

两个传感器采样率不同怎么办？比如IMU是100Hz，GPS是10Hz。你需要把GPS数据插值到IMU的时间轴上。

import numpy as np

def align_timestamps(source_time, source_data, target_time):
    """
    将source_data插值到target_time的时间轴上
    """
    aligned = np.interp(target_time, source_time, source_data)
    return aligned

# 使用示例
imu_time = np.array([0.0, 0.01, 0.02, 0.03, ...])  # 100Hz
gps_time = np.array([0.0, 0.1, 0.2, 0.3, ...])     # 10Hz
gps_lat = np.array([...])

# 把GPS纬度插值到IMU时间轴上
gps_lat_aligned = align_timestamps(gps_time, gps_lat, imu_time)

1.3 缺失值处理——数据不完整怎么办？

传感器掉线、通信丢包、存储故障，都会导致数据缺失。你想想看，一个时间序列里突然少了几行，算法直接崩了。

常见的缺失值处理方法：

方法	适用场景	优点	缺点
直接删除	缺失比例很小（<5%）	简单快速	可能丢失有用信息
前向填充	缓慢变化的信号	保持连续性	引入滞后
线性插值	近似线性变化	平滑过渡	对突变不友好
样条插值	非线性变化	更自然	计算量大
模型预测	有规律的数据	精度高	实现复杂

import pandas as pd
import numpy as np

# 检测缺失值
print(df.isnull().sum())

# 方法1：直接删除
df_clean = df.dropna()

# 方法2：前向填充
df['accel_x'].fillna(method='ffill', inplace=True)

# 方法3：线性插值
df['accel_x'].interpolate(method='linear', inplace=True)

# 方法4：限制连续缺失长度
# 如果连续缺失超过5个点，就不插值了
df['accel_x'] = df['accel_x'].interpolate(limit=5, limit_direction='both')

1.4 异常值检测——揪出捣乱的数据点

传感器被撞了一下、电磁干扰、ADC饱和，都会产生异常值。这些点如果不处理，滤波算法会被带偏。

异常值检测，说白了就是找那些「不合理」的点。

1.4.1 基于统计的方法

最常用的是3σ原则和四分位距法。

import numpy as np

def detect_outliers_3sigma(data, threshold=3):
    """
    基于3σ原则检测异常值
    """
    mean = np.mean(data)
    std = np.std(data)
    lower = mean - threshold * std
    upper = mean + threshold * std
    outliers = (data < lower) | (data > upper)
    return outliers

def detect_outliers_iqr(data, multiplier=1.5):
    """
    基于四分位距检测异常值
    """
    q1 = np.percentile(data, 25)
    q3 = np.percentile(data, 75)
    iqr = q3 - q1
    lower = q1 - multiplier * iqr
    upper = q3 + multiplier * iqr
    outliers = (data < lower) | (data > upper)
    return outliers

1.4.2 基于物理约束的方法

这个方法更靠谱。你知道传感器的量程，超出量程的就是异常。

def detect_outliers_physical(data, min_val, max_val):
    """
    基于物理量程检测异常值
    """
    outliers = (data < min_val) | (data > max_val)
    return outliers

# 例如：加速度计量程±16g
accel_outliers = detect_outliers_physical(df['accel_x'], -16.0, 16.0)

1.4.3 基于变化率的方法

有些异常值在量程内，但变化率离谱。比如温度传感器，一秒内跳了50度，这肯定有问题。

def detect_outliers_rate(data, max_rate):
    """
    基于变化率检测异常值
    """
    diff = np.abs(np.diff(data))
    # 第一个点没法判断，默认正常
    outliers = np.concatenate(([False], diff > max_rate))
    return outliers

# 温度变化率不超过5度/秒
temp_outliers = detect_outliers_rate(df['temperature'], 5.0)

1.5 实战建议——数据预处理的流水线

好了，知识点讲完了。咱们串一下，一个完整的数据预处理流水线应该怎么做。

1. 读取原始数据：CSV用pandas，二进制用struct
2. 时间戳归一化：统一转成Unix时间戳
3. 多传感器对齐：以最高采样率的传感器为基准，插值对齐
4. 缺失值处理：先看缺失比例，小则删除，大则插值
5. 异常值检测：物理约束 + 统计方法 + 变化率，三重保险
6. 数据保存：处理好的数据存成新文件，别覆盖原始数据

嗯，这一章就到这里。下一章咱们聊聊时域滤波，从最简单的移动平均开始。

记住一句话：数据预处理做得好，后面滤波事半功倍。预处理偷懒，后面全是坑。