can_rxtx.py

import struct
import time

import board
import digitalio

class Deadline:
    def __init__(self, relative=None, relative_ns=None):
        self.deadline = time.monotonic_ns()
        self.reset(relative, relative_ns)

    def reset(self, relative=None, relative_ns=None):
        if relative_ns is None:
            relative_ns = round(relative * 1e9)
        self.deadline += relative_ns

    @property
    def expired(self):
        return time.monotonic_ns() > self.deadline


def print_msg(m):
    if m is None:
        print(m)
    elif isinstance(m, Message):
        print ("MSG", hex(m.id), m.extended, ":".join("%02x" % d for d in m.data))
    else:
        print ("RTR", hex(m.id), m.extended, m.length)


baudrate = 250_000
if hasattr(board, 'CAN_TX'):
    from canio import CAN, Message, BusState
    if hasattr(board, 'CAN_STANDBY'):
        standby = digitalio.DigitalInOut(board.CAN_STANDBY)
        standby.switch_to_output(False)

    c = CAN(rx=board.CAN_RX, tx=board.CAN_TX, baudrate=baudrate)
else:
    from adafruit_mcp2515.canio import Message, BusState
    from adafruit_mcp2515 import MCP2515 as CAN
    cs = digitalio.DigitalInOut(board.D5)
    cs.switch_to_output()
    c = CAN(board.SPI(), cs, baudrate=2*baudrate) ## my board has an 8MHz xtal

l = c.listen()
m = Message(id=0x408, data=b'', extended=True)
d = Deadline(relative=0.25)
while True:
    if d.expired:
        state = c.state
        if state != BusState.ERROR_ACTIVE:
            print(state)
            if state == BusState.BUS_OFF:
                b.restart()
        m.data = struct.pack('Q', time.monotonic_ns())
        c.send(m)
        d.reset(0.25)
    if l.in_waiting():
        m = l.receive()
        print_msg(m)