canml.canmlio Module

This module provides the core APIs for decoding BLF files:

canmlio: Enhanced CAN BLF processing toolkit for production use. Module: canml/canmlio.py

Features:

• Merge multiple DBCs with namespace collision avoidance and caching.

• Stream-decode large BLF files into pandas DataFrame chunks.

• Full-file loading with uniform timestamp spacing, interpolation, and sorting.

• Signal/message filtering by CAN ID or signal name.

• Automatic injection of missing signals with dtype preservation.

• Incremental CSV & Parquet export with side-car metadata JSON.

• Generic support for enums and custom signal attributes.

• Configurable progress bars via tqdm.

class canml.canmlio.CanmlConfig(chunk_size: int = 10000, progress_bar: bool = True, dtype_map: Dict[str, Any] | None = None, sort_timestamps: bool = False, force_uniform_timing: bool = False, interval_seconds: float = 0.01, interpolate_missing: bool = False)

Bases: object

Configuration for BLF processing.

chunk_size

Number of messages per DataFrame chunk.

Type:

int

progress_bar

Show tqdm progress bar if True.

Type:

bool

dtype_map

Mapping from signal name to pandas dtype.

Type:

Dict[str, Any] | None

sort_timestamps

Sort final DataFrame by timestamp.

Type:

bool

force_uniform_timing

Override timestamps with uniform spacing.

Type:

bool

interval_seconds

Interval between timestamps for uniform timing.

Type:

float

interpolate_missing

Interpolate missing signal values if True.

Type:

bool

chunk_size: int = 10000

dtype_map: Dict[str, Any] | None = None

force_uniform_timing: bool = False

interpolate_missing: bool = False

interval_seconds: float = 0.01

progress_bar: bool = True

sort_timestamps: bool = False

canml.canmlio.iter_blf_chunks(blf_path: str, db: Database, config: CanmlConfig, filter_ids: Set[int] | None = None, filter_signals: Iterable[Any] | None = None) → Iterator[DataFrame]

Stream-decode a BLF file into DataFrame chunks.

Parameters:

• blf_path – Path to .blf log.

• db – Loaded CantoolsDatabase.

• config – CanmlConfig for chunk_size & progress_bar.

• filter_ids – Optional set of arbitration IDs to include.

• filter_signals – Optional iterable of signal identifiers to include.

Yields:

pandas.DataFrame chunks containing decoded signals + ‘timestamp’.

canml.canmlio.load_blf(blf_path: str, db: Database | str | List[str], config: CanmlConfig | None = None, message_ids: Set[int] | None = None, expected_signals: Iterable[Any] | None = None) → DataFrame

Load a BLF log into a pandas DataFrame with decoding, filtering, timing normalization, missing-signal injection, and metadata.

Parameters:

• blf_path – Path to .blf log file.

• db – CantoolsDatabase instance or path(s) to DBC file(s).

• config – CanmlConfig instance (optional).

• message_ids – Set of arbitration IDs to include (None = all).

• expected_signals – Iterable of signals to include (None = all).

Returns:

pandas.DataFrame with columns [‘timestamp’, …signals].

canml.canmlio.load_dbc_files(dbc_paths: str | List[str], prefix_signals: bool = False) → Database

Public API: Load and merge DBC file(s) into a CantoolsDatabase.

Parameters:

• dbc_paths – Path or list of paths to .dbc files.

• prefix_signals – If True, prefix each signal name with its message name.

Returns:

A CantoolsDatabase containing all loaded definitions.

canml.canmlio.to_csv(df_or_iter: DataFrame | Iterable[DataFrame], output_path: str, mode: str = 'w', header: bool = True, pandas_kwargs: Dict[str, Any] | None = None, columns: List[str] | None = None, metadata_path: str | None = None) → None

Write DataFrame or iterable of DataFrames to CSV, exporting metadata JSON.

Parameters:

• df_or_iter – Single DataFrame or iterable of chunks.

• output_path – CSV file path.

• mode – ‘w’ or ‘a’.

• header – Write header row if True.

• pandas_kwargs – Extra pandas.to_csv kwargs.

• columns – Optional list of columns (subset/order).

• metadata_path – Path to side-car JSON with df.attrs[“signal_attributes”].

canml.canmlio.to_parquet(df: DataFrame, output_path: str, compression: str = 'snappy', pandas_kwargs: Dict[str, Any] | None = None, metadata_path: str | None = None) → None

Write DataFrame to Parquet with optional metadata export.

Parameters:

• df – DataFrame to write.

• output_path – Path to .parquet file.

• compression – Parquet codec (snappy, gzip, zstd).

• pandas_kwargs – Extra pandas.to_parquet kwargs.

• metadata_path – Path to side-car JSON with df.attrs[“signal_attributes”].