日志数据提取实战:使用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²。这种数据不处理,后面分析全白搭。
清洗步骤一般包括:
- 去除全空行:有些时间戳对应的数据全部为NaN
- 过滤异常值:比如加速度计读数超过±50m/s²的,基本可以判定为异常
- 插值填充:对于少量缺失值,用前后数据线性插值
代码示例:
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]做对比分析。
避坑指南:我曾经直接用最近邻插值重采样,结果在剧烈机动时数据出现了阶梯状跳跃。后来改用线性插值,平滑多了。如果你的数据噪声很大,可以考虑用三次样条插值。
知识体系总览
下面这张图总结了整个数据提取流程。你可以把它当作操作手册,每次处理日志时对照着来:
完整代码整合
把上面几步串起来,就是一个完整的提取脚本。我建议你把它保存成模板,以后每次处理日志直接改文件名就行:
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日志都只需要一行调用。是不是省事多了?
数据提取是日志分析的地基。地基没打好,后面画再漂亮的图、做再复杂的分析都是空中楼阁。把这一步练扎实了,后面调参的时候你会感谢自己的。