canml.canmlio Module

This module provides the core APIs for decoding BLF files:

canmlio: Enhanced CAN BLF processing toolkit for production use.

Features:

• Merge multiple DBCs with namespace collision avoidance (optional prefixing).

• Stream‐decode large BLF files in pandas DataFrame chunks.

• Full‐file loading with optional uniform timestamp spacing.

• Signal‐ and message‐level filtering.

• Automatic injection of expected signals (NaN‐filled if missing).

• Incremental CSV export and Parquet export.

• Progress bars via tqdm.

Example: ```python from canml.canmlio import (

load_dbc_files, load_blf, to_csv

)

# Merge powertrain and chassis DBCs, prefix signals to avoid collisions db = load_dbc_files([“pt.dbc”,”chassis.dbc”], prefix_signals=True)

# Load BLF, only messages 0x100 & 0x200, inject expected signals df = load_blf(

blf_path=”log.blf”, db=db, message_ids={0x100,0x200}, expected_signals=[“EngineData_EngineRPM”,”BrakeStatus_ABSActive”], force_uniform_timing=True

)

to_csv(df, “decoded.csv”) ```

canml.canmlio.iter_blf_chunks(blf_path: str, db: Database, chunk_size: int = 10000, filter_ids: Set[int] | None = None) → Iterator[DataFrame]

Stream‐decode a BLF file in manageable pandas DataFrame chunks.

Parameters:

• blf_path – Path to the BLF log.

• db – cantools Database with message definitions (possibly prefixed).

• chunk_size – Rows per DataFrame chunk.

• filter_ids – If set, only decode messages with these arbitration IDs.

Yields:

DataFrame chunks with decoded signals + timestamp column.

Raises:

FileNotFoundError – If BLF file not found.

canml.canmlio.load_blf(blf_path: str, db: Database | str | List[str], message_ids: Set[int] | None = None, expected_signals: List[str] | None = None, force_uniform_timing: bool = False, interval_seconds: float = 0.01) → DataFrame

Load an entire BLF file into a DataFrame, with optional filters and signal injection.

Parameters:

• blf_path – Path to the BLF log.

• db – Database instance or DBC path(s).

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

• expected_signals – List of signal names to ensure exist (fill NaN if missing).

• force_uniform_timing – If True, override timestamps with uniform spacing.

• interval_seconds – Interval for uniform timing.

Returns:

A DataFrame with ‘timestamp’ + decoded signal columns.

Raises:

FileNotFoundError – If files missing.

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

Load and merge one or more DBC files into a single Database. Optionally prefix signal names with message names to avoid collisions.

Parameters:

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

• prefix_signals – If True, automatically rename signals to “<MessageName>_<SignalName>”.

Returns:

A cantools Database with all definitions loaded.

Raises:

• FileNotFoundError – If any DBC file is missing.

• ValueError – If loading fails.

canml.canmlio.to_csv(df_or_iter: DataFrame | Iterator[DataFrame], output_path: str, mode: str = 'w', header: bool = True) → None

Write DataFrame or iterator of DataFrames to CSV incrementally.

Parameters:

• df_or_iter – DataFrame or iterator of DataFrames.

• output_path – Destination CSV file.

• mode – Write mode (‘w’ or ‘a’).

• header – Write header for first block.

Raises:

ValueError – If write fails.

canml.canmlio.to_parquet(df: DataFrame, output_path: str, compression: str = 'snappy') → None

Write a DataFrame to Parquet.

Parameters:

• df – pandas DataFrame.

• output_path – ‘.parquet’ file path.

• compression – Parquet codec.

Raises:

ValueError – If write fails.