The various programs share a lot of code, so let's look at what the building blocks are.

Begin by importing the modules that are needed by our code:

import struct
import time

import board
import canio
import digitalio

1
2
3
4
5
6

import struct
import time

import board
import canio
import digitalio

Create the necessary digital pin settings needed to enable the CAN Transceiver chip:

# If the CAN transceiver has a standby pin, bring it out of standby mode
if hasattr(board, 'CAN_STANDBY'):
    standby = digitalio.DigitalInOut(board.CAN_STANDBY)
    standby.switch_to_output(False)

# If the CAN transceiver is powered by a boost converter, turn on its supply
if hasattr(board, 'BOOST_ENABLE'):
    boost_enable = digitalio.DigitalInOut(board.BOOST_ENABLE)
    boost_enable.switch_to_output(True)

1
2
3
4
5
6
7
8
9

# If the CAN transceiver has a standby pin, bring it out of standby mode
if hasattr(board, 'CAN_STANDBY'):
    standby = digitalio.DigitalInOut(board.CAN_STANDBY)
    standby.switch_to_output(False)

# If the CAN transceiver is powered by a boost converter, turn on its supply
if hasattr(board, 'BOOST_ENABLE'):
    boost_enable = digitalio.DigitalInOut(board.BOOST_ENABLE)
    boost_enable.switch_to_output(True)

Create the CAN bus object. Note that all devices on the same bus need to agree on the baudrate!

can = canio.CAN(rx=board.CAN_RX, tx=board.CAN_TX, baudrate=250_000, auto_restart=True)

1

can = canio.CAN(rx=board.CAN_RX, tx=board.CAN_TX, baudrate=250_000, auto_restart=True)

Construct a listener object. This listener will ONLY receive messages sent to the ID 0x408. If no matches= was specified, it would receive all messages. The timeout= governs how long the listener will wait for a message. A Match object can also specify an optional mask to allow a range of related IDs to be received—see the full documentation for more details.

listener = can.listen(matches=[canio.Match(0x408)], timeout=.1)

1

listener = can.listen(matches=[canio.Match(0x408)], timeout=.1)

Now we're ready for the main loop of our program:

while True:
    ...

1
2

while True:
    ...

The CAN object's state monitors the health of the bus. The confusingly-named ERROR_ACTIVE state actually indicates that all is well. A node that is ERROR_PASSIVE will not transmit messages, and one that is BUS_OFF will neither transmit messages nor acknowledge messages from other nodes. Because we specified auto_restart=True when we created our CAN object, our node will automatically restart itself a short time after entering the BUS_OFF state.

bus_state = can.state
if bus_state != old_bus_state:
    print(f"Bus state changed to {bus_state}")
    old_bus_state = bus_state

1
2
3
4

bus_state = can.state
if bus_state != old_bus_state:
    print(f"Bus state changed to {bus_state}")
    old_bus_state = bus_state

Create and send a message. In this case, we use the struct module to pack our integer data into a sequence of 8 bytes. Messages can range from 0 to 8 bytes of data.

message = canio.Message(id=0x408, data=struct.pack("<II", count, now_ms))
can.send(message)

1
2

message = canio.Message(id=0x408, data=struct.pack("<II", count, now_ms))
can.send(message)

Receive a message. One of several things can happen, and we need to deal with them:

• If no message is received before the timeout, message will be None
• If we were listening for more than one message ID, we would want to look at message.id and make decisions based on it.
• A message could come in, but not have the expected structure. Here, if the message is not the expected 8 bytes long, we ignore it
• If the message has the expected length, we can take the individual pieces of data out using struct.unpack, and act on them.

message = listener.receive()
if message is None:
    print("No messsage received within timeout")
    continue

data = message.data
if len(data) != 8:
    print(f"Unusual message length {len(data)}")
    continue
    
count, now_ms = struct.unpack("<II", data)
print(f"received message: count={count} now_ms={now_ms}")

1
2
3
4
5
6
7
8
9
10
11
12

message = listener.receive()
if message is None:
    print("No messsage received within timeout")
    continue

data = message.data
if len(data) != 8:
    print(f"Unusual message length {len(data)}")
    continue
    
count, now_ms = struct.unpack("<II", data)
print(f"received message: count={count} now_ms={now_ms}")