📚 课程目录第4章常用数据格式解析与转换

4. 常用数据格式解析与转换

飞控HIL测试中，数据来源非常杂：Simulink/Stateflow仿真侧常导出 MAT，地面站与自定义遥测脚本常写 CSV，NI采集设备与部分台架记录仪输出 TDMS，CAN/CAN FD总线记录仪输出 BLF，整车或台架级测量系统则常用 MDF/MF4。做回放与数据分析时，第一步不是画曲线，而是把这些异构文件统一成“可被同一套脚本读取”的中间结构。本节围绕“格式特征—读写方式—转换策略—批量脚本”展开，重点解决字段对齐、时间基准统一和批量处理三个问题。

一、五种格式的定位与关键差异

MAT 是 MATLAB 的原生二进制容器，适合保存结构体、时间序列、多维数组，读取后可直接得到变量名与维度信息，但跨语言读取需要 h5py（v7.3）或 scipy.io（v7 及以下）。CSV 是纯文本，通用性最好，但无类型信息、无单位、无时间基准，列名容易因导出工具不同而变化。TDMS 是 NI 的层次化二进制格式，按“文件—通道组—通道”组织，天然带属性（单位、采样率、量程），适合高频同步采集。BLF 是 Vector 的二进制日志，核心是报文对象（CAN ID、DLC、数据字节、时间戳），适合总线回放，但不直接表达物理量。MDF（.mf4/.dat）是 ASAM 标准测量格式，按“通道组—通道”组织，带丰富元数据，适合多源同步测量。

二、读写要点与常见坑

MAT 读取时先判断版本：v7.3 用 h5py.File(path,'r')，其余用 scipy.io.loadmat。注意 MATLAB 的列主序与 Python 行主序差异，转置错误会导致时间轴与数据错位。CSV 读取建议显式指定 dtype 与 parse_dates，并检查是否存在 BOM 头、分号分隔、千分位逗号。TDMS 用 nptdms.TdmsFile.read()，通过 channel.properties['unit_string'] 取单位，channel.properties['wf_increment'] 取采样间隔。BLF 用 can.BLFReader 迭代报文，需自行按 DBC 解码为物理量。MDF 用 asammdf.MDF，mdf.get('ChannelName') 返回带 timestamps 与 samples 的信号对象。

时间基准是转换中最容易出错的地方。BLF 时间戳通常是绝对时间（如 Unix 或上电计时），MAT 可能是仿真步长索引，TDMS 是相对采集开始的秒数。统一时必须以“触发时刻 t0”为 0 点，把各源时间戳减去各自 t0，再对齐到同一时间轴。

三、转换策略：先对齐，再合并

推荐采用“逐源解析 → 统一时间轴 → 重采样 → 合并”的四步策略。假设 CAN 报文周期 10 ms、TDMS 采集 1 kHz、MAT 仿真步长 5 ms，统一时间轴可取 1 ms 栅格。对慢速信号用零阶保持（前向填充），对连续物理量用线性插值，对开关量用最近邻。合并时以字典形式写入 signals，键名建议采用“源_通道”命名，避免同名冲突。

import numpy as np, pandas as pd
from asammdf import MDF
from nptdms import TdmsFile
import can

def load_mdf(path):
    mdf = MDF(path)
    out = {}
    for ch in mdf.channels_db:
        sig = mdf.get(ch)
        out[ch] = {"t": sig.timestamps, "v": sig.samples,
                   "unit": sig.unit or ""}
    return out

def load_tdms(path):
    td = TdmsFile.read(path)
    out = {}
    for grp in td.groups():
        for ch in grp.channels():
            dt = ch.properties.get("wf_increment", 1.0)
            out[f"{grp.name}_{ch.name}"] = {
                "t": np.arange(len(ch)) * dt,
                "v": ch[:], "unit": ch.properties.get("unit_string","")}
    return out

def resample(sig, t_grid):
    return np.interp(t_grid, sig["t"], sig["v"])

四、批量处理脚本的组织方式

批量处理的核心是“目录扫描 + 格式分派 + 异常隔离”。建议按扩展名分派到不同 loader，单个文件失败时记录日志并继续，不要中断整批任务。输出统一为 Parquet 或 HDF5，既保留列式读取性能，又能携带 meta 属性。下面给出一个可直接改造的批处理骨架：

from pathlib import Path
import pandas as pd, traceback

LOADERS = {".mf4": load_mdf, ".tdms": load_tdms,
           ".mat": load_mat, ".csv": load_csv, ".blf": load_blf}

def batch_convert(src_dir, dst_dir, t_grid):
    src, dst = Path(src_dir), Path(dst_dir)
    dst.mkdir(parents=True, exist_ok=True)
    log = []
    for f in sorted(src.rglob("*")):
        if f.suffix.lower() not in LOADERS:
            continue
        try:
            raw = LOADERS[f.suffix.lower()](str(f))
            df = pd.DataFrame({"time_s": t_grid})
            for name, sig in raw.items():
                df[name] = resample(sig, t_grid)
            df.to_parquet(dst / f"{f.stem}.parquet", index=False)
            log.append((f.name, "OK", len(df)))
        except Exception as e:
            log.append((f.name, "FAIL", str(e)))
            traceback.print_exc()
    return pd.DataFrame(log, columns=["file","status","detail"])

批量转换后务必做一次“抽样回读校验”：随机抽取 3 个输出文件，检查时间列单调递增、关键通道量程未溢出、单位字段非空。这一步能提前发现 90% 以上的静默错误。

五、结果判读与常见异常

转换完成后，用“时间轴一致性、通道完整性、量程合理性”三项做快速判读。若时间列出现负值或跳变，通常是 t0 未对齐；若某通道全为 NaN，多为列名映射失败或 DBC 未加载；若幅值整体偏大 100 倍，检查单位换算（如 rad 与 deg、g 与 m/s²）。把这些检查写成断言，嵌入批处理脚本末尾，即可让每次转换都自带质量门禁。

格式	推荐读取库	关键元数据	典型风险
MAT	scipy.io / h5py	变量名、维度	列主序转置错误
CSV	pandas	列名、分隔符	BOM、千分位、无单位
TDMS	nptdms	unit_string、wf_increment	组名冲突、采样率缺失
BLF	python-can	CAN ID、DLC、时间戳	未按 DBC 解码
MDF/MF4	asammdf	通道组、单位、量程	多源时间基准不一致

把上述流程固化为团队脚本后，HIL 回放前的数据准备时间可从“逐文件手工导出”压缩到“一条命令批量转换”，并且每次转换都留下可追溯的日志与校验结果，为后续的自动判据和报告生成提供稳定输入。