如何通过Python/asammdf最小化MDF文件体积？技术实践建议咨询

你的问题很典型——从DataFrame转MDF时，默认处理方式会丢失原始MDF的高效存储特性，直接导致体积暴增。下面是几个核心优化方向，完全满足你“不下采样、保留所有通道”的需求：

1. 严格复用原始通道的数据类型 #

DataFrame在加载时经常会自动扩张数据类型（比如把int8转成int64，float32转成float64），这会直接让存储体积翻倍甚至更多。解决方法是从原始MDF对象中提取每个通道的原生数据类型，创建Signal时强制匹配：

# 替代原来的DataFrame加载方式，直接从原始MDF获取通道信息
mdf_obj = MDF(mdf_file_path)
new_mdf = MDF()

for channel in mdf_obj.channels:
    # 获取原始通道的数据类型、时间戳和样本
    original_dtype = channel.data_type
    timestamps = channel.timestamps
    samples = channel.samples
    
    # 严格按原始类型创建Signal
    sig = Signal(
        samples=samples.astype(original_dtype),
        timestamps=timestamps,
        name=channel.name,
        data_type=original_dtype
    )
    new_mdf.append(sig)

2. 按采样率分组存储，共享时间戳 #

原始MDF会自动将相同采样率的通道归为一组，共享一套时间戳；但从DataFrame转的时候，每个通道都会单独存储全量时间戳，这是体积暴增的关键原因之一。我们可以按采样率对通道分组，复用时间戳：

from collections import defaultdict

mdf_obj = MDF(mdf_file_path)
new_mdf = MDF()
# 按采样率分组通道（处理浮点精度问题，适当取整）
channels_by_rate = defaultdict(list)

for channel in mdf_obj.channels:
    sample_rate = round(channel.sample_rate, 6) if channel.sample_rate else 0
    channels_by_rate[sample_rate].append(channel)

for rate, channels in channels_by_rate.items():
    # 同采样率通道时间戳一致，复用第一组的时间戳
    shared_timestamps = channels[0].timestamps
    for channel in channels:
        sig = Signal(
            samples=channel.samples.astype(channel.data_type),
            timestamps=shared_timestamps,
            name=channel.name,
            data_type=channel.data_type
        )
        new_mdf.append(sig)

3. 启用MDF4的压缩功能 #

asammdf默认保存时可能未启用压缩，而MDF4支持zlib压缩，能大幅减小体积。保存时指定压缩参数：

import zlib

# 保存时启用最高级别压缩
new_mdf.save(new_mdf_output_path, compression=zlib.Z_BEST_COMPRESSION)

4. 针对固定值字符串通道的极致优化 #

对于所有值都相同的字符串通道，不需要存储每个样本的重复值，asammdf支持标记为常量信号，内部会只存储一次值：

for channel in mdf_obj.channels:
    # 判断是否为字符串类型且所有值相同
    if channel.data_type == 'string':
        unique_vals = np.unique(channel.samples)
        if len(unique_vals) == 1:
            # 创建常量信号，仅传入一个样本值
            sig = Signal(
                samples=np.array([unique_vals[0]], dtype='U'),
                timestamps=channel.timestamps,
                name=channel.name,
                data_type='string',
                is_constant=True
            )
            new_mdf.append(sig)
            continue
    # 其他通道按正常逻辑处理
    sig = Signal(
        samples=channel.samples.astype(channel.data_type),
        timestamps=channel.timestamps,
        name=channel.name,
        data_type=channel.data_type
    )
    new_mdf.append(sig)

整合所有优化的完整代码 #

把上面的优化点整合起来，最终的代码如下：

import numpy as np
import zlib
from collections import defaultdict
from asammdf import MDF, Signal

mdf_file_path = r"C:\Users\user\Desktop\input.mdf"
new_mdf_output_path = r"C:\Users\user\Desktop\output.mdf"

mdf_obj = MDF(mdf_file_path)
new_mdf = MDF()
channels_by_rate = defaultdict(list)

# 按采样率分组通道
for channel in mdf_obj.channels:
    sample_rate = round(channel.sample_rate, 6) if channel.sample_rate else 0
    channels_by_rate[sample_rate].append(channel)

for rate, channels in channels_by_rate.items():
    shared_timestamps = channels[0].timestamps
    for channel in channels:
        # 处理固定值字符串通道
        if channel.data_type == 'string':
            unique_vals = np.unique(channel.samples)
            if len(unique_vals) == 1:
                sig = Signal(
                    samples=np.array([unique_vals[0]], dtype='U'),
                    timestamps=shared_timestamps,
                    name=channel.name,
                    data_type='string',
                    is_constant=True
                )
                new_mdf.append(sig)
                continue
        
        # 其他通道严格复用原始类型
        sig = Signal(
            samples=channel.samples.astype(channel.data_type),
            timestamps=shared_timestamps,
            name=channel.name,
            data_type=channel.data_type
        )
        new_mdf.append(sig)

# 启用最高压缩保存
new_mdf.save(new_mdf_output_path, compression=zlib.Z_BEST_COMPRESSION)

这些优化组合下来，生成的MDF体积应该会和原始文件接近甚至更小，同时完全保留所有通道和数据精度。

内容的提问来源于stack exchange，提问作者jB777