API¶
MDF¶
This class acts as a proxy for the MDF2, MDF3 and MDF4 classes. All attribute access is delegated to the underlying _mdf attribute (MDF2, MDF3 or MDF4 object). See MDF3 and MDF4 for available extra methods (MDF2 and MDF3 share the same implementation).
An empty MDF file is created if the name argument is not provided. If the name argument is provided then the file must exist in the filesystem, otherwise an exception is raised.
The best practice is to use the MDF as a context manager. This way all resources are released correctly in case of exceptions.
with MDF(r'test.mdf') as mdf_file:
# do something
- class asammdf.mdf.MDF(name: InputType | None = None, version: str = '4.10', channels: list[str] | None = None, **kwargs)[source]¶
Unified access to MDF v3 and v4 files. Underlying _mdf’s attributes and methods are linked to the MDF object via setattr. This is done to expose them to the user code and for performance considerations.
- Parameters:
- namestring | BytesIO | zipfile.ZipFile | bz2.BZ2File | gzip.GzipFile
mdf file name (if provided it must be a real file name), file-like object or compressed file opened as Python object
Changed in version 6.2.0: added support for zipfile.ZipFile, bz2.BZ2File and gzip.GzipFile
- versionstring
mdf file version from (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’, ‘3.30’, ‘4.00’, ‘4.10’, ‘4.11’, ‘4.20’); default ‘4.10’. This argument is only used for MDF objects created from scratch; for MDF objects created from a file the version is set to file version
- channels (None)iterable
channel names that will used for selective loading. This can dramatically improve the file loading time. Default None -> load all channels
New in version 6.1.0.
Changed in version 6.3.0: make the default None
- use_display_names (**kwargs)bool
keyword only argument: for MDF4 files parse the XML channel comment to search for the display name; XML parsing is quite expensive so setting this to False can decrease the loading times very much; default False
- remove_source_from_channel_names (**kwargs)bool
remove source from channel names (“SpeedXCP3” -> “Speed”)
- copy_on_get (**kwargs)bool
copy arrays in the get method; default True
- expand_zippedfile (**kwargs)bool
only for bz2.BZ2File and gzip.GzipFile, load the file content into a BytesIO before parsing (avoids the huge performance penalty of doing random reads from the zipped file); default True
- raise_on_multiple_occurrences (**kwargs)bool
raise exception when there are multiple channel occurrences in the file and the get call is ambiguous; default True
New in version 7.0.0.
- temporary_folder (**kwargs)str | pathlib.Path
folder to use for temporary files
New in version 7.0.0.
Examples
>>> mdf = MDF(version='3.30') # new MDF object with version 3.30 >>> mdf = MDF('path/to/file.mf4') # MDF loaded from file >>> mdf = MDF(BytesIO(data)) # MDF from file contents >>> mdf = MDF(zipfile.ZipFile('data.zip')) # MDF creating using the first valid MDF from archive >>> mdf = MDF(bz2.BZ2File('path/to/data.bz2', 'rb')) # MDF from bz2 object >>> mdf = MDF(gzip.GzipFile('path/to/data.gzip', 'rb')) # MDF from gzip object
- cleanup_timestamps(minimum: float, maximum: float, exp_min: int = -15, exp_max: int = 15, version: str | None = None, progress=None) MDF [source]¶
convert MDF to other version
New in version 5.22.0.
- Parameters:
- minimumfloat
minimum plausible time stamp
- maximumfloat
maximum plausible time stamp
- exp_min (-15)int
minimum plausible exponent used for the time stamps float values
- exp_max (15)int
maximum plausible exponent used for the time stamps float values
- versionstr
new mdf file version from (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’, ‘3.30’, ‘4.00’, ‘4.10’, ‘4.11’, ‘4.20’); default the same as the input file
- Returns:
- outMDF
new MDF object
- static concatenate(files: Sequence[MDF | InputType], version: str = '4.10', sync: bool = True, add_samples_origin: bool = False, direct_timestamp_continuation: bool = False, progress=None, **kwargs) MDF [source]¶
concatenates several files. The files must have the same internal structure (same number of groups, and same channels in each group).
The order of the input files is always preserved, only the samples timestamps are influenced by the
sync
argument.- Parameters:
- fileslist | tuple
list of MDF file names or MDF, zipfile.ZipFile, bz2.BZ2File or gzip.GzipFile instances
..versionchanged:: 6.2.0
added support for zipfile.ZipFile, bz2.BZ2File and gzip.GzipFile
- versionstr
merged file version
- syncbool
sync the files based on the start of measurement, default True. The order of the input files is preserved, only the samples timestamps are influenced by this argument
- add_samples_originbool
option to create a new “__samples_origin” channel that will hold the index of the measurement from where each timestamp originated
- direct_timestamp_continuation (False)bool
the time stamps from the next file will be added right after the last time stamp from the previous file; default False
..versionadded:: 6.0.0
- kwargs
use_display_names (False) : bool
- Returns:
- concatenateMDF
new MDF object with concatenated channels
- Raises:
- MdfExceptionif there are inconsistencies between the files
Examples
>>> conc = MDF.concatenate( [ 'path/to/file.mf4', MDF(BytesIO(data)), MDF(zipfile.ZipFile('data.zip')), MDF(bz2.BZ2File('path/to/data.bz2', 'rb')), MDF(gzip.GzipFile('path/to/data.gzip', 'rb')), ], version='4.00', sync=False, )
- configure(*, from_other: MDF_v2_v3_v4 | None = None, read_fragment_size: int | None = None, write_fragment_size: int | None = None, use_display_names: bool | None = None, single_bit_uint_as_bool: bool | None = None, integer_interpolation: IntInterpolationModeType | IntegerInterpolation | None = None, copy_on_get: bool | None = None, float_interpolation: FloatInterpolationModeType | FloatInterpolation | None = None, raise_on_multiple_occurrences: bool | None = None, temporary_folder: str | None = None, fill_0_for_missing_computation_channels: bool | None = None) None [source]¶
configure MDF parameters
The default values for the options are the following: * read_fragment_size = 0 * write_fragment_size = 4MB * use_display_names = False * single_bit_uint_as_bool = False * integer_interpolation = 0 (fill - use previous sample) * float_interpolation = 1 (linear interpolation) * copy_on_get = False * raise_on_multiple_occurrences = True * temporary_folder = “” * fill_0_for_missing_computation_channels = False
- Parameters:
- read_fragment_sizeint
size hint of split data blocks, default 8MB; if the initial size is smaller, then no data list is used. The actual split size depends on the data groups’ records size
- write_fragment_sizeint
size hint of split data blocks, default 4MB; if the initial size is smaller, then no data list is used. The actual split size depends on the data groups’ records size. Maximum size is 4MB to ensure compatibility with CANape
- use_display_namesbool
search for display name in the Channel XML comment
- single_bit_uint_as_boolbool
return single bit channels are np.bool arrays
- integer_interpolationint
interpolation mode for integer channels:
0 - repeat previous sample
1 - use linear interpolation
2 - hybrid interpolation: channels with integer data type (raw values) that have a conversion that outputs float values will use linear interpolation, otherwise the previous sample is used
Changed in version 6.2.0: added hybrid mode interpolation
- copy_on_getbool
copy arrays in the get method
- float_interpolationint
interpolation mode for float channels:
0 - repeat previous sample
1 - use linear interpolation
New in version 6.2.0.
- raise_on_multiple_occurrencesbool
raise exception when there are multiple channel occurrences in the file and the get call is ambiguous; default True
New in version 6.2.0.
- from_otherMDF
copy configuration options from other MDF
New in version 6.2.0.
- temporary_folderstr
default folder for temporary files
New in version 7.0.0.
- fill_0_for_missing_computation_channelsbool
when a channel required by a computed channel is missing, then fill with 0 values. If false then the computation will fail and the computed channel will be marked as not existing.
New in version 7.1.0.
- convert(version: str, progress=None) MDF [source]¶
convert MDF to other version
- Parameters:
- versionstr
new mdf file version from (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’, ‘3.30’, ‘4.00’, ‘4.10’, ‘4.11’, ‘4.20’); default ‘4.10’
- Returns:
- outMDF
new MDF object
- cut(start: float | None = None, stop: float | None = None, whence: int = 0, version: str | None = None, include_ends: bool = True, time_from_zero: bool = False, progress=None) MDF [source]¶
cut MDF file. start and stop limits are absolute values or values relative to the first timestamp depending on the whence argument.
- Parameters:
- startfloat
start time, default None. If None then the start of measurement is used
- stopfloat
stop time, default None. If None then the end of measurement is used
- whenceint
how to search for the start and stop values
0 : absolute
1 : relative to first timestamp
- versionstr
new mdf file version from (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’, ‘3.30’, ‘4.00’, ‘4.10’, ‘4.11’, 4.20’); default None and in this case the original file version is used
- include_endsbool
include the start and stop timestamps after cutting the signal. If start and stop are found in the original timestamps, then the new samples will be computed using interpolation. Default True
- time_from_zerobool
start time stamps from 0s in the cut measurement
- Returns:
- outMDF
new MDF object
- export(fmt: Literal['asc', 'csv', 'hdf5', 'mat', 'parquet'], filename: StrPathType | None = None, progress=None, **kwargs) None [source]¶
export MDF to other formats. The MDF file name is used is available, else the filename argument must be provided.
The pandas export option was removed. you should use the method to_dataframe instead.
- Parameters:
- fmtstring
can be one of the following:
csv : CSV export that uses the “,” delimiter. This option will generate a new csv file for each data group (<MDFNAME>_DataGroup_<cntr>.csv)
hdf5 : HDF5 file output; each MDF data group is mapped to a HDF5 group with the name ‘DataGroup_<cntr>’ (where <cntr> is the index)
mat : Matlab .mat version 4, 5 or 7.3 export. If single_time_base==False the channels will be renamed in the mat file to ‘D<cntr>_<channel name>’. The channel group master will be renamed to ‘DM<cntr>_<channel name>’ ( <cntr> is the data group index starting from 0)
parquet : export to Apache parquet format
asc: Vector ASCII format for bus logging
New in version 7.3.3.
- filenamestring | pathlib.Path
export file name
- **kwargs
single_time_base: resample all channels to common time base, default False
raster: float time raster for resampling. Valid if single_time_base is True
time_from_zero: adjust time channel to start from 0
use_display_names: use display name instead of standard channel name, if available.
empty_channels: behaviour for channels without samples; the options are skip or zeros; default is skip
format: only valid for mat export; can be ‘4’, ‘5’ or ‘7.3’, default is ‘5’
oned_as: only valid for mat export; can be ‘row’ or ‘column’
keep_arrays : keep arrays and structure channels as well as the component channels. If True this can be very slow. If False only the component channels are saved, and their names will be prefixed with the parent channel.
reduce_memory_usage : bool reduce memory usage by converting all float columns to float32 and searching for minimum dtype that can reprezent the values found in integer columns; default False
compression : str compression to be used
for
parquet
: “GZIP” or “SNAPPY”for
hfd5
: “gzip”, “lzf” or “szip”for
mat
: bool
time_as_date (False) : bool export time as local timezone datetimee; only valid for CSV export
New in version 5.8.0.
ignore_value2text_conversions (False) : bool valid only for the channels that have value to text conversions and if raw=False. If this is True then the raw numeric values will be used, and the conversion will not be applied.
New in version 5.8.0.
raw (False) : bool export all channels using the raw values
New in version 6.0.0.
delimiter (‘,’) : str only valid for CSV: see cpython documentation for csv.Dialect.delimiter
New in version 6.2.0.
doublequote (True) : bool only valid for CSV: see cpython documentation for csv.Dialect.doublequote
New in version 6.2.0.
escapechar (None) : str only valid for CSV: see cpython documentation for csv.Dialect.escapechar
New in version 6.2.0.
lineterminator (”\r\n”) : str only valid for CSV: see cpython documentation for csv.Dialect.lineterminator
New in version 6.2.0.
quotechar (‘”’) : str only valid for CSV: see cpython documentation for csv.Dialect.quotechar
New in version 6.2.0.
quoting (“MINIMAL”) : str only valid for CSV: see cpython documentation for csv.Dialect.quoting. Use the last part of the quoting constant name
New in version 6.2.0.
add_units (False) : bool only valid for CSV: add the channel units on the second row of the CSV file
New in version 7.1.0.
- extract_bus_logging(database_files: dict[BusType, Iterable[DbcFileType]], version: str | None = None, ignore_invalid_signals: bool | None = None, consolidated_j1939: bool | None = None, ignore_value2text_conversion: bool = True, prefix: str = '', progress=None) MDF [source]¶
extract all possible CAN signal using the provided databases.
Changed in version 6.0.0 from extract_can_logging
- Parameters:
- database_filesdict
each key will contain an iterable of database files for that bus type. The supported bus types are “CAN”, “LIN”. The iterables will contain the (databases, valid bus) pairs. The database can be a str, pathlib.Path or canamtrix.CanMatrix object. The valid bus is an integer specifying for which bus channel the database can be applied; 0 means any bus channel.
Changed in version 6.0.0: added canmatrix.CanMatrix type
Changed in version 6.3.0: added bus channel fileter
- version (None)str
output file version
- ignore_invalid_signals (None)bool | None
ignore signals that have all samples equal to their maximum value
New in version 5.7.0.
Deprecated since version 7.0.2: this argument is no longer used and will be removed in the future
- consolidated_j1939 (None)bool | None
handle PGNs from all the messages as a single instance
New in version 5.7.0.
Deprecated since version 7.2.0: this argument is no longer used and will be removed in the future
- ignore_value2text_conversion (True): bool
ignore value to text conversions
New in version 5.23.0.
- prefix (“”)str
prefix that will added to the channel group names and signal names in the output file
New in version 6.3.0.
- Returns:
- mdfMDF
new MDF file that contains the succesfully extracted signals
Examples
>>> "extrac CAN and LIN bus logging" >>> mdf = asammdf.MDF(r'bus_logging.mf4') >>> databases = { ... "CAN": [("file1.dbc", 0), ("file2.arxml", 2)], ... "LIN": [("file3.dbc", 0)], ... } >>> extracted = mdf.extract_bus_logging(database_files=database_files) >>> ... >>> "extrac just LIN bus logging" >>> mdf = asammdf.MDF(r'bus_logging.mf4') >>> databases = { ... "LIN": [("file3.dbc", 0)], ... } >>> extracted = mdf.extract_bus_logging(database_files=database_files)
- filter(channels: ChannelsType, version: str | None = None, progress=None) MDF [source]¶
return new MDF object that contains only the channels listed in channels argument
- Parameters:
- channelslist
list of items to be filtered; each item can be :
a channel name string
(channel name, group index, channel index) list or tuple
(channel name, group index) list or tuple
(None, group index, channel index) list or tuple
- versionstr
new mdf file version from (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’, ‘3.30’, ‘4.00’, ‘4.10’, ‘4.11’, ‘4.20’); default None and in this case the original file version is used
- Returns:
- mdfMDF
new MDF file
Examples
>>> from asammdf import MDF, Signal >>> import numpy as np >>> t = np.arange(5) >>> s = np.ones(5) >>> mdf = MDF() >>> for i in range(4): ... sigs = [Signal(s*(i*10+j), t, name='SIG') for j in range(1,4)] ... mdf.append(sigs) ... >>> filtered = mdf.filter(['SIG', ('SIG', 3, 1), ['SIG', 2], (None, 1, 2)]) >>> for gp_nr, ch_nr in filtered.channels_db['SIG']: ... print(filtered.get(group=gp_nr, index=ch_nr)) ... <Signal SIG: samples=[ 1. 1. 1. 1. 1.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> <Signal SIG: samples=[ 31. 31. 31. 31. 31.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> <Signal SIG: samples=[ 21. 21. 21. 21. 21.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> <Signal SIG: samples=[ 12. 12. 12. 12. 12.] timestamps=[0 1 2 3 4] unit="" info=None comment="">
- get_group(index: int, raster: RasterType | None = None, time_from_zero: bool = True, empty_channels: EmptyChannelsType = 'skip', keep_arrays: bool = False, use_display_names: bool = False, time_as_date: bool = False, reduce_memory_usage: bool = False, raw: bool = False, ignore_value2text_conversions: bool = False, only_basenames: bool = False) pd.DataFrame [source]¶
get channel group as pandas DataFrames. If there are multiple occurrences for the same channel name, then a counter will be used to make the names unique (<original_name>_<counter>)
- Parameters:
- indexint
channel group index
- use_display_namesbool
use display name instead of standard channel name, if available.
- reduce_memory_usagebool
reduce memory usage by converting all float columns to float32 and searching for minimum dtype that can reprezent the values found in integer columns; default False
- raw (False)bool
the dataframe will contain the raw channel values
New in version 5.7.0.
- ignore_value2text_conversions (False)bool
valid only for the channels that have value to text conversions and if raw=False. If this is True then the raw numeric values will be used, and the conversion will not be applied.
New in version 5.8.0.
- keep_arrays (False)bool
keep arrays and structure channels as well as the component channels. If True this can be very slow. If False only the component channels are saved, and their names will be prefixed with the parent channel.
New in version 5.8.0.
- empty_channels (“skip”)str
behaviour for channels without samples; the options are skip or zeros; default is skip
New in version 5.8.0.
- only_basenames (False)bool
use just the field names, without prefix, for structures and channel arrays
New in version 5.13.0.
- rasterfloat | np.array | str
new raster that can be
a float step value
a channel name who’s timestamps will be used as raster (starting with asammdf 5.5.0)
an array (starting with asammdf 5.5.0)
see resample for examples of using this argument
- Returns:
- dfpandas.DataFrame
- iter_channels(skip_master: bool = True, copy_master: bool = True, raw: bool = False) Iterator[Signal] [source]¶
generator that yields a Signal for each non-master channel
- Parameters:
- skip_masterbool
do not yield master channels; default True
- copy_masterbool
copy master for each yielded channel True
- rawbool
return raw channels instead of converted; default False
- iter_get(name: str | None = None, group: int | None = None, index: int | None = None, raster: float | None = None, samples_only: typing_extensions.Literal[False] = False, raw: bool = False) Iterator[Signal] [source]¶
- iter_get(name: str | None = None, group: int | None = None, index: int | None = None, raster: float | None = None, samples_only: typing_extensions.Literal[True] = False, raw: bool = False) Iterator[tuple[NDArray[Any], NDArray[Any] | None]]
iterator over a channel
This is usefull in case of large files with a small number of channels.
If the raster keyword argument is not None the output is interpolated accordingly
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- rasterfloat
time raster in seconds
- samples_onlybool
- if True return only the channel samples as numpy array; if
False return a Signal object
- rawbool
return channel samples without appling the conversion rule; default False
- iter_groups(raster: RasterType | None = None, time_from_zero: bool = True, empty_channels: EmptyChannelsType = 'skip', keep_arrays: bool = False, use_display_names: bool = False, time_as_date: bool = False, reduce_memory_usage: bool = False, raw: bool = False, ignore_value2text_conversions: bool = False, only_basenames: bool = False) Iterator[pd.DataFrame] [source]¶
generator that yields channel groups as pandas DataFrames. If there are multiple occurrences for the same channel name inside a channel group, then a counter will be used to make the names unique (<original_name>_<counter>)
- Parameters:
- use_display_namesbool
use display name instead of standard channel name, if available.
New in version 5.21.0.
- reduce_memory_usagebool
reduce memory usage by converting all float columns to float32 and searching for minimum dtype that can reprezent the values found in integer columns; default False
New in version 5.21.0.
- raw (False)bool
the dataframe will contain the raw channel values
New in version 5.21.0.
- ignore_value2text_conversions (False)bool
valid only for the channels that have value to text conversions and if raw=False. If this is True then the raw numeric values will be used, and the conversion will not be applied.
New in version 5.21.0.
- keep_arrays (False)bool
keep arrays and structure channels as well as the component channels. If True this can be very slow. If False only the component channels are saved, and their names will be prefixed with the parent channel.
New in version 5.21.0.
- empty_channels (“skip”)str
behaviour for channels without samples; the options are skip or zeros; default is skip
New in version 5.21.0.
- only_basenames (False)bool
use just the field names, without prefix, for structures and channel arrays
New in version 5.21.0.
- rasterfloat | np.array | str
new raster that can be
a float step value
a channel name who’s timestamps will be used as raster (starting with asammdf 5.5.0)
an array (starting with asammdf 5.5.0)
see resample for examples of using this argument
New in version 5.21.0.
- iter_to_dataframe(channels: ChannelsType | None = None, raster: RasterType | None = None, time_from_zero: bool = True, empty_channels: EmptyChannelsType = 'skip', keep_arrays: bool = False, use_display_names: bool = False, time_as_date: bool = False, reduce_memory_usage: bool = False, raw: bool = False, ignore_value2text_conversions: bool = False, use_interpolation: bool = True, only_basenames: bool = False, chunk_ram_size: int = 209715200, interpolate_outwards_with_nan: bool = False, numeric_1D_only: bool = False, progress=None) Iterator[pd.DataFrame] [source]¶
generator that yields pandas DataFrame’s that should not exceed 200MB of RAM
New in version 5.15.0.
- Parameters:
- channelslist
list of items to be filtered (default None); each item can be :
a channel name string
(channel name, group index, channel index) list or tuple
(channel name, group index) list or tuple
(None, group index, channel index) list or tuple
- rasterfloat | np.array | str
new raster that can be
a float step value
a channel name who’s timestamps will be used as raster (starting with asammdf 5.5.0)
an array (starting with asammdf 5.5.0)
see resample for examples of using this argument
- time_from_zerobool
adjust time channel to start from 0; default True
- empty_channelsstr
behaviour for channels without samples; the options are skip or zeros; default is skip
- use_display_namesbool
use display name instead of standard channel name, if available.
- keep_arraysbool
keep arrays and structure channels as well as the component channels. If True this can be very slow. If False only the component channels are saved, and their names will be prefixed with the parent channel.
- time_as_datebool
the dataframe index will contain the datetime timestamps according to the measurement start time; default False. If True then the argument
time_from_zero
will be ignored.- reduce_memory_usagebool
reduce memory usage by converting all float columns to float32 and searching for minimum dtype that can reprezent the values found in integer columns; default False
- raw (False)bool
the columns will contain the raw values
- ignore_value2text_conversions (False)bool
valid only for the channels that have value to text conversions and if raw=False. If this is True then the raw numeric values will be used, and the conversion will not be applied.
- use_interpolation (True)bool
option to perform interpoaltions when multiple timestamp raster are present. If False then dataframe columns will be automatically filled with NaN’s were the dataframe index values are not found in the current column’s timestamps
- only_basenames (False)bool
use jsut the field names, without prefix, for structures and channel arrays
- interpolate_outwards_with_nanbool
use NaN values for the samples that lie outside of the original signal’s timestamps
- chunk_ram_sizeint
desired data frame RAM usage in bytes; default 200 MB
- numeric_1D_only (False)bool
only keep the 1D-columns that have numeric values
New in version 7.0.0.
- Returns:
- dataframepandas.DataFrame
yields pandas DataFrame’s that should not exceed 200MB of RAM
- resample(raster: RasterType, version: str | None = None, time_from_zero: bool = False, progress=None) MDF [source]¶
resample all channels using the given raster. See configure to select the interpolation method for interger channels
- Parameters:
- rasterfloat | np.array | str
new raster that can be
a float step value
a channel name who’s timestamps will be used as raster (starting with asammdf 5.5.0)
an array (starting with asammdf 5.5.0)
- versionstr
new mdf file version from (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’, ‘3.30’, ‘4.00’, ‘4.10’, ‘4.11’, ‘4.20’); default None and in this case the original file version is used
- time_from_zerobool
start time stamps from 0s in the cut measurement
- Returns:
- mdfMDF
new MDF with resampled channels
Examples
>>> from asammdf import MDF, Signal >>> import numpy as np >>> mdf = MDF() >>> sig = Signal(name='S1', samples=[1,2,3,4], timestamps=[1,2,3,4]) >>> mdf.append(sig) >>> sig = Signal(name='S2', samples=[1,2,3,4], timestamps=[1.1, 3.5, 3.7, 3.9]) >>> mdf.append(sig) >>> resampled = mdf.resample(raster=0.1) >>> resampled.select(['S1', 'S2']) [<Signal S1: samples=[1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3 3 3 3 3 3 3 3 3 4] timestamps=[1. 1.1 1.2 1.3 1.4 1.5 1.6 1.7 1.8 1.9 2. 2.1 2.2 2.3 2.4 2.5 2.6 2.7 2.8 2.9 3. 3.1 3.2 3.3 3.4 3.5 3.6 3.7 3.8 3.9 4. ] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> , <Signal S2: samples=[1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 2 2 3 3 4 4] timestamps=[1. 1.1 1.2 1.3 1.4 1.5 1.6 1.7 1.8 1.9 2. 2.1 2.2 2.3 2.4 2.5 2.6 2.7 2.8 2.9 3. 3.1 3.2 3.3 3.4 3.5 3.6 3.7 3.8 3.9 4. ] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> ] >>> resampled = mdf.resample(raster='S2') >>> resampled.select(['S1', 'S2']) [<Signal S1: samples=[1 3 3 3] timestamps=[1.1 3.5 3.7 3.9] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> , <Signal S2: samples=[1 2 3 4] timestamps=[1.1 3.5 3.7 3.9] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> ] >>> resampled = mdf.resample(raster=[1.9, 2.0, 2.1]) >>> resampled.select(['S1', 'S2']) [<Signal S1: samples=[1 2 2] timestamps=[1.9 2. 2.1] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> , <Signal S2: samples=[1 1 1] timestamps=[1.9 2. 2.1] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> ] >>> resampled = mdf.resample(raster='S2', time_from_zero=True) >>> resampled.select(['S1', 'S2']) [<Signal S1: samples=[1 3 3 3] timestamps=[0. 2.4 2.6 2.8] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> , <Signal S2: samples=[1 2 3 4] timestamps=[0. 2.4 2.6 2.8] invalidation_bits=None unit="" conversion=None source=Source(name='Python', path='Python', comment='', source_type=4, bus_type=0) comment="" mastermeta="('time', 1)" raw=True display_names={} attachment=()> ]
- static scramble(name: str | PathLike[str], skip_attachments: bool = False, progress=None, **kwargs) Path [source]¶
scramble text blocks and keep original file structure
- Parameters:
- namestr | pathlib.Path
file name
- skip_attachmentsbool
skip scrambling of attachments data if True
New in version 5.9.0.
- Returns:
- namepathlib.Path
scrambled file name
- search(pattern: str, mode: Literal['plain', 'regex', 'wildcard'] | SearchMode = SearchMode.plain, case_insensitive: bool = False) list[str] [source]¶
search channels
New in version 7.0.0.
- Parameters:
- patternstr
search pattern
- modeLiteral[“plain”, “regex”, “wildcard”] or SearchMode, optional
search mode, by default SearchMode.plain
plain : normal name search
regex : regular expression based search
wildcard : wildcard based search
- case_insensitivebool, optional
case sensitivity for the channel name search, by default False
- Returns:
- list[str]
name of the channels
- Raises:
- ValueError
unsupported search mode
Examples
>>> mdf = MDF(file_name) >>> mdf.search('*veh*speed*', case_insensitive=True, mode='wildcard') # case insensitive wildcard based search ['vehicleAverageSpeed', 'vehicleInstantSpeed', 'targetVehicleAverageSpeed', 'targetVehicleInstantSpeed'] >>> mdf.search('^vehicle.*Speed$', case_insensitive=False, mode='regex') # case sensitive regex based search ['vehicleAverageSpeed', 'vehicleInstantSpeed']
- select(channels: ChannelsType, record_offset: int = 0, raw: bool = False, copy_master: bool = True, ignore_value2text_conversions: bool = False, record_count: int | None = None, validate: bool = False) list[Signal] [source]¶
retrieve the channels listed in channels argument as Signal objects
Note
the dataframe argument was removed in version 5.8.0 use the
to_dataframe
method instead- Parameters:
- channelslist
list of items to be filtered; each item can be :
a channel name string
(channel name, group index, channel index) list or tuple
(channel name, group index) list or tuple
(None, group index, channel index) list or tuple
- record_offsetint
record number offset; optimization to get the last part of signal samples
- rawbool
get raw channel samples; default False
- copy_masterbool
option to get a new timestamps array for each selected Signal or to use a shared array for channels of the same channel group; default True
- ignore_value2text_conversions (False)bool
valid only for the channels that have value to text conversions and if raw=False. If this is True then the raw numeric values will be used, and the conversion will not be applied.
New in version 5.8.0.
- validate (False)bool
consider the invalidation bits
New in version 5.16.0.
- Returns:
- signalslist
list of Signal objects based on the input channel list
Examples
>>> from asammdf import MDF, Signal >>> import numpy as np >>> t = np.arange(5) >>> s = np.ones(5) >>> mdf = MDF() >>> for i in range(4): ... sigs = [Signal(s*(i*10+j), t, name='SIG') for j in range(1,4)] ... mdf.append(sigs) ... >>> # select SIG group 0 default index 1 default, SIG group 3 index 1, SIG group 2 index 1 default and channel index 2 from group 1 ... >>> mdf.select(['SIG', ('SIG', 3, 1), ['SIG', 2], (None, 1, 2)]) [<Signal SIG: samples=[ 1. 1. 1. 1. 1.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> , <Signal SIG: samples=[ 31. 31. 31. 31. 31.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> , <Signal SIG: samples=[ 21. 21. 21. 21. 21.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> , <Signal SIG: samples=[ 12. 12. 12. 12. 12.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> ]
- static stack(files: Sequence[MDF | InputType], version: str = '4.10', sync: bool = True, progress=None, **kwargs) MDF [source]¶
stack several files and return the stacked MDF object
- Parameters:
- fileslist | tuple
list of MDF file names or MDF, zipfile.ZipFile, bz2.BZ2File or gzip.GzipFile instances
..versionchanged:: 6.2.0
added support for zipfile.ZipFile, bz2.BZ2File and gzip.GzipFile
- versionstr
merged file version
- syncbool
sync the files based on the start of measurement, default True
- kwargs
use_display_names (False) : bool
- Returns:
- stackedMDF
new MDF object with stacked channels
Examples
>>> stacked = MDF.stack( [ 'path/to/file.mf4', MDF(BytesIO(data)), MDF(zipfile.ZipFile('data.zip')), MDF(bz2.BZ2File('path/to/data.bz2', 'rb')), MDF(gzip.GzipFile('path/to/data.gzip', 'rb')), ], version='4.00', sync=False, )
- property start_time: datetime¶
getter and setter the measurement start timestamp
- Returns:
- timestampdatetime.datetime
start timestamp
- to_dataframe(channels: ChannelsType | None = None, raster: RasterType | None = None, time_from_zero: bool = True, empty_channels: EmptyChannelsType = 'skip', keep_arrays: bool = False, use_display_names: bool = False, time_as_date: bool = False, reduce_memory_usage: bool = False, raw: bool = False, ignore_value2text_conversions: bool = False, use_interpolation: bool = True, only_basenames: bool = False, interpolate_outwards_with_nan: bool = False, numeric_1D_only: bool = False, progress=None) pd.DataFrame [source]¶
generate pandas DataFrame
- Parameters:
- channelslist
list of items to be filtered (default None); each item can be :
a channel name string
(channel name, group index, channel index) list or tuple
(channel name, group index) list or tuple
(None, group index, channel index) list or tuple
- rasterfloat | np.array | str
new raster that can be
a float step value
a channel name who’s timestamps will be used as raster (starting with asammdf 5.5.0)
an array (starting with asammdf 5.5.0)
see resample for examples of using this argument
- time_from_zerobool
adjust time channel to start from 0; default True
- empty_channelsstr
behaviour for channels without samples; the options are skip or zeros; default is skip
- use_display_namesbool
use display name instead of standard channel name, if available.
- keep_arraysbool
keep arrays and structure channels as well as the component channels. If True this can be very slow. If False only the component channels are saved, and their names will be prefixed with the parent channel.
- time_as_datebool
the dataframe index will contain the datetime timestamps according to the measurement start time; default False. If True then the argument
time_from_zero
will be ignored.- reduce_memory_usagebool
reduce memory usage by converting all float columns to float32 and searching for minimum dtype that can reprezent the values found in integer columns; default False
- raw (False)bool
the columns will contain the raw values
New in version 5.7.0.
- ignore_value2text_conversions (False)bool
valid only for the channels that have value to text conversions and if raw=False. If this is True then the raw numeric values will be used, and the conversion will not be applied.
New in version 5.8.0.
- use_interpolation (True)bool
option to perform interpoaltions when multiple timestamp raster are present. If False then dataframe columns will be automatically filled with NaN’s were the dataframe index values are not found in the current column’s timestamps
New in version 5.11.0.
- only_basenames (False)bool
use just the field names, without prefix, for structures and channel arrays
New in version 5.13.0.
- interpolate_outwards_with_nanbool
use NaN values for the samples that lie outside of the original signal’s timestamps
New in version 5.15.0.
- Returns:
- dataframepandas.DataFrame
- whereis(channel: str, source_name: str | None = None, source_path: str | None = None, acq_name: str | None = None) tuple[tuple[int, int], ...] [source]¶
get occurrences of channel name in the file
- Parameters:
- channelstr
channel name string
- source_namestr, optional
filter occurrences on source name, by default None
- source_pathstr, optional
filter occurrences on source path, by default None
- acq_namestr, optional
filter occurrences on channel group acquisition name, by default None
New in version 6.0.0.
- Returns:
- tuple[tuple[int, int], …]
(gp_idx, cn_idx) pairs
Examples
>>> mdf = MDF(file_name) >>> mdf.whereis('VehicleSpeed') # "VehicleSpeed" exists in the file ((1, 2), (2, 4)) >>> mdf.whereis('VehicleSPD') # "VehicleSPD" doesn't exist in the file ()
MDF3¶
- class asammdf.blocks.mdf_v3.MDF3(name: BufferedReader | BytesIO | StrPathType | None = None, version: str = '3.30', channels: list[str] | None = None, **kwargs)[source]
The header attibute is a HeaderBlock.
The groups attribute is a list of dicts, each one with the following keys
data_group
- DataGroup objectchannel_group
- ChannelGroup objectchannels
- list of Channel objects with the same order as found in the mdf filechannel_dependencies
- list of ChannelArrayBlock in case of channel arrays; list of Channel objects in case of structure channel compositiondata_block
- address of data blockdata_location
- integer code for data location (original file, temporary file or memory)data_block_addr
- list of raw samples starting addressesdata_block_type
- list of codes for data block typedata_block_size
- list of raw samples block sizesorted
- sorted indicator flagrecord_size
- dict that maps record ID’s to record sizes in bytessize
- total size of data block for the current grouptrigger
- Trigger object for current group
- Parameters:
- namestring | pathlib.Path
mdf file name (if provided it must be a real file name) or file-like object
- versionstring
mdf file version (‘2.00’, ‘2.10’, ‘2.14’, ‘3.00’, ‘3.10’, ‘3.20’ or ‘3.30’); default ‘3.30’
- callbackfunction
keyword only argument: function to call to update the progress; the function must accept two arguments (the current progress and maximum progress value)
- Attributes:
- attachmentslist
list of file attachments
- channels_dbdict
used for fast channel access by name; for each name key the value is a list of (group index, channel index) tuples
- groupslist
list of data group dicts
- headerHeaderBlock
mdf file header
- identificationFileIdentificationBlock
mdf file start block
- last_call_infodict | None
a dict to hold information about the last called method.
New in version 5.12.0.
- masters_dbdict
- used for fast master channel access; for each group index key the value
is the master channel index
- memorystr
memory optimization option
- namestring
mdf file name
- versionstr
mdf version
- add_trigger(group: int, timestamp: float, pre_time: float = 0, post_time: float = 0, comment: str = '') None [source]
add trigger to data group
- Parameters:
- groupint
group index
- timestampfloat
trigger time
- pre_timefloat
trigger pre time; default 0
- post_timefloat
trigger post time; default 0
- commentstr
trigger comment
- append(signals: list[Signal] | Signal | DataFrame, acq_name: str | None = None, acq_source: Source | None = None, comment: str = 'Python', common_timebase: bool = False, units: dict[str, str | bytes] | None = None) int | None [source]
Appends a new data group.
For channel dependencies type Signals, the samples attribute must be a numpy.recarray
- Parameters:
- signalslist | Signal | pandas.DataFrame
list of Signal objects, or a single Signal object, or a pandas DataFrame object. All bytes columns in the pandas DataFrame must be latin-1 encoded
- acq_namestr
channel group acquisition name
- acq_sourceSource
channel group acquisition source
- commentstr
channel group comment; default ‘Python’
- common_timebasebool
flag to hint that the signals have the same timebase. Only set this if you know for sure that all appended channels share the same time base
- unitsdict
will contain the signal units mapped to the signal names when appending a pandas DataFrame
Examples
>>> # case 1 conversion type None >>> s1 = np.array([1, 2, 3, 4, 5]) >>> s2 = np.array([-1, -2, -3, -4, -5]) >>> s3 = np.array([0.1, 0.04, 0.09, 0.16, 0.25]) >>> t = np.array([0.001, 0.002, 0.003, 0.004, 0.005]) >>> names = ['Positive', 'Negative', 'Float'] >>> units = ['+', '-', '.f'] >>> info = {} >>> s1 = Signal(samples=s1, timestamps=t, unit='+', name='Positive') >>> s2 = Signal(samples=s2, timestamps=t, unit='-', name='Negative') >>> s3 = Signal(samples=s3, timestamps=t, unit='flts', name='Floats') >>> mdf = MDF3('new.mdf') >>> mdf.append([s1, s2, s3], comment='created by asammdf v1.1.0') >>> # case 2: VTAB conversions from channels inside another file >>> mdf1 = MDF3('in.mdf') >>> ch1 = mdf1.get("Channel1_VTAB") >>> ch2 = mdf1.get("Channel2_VTABR") >>> sigs = [ch1, ch2] >>> mdf2 = MDF3('out.mdf') >>> mdf2.append(sigs, comment='created by asammdf v1.1.0') >>> df = pd.DataFrame.from_dict({'s1': np.array([1, 2, 3, 4, 5]), 's2': np.array([-1, -2, -3, -4, -5])}) >>> units = {'s1': 'V', 's2': 'A'} >>> mdf2.append(df, units=units)
- close() None [source]
if the MDF was created with memory=’minimum’ and new channels have been appended, then this must be called just before the object is not used anymore to clean-up the temporary file
- extend(index: int, signals: list[tuple[NDArray[Any], None]]) None [source]
Extend a group with new samples. signals contains (values, invalidation_bits) pairs for each extended signal. Since MDF3 does not support invalidation bits, the second item of each pair must be None. The first pair is the master channel’s pair, and the next pairs must respect the same order in which the signals were appended. The samples must have raw or physical values according to the Signals used for the initial append.
- Parameters:
- indexint
group index
- signalslist
list of (numpy.ndarray, None) objects
Examples
>>> # case 1 conversion type None >>> s1 = np.array([1, 2, 3, 4, 5]) >>> s2 = np.array([-1, -2, -3, -4, -5]) >>> s3 = np.array([0.1, 0.04, 0.09, 0.16, 0.25]) >>> t = np.array([0.001, 0.002, 0.003, 0.004, 0.005]) >>> names = ['Positive', 'Negative', 'Float'] >>> units = ['+', '-', '.f'] >>> s1 = Signal(samples=s1, timestamps=t, unit='+', name='Positive') >>> s2 = Signal(samples=s2, timestamps=t, unit='-', name='Negative') >>> s3 = Signal(samples=s3, timestamps=t, unit='flts', name='Floats') >>> mdf = MDF3('new.mdf') >>> mdf.append([s1, s2, s3], comment='created by asammdf v1.1.0') >>> t = np.array([0.006, 0.007, 0.008, 0.009, 0.010]) >>> mdf2.extend(0, [(t, None), (s1.samples, None), (s2.samples, None), (s3.samples, None)])
- get(name: str | None = None, group: int | None = None, index: int | None = None, raster: RasterType | None = None, samples_only: typing_extensions.Literal[False] = False, data: bytes | None = None, raw: bool = False, ignore_invalidation_bits: bool = False, record_offset: int = 0, record_count: int | None = None, skip_channel_validation: bool = False) Signal [source]
- get(name: str | None = None, group: int | None = None, index: int | None = None, raster: RasterType | None = None, samples_only: typing_extensions.Literal[True] = False, data: bytes | None = None, raw: bool = False, ignore_invalidation_bits: bool = False, record_offset: int = 0, record_count: int | None = None, skip_channel_validation: bool = False) tuple[NDArray[Any], None]
Gets channel samples. Channel can be specified in two ways:
using the first positional argument name
if there are multiple occurrences for this channel then the group and index arguments can be used to select a specific group.
if there are multiple occurrences for this channel and either the group or index arguments is None then a warning is issued
using the group number (keyword argument group) and the channel number (keyword argument index). Use info method for group and channel numbers
If the raster keyword argument is not None the output is interpolated accordingly.
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- rasterfloat
time raster in seconds
- samples_onlybool
if True return only the channel samples as numpy array; if False return a Signal object
- databytes
prevent redundant data read by providing the raw data group samples
- rawbool
return channel samples without applying the conversion rule; default False
- ignore_invalidation_bitsbool
only defined to have the same API with the MDF v4
- record_offsetint
if data=None use this to select the record offset from which the group data should be loaded
- skip_channel_validation (False)bool
skip validation of channel name, group index and channel index; defualt False. If True, the caller has to make sure that the group and index arguments are provided and are correct.
..versionadded:: 7.0.0
- Returns:
- res(numpy.array, None) | Signal
returns Signal if samples_only*=*False (default option), otherwise returns a (numpy.array, None) tuple (for compatibility with MDF v4 class.
The Signal samples are
numpy recarray for channels that have CDBLOCK or BYTEARRAY type channels
numpy array for all the rest
- Raises:
- MdfException
- if the channel name is not found
- if the group index is out of range
- if the channel index is out of range
Examples
>>> from asammdf import MDF, Signal >>> import numpy as np >>> t = np.arange(5) >>> s = np.ones(5) >>> mdf = MDF(version='3.30') >>> for i in range(4): ... sigs = [Signal(s*(i*10+j), t, name='Sig') for j in range(1, 4)] ... mdf.append(sigs) ... >>> # first group and channel index of the specified channel name ... >>> mdf.get('Sig') UserWarning: Multiple occurrences for channel "Sig". Using first occurrence from data group 4. Provide both "group" and "index" arguments to select another data group <Signal Sig: samples=[ 1. 1. 1. 1. 1.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> # first channel index in the specified group ... >>> mdf.get('Sig', 1) <Signal Sig: samples=[ 11. 11. 11. 11. 11.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> # channel named Sig from group 1 channel index 2 ... >>> mdf.get('Sig', 1, 2) <Signal Sig: samples=[ 12. 12. 12. 12. 12.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> # channel index 1 or group 2 ... >>> mdf.get(None, 2, 1) <Signal Sig: samples=[ 21. 21. 21. 21. 21.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> mdf.get(group=2, index=1) <Signal Sig: samples=[ 21. 21. 21. 21. 21.] timestamps=[0 1 2 3 4] unit="" info=None comment="">
- get_channel_comment(name: str | None = None, group: int | None = None, index: int | None = None) str [source]
Gets channel comment. Channel can be specified in two ways:
using the first positional argument name
if there are multiple occurrences for this channel then the group and index arguments can be used to select a specific group.
if there are multiple occurrences for this channel and either the group or index arguments is None then a warning is issued
using the group number (keyword argument group) and the channel number (keyword argument index). Use info method for group and channel numbers
If the raster keyword argument is not None the output is interpolated accordingly.
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- Returns:
- commentstr
found channel comment
- get_channel_name(group: int, index: int) str [source]
Gets channel name.
- Parameters:
- groupint
0-based group index
- indexint
0-based channel index
- Returns:
- namestr
found channel name
- get_channel_unit(name: str | None = None, group: int | None = None, index: int | None = None) str [source]
Gets channel unit.
Channel can be specified in two ways:
using the first positional argument name
if there are multiple occurrences for this channel then the group and index arguments can be used to select a specific group.
if there are multiple occurrences for this channel and either the group or index arguments is None then a warning is issued
using the group number (keyword argument group) and the channel number (keyword argument index). Use info method for group and channel numbers
If the raster keyword argument is not None the output is interpolated accordingly.
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- Returns:
- unitstr
found channel unit
- get_master(index: int, data: bytes | None = None, raster: RasterType | None = None, record_offset: int = 0, record_count: int | None = None, one_piece: bool = False) NDArray[Any] [source]
returns master channel samples for given group
- Parameters:
- indexint
group index
- data(bytes, int)
(data block raw bytes, fragment offset); default None
- rasterfloat
raster to be used for interpolation; default None
Deprecated since version 5.13.0.
- record_offsetint
if data=None use this to select the record offset from which the group data should be loaded
- Returns:
- tnumpy.array
master channel samples
- info() dict[str, Any] [source]
get MDF information as a dict
Examples
>>> mdf = MDF3('test.mdf') >>> mdf.info()
- iter_get_triggers() Iterator[TriggerInfoDict] [source]
generator that yields triggers
- Returns:
- trigger_infodict
trigger information with the following keys:
comment : trigger comment
time : trigger time
pre_time : trigger pre time
post_time : trigger post time
index : trigger index
group : data group index of trigger
- save(dst: StrPathType, overwrite: bool = False, compression: CompressionType = 0, progress=None, add_history_block: bool = True) Path | None [source]
Save MDF to dst. If overwrite is True then the destination file is overwritten, otherwise the file name is appended with ‘.<cntr>’, were ‘<cntr>’ is the first counter that produces a new file name (that does not already exist in the filesystem).
- Parameters:
- dststr | pathlib.Path
destination file name
- overwritebool
overwrite flag, default False
- compressionint
does nothing for mdf version3; introduced here to share the same API as mdf version 4 files
- Returns:
- output_filestr
output file name
- property start_time: datetime
getter and setter the measurement start timestamp
- Returns:
- timestampdatetime.datetime
start timestamp
MDF version 2 & 3 blocks¶
MDF4¶
- class asammdf.blocks.mdf_v4.MDF4(name: BufferedReader | BytesIO | StrPathType | None = None, version: str = '4.10', channels: list[str] | None = None, **kwargs)[source]
The header attibute is a HeaderBlock.
The groups attribute is a list of dicts, each one with the following keys:
data_group
- DataGroup objectchannel_group
- ChannelGroup objectchannels
- list of Channel objects with the same order as found in the mdf filechannel_dependencies
- list of ChannelArrayBlock in case of channel arrays; list of Channel objects in case of structure channel compositiondata_block
- address of data blockdata_location
- integer code for data location (original file, temporary file or memory)data_block_addr
- list of raw samples starting addressesdata_block_type
- list of codes for data block typedata_block_size
- list of raw samples block sizesorted
- sorted indicator flagrecord_size
- dict that maps record ID’s to record sizes in bytes (including invalidation bytes)param
- row size used for transposition, in case of transposed zipped blocks
- Parameters:
- namestring
mdf file name (if provided it must be a real file name) or file-like object
- versionstring
mdf file version (‘4.00’, ‘4.10’, ‘4.11’, ‘4.20’); default ‘4.10’
- kwargs
- use_display_names (True)bool
keyword only argument: for MDF4 files parse the XML channel comment to search for the display name; XML parsing is quite expensive so setting this to False can decrease the loading times very much; default True
- remove_source_from_channel_names (True)bool
- copy_on_get (True)bool
copy channel values (np.array) to avoid high memory usage
- compact_vlsd (False)bool
use slower method to save the exact sample size for VLSD channels
- column_storage (True)bool
use column storage for MDF version >= 4.20
- passwordbytes | str
use this password to decode encrypted attachments
- Attributes:
- attachmentslist
list of file attachments
- channels_dbdict
used for fast channel access by name; for each name key the value is a list of (group index, channel index) tuples
- eventslist
list event blocks
- file_commentTextBlock
file comment TextBlock
- file_historylist
list of (FileHistory, TextBlock) pairs
- groupslist
list of data group dicts
- headerHeaderBlock
mdf file header
- identificationFileIdentificationBlock
mdf file start block
- last_call_infodict | None
a dict to hold information about the last called method.
New in version 5.12.0.
- masters_dbdict
- used for fast master channel access; for each group index key the value
is the master channel index
- namestring
mdf file name
- versionstr
mdf version
- append(signals: list[Signal] | Signal | DataFrame, acq_name: str | None = None, acq_source: Source | None = None, comment: str = 'Python', common_timebase: bool = False, units: dict[str, str | bytes] | None = None) int | None [source]
Appends a new data group.
For channel dependencies type Signals, the samples attribute must be a numpy.recarray
- Parameters:
- signalslist | Signal | pandas.DataFrame
list of Signal objects, or a single Signal object, or a pandas DataFrame object. All bytes columns in the pandas DataFrame must be utf-8 encoded
- acq_namestr
channel group acquisition name
- acq_sourceSource
channel group acquisition source
- commentstr
channel group comment; default ‘Python’
- common_timebasebool
flag to hint that the signals have the same timebase. Only set this if you know for sure that all appended channels share the same time base
- unitsdict
will contain the signal units mapped to the signal names when appending a pandas DataFrame
Examples
>>> # case 1 conversion type None >>> s1 = np.array([1, 2, 3, 4, 5]) >>> s2 = np.array([-1, -2, -3, -4, -5]) >>> s3 = np.array([0.1, 0.04, 0.09, 0.16, 0.25]) >>> t = np.array([0.001, 0.002, 0.003, 0.004, 0.005]) >>> names = ['Positive', 'Negative', 'Float'] >>> units = ['+', '-', '.f'] >>> info = {} >>> s1 = Signal(samples=s1, timestamps=t, unit='+', name='Positive') >>> s2 = Signal(samples=s2, timestamps=t, unit='-', name='Negative') >>> s3 = Signal(samples=s3, timestamps=t, unit='flts', name='Floats') >>> mdf = MDF4('new.mdf') >>> mdf.append([s1, s2, s3], comment='created by asammdf v4.0.0') >>> # case 2: VTAB conversions from channels inside another file >>> mdf1 = MDF4('in.mf4') >>> ch1 = mdf1.get("Channel1_VTAB") >>> ch2 = mdf1.get("Channel2_VTABR") >>> sigs = [ch1, ch2] >>> mdf2 = MDF4('out.mf4') >>> mdf2.append(sigs, comment='created by asammdf v4.0.0') >>> mdf2.append(ch1, comment='just a single channel') >>> df = pd.DataFrame.from_dict({'s1': np.array([1, 2, 3, 4, 5]), 's2': np.array([-1, -2, -3, -4, -5])}) >>> units = {'s1': 'V', 's2': 'A'} >>> mdf2.append(df, units=units)
- attach(data: bytes, file_name: str | None = None, hash_sum: bytes | None = None, comment: str = '', compression: bool = True, mime: str = 'application/octet-stream', embedded: bool = True, password: str | bytes | None = None) int [source]
attach embedded attachment as application/octet-stream.
- Parameters:
- databytes
data to be attached
- file_namestr
string file name
- hash_sumbytes
md5 of the data
- commentstr
attachment comment
- compressionbool
use compression for embedded attachment data
- mimestr
mime type string
- embeddedbool
attachment is embedded in the file
- passwordstr | bytes | None , default None
password used to encrypt the data using AES256 encryption
New in version 7.0.0.
- Returns:
- indexint
new attachment index
- close() None [source]
if the MDF was created with memory=False and new channels have been appended, then this must be called just before the object is not used anymore to clean-up the temporary file
- extend(index: int, signals: list[tuple[NDArray[Any], NDArray[Any] | None]]) None [source]
Extend a group with new samples. signals contains (values, invalidation_bits) pairs for each extended signal. The first pair is the master channel’s pair, and the next pairs must respect the same order in which the signals were appended. The samples must have raw or physical values according to the Signals used for the initial append.
- Parameters:
- indexint
group index
- signalslist
list on (numpy.ndarray, numpy.ndarray) objects
Examples
>>> # case 1 conversion type None >>> s1 = np.array([1, 2, 3, 4, 5]) >>> s2 = np.array([-1, -2, -3, -4, -5]) >>> s3 = np.array([0.1, 0.04, 0.09, 0.16, 0.25]) >>> t = np.array([0.001, 0.002, 0.003, 0.004, 0.005]) >>> names = ['Positive', 'Negative', 'Float'] >>> units = ['+', '-', '.f'] >>> s1 = Signal(samples=s1, timestamps=t, unit='+', name='Positive') >>> s2 = Signal(samples=s2, timestamps=t, unit='-', name='Negative') >>> s3 = Signal(samples=s3, timestamps=t, unit='flts', name='Floats') >>> mdf = MDF4('new.mdf') >>> mdf.append([s1, s2, s3], comment='created by asammdf v1.1.0') >>> t = np.array([0.006, 0.007, 0.008, 0.009, 0.010]) >>> # extend without invalidation bits >>> mdf2.extend(0, [(t, None), (s1, None), (s2, None), (s3, None)]) >>> # some invaldiation btis >>> s1_inv = np.array([0,0,0,1,1], dtype=np.bool) >>> mdf2.extend(0, [(t, None), (s1.samples, None), (s2.samples, None), (s3.samples, None)])
- extract_attachment(index: int | None = None, password: str | bytes | None = None) tuple[bytes, Path, bytes] [source]
extract attachment data by index. If it is an embedded attachment, then this method creates the new file according to the attachment file name information
- Parameters:
- indexint
attachment index; default None
- passwordstr | bytes | None, default None
password used to encrypt the data using AES256 encryption
New in version 7.0.0.
- Returns:
- data(bytes, pathlib.Path)
tuple of attachment data and path
- get(name: str | None = None, group: int | None = None, index: int | None = None, raster: RasterType | None = None, samples_only: typing_extensions.Literal[False] = False, data: bytes | None = None, raw: bool = False, ignore_invalidation_bits: bool = False, record_offset: int = 0, record_count: int | None = None, skip_channel_validation: bool = False) Signal [source]
- get(name: str | None = None, group: int | None = None, index: int | None = None, raster: RasterType | None = None, samples_only: typing_extensions.Literal[True] = False, data: bytes | None = None, raw: bool = False, ignore_invalidation_bits: bool = False, record_offset: int = 0, record_count: int | None = None, skip_channel_validation: bool = False) tuple[NDArray[Any], NDArray[Any]]
Gets channel samples. The raw data group samples are not loaded to memory so it is advised to use
filter
orselect
instead of performing severalget
calls.Channel can be specified in two ways:
using the first positional argument name
if there are multiple occurrences for this channel then the group and index arguments can be used to select a specific group.
if there are multiple occurrences for this channel and either the group or index arguments is None then a warning is issued
using the group number (keyword argument group) and the channel number (keyword argument index). Use info method for group and channel numbers
If the raster keyword argument is not None the output is interpolated accordingly
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- rasterfloat
time raster in seconds
- samples_onlybool
- if True return only the channel samples as numpy array; if
False return a Signal object
- databytes
prevent redundant data read by providing the raw data group samples
- rawbool
return channel samples without applying the conversion rule; default False
- ignore_invalidation_bitsbool
option to ignore invalidation bits
- record_offsetint
if data=None use this to select the record offset from which the group data should be loaded
- record_countint
number of records to read; default None and in this case all available records are used
- skip_channel_validation (False)bool
skip validation of channel name, group index and channel index; defualt False. If True, the caller has to make sure that the group and index arguments are provided and are correct.
..versionadded:: 7.0.0
- Returns:
- res(numpy.array, numpy.array) | Signal
returns Signal if samples_only*=*False (default option), otherwise returns a (numpy.array, numpy.array) tuple of samples and invalidation bits. If invalidation bits are not used or if ignore_invalidation_bits if False, then the second item will be None.
The Signal samples are:
numpy recarray for channels that have composition/channel array address or for channel of type CANOPENDATE, CANOPENTIME
numpy array for all the rest
- Raises:
- MdfException
- if the channel name is not found
- if the group index is out of range
- if the channel index is out of range
Examples
>>> from asammdf import MDF, Signal >>> import numpy as np >>> t = np.arange(5) >>> s = np.ones(5) >>> mdf = MDF(version='4.10') >>> for i in range(4): ... sigs = [Signal(s*(i*10+j), t, name='Sig') for j in range(1, 4)] ... mdf.append(sigs) ... >>> # first group and channel index of the specified channel name ... >>> mdf.get('Sig') UserWarning: Multiple occurrences for channel "Sig". Using first occurrence from data group 4. Provide both "group" and "index" arguments to select another data group <Signal Sig: samples=[ 1. 1. 1. 1. 1.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> # first channel index in the specified group ... >>> mdf.get('Sig', 1) <Signal Sig: samples=[ 11. 11. 11. 11. 11.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> # channel named Sig from group 1 channel index 2 ... >>> mdf.get('Sig', 1, 2) <Signal Sig: samples=[ 12. 12. 12. 12. 12.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> # channel index 1 or group 2 ... >>> mdf.get(None, 2, 1) <Signal Sig: samples=[ 21. 21. 21. 21. 21.] timestamps=[0 1 2 3 4] unit="" info=None comment=""> >>> mdf.get(group=2, index=1) <Signal Sig: samples=[ 21. 21. 21. 21. 21.] timestamps=[0 1 2 3 4] unit="" info=None comment="">
- get_bus_signal(bus: BusType, name: str, database: CanMatrix | StrPathType | None = None, ignore_invalidation_bits: bool = False, data: bytes | None = None, raw: bool = False, ignore_value2text_conversion: bool = True) Signal [source]
get a signal decoded from a raw bus logging. The currently supported buses are CAN and LIN (LDF databases are not supported, they need to be converted to DBC and feed to this function)
New in version 6.0.0.
- Parameters:
- busstr
“CAN” or “LIN”
- namestr
signal name
- databasestr
path of external CAN/LIN database file (.dbc or .arxml) or canmatrix.CanMatrix; default None
Changed in version 6.0.0: db and database arguments were merged into this single argument
- ignore_invalidation_bitsbool
option to ignore invalidation bits
- rawbool
return channel samples without applying the conversion rule; default False
- ignore_value2text_conversionbool
return channel samples without values that have a description in .dbc or .arxml file True
- Returns:
- sigSignal
Signal object with the physical values
- get_can_signal(name: str, database: CanMatrix | StrPathType | None = None, ignore_invalidation_bits: bool = False, data: bytes | None = None, raw: bool = False, ignore_value2text_conversion: bool = True) Signal [source]
get CAN message signal. You can specify an external CAN database ( database argument) or canmatrix database object that has already been loaded from a file (db argument).
The signal name can be specified in the following ways
CAN<ID>.<MESSAGE_NAME>.<SIGNAL_NAME>
- the ID value starts from 1 and must match the ID found in the measurement (the source CAN bus ID) Example: CAN1.Wheels.FL_WheelSpeedCAN<ID>.CAN_DataFrame_<MESSAGE_ID>.<SIGNAL_NAME>
- the ID value starts from 1 and the MESSAGE_ID is the decimal message ID as found in the database. Example: CAN1.CAN_DataFrame_218.FL_WheelSpeed<MESSAGE_NAME>.<SIGNAL_NAME>
- in this case the first occurrence of the message name and signal are returned (the same message could be found on multiple CAN buses; for example on CAN1 and CAN3) Example: Wheels.FL_WheelSpeedCAN_DataFrame_<MESSAGE_ID>.<SIGNAL_NAME>
- in this case the first occurrence of the message name and signal are returned (the same message could be found on multiple CAN buses; for example on CAN1 and CAN3). Example: CAN_DataFrame_218.FL_WheelSpeed<SIGNAL_NAME>
- in this case the first occurrence of the signal name is returned (the same signal name could be found in multiple messages and on multiple CAN buses). Example: FL_WheelSpeed
- Parameters:
- namestr
signal name
- databasestr
path of external CAN database file (.dbc or .arxml) or canmatrix.CanMatrix; default None
Changed in version 6.0.0: db and database arguments were merged into this single argument
- ignore_invalidation_bitsbool
option to ignore invalidation bits
- rawbool
return channel samples without applying the conversion rule; default False
- ignore_value2text_conversionbool
return channel samples without values that have a description in .dbc or .arxml file True
- Returns:
- sigSignal
Signal object with the physical values
- get_channel_comment(name: str | None = None, group: int | None = None, index: int | None = None) str [source]
Gets channel comment.
Channel can be specified in two ways:
using the first positional argument name
if there are multiple occurrences for this channel then the group and index arguments can be used to select a specific group.
if there are multiple occurrences for this channel and either the group or index arguments is None then a warning is issued
using the group number (keyword argument group) and the channel number (keyword argument index). Use info method for group and channel numbers
If the raster keyword argument is not None the output is interpolated accordingly.
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- Returns:
- commentstr
found channel comment
- get_channel_name(group: int, index: int) str [source]
Gets channel name.
- Parameters:
- groupint
0-based group index
- indexint
0-based channel index
- Returns:
- namestr
found channel name
- get_channel_unit(name: str | None = None, group: int | None = None, index: int | None = None) str [source]
Gets channel unit.
Channel can be specified in two ways:
using the first positional argument name
if there are multiple occurrences for this channel then the group and index arguments can be used to select a specific group.
if there are multiple occurrences for this channel and either the group or index arguments is None then a warning is issued
using the group number (keyword argument group) and the channel number (keyword argument index). Use info method for group and channel numbers
If the raster keyword argument is not None the output is interpolated accordingly.
- Parameters:
- namestring
name of channel
- groupint
0-based group index
- indexint
0-based channel index
- Returns:
- unitstr
found channel unit
- get_invalidation_bits(group_index: int, channel: Channel, fragment: tuple[bytes, int, int, ReadableBufferType | None]) NDArray[bool_] [source]
get invalidation indexes for the channel
- Parameters:
- group_indexint
group index
- channelChannel
channel object
- fragment(bytes, int)
(fragment bytes, fragment offset)
- Returns:
- invalidation_bitsiterable
iterable of valid channel indexes; if all are valid None is returned
- get_lin_signal(name: str, database: CanMatrix | StrPathType | None = None, ignore_invalidation_bits: bool = False, data: bytes | None = None, raw: bool = False, ignore_value2text_conversion: bool = True) Signal [source]
get LIN message signal. You can specify an external LIN database ( database argument) or canmatrix database object that has already been loaded from a file (db argument).
The signal name can be specified in the following ways
LIN_Frame_<MESSAGE_ID>.<SIGNAL_NAME>
- Example: LIN_Frame_218.FL_WheelSpeed<MESSAGE_NAME>.<SIGNAL_NAME>
- Example: Wheels.FL_WheelSpeed<SIGNAL_NAME>
- Example: FL_WheelSpeed
New in version 6.0.0.
- Parameters:
- namestr
signal name
- databasestr
path of external LIN database file (.dbc, .arxml or .ldf) or canmatrix.CanMatrix; default None
- ignore_invalidation_bitsbool
option to ignore invalidation bits
- rawbool
return channel samples without applying the conversion rule; default False
- ignore_value2text_conversionbool
return channel samples without values that have a description in .dbc, .arxml or .ldf file True
- Returns:
- sigSignal
Signal object with the physical values
- get_master(index: int, data: bytes | None = None, raster: RasterType | None = None, record_offset: int = 0, record_count: int | None = None, one_piece: bool = False) NDArray[Any] [source]
returns master channel samples for given group
- Parameters:
- indexint
group index
- data(bytes, int, int, bytes|None)
(data block raw bytes, fragment offset, count, invalidation bytes); default None
- rasterfloat
raster to be used for interpolation; default None
Deprecated since version 5.13.0.
- record_offsetint
if data=None use this to select the record offset from which the group data should be loaded
- record_countint
number of records to read; default None and in this case all available records are used
- Returns:
- tnumpy.array
master channel samples
- info() dict[str, Any] [source]
get MDF information as a dict
Examples
>>> mdf = MDF4('test.mdf') >>> mdf.info()
- save(dst: WritableBufferType | StrPathType, overwrite: bool = False, compression: CompressionType = 0, progress=None, add_history_block: bool = True) Path [source]
Save MDF to dst. If overwrite is True then the destination file is overwritten, otherwise the file name is appended with ‘.<cntr>’, were ‘<cntr>’ is the first counter that produces a new file name (that does not already exist in the filesystem)
- Parameters:
- dststr
destination file name, Default ‘’
- overwritebool
overwrite flag, default False
- compressionint
use compressed data blocks, default 0; valid since version 4.10
0 - no compression
1 - deflate (slower, but produces smaller files)
2 - transposition + deflate (slowest, but produces the smallest files)
- add_history_blockbool
option to add file historyu block
- Returns:
- output_filepathlib.Path
path to saved file
- property start_time: datetime
getter and setter the measurement start timestamp
- Returns:
- timestampdatetime.datetime
start timestamp
MDF version 4 blocks¶
Signal¶
- class asammdf.signal.Signal(samples: ArrayLike | None = None, timestamps: ArrayLike | None = None, unit: str = '', name: str = '', conversion: dict[str, Any] | ChannelConversionType | None = None, comment: str = '', raw: bool = True, master_metadata: tuple[str, SyncType] | None = None, display_names: dict[str, str] | None = None, attachment: tuple[bytes, str | None, str | None] | None = None, source: SourceType | None = None, bit_count: int | None = None, invalidation_bits: ArrayLike | None = None, encoding: str | None = None, group_index: int = -1, channel_index: int = -1, flags: Flags = SignalFlags.no_flags)[source]¶
The Signal represents a channel described by it’s samples and timestamps. It can perform arithmetic operations against other Signal or numeric types. The operations are computed in respect to the timestamps (time correct). The non-float signals are not interpolated, instead the last value relative to the current timestamp is used. samples, timestamps and name are mandatory arguments.
- Parameters:
- samplesnumpy.array | list | tuple
signal samples
- timestampsnumpy.array | list | tuple
signal timestamps
- unitstr
signal unit
- namestr
signal name
- conversiondict | channel conversion block
dict that contains extra conversion information about the signal , default None
- commentstr
signal comment, default ‘’
- rawbool
signal samples are raw values, with no physical conversion applied
- master_metadatalist
master name and sync type
- display_namesdict
display names used by mdf version 3
- attachmentbytes, name
channel attachment and name from MDF version 4
- sourceSource
source information named tuple
- bit_countint
bit count; useful for integer channels
- invalidation_bitsnumpy.array | None
channel invalidation bits, default None
- encodingstr | None
encoding for string signals; default None
- flagsSignal.Flags
flags for user defined attributes and stream sync
- Flags¶
alias of
SignalFlags
- astype(np_type: dtype[Any] | None | Type[Any] | _SupportsDType[dtype[Any]] | str | Tuple[Any, int] | Tuple[Any, SupportsIndex | Sequence[SupportsIndex]] | List[Any] | _DTypeDict | Tuple[Any, Any]) Signal [source]¶
returns new Signal with samples of dtype np_type
- Parameters:
- np_typenp.dtype
new numpy dtye
- Returns:
- signalSignal
new Signal with the samples of np_type dtype
- cut(start: float | None = None, stop: float | None = None, include_ends: bool = True, integer_interpolation_mode: IntInterpolationModeType | IntegerInterpolation = IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE, float_interpolation_mode: FloatInterpolationModeType | FloatInterpolation = FloatInterpolation.LINEAR_INTERPOLATION) Signal [source]¶
Cuts the signal according to the start and stop values, by using the insertion indexes in the signal’s time axis.
- Parameters:
- startfloat
start timestamp for cutting
- stopfloat
stop timestamp for cutting
- include_endsbool
include the start and stop timestamps after cutting the signal. If start and stop are found in the original timestamps, then the new samples will be computed using interpolation. Default True
- integer_interpolation_modeint
interpolation mode for integer signals; default 0
0 - repeat previous samples
1 - linear interpolation
2 - hybrid interpolation: channels with integer data type (raw values) that have a conversion that outputs float values will use linear interpolation, otherwise the previous sample is used
New in version 6.2.0.
- float_interpolation_modeint
interpolation mode for float channels; default 1
0 - repeat previous sample
1 - use linear interpolation
New in version 6.2.0.
- Returns:
- resultSignal
new Signal cut from the original
Examples
>>> new_sig = old_sig.cut(1.0, 10.5) >>> new_sig.timestamps[0], new_sig.timestamps[-1] 0.98, 10.48
- extend(other: Signal) Signal [source]¶
extend signal with samples from another signal
- Parameters:
- otherSignal
- Returns:
- signalSignal
new extended Signal
- interp(new_timestamps: NDArray[Any], integer_interpolation_mode: IntInterpolationModeType | IntegerInterpolation = IntegerInterpolation.REPEAT_PREVIOUS_SAMPLE, float_interpolation_mode: FloatInterpolationModeType | FloatInterpolation = FloatInterpolation.LINEAR_INTERPOLATION) Signal [source]¶
returns a new Signal interpolated using the new_timestamps
- Parameters:
- new_timestampsnp.array
timestamps used for interpolation
- integer_interpolation_modeint
interpolation mode for integer signals; default 0
0 - repeat previous samples
1 - linear interpolation
2 - hybrid interpolation: channels with integer data type (raw values) that have a conversion that outputs float values will use linear interpolation, otherwise the previous sample is used
New in version 6.2.0.
- float_interpolation_modeint
interpolation mode for float channels; default 1
0 - repeat previous sample
1 - use linear interpolation
New in version 6.2.0.
- Returns:
- signalSignal
new interpolated Signal
- physical() Signal [source]¶
get the physical samples values
- Returns:
- physSignal
new Signal with physical values
- plot(validate: bool = True, index_only: bool = False) None [source]¶
plot Signal samples. Pyqtgraph is used if it is available; in this case see the GUI plot documentation to see the available commands
- Parameters:
- validate (True): bool
apply the invalidation bits
- index_only (False)bool
use index based X axis. This can be useful if the master (usually time based) is corrupted with NaN, inf or if it is not strictly increasing