tessl/pypi-python-can

Controller Area Network interface module for Python providing common abstractions for CAN hardware devices and message handling utilities

—

Pending

Overview

Eval results

Files

File I/O and Logging

Name: tessl/pypi-python-can
Author: tessl

Comprehensive logging and replay system supporting multiple file formats with both reader and writer implementations. Enables recording CAN traffic for analysis, replay for testing, and conversion between different logging formats.

Capabilities

Generic Logging

Base logging functionality with automatic format detection and writer management.

class Logger:
    def __init__(self, filename: str):
        """
        Create a logger that writes to the specified file.
        Format determined by file extension.
        
        Parameters:
        - filename: Output file path (extension determines format)
        """
    
    def __call__(self, msg: Message) -> None:
        """Log a message (callable interface for use with Notifier)."""
    
    def stop(self) -> None:
        """Stop logging and close files."""

class BaseRotatingLogger:
    """Base class for size or time-based log rotation."""

class SizedRotatingLogger(BaseRotatingLogger):
    def __init__(self, base_filename: str, max_bytes: int):
        """
        Logger with automatic file rotation based on size.
        
        Parameters:
        - base_filename: Base name for log files
        - max_bytes: Maximum size before rotation
        """

Generic Playback

Message playback and synchronization utilities for replaying recorded CAN traffic.

class LogReader:
    def __init__(self, filename: str):
        """
        Read messages from log file. Format auto-detected from extension.
        
        Parameters:
        - filename: Input file path
        """
    
    def __iter__(self):
        """Iterate over messages in the log file."""

class MessageSync:
    def __init__(self, messages, timestamps=None, gap=0.0001):
        """
        Synchronize message playback with real-time or custom timing.
        
        Parameters:
        - messages: Iterable of messages
        - timestamps: Custom timestamp sequence
        - gap: Minimum gap between messages (seconds)
        """

ASC Format (Vector)

Vector CANalyzer/CANoe ASC log format support.

class ASCReader:
    def __init__(self, filename: str):
        """Read Vector ASC log files."""
    
    def __iter__(self):
        """Iterate over messages."""

class ASCWriter:
    def __init__(self, filename: str, channel: int = 1):
        """
        Write Vector ASC log files.
        
        Parameters:
        - filename: Output file path
        - channel: CAN channel number for logs
        """
    
    def on_message_received(self, msg: Message) -> None:
        """Write message to ASC file."""

BLF Format (Vector Binary)

Vector Binary Logging Format for high-performance logging.

class BLFReader:
    def __init__(self, filename: str):
        """Read Vector BLF (Binary Logging Format) files."""
    
    def __iter__(self):
        """Iterate over messages."""

class BLFWriter:
    def __init__(self, filename: str, channel: int = 1):
        """
        Write Vector BLF files.
        
        Parameters:
        - filename: Output file path  
        - channel: CAN channel number
        """
    
    def on_message_received(self, msg: Message) -> None:
        """Write message to BLF file."""

CSV Format

Comma-separated values format for easy analysis in spreadsheet applications.

class CSVReader:
    def __init__(self, filename: str):
        """Read CSV log files with standard CAN message format."""
    
    def __iter__(self):
        """Iterate over messages."""

class CSVWriter:
    def __init__(self, filename: str):
        """Write CSV log files with message details in columns."""
    
    def on_message_received(self, msg: Message) -> None:
        """Write message to CSV file."""

Candump Format (Linux)

Linux SocketCAN candump/canutils log format.

class CanutilsLogReader:
    def __init__(self, filename: str):
        """Read Linux candump format log files."""
    
    def __iter__(self):
        """Iterate over messages."""

class CanutilsLogWriter:
    def __init__(self, filename: str):
        """Write Linux candump format log files."""
    
    def on_message_received(self, msg: Message) -> None:
        """Write message in candump format."""

MF4 Format (ASAM MDF)

ASAM Measurement Data Format version 4 for automotive measurement data.

class MF4Reader:
    def __init__(self, filename: str):
        """Read ASAM MF4 measurement files."""
    
    def __iter__(self):
        """Iterate over messages."""

class MF4Writer:
    def __init__(self, filename: str):
        """Write ASAM MF4 measurement files."""
    
    def on_message_received(self, msg: Message) -> None:
        """Write message to MF4 file."""

SQLite Database

SQLite database format for efficient querying and analysis.

class SqliteReader:
    def __init__(self, filename: str):
        """Read messages from SQLite database."""
    
    def __iter__(self):
        """Iterate over messages."""

class SqliteWriter:
    def __init__(self, filename: str, table_name: str = 'messages'):
        """
        Write messages to SQLite database.
        
        Parameters:
        - filename: Database file path
        - table_name: Table name for messages
        """
    
    def on_message_received(self, msg: Message) -> None:
        """Write message to database."""

TRC Format

TRC trace format support with version selection.

class TRCFileVersion(Enum):
    """TRC file format versions."""
    V1_0 = "1.0"
    V1_1 = "1.1"
    V2_0 = "2.0"

class TRCReader:
    def __init__(self, filename: str):
        """Read TRC trace files."""
    
    def __iter__(self):
        """Iterate over messages."""

class TRCWriter:
    def __init__(self, filename: str, version: TRCFileVersion = TRCFileVersion.V2_0):
        """
        Write TRC trace files.
        
        Parameters:
        - filename: Output file path
        - version: TRC format version
        """
    
    def on_message_received(self, msg: Message) -> None:
        """Write message to TRC file."""

Console Output

Print messages to console for real-time monitoring.

class Printer:
    def __init__(self, filename=None):
        """
        Print messages to console or file.
        
        Parameters:
        - filename: Optional file for output (None for stdout)
        """
    
    def on_message_received(self, msg: Message) -> None:
        """Print message in human-readable format."""

Format Registry

Automatic format detection and writer/reader selection.

MESSAGE_READERS: dict[str, type]
"""Maps file extensions to reader classes."""

MESSAGE_WRITERS: dict[str, type]  
"""Maps file extensions to writer classes."""

Usage Examples

Basic Logging

import can

# Create bus and logger
bus = can.Bus(channel='can0', interface='socketcan')
logger = can.Logger('traffic.blf')

# Log messages manually
for _ in range(100):
    msg = bus.recv(timeout=1.0)
    if msg:
        logger(msg)

logger.stop()
bus.shutdown()

Automatic Logging with Notifier

import can
import time

bus = can.Bus(channel='can0', interface='socketcan')

# Multiple loggers simultaneously
loggers = [
    can.Logger('traffic.blf'),    # Binary format
    can.Logger('traffic.asc'),    # ASCII format  
    can.Logger('traffic.csv'),    # CSV format
    can.Printer()                 # Console output
]

# Start logging
notifier = can.Notifier(bus, loggers)

# Let it log for 30 seconds
time.sleep(30)

# Stop logging
notifier.stop()
for logger in loggers:
    if hasattr(logger, 'stop'):
        logger.stop()

bus.shutdown()

Log Playback

import can
import time

# Read from log file
log_reader = can.LogReader('recorded_traffic.blf')

# Create playback bus
bus = can.Bus(channel='test', interface='virtual')

# Replay messages with original timing
for msg in can.MessageSync(log_reader, gap=0.0001):
    bus.send(msg)
    # MessageSync handles timing automatically

bus.shutdown()

Log Conversion

import can

# Convert between formats
input_reader = can.io.ASCReader('input.asc')
output_writer = can.io.BLFWriter('output.blf')

for msg in input_reader:
    output_writer.on_message_received(msg)

output_writer.stop()

Size-Based Log Rotation

import can
import time

bus = can.Bus(channel='can0', interface='socketcan')

# Rotate logs every 10MB
rotating_logger = can.SizedRotatingLogger(
    base_filename='can_traffic.blf',
    max_bytes=10 * 1024 * 1024  # 10MB
)

notifier = can.Notifier(bus, [rotating_logger])

# Log for extended period with automatic rotation
time.sleep(3600)  # 1 hour

notifier.stop()
rotating_logger.stop()
bus.shutdown()

SQLite Analysis

import can
import sqlite3

# Log to SQLite database
bus = can.Bus(channel='can0', interface='socketcan')
db_logger = can.SqliteWriter('can_data.db')

notifier = can.Notifier(bus, [db_logger])
time.sleep(60)  # Log for 1 minute
notifier.stop()
db_logger.stop()
bus.shutdown()

# Analyze logged data
conn = sqlite3.connect('can_data.db')
cursor = conn.cursor()

# Find most common message IDs
cursor.execute("""
    SELECT arbitration_id, COUNT(*) as count 
    FROM messages 
    GROUP BY arbitration_id 
    ORDER BY count DESC 
    LIMIT 10
""")

for row in cursor.fetchall():
    print(f"ID: 0x{row[0]:X}, Count: {row[1]}")

conn.close()

Types

from typing import Dict, Type, Iterator, Optional, Any
from abc import ABC, abstractmethod
from enum import Enum

class Listener(ABC):
    """Base class for message listeners/loggers."""
    
    @abstractmethod
    def on_message_received(self, msg: Message) -> None: ...
    
    def stop(self) -> None: ...

# Format registries
MESSAGE_READERS: Dict[str, Type] = {
    '.asc': ASCReader,
    '.blf': BLFReader, 
    '.csv': CSVReader,
    '.log': CanutilsLogReader,
    '.mf4': MF4Reader,
    '.db': SqliteReader,
    '.trc': TRCReader
}

MESSAGE_WRITERS: Dict[str, Type] = {
    '.asc': ASCWriter,
    '.blf': BLFWriter,
    '.csv': CSVWriter, 
    '.log': CanutilsLogWriter,
    '.mf4': MF4Writer,
    '.db': SqliteWriter,
    '.trc': TRCWriter
}

Install with Tessl CLI