canml Module

This is the top‐level package for canml.

class canml.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 options for BLF processing.

Parameters:

• chunk_size (int) – Number of messages per chunk. Defaults to 10000.

• progress_bar (bool) – Show tqdm bar if True. Defaults to True.

• dtype_map (Optional[Dict[str, Any]]) – Signal-to-dtype map. Defaults to None.

• sort_timestamps (bool) – Sort by timestamp. Defaults to False.

• force_uniform_timing (bool) – Uniform spacing of timestamps. Defaults to False.

• interval_seconds (float) – Uniform interval seconds. Defaults to 0.01.

• interpolate_missing (bool) – Interpolate missing signals. Defaults to False.

Raises:

ValueError – If chunk_size or interval_seconds <= 0.

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.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 pandas DataFrame chunks.

Logs total vs dropped message counts.

canml.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 an entire BLF file into a DataFrame.

Supports:

• ID and signal filtering

• Timestamp sorting and uniform spacing

• Missing signal injection with dtype preservation

• Metadata attributes and enum conversion

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

Load and merge DBC files with optional prefixing.

canml.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 chunks to CSV with side-car metadata JSON.

Parameters:

• df_or_iter (DataFrame or iterable) – Data to write.

• output_path (str) – Destination CSV file path.

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

• header (bool) – Include header in CSV.

• pandas_kwargs (dict) – Extra pandas.to_csv args.

• columns (list) – Subset of columns to write.

• metadata_path (str) – Path to JSON for signal_attributes.

canml.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 side-car metadata JSON.

Parameters:

• df (DataFrame) – Data to write.

• output_path (str) – Destination .parquet file path.

• compression (str) – Parquet codec.

• pandas_kwargs (dict) – Extra pandas.to_parquet args.

• metadata_path (str) – JSON path for signal_attributes.