Skip to content
Back to top
View this page

# Internal API

Here we document the odds and ends that are more helpful for creating your own interfaces or listeners but generally shouldn’t be required to interact with python-can.

## BusABC

The `BusABC` class, as the name suggests, provides an abstraction of a CAN bus. The bus provides a wrapper around a physical or virtual CAN Bus.

An interface specific instance of the `BusABC` is created by the `Bus` class, see Bus for the user facing API.

## Extending the `BusABC` class

Concrete implementations must implement the following:

  - `send()` to send individual messages

  - `_recv_internal()` to receive individual messages (see note below!)

  - set the `channel_info` attribute to a string describing the underlying bus and/or channel

They might implement the following:

  - `flush_tx_buffer()` to allow discarding any messages yet to be sent

  - `shutdown()` to override how the bus should shut down

  - `_send_periodic_internal()` to override the software based periodic sending and push it down to the kernel or hardware.

  - `_apply_filters()` to apply efficient filters to lower level systems like the OS kernel or hardware.

  - `_detect_available_configs()` to allow the interface to report which configurations are currently available for new connections.

  - `state()` property to allow reading and/or changing the bus state.

Note

TL;DR: Only override `_recv_internal()`, never `recv()` directly.

Previously, concrete bus classes had to override `recv()` directly instead of `_recv_internal()`, but that has changed to allow the abstract base class to handle in-software message filtering as a fallback. All internal interfaces now implement that new behaviour. Older (custom) interfaces might still be implemented like that and thus might not provide message filtering:

Concrete instances are usually created by `can.Bus()` which takes the users configuration into account.

### Bus Internals

Several methods are not documented in the main `can.BusABC` as they are primarily useful for library developers as opposed to library users.

abstractmethod BusABC.__init__(channel, can_filters=None, **kwargs)

  Construct and open a CAN bus instance of the specified type.

  Subclasses should call though this method with all given parameters as it handles generic tasks like applying filters.

  Parameters:

    - channel (int | str | Sequence[int]) – The can interface identifier. Expected type is backend dependent.

    - can_filters (Sequence[CanFilter] | None) – See `set_filters()` for details.

    - kwargs (dict) – Any backend dependent configurations are passed in this dictionary

  Raises:

    - ValueError – If parameters are out of range

    - CanInterfaceNotImplementedError – If the driver cannot be accessed

    - CanInitializationError – If the bus cannot be initialized

BusABC.__iter__()

  Allow iteration on messages as they are received.

      for msg in bus:
          print(msg)

  Yields:

    `Message` msg objects.

  Return type:

    Iterator[Message]

BusABC.__str__()

  Return str(self).

  Return type:

    str

BusABC.__weakref__

  list of weak references to the object

BusABC._recv_internal(timeout)

  Read a message from the bus and tell whether it was filtered. This methods may be called by `recv()` to read a message multiple times if the filters set by `set_filters()` do not match and the call has not yet timed out.

  New implementations should always override this method instead of `recv()`, to be able to take advantage of the software based filtering provided by `recv()` as a fallback. This method should never be called directly.

  Note

  This method is not an @abstractmethod (for now) to allow older external implementations to continue using their existing `recv()` implementation.

  Note

  The second return value (whether filtering was already done) may change over time for some interfaces, like for example in the Kvaser interface. Thus it cannot be simplified to a constant value.

  Parameters:

    timeout (float) – seconds to wait for a message, see `send()`

  Returns:

    1. a message that was read or None on timeout

    2. a bool that is True if message filtering has already been done and else False

  Raises:

    - CanOperationError – If an error occurred while reading

    - NotImplementedError – if the bus provides it’s own `recv()` implementation (legacy implementation)

  Return type:

    tuple[Message | None, bool]

BusABC._apply_filters(filters)

  Hook for applying the filters to the underlying kernel or hardware if supported/implemented by the interface.

  Parameters:

    filters (Sequence[CanFilter] | None) – See `set_filters()` for details.

  Return type:

    None

BusABC._send_periodic_internal(msgs, period, duration=None, autostart=True, modifier_callback=None)

  Default implementation of periodic message sending using threading.

  Override this method to enable a more efficient backend specific approach.

  Parameters:

    - msgs (Sequence[Message] | Message) – Messages to transmit

    - period (float) – Period in seconds between each message

    - duration (float | None) – The duration between sending each message at the given rate. If no duration is provided, the task will continue indefinitely.

    - autostart (bool) – If True (the default) the sending task will immediately start after creation. Otherwise, the task has to be started by calling the tasks `start()` method on it.

    - modifier_callback (Callable[[Message], None] | None)

  Returns:

    A started task instance. Note the task can be stopped (and depending on the backend modified) by calling the `stop()` method.

  Return type:

    CyclicSendTaskABC

static BusABC._detect_available_configs()

  Detect all configurations/channels that this interface could currently connect with.

  This might be quite time consuming.

  May not to be implemented by every interface on every platform.

  Returns:

    an iterable of dicts, each being a configuration suitable for usage in the interface’s bus constructor.

  Return type:

    Sequence[AutoDetectedConfig]

## About the IO module

Handling of the different file formats is implemented in `can.io`. Each file/IO type is within a separate module and ideally implements both a Reader and a Writer. The reader extends `can.io.generic.MessageReader`, while the writer extends `can.io.generic.MessageWriter`, a subclass of the `can.Listener`, to be able to be passed directly to a `can.Notifier`.

### Adding support for new file formats

This assumes that you want to add a new file format, called canstore. Ideally add both reading and writing support for the new file format, although this is not strictly required.

1. Create a new module: can/io/canstore.py (or simply copy some existing one like can/io/csv.py)

2. Implement a reader `CanstoreReader` which extends `can.io.generic.MessageReader`. Besides from a constructor, only `__iter__(self)` needs to be implemented.

3. Implement a writer `CanstoreWriter` which extends `can.io.generic.MessageWriter`. Besides from a constructor, only `on_message_received(self, msg)` needs to be implemented.

4. Add a case to `can.io.player.LogReader`’s `__new__()`.

5. Document the two new classes (and possibly additional helpers) with docstrings and comments. Please mention features and limitations of the implementation.

6. Add a short section to the bottom of doc/listeners.rst.

7. Add tests where appropriate, for example by simply adding a test case called class TestCanstoreFileFormat(ReaderWriterTest) to test/logformats_test.py. That should already handle all of the general testing. Just follow the way the other tests in there do it.

8. Add imports to can/__init__py and can/io/__init__py so that the new classes can be simply imported as from can import CanstoreReader, CanstoreWriter.

### IO Utilities

This module provides abstract base classes for CAN message reading and writing operations to various file formats.

Note

All classes in this module are abstract and should be subclassed to implement specific file format handling.

class can.io.generic._IoTypeVar

  type parameter used in generic classes `MessageReader` and `MessageWriter`

  alias of TypeVar(‘_IoTypeVar’, bound=`IO`[`Any`] | `TextIOWrapper` | `BufferedIOBase`)

class can.io.generic.MessageWriter(file, **kwargs)

  Bases: `AbstractContextManager`[MessageWriter], `Listener`, `ABC`

  Abstract base class for all CAN message writers.

  This class serves as a foundation for implementing different message writer formats. It combines context manager capabilities with the message listener interface.

  Parameters:

    - file (str | os.PathLike[str]) – Path-like object or string representing the output file location

    - kwargs (Any) – Additional keyword arguments for specific writer implementations

  abstractmethod stop()

    Stop handling messages and cleanup any resources.

    Return type:

      None

class can.io.generic.SizedMessageWriter(file, **kwargs)

  Bases: `MessageWriter`, `ABC`

  Abstract base class for message writers that can report their file size.

  This class extends `MessageWriter` with the ability to determine the size of the output file.

  Parameters:

    - file (str | os.PathLike[str])

    - kwargs (Any)

  abstractmethod file_size()

    Get the current size of the output file in bytes.

    Returns:

      The size of the file in bytes

    Return type:

      int

class can.io.generic.FileIOMessageWriter(file, **kwargs)

  Bases: `SizedMessageWriter`, `Generic`[`_IoTypeVar`]

  Base class for writers that operate on file descriptors.

  This class provides common functionality for writers that work with file objects.

  Parameters:

    - file (_IoTypeVar) – A path-like object or file object to write to

    - kwargs (Any) – Additional keyword arguments for specific writer implementations

  Variables:

    file – The file object being written to

  stop()

    Close the file and stop writing.

    Return type:

      None

  file_size()

    Get the current file size.

    Return type:

      int

class can.io.generic.TextIOMessageWriter(file, mode='w', **kwargs)

  Bases: `FileIOMessageWriter`[`TextIO` | `TextIOWrapper`], `ABC`

  Text-based message writer implementation.

  Parameters:

    - file (_IoTypeVar) – Text file to write to

    - mode (Union[OpenTextModeUpdating, OpenTextModeWriting]) – File open mode for text operations

    - kwargs (Any) – Additional arguments like encoding

class can.io.generic.BinaryIOMessageWriter(file, mode='wb', **kwargs)

  Bases: `FileIOMessageWriter`[`BinaryIO` | `BufferedIOBase`], `ABC`

  Binary file message writer implementation.

  Parameters:

    - file (_IoTypeVar) – Binary file to write to

    - mode (Union[OpenBinaryModeUpdating, OpenBinaryModeWriting]) – File open mode for binary operations

    - kwargs (Any) – Additional implementation specific arguments

class can.io.generic.MessageReader(file, **kwargs)

  Bases: `AbstractContextManager`[MessageReader], `Iterable`[`Message`], `ABC`

  Abstract base class for all CAN message readers.

  This class serves as a foundation for implementing different message reader formats. It combines context manager capabilities with iteration interface.

  Parameters:

    - file (str | os.PathLike[str]) – Path-like object or string representing the input file location

    - kwargs (Any) – Additional keyword arguments for specific reader implementations

  abstractmethod stop()

    Stop reading messages and cleanup any resources.

    Return type:

      None

class can.io.generic.FileIOMessageReader(file, **kwargs)

  Bases: `MessageReader`, `Generic`[`_IoTypeVar`]

  Base class for readers that operate on file descriptors.

  This class provides common functionality for readers that work with file objects.

  Parameters:

    - file (_IoTypeVar) – A path-like object or file object to read from

    - kwargs (Any) – Additional keyword arguments for specific reader implementations

  Variables:

    file – The file object being read from

  stop()

    Stop reading messages and cleanup any resources.

    Return type:

      None

class can.io.generic.TextIOMessageReader(file, mode='r', **kwargs)

  Bases: `FileIOMessageReader`[`TextIO` | `TextIOWrapper`], `ABC`

  Text-based message reader implementation.

  Parameters:

    - file (_IoTypeVar) – Text file to read from

    - mode (OpenTextModeReading) – File open mode for text operations

    - kwargs (Any) – Additional arguments like encoding

class can.io.generic.BinaryIOMessageReader(file, mode='rb', **kwargs)

  Bases: `FileIOMessageReader`[`BinaryIO` | `BufferedIOBase`], `ABC`

  Binary file message reader implementation.

  Parameters:

    - file (_IoTypeVar) – Binary file to read from

    - mode (OpenBinaryModeReading) – File open mode for binary operations

    - kwargs (Any) – Additional implementation specific arguments

## Other Utilities

Utilities and configuration file parsing.

can.util.cast_from_string(string_val)

  Perform trivial type conversion from `str` values.

  Parameters:

    string_val (str) – the string, that shall be converted

  Return type:

    str | int | float | bool

can.util.channel2int(channel)

  Try to convert the channel to an integer.

  Parameters:

    channel (int | str | Sequence[int] | None) – Channel string (e.g. “can0”, “CAN1”) or an integer

  Returns:

    Channel integer or `None` if unsuccessful

  Return type:

    int | None

can.util.check_or_adjust_timing_clock(timing, valid_clocks)

  Adjusts the given timing instance to have an f_clock value that is within the allowed values specified by valid_clocks. If the f_clock value of timing is already within valid_clocks, then timing is returned unchanged.

  Parameters:

    - timing (T2) – The `BitTiming` or `BitTimingFd` instance to adjust.

    - valid_clocks (Iterable[int]) – An iterable of integers representing the valid f_clock values that the timing instance can be changed to. The order of the values in valid_clocks determines the priority in which they are tried, with earlier values being tried before later ones.

  Returns:

    A new `BitTiming` or `BitTimingFd` instance with an f_clock value within valid_clocks.

  Raises:

    CanInitializationError – If no compatible f_clock value can be found within valid_clocks.

  Return type:

    T2

can.util.deprecated_args_alias(deprecation_start, deprecation_end=None, **aliases)

  Allows to rename/deprecate a function kwarg(s) and optionally have the deprecated kwarg(s) set as alias(es)

  Example:

      @deprecated_args_alias("1.2.0", oldArg="new_arg", anotherOldArg="another_new_arg")
      def library_function(new_arg, another_new_arg):
          pass

      @deprecated_args_alias(
          deprecation_start="1.2.0",
          deprecation_end="3.0.0",
          oldArg="new_arg",
          obsoleteOldArg=None,
      )
      def library_function(new_arg):
          pass

  Parameters:

    - deprecation_start (str) – The python-can version, that introduced the `DeprecationWarning`.

    - deprecation_end (str | None) – The python-can version, that marks the end of the deprecation period.

    - aliases (str | None) – keyword arguments, that map the deprecated argument names to the new argument names or `None`.

  Return type:

    Callable[[Callable[[~P1], T1]], Callable[[~P1], T1]]

can.util.dlc2len(dlc)

  Calculate the data length from DLC.

  Parameters:

    dlc (int) – DLC (0-15)

  Returns:

    Data length in number of bytes (0-64)

  Return type:

    int

can.util.len2dlc(length)

  Calculate the DLC from data length.

  Parameters:

    length (int) – Length in number of bytes (0-64)

  Returns:

    DLC (0-15)

  Return type:

    int

can.util.load_config(path=None, config=None, context=None)

  Returns a dict with configuration details which is loaded from (in this order):

  - config

  - can.rc

  - Environment variables CAN_INTERFACE, CAN_CHANNEL, CAN_BITRATE

  - Config files `/etc/can.conf` or `~/.can` or `~/.canrc` where the latter may add or replace values of the former.

  Interface can be any of the strings from `can.VALID_INTERFACES` for example: kvaser, socketcan, pcan, usb2can, ixxat, nican, virtual.

  Note

  The key `bustype` is copied to `interface` if that one is missing and does never appear in the result.

  Parameters:

    - path (str | PathLike[str] | None) – Optional path to config file.

    - config (dict[str, Any] | None) – A dict which may set the ‘interface’, and/or the ‘channel’, or neither. It may set other values that are passed through.

    - context (str | None) – Extra ‘context’ pass to config sources. This can be used to section other than ‘default’ in the configuration file.

  Returns:

    A config dictionary that should contain ‘interface’ & ‘channel’:

        {
            'interface': 'python-can backend interface to use',
            'channel': 'default channel to use',
            # possibly more
        }

    Note `None` will be used if all the options are exhausted without finding a value.

    All unused values are passed from `config` over to this.

  Raises:

    CanInterfaceNotImplementedError if the `interface` name isn’t recognized

  Return type:

    BusConfig

can.util.load_environment_config(context=None)

  Loads config dict from environmental variables (if set):

  - CAN_INTERFACE

  - CAN_CHANNEL

  - CAN_BITRATE

  - CAN_CONFIG

  if context is supplied, “_{context}” is appended to the environment variable name we will look at. For example if context=”ABC”:

  - CAN_INTERFACE_ABC

  - CAN_CHANNEL_ABC

  - CAN_BITRATE_ABC

  - CAN_CONFIG_ABC

  Parameters:

    context (str | None)

  Return type:

    dict[str, str]

can.util.load_file_config(path=None, section='default')

  Loads configuration from file with following content:

      [default]
      interface = socketcan
      channel = can0

  Parameters:

    - path (str | PathLike[str] | None) – path to config file. If not specified, several sensible default locations are tried depending on platform.

    - section (str) – name of the section to read configuration from.

  Return type:

    dict[str, str]

can.util.set_logging_level(level_name)

  Set the logging level for the “can” logger.

  Parameters:

    level_name (str) – One of: ‘critical’, ‘error’, ‘warning’, ‘info’, ‘debug’, ‘subdebug’, or the value `None` (=default). Defaults to ‘debug’.

  Return type:

    None

can.util.time_perfcounter_correlation()

  Get the perf_counter value nearest to when time.time() is updated

  Computed if the default timer used by time.time on this platform has a resolution higher than 10μs, otherwise the current time and perf_counter is directly returned. This was chosen as typical timer resolution on Linux/macOS is ~1μs, and the Windows platform can vary from ~500μs to 10ms.

  Note this value is based on when time.time() is observed to update from Python, it is not directly returned by the operating system.

  Returns:

    (t, performance_counter) time.time value and time.perf_counter value when the time.time is updated

  Return type:

    tuple[float, float]