python解析mdf或mf4文件利器之asammdf用法

pip install asammdf

from asammdf import MDF

f = r"xxx.mdf"
mdf = MDF(f)
if '信号名' in mdf:
    #注意，一个mdf里可能有重名的信号，这个时候要加上group_index和channel_index，比如mdf.get('信号名',3,2)，至于后面两个索引号怎么获取，这个就需要看后面讲的的channels_db了
    signal = mdf.get('信号名')
    data = signal.samples
    timestamps = signal.timestamps

chn_db = mdf.channels_db

#代码接上
df = mdf.to_dataframe()

for group_index, (virtual_group_index, virtual_group) in enumerate(
            mdf_ins.virtual_groups.items()
        ):
            if virtual_group.cycles_nr == 0 and empty_channels == "skip":
                continue

            channels = [
                (None, gp_index, ch_index)
                for gp_index, channel_indexes in mdf_ins.included_channels(
                    virtual_group_index
                )[virtual_group_index].items()
                for ch_index in channel_indexes
                if ch_index != mdf_ins.masters_db.get(gp_index, None)
            ]

            old_signals = [
                signal
                for signal in mdf_ins.select(
                    channels, raw=True, copy_master=False, validate=False
                )
            ]

from asammdf import MDF
mdf1 = MDF(r"f1.mf4")
mdf2 = MDF(r"f2.mf4")

mdf = MDF.concatenate([mdf1,mdf2])

mdf.export(format,filename)

# map signals
xs = np.linspace(-1, 1, 50)
ys = np.linspace(-1, 1, 50)
X, Y = np.meshgrid(xs, ys)
vals = np.linspace(0, 180.0 / np.pi, 100)
phi = np.ones((len(vals), 50, 50), dtype=np.float64)
for i, val in enumerate(vals):
    phi[i] *= val
R = 1 - np.sqrt(X**2 + Y**2)
samples = np.cos(2 * np.pi * X + phi) * R

timestamps = np.arange(0, 2, 0.02)

s_map = Signal(
    samples=samples, timestamps=timestamps, name="Variable Map Signal", unit="dB"
)
s_map.plot()

import numpy as np

from asammdf import Signal

# create 3 Signal objects with different time stamps

# unit8 with 100ms time raster
timestamps = np.array([0.1 * t for t in range(5)], dtype=np.float32)
s_uint8 = Signal(
    samples=np.array([t for t in range(5)], dtype=np.uint8),
    timestamps=timestamps,
    name="Uint8_Signal",
    unit="u1",
)

# int32 with 50ms time raster
timestamps = np.array([0.05 * t for t in range(10)], dtype=np.float32)
s_int32 = Signal(
    samples=np.array(list(range(-500, 500, 100)), dtype=np.int32),
    timestamps=timestamps,
    name="Int32_Signal",
    unit="i4",
)

# float64 with 300ms time raster
timestamps = np.array([0.3 * t for t in range(3)], dtype=np.float32)
s_float64 = Signal(
    samples=np.array(list(range(2000, -1000, -1000)), dtype=np.int32),
    timestamps=timestamps,
    name="Float64_Signal",
    unit="f8",
)

prod = s_float64 * s_uint8
prod.name = "Uint8_Signal * Float64_Signal"
prod.unit = "*"
prod.plot()

pow2 = s_uint8**2
pow2.name = "Uint8_Signal ^ 2"
pow2.unit = "u1^2"
pow2.plot()

allsum = s_uint8 + s_int32 + s_float64
allsum.name = "Uint8_Signal + Int32_Signal + Float64_Signal"
allsum.unit = "+"
allsum.plot()

# inplace operations
pow2 *= -1
pow2.name = "- Uint8_Signal ^ 2"
pow2.plot()

# cut signal
s_int32.plot()
cut_signal = s_int32.cut(start=0.2, stop=0.35)
cut_signal.plot()

from asammdf import MDF
from utils import get_lines_from_txt
f = r"C:\Users\c-master1\Desktop\下载\AnalysisTools\YIQI_2022-08-06_14-37-50.mf4"
f1 = r"C:\Users\c-master1\Downloads\2022-11-21_16-59-29.mf4"
f2 = r"C:\Users\c-master1\Downloads\Recorder_2022-11-17_14-55-00.MDF"
clist_f = r"C:\Users\c-master1\Downloads\数据分析用信号列表(2)(2).txt"

channel_list = ['ABV_Ki_Out','ACsurge_Flag_b','csABPV_posVlv_perc']
channel_list = get_lines_from_txt(clist_f)
mdf = MDF(f2)
mdf = mdf.filter(channel_list)
signals = mdf.select(channel_list)
max = 0
name = ""
for sig in signals:
    if len(sig.samples) > max:
        max = len(sig.samples)
        name = sig.name
mdf1 = mdf.resample(raster=name)
df1 = mdf1.to_dataframe()
df1

mdf.configure(float_interpolation=0,integer_interpolation=0)

# -*- coding: utf-8 -*-
"""
*asammdf* MDF usage example
"""
import numpy as np

from asammdf import MDF, Signal

# create 3 Signal objects

timestamps = np.array([0.1, 0.2, 0.3, 0.4, 0.5], dtype=np.float32)

# unit8
s_uint8 = Signal(
    samples=np.array([0, 1, 2, 3, 4], dtype=np.uint8),
    timestamps=timestamps,
    name="Uint8_Signal",
    unit="u1",
)
# int32
s_int32 = Signal(
    samples=np.array([-20, -10, 0, 10, 20], dtype=np.int32),
    timestamps=timestamps,
    name="Int32_Signal",
    unit="i4",
)

# float64
s_float64 = Signal(
    samples=np.array([-20, -10, 0, 10, 20], dtype=np.float64),
    timestamps=timestamps,
    name="Float64_Signal",
    unit="f8",
)

# create empty MDf version 4.00 file
with MDF(version="4.10") as mdf4:

    # append the 3 signals to the new file
    signals = [s_uint8, s_int32, s_float64]
    mdf4.append(signals, comment="Created by Python")

    # save new file
    mdf4.save("my_new_file.mf4", overwrite=True)

    # convert new file to mdf version 3.10
    mdf3 = mdf4.convert(version="3.10")
    print(mdf3.version)

    # get the float signal
    sig = mdf3.get("Float64_Signal")
    print(sig)

    # cut measurement from 0.3s to end of measurement
    mdf4_cut = mdf4.cut(start=0.3)
    mdf4_cut.get("Float64_Signal").plot()

    # cut measurement from start of measurement to 0.4s
    mdf4_cut = mdf4.cut(stop=0.45)
    mdf4_cut.get("Float64_Signal").plot()

    # filter some signals from the file
    mdf4 = mdf4.filter(["Int32_Signal", "Uint8_Signal"])

    # save using zipped transpose deflate blocks
    mdf4.save("out.mf4", compression=2, overwrite=True)

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'a': np.arrange(10),
    'b': np.arrange(10),
    'c': np.arrange(10),
})
mdf_f = MDF()
mdf_f.append(df)
mdf_f.save("xx.mf4",overwrite=True)