日志数据提取实战:使用Python读取ULG日志

说实话,很多做飞控调优的朋友,第一步就卡在了数据提取上。

手里拿着飞控的ULG日志文件,却不知道怎么把里面的姿态角、角速度这些关键数据弄出来。我刚开始做飞控时也这样,对着二进制文件发愣。后来摸清了门道,发现其实就那么几步。

今天咱们就把这事彻底搞定。

ULG日志文件结构速览

ULG是PX4飞控的二进制日志格式。它不像CSV那样直接能打开看,但好处是体积小、存储效率高。

一个ULG文件内部,说白了就是一系列消息包的集合。每个消息包包含:

  • 消息头:标识这个消息是什么类型(比如IMU数据、GPS数据)
  • 时间戳:微秒级精度,记录数据采集的时刻
  • 消息体:真正的数据内容,比如三轴角速度、加速度计读数

我习惯把ULG想象成一个“数据集装箱”。里面整整齐齐码放着不同种类的数据箱子,每个箱子都贴着标签和时间。我们要做的,就是找到需要的箱子,把里面的东西拿出来。

核心要点:ULG日志是二进制消息流,不是文本文件。必须用专门的解析工具来读取。

环境准备:装好pyulog

Python生态里有个现成的库叫pyulog,专门干这个活。安装很简单:

pip install pyulog

嗯,这里要注意。如果你用的是Python 3.8以上版本,直接装就行。我遇到过有人用Python 2.7来装,结果各种报错——现在都什么年代了,赶紧升级吧。

第一步:读取ULG文件

代码其实就几行。咱们直接上手:

from pyulog import ULog

# 加载日志文件
ulog = ULog('flight_log_001.ulg')

# 看看里面有哪些数据主题
print(ulog.data_list)

运行后,你会看到类似这样的输出:

['sensor_accel', 'sensor_gyro', 'vehicle_attitude', 'vehicle_gps_position', ...]

这些就是飞控记录的所有数据主题。每个主题对应一类传感器或状态数据。

小技巧:如果你不确定日志里有哪些主题,可以用ulog.msg_dict查看更详细的信息,包括每个主题包含的字段名和数据类型。

第二步:提取时间戳、姿态角、角速度、加速度计数据

咱们这次要提取四个核心数据:时间戳、姿态角、角速度、加速度计。对应的主题分别是:

数据类别 ULG主题名 关键字段
时间戳 所有主题都包含 timestamp
姿态角 vehicle_attitude roll, pitch, yaw
角速度 sensor_gyro x, y, z
加速度计 sensor_accel x, y, z

提取代码也很直接:

# 获取姿态数据
attitude = ulog.get_dataset('vehicle_attitude')
time_att = attitude.data['timestamp']
roll = attitude.data['roll']
pitch = attitude.data['pitch']
yaw = attitude.data['yaw']

# 获取角速度数据
gyro = ulog.get_dataset('sensor_gyro')
time_gyro = gyro.data['timestamp']
gyro_x = gyro.data['x']
gyro_y = gyro.data['y']
gyro_z = gyro.data['z']

# 获取加速度计数据
accel = ulog.get_dataset('sensor_accel')
time_accel = accel.data['timestamp']
accel_x = accel.data['x']
accel_y = accel.data['y']
accel_z = accel.data['z']

你看,是不是很简单?每个数据集都是一个字典,data字段里存着所有数据列。

注意:不同主题的时间戳是独立的。姿态数据可能每秒100条,而加速度计数据可能每秒800条。直接合并会出问题,后面我们会讲怎么对齐。

第三步:数据清洗——处理缺失值和异常值

日志里难免有脏数据。我遇到过几次,飞控在剧烈震动时,加速度计数据突然跳出一个离谱的值,比如1000m/s²。这种数据不处理,后面分析全白搭。

清洗步骤一般包括:

  1. 去除全空行:有些时间戳对应的数据全部为NaN
  2. 过滤异常值:比如加速度计读数超过±50m/s²的,基本可以判定为异常
  3. 插值填充:对于少量缺失值,用前后数据线性插值

代码示例:

import numpy as np

def clean_sensor_data(time, data, max_value=50.0):
    # 转为numpy数组
    time = np.array(time)
    data = np.array(data)
    
    # 标记异常值
    mask = np.all(np.abs(data) < max_value, axis=1)
    
    # 去除异常值和对应的时刻
    time_clean = time[mask]
    data_clean = data[mask]
    
    return time_clean, data_clean

# 清洗加速度计数据
time_accel_clean, accel_clean = clean_sensor_data(
    time_accel, 
    np.column_stack((accel_x, accel_y, accel_z))
)

我曾经遇到过一个案例,飞控日志里角速度数据每隔几秒就跳一个零值。后来发现是SD卡写入时丢包了。用这种清洗方法,把零值附近的数据都标记出来,问题一目了然。

第四步:重采样——统一时间基准

不同传感器的采样率不一样。加速度计通常是800Hz,陀螺仪也是800Hz,但姿态解算只有100Hz。要分析它们之间的关系,必须重采样到统一的时间基准。

我个人习惯把所有数据重采样到100Hz,因为姿态控制环的更新频率通常就是100Hz。这样分析出来的结果更贴近实际控制效果。

from scipy import interpolate

def resample_to_uniform(time_src, data_src, target_freq=100.0):
    # 生成目标时间轴
    t_start = time_src[0]
    t_end = time_src[-1]
    t_target = np.arange(t_start, t_end, 1e6 / target_freq)  # 微秒单位
    
    # 线性插值
    f_interp = interpolate.interp1d(time_src, data_src, axis=0, 
                                    bounds_error=False, fill_value='extrapolate')
    data_resampled = f_interp(t_target)
    
    return t_target, data_resampled

# 将所有数据重采样到100Hz
t_uniform, accel_uniform = resample_to_uniform(time_accel_clean, accel_clean)
_, gyro_uniform = resample_to_uniform(time_gyro, np.column_stack((gyro_x, gyro_y, gyro_z)))
_, attitude_uniform = resample_to_uniform(time_att, np.column_stack((roll, pitch, yaw)))

重采样之后,所有数据的时间轴就对齐了。你可以直接拿accel_uniform[i]和attitude_uniform[i]做对比分析。

避坑指南:我曾经直接用最近邻插值重采样,结果在剧烈机动时数据出现了阶梯状跳跃。后来改用线性插值,平滑多了。如果你的数据噪声很大,可以考虑用三次样条插值。

知识体系总览

下面这张图总结了整个数据提取流程。你可以把它当作操作手册,每次处理日志时对照着来:

ULG日志数据提取与预处理流程 读取ULG文件 提取时间戳、姿态角、角速度、加速度计 数据清洗:去异常、插值填充 重采样到统一时间基准 对齐后的结构化数据 输入 .ulg 二进制日志文件 来自PX4飞控 输出 时间对齐的Numpy数组 可直接用于分析绘图 关键工具 pyulog + numpy + scipy 三件套搞定

完整代码整合

把上面几步串起来,就是一个完整的提取脚本。我建议你把它保存成模板,以后每次处理日志直接改文件名就行:

import numpy as np
from pyulog import ULog
from scipy import interpolate

def extract_and_process_ulog(filepath, target_freq=100.0):
    # 1. 读取
    ulog = ULog(filepath)
    
    # 2. 提取原始数据
    att = ulog.get_dataset('vehicle_attitude')
    gyro = ulog.get_dataset('sensor_gyro')
    accel = ulog.get_dataset('sensor_accel')
    
    # 3. 清洗加速度计(示例:过滤超过±50m/s²的数据)
    accel_raw = np.column_stack((accel.data['x'], accel.data['y'], accel.data['z']))
    mask = np.all(np.abs(accel_raw) < 50.0, axis=1)
    time_accel_clean = np.array(accel.data['timestamp'])[mask]
    accel_clean = accel_raw[mask]
    
    # 4. 重采样到统一时间轴
    t_start = max(att.data['timestamp'][0], gyro.data['timestamp'][0], time_accel_clean[0])
    t_end = min(att.data['timestamp'][-1], gyro.data['timestamp'][-1], time_accel_clean[-1])
    t_uniform = np.arange(t_start, t_end, 1e6 / target_freq)
    
    def resample(t_src, data_src):
        f = interpolate.interp1d(t_src, data_src, axis=0, 
                                 bounds_error=False, fill_value='extrapolate')
        return f(t_uniform)
    
    # 返回对齐后的数据
    return {
        'time': t_uniform,
        'attitude': resample(att.data['timestamp'], 
                             np.column_stack((att.data['roll'], att.data['pitch'], att.data['yaw']))),
        'gyro': resample(gyro.data['timestamp'],
                         np.column_stack((gyro.data['x'], gyro.data['y'], gyro.data['z']))),
        'accel': resample(time_accel_clean, accel_clean)
    }

# 使用示例
data = extract_and_process_ulog('flight_log_001.ulg')
print(f"提取完成,共 {len(data['time'])} 个时间点")

你想想看,有了这个函数,以后处理任何ULG日志都只需要一行调用。是不是省事多了?

数据提取是日志分析的地基。地基没打好,后面画再漂亮的图、做再复杂的分析都是空中楼阁。把这一步练扎实了,后面调参的时候你会感谢自己的。