Virtual#

The virtual interface can be used as a way to write OS and driver independent tests. Any VirtualBus instances connecting to the same channel (from within the same Python process) will receive each others messages.

If messages shall be sent across process or host borders, consider using the Multicast IP Interface and refer to Virtual Interfaces for a comparison and general discussion of different virtual interfaces.

Example#

import can

bus1 = can.interface.Bus('test', bustype='virtual')
bus2 = can.interface.Bus('test', bustype='virtual')

msg1 = can.Message(arbitration_id=0xabcde, data=[1,2,3])
bus1.send(msg1)
msg2 = bus2.recv()

#assert msg1 == msg2
assert msg1.arbitration_id == msg2.arbitration_id
assert msg1.data == msg2.data
assert msg1.timestamp != msg2.timestamp

import can

bus1 = can.interface.Bus('test', bustype='virtual', preserve_timestamps=True)
bus2 = can.interface.Bus('test', bustype='virtual')

msg1 = can.Message(timestamp=1639740470.051948, arbitration_id=0xabcde, data=[1,2,3])

# Messages sent on bus1 will have their timestamps preserved when received
# on bus2
bus1.send(msg1)
msg2 = bus2.recv()

assert msg1.arbitration_id == msg2.arbitration_id
assert msg1.data == msg2.data
assert msg1.timestamp == msg2.timestamp

# Messages sent on bus2 will not have their timestamps preserved when
# received on bus1
bus2.send(msg1)
msg3 = bus1.recv()

assert msg1.arbitration_id == msg3.arbitration_id
assert msg1.data == msg3.data
assert msg1.timestamp != msg3.timestamp

Bus Class Documentation#

class can.interfaces.virtual.VirtualBus(channel=None, receive_own_messages=False, rx_queue_size=0, preserve_timestamps=False, **kwargs)[source]#

A virtual CAN bus using an internal message queue. It can be used for example for testing.

In this interface, a channel is an arbitrary object used as an identifier for connected buses.

Implements can.BusABC._detect_available_configs(); see _detect_available_configs() for how it behaves here.

Note

The timeout when sending a message applies to each receiver individually. This means that sending can block up to 5 seconds if a message is sent to 5 receivers with the timeout set to 1.0.

Warning

This interface guarantees reliable delivery and message ordering, but does not implement rate limiting or ID arbitration/prioritization under high loads. Please refer to the section Virtual Interfaces for more information on this and a comparison to alternatives.

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 – See set_filters() for details.

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

  • receive_own_messages (bool) –

  • rx_queue_size (int) –

  • preserve_timestamps (bool) –

Raises
Return type

None

static _detect_available_configs()[source]#

Returns all currently used channels as well as one other currently unused channel.

Note

This method will run into problems if thousands of autodetected buses are used at once.

Return type

List[can.typechecking.AutoDetectedConfig]

send(msg, timeout=None)[source]#

Transmit a message to the CAN bus.

Override this method to enable the transmit path.

Parameters

  • msg (Message) – A message object.

  • timeout (Optional[float]) – If > 0, wait up to this many seconds for message to be ACK’ed or for transmit queue to be ready depending on driver implementation. If timeout is exceeded, an exception will be raised. Might not be supported by all interfaces. None blocks indefinitely.

Raises

CanOperationError – If an error occurred while sending

Return type

None

shutdown()[source]#

Called to carry out any interface specific cleanup required in shutting down a bus.

Return type

None