Hide navigation sidebar
Hide table of contents sidebar
Back to top
Toggle table of contents sidebar

# 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.

abstract 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 (Any) – The can interface identifier. Expected type is backend dependent.

    - can_filters (Sequence[CanFilter | CanFilterExtended] | 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 | CanFilterExtended] | None) – See `set_filters()` for details.

  Return type:

    None

BusABC._send_periodic_internal(msgs, period, duration=None, 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.

    - 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:

    List[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 usually extends `can.io.generic.BaseIOHandler`, while the writer often additionally extends `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 often extends `can.io.generic.BaseIOHandler`, but does not have to). Besides from a constructor, only `__iter__(self)` needs to be implemented.

3. Implement a writer `CanstoreWriter` (which often extends `can.io.generic.BaseIOHandler` and `can.Listener`, but does not have to). 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

Contains generic base classes for file IO.

class can.io.generic.BaseIOHandler(file, mode='rt', **kwargs)

  A generic file handler that can be used for reading and writing.

  Can be used as a context manager.

  Attr file:

    the file-like object that is kept internally, or None if none was opened

  Parameters:

    - file (TextIO | BinaryIO | GzipFile | None) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

  stop()

    Closes the underlying file-like object and flushes it, if it was opened in write mode.

    Return type:

      None

class can.io.generic.BinaryIOMessageReader(file, mode='rt', **kwargs)
  Parameters:

    - file (BinaryIO | GzipFile) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

class can.io.generic.BinaryIOMessageWriter(file, mode='wt', **kwargs)
  Parameters:

    - file (BinaryIO | GzipFile) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

class can.io.generic.FileIOMessageWriter(file, mode='wt', **kwargs)

  A specialized base class for all writers with file descriptors.

  Parameters:

    - file (TextIO | BinaryIO | GzipFile) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

  file_size()

    Return an estimate of the current file size in bytes.

    Return type:

      int

class can.io.generic.MessageReader(file, mode='rt', **kwargs)

  The base class for all readers.

  Parameters:

    - file (TextIO | BinaryIO | GzipFile | None) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

class can.io.generic.MessageWriter(file, mode='rt', **kwargs)

  The base class for all writers.

  Parameters:

    - file (TextIO | BinaryIO | GzipFile | None) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

class can.io.generic.TextIOMessageReader(file, mode='rt', **kwargs)
  Parameters:

    - file (TextIO) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

class can.io.generic.TextIOMessageWriter(file, mode='wt', **kwargs)
  Parameters:

    - file (TextIO) – a path-like object to open a file, a file-like object to be used as a file or None to not use a file at all

    - mode (str) – the mode that should be used to open the file, see `open()`, ignored if file is None

    - kwargs (Any) –

## 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 (str | 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 (TextIO | BinaryIO | GzipFile | 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 (TextIO | BinaryIO | GzipFile | 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]