Notifier and Listeners

Notifier

The Notifier object is used as a message distributor for a bus. The Notifier uses an event loop or creates a thread to read messages from the bus and distributes them to listeners.

class can.Notifier(bus, listeners, timeout=1.0, loop=None)

Manages the distribution of Message instances to listeners.

Supports multiple buses and listeners.

Note

Remember to call stop() after all messages are received as many listeners carry out flush operations to persist data.

Parameters:

• bus (BusABC | list[BusABC]) – A Bus or a list of buses to consume messages from.

• listeners (Iterable[Listener | Callable[[Message], Awaitable[None] | None]]) – An iterable of Listener or callables that receive a Message and return nothing.

• timeout (float) – An optional maximum number of seconds to wait for any Message.

• loop (AbstractEventLoop | None) – An asyncio event loop to schedule the listeners in.

Raises:

ValueError – If a passed in bus is already assigned to an active Notifier.

add_bus(bus)

Add a bus for notification.

Parameters:

bus (BusABC) – CAN bus instance.

Raises:

ValueError – If the bus is already assigned to an active Notifier.

Return type:

None

add_listener(listener)

Add new Listener to the notification list. If it is already present, it will be called two times each time a message arrives.

Parameters:

listener (Listener | Callable[[Message], Awaitable[None] | None]) – Listener to be added to the list to be notified

Return type:

None

property bus: BusABC | tuple[BusABC, ...]

Return the associated bus or a tuple of buses.

exception: Exception | None

Exception raised in thread

static find_instances(bus)

Find Notifier instances associated with a given CAN bus.

This method searches the registry for the Notifier that is linked to the specified bus. If the bus is found, the corresponding Notifier instances are returned. If the bus is not found in the registry, an empty tuple is returned.

Parameters:

bus (BusABC) – The CAN bus for which to find the associated Notifier .

Returns:

A tuple of Notifier instances associated with the given bus.

Return type:

tuple[Notifier, …]

remove_listener(listener)

Remove a listener from the notification list. This method throws an exception if the given listener is not part of the stored listeners.

Parameters:

listener (Listener | Callable[[Message], Awaitable[None] | None]) – Listener to be removed from the list to be notified

Raises:

ValueError – if listener was never added to this notifier

Return type:

None

stop(timeout=5.0)

Stop notifying Listeners when new Message objects arrive and call stop() on each Listener.

Parameters:

timeout (float) – Max time in seconds to wait for receive threads to finish. Should be longer than timeout given at instantiation.

Return type:

None

property stopped: bool

Return True, if Notifier was properly shut down with stop().

Listener

The Listener class is an “abstract” base class for any objects which wish to register to receive notifications of new messages on the bus. A Listener can be used in two ways; the default is to call the Listener with a new message, or by calling the method on_message_received.

Listeners are registered with Notifier object(s) which ensure they are notified whenever a new message is received.

#!/usr/bin/env python

import time

import can


def main():
    with can.Bus(interface="virtual", receive_own_messages=True) as bus:
        print_listener = can.Printer()
        with can.Notifier(bus, listeners=[print_listener]):
            # using Notifier as a context manager automatically calls `Notifier.stop()`
            # at the end of the `with` block
            bus.send(can.Message(arbitration_id=1, is_extended_id=True))
            bus.send(can.Message(arbitration_id=2, is_extended_id=True))
            bus.send(can.Message(arbitration_id=1, is_extended_id=False))
            time.sleep(1.0)


if __name__ == "__main__":
    main()

Subclasses of Listener that do not override on_message_received will cause NotImplementedError to be thrown when a message is received on the CAN bus.

class can.Listener

The basic listener that can be called directly to handle some CAN message:

listener = SomeListener()
msg = my_bus.recv()

# now either call
listener(msg)
# or
listener.on_message_received(msg)

# Important to ensure all outputs are flushed
listener.stop()

on_error(exc)

This method is called to handle any exception in the receive thread.

Parameters:

exc (Exception) – The exception causing the thread to stop

Return type:

None

abstractmethod on_message_received(msg)

This method is called to handle the given message.

Parameters:

msg (Message) – the delivered message

Return type:

None

stop()

Stop handling new messages, carry out any final tasks to ensure data is persisted and cleanup any open resources.

Concrete implementations override.

Return type:

None

There are some listeners that already ship together with python-can and are listed below. Some of them allow messages to be written to files, and the corresponding file readers are also documented here.

Note

Please note that writing and the reading a message might not always yield a completely unchanged message again, since some properties are not (yet) supported by some file formats.

Note

Additional file formats for both reading/writing log files can be added via a plugin reader/writer. An external package can register a new reader by using the can.io.message_reader entry point. Similarly, a writer can be added using the can.io.message_writer entry point.

The format of the entry point is reader_name=module:classname where classname is a concrete implementation of MessageReader or MessageWriter.

entry_points={
    'can.io.message_reader': [
       '.asc = my_package.io.asc:ASCReader'
   ]
},

BufferedReader

class can.BufferedReader

A BufferedReader is a subclass of Listener which implements a message buffer: that is, when the can.BufferedReader instance is notified of a new message it pushes it into a queue of messages waiting to be serviced. The messages can then be fetched with get_message().

Putting in messages after stop() has been called will raise an exception, see on_message_received().

Attr is_stopped:

True if the reader has been stopped

get_message(timeout=0.5)

Attempts to retrieve the message that has been in the queue for the longest amount of time (FIFO). If no message is available, it blocks for given timeout or until a message is received (whichever is shorter), or else returns None. This method does not block after can.BufferedReader.stop() has been called.

Parameters:

timeout (float) – The number of seconds to wait for a new message.

Returns:

the received can.Message or None, if the queue is empty.

Return type:

Message | None

on_message_received(msg)

Append a message to the buffer.

Raises:

BufferError if the reader has already been stopped

Parameters:

msg (Message)

Return type:

None

stop()

Prohibits any more additions to this reader.

Return type:

None

class can.AsyncBufferedReader(**kwargs)

A message buffer for use with asyncio.

See Asyncio support for how to use with can.Notifier.

Can also be used as an asynchronous iterator:

async for msg in reader:
    print(msg)

Parameters:

kwargs (Any)

async get_message()

Retrieve the latest message when awaited for:

msg = await reader.get_message()

Returns:

The CAN message.

Return type:

Message

on_message_received(msg)

Append a message to the buffer.

Must only be called inside an event loop!

Parameters:

msg (Message)

Return type:

None

stop()

Stop handling new messages, carry out any final tasks to ensure data is persisted and cleanup any open resources.

Concrete implementations override.

Return type:

None

RedirectReader

class can.RedirectReader(bus, *args, **kwargs)

A RedirectReader sends all received messages to another Bus.

Parameters:

• bus (BusABC)

• args (Any)

• kwargs (Any)

on_message_received(msg)

This method is called to handle the given message.

Parameters:

msg (Message) – the delivered message

Return type:

None