I have a CAN bus (PCAN) with several inputs. I try to read the inputs in python and print them on the console. The first message I get from the bus is correct, however if I change the state on the input, the data in the message doesn't change and keeps spitting the first data it got. In PCAN-view I can see the data change, so it isn't a hardware failure.
My code:
import can
from time import sleep
def main():
bus = can.interface.Bus(bustype='pcan', channel='PCAN_USBBUS1', bitrate=500000)
try:
while True:
# msg = can.Message(arbitration_id=0x232, data=[0x00], is_extended_id=False)
# try:
# bus.send(msg)
# print("message sent on {}".format(bus.channel_info))
# except can.CanError:
# print("message not sent!")
msg = bus.recv(None)
try:
if msg.arbitration_id == 0x1B2:
print(msg)
if msg.arbitration_id == 0x1B3:
print(msg)
if msg.arbitration_id == 0x1B4:
print(msg)
if msg.arbitration_id == 0x1B5:
print(msg)
except AttributeError:
print("Nothing received this time")
sleep(0.2)
except KeyboardInterrupt:
print("Program Exited")
except can.CanError:
print("Message NOT sent")
bus.shutdown()
if __name__ == '__main__':
main()
I tried sending a message before each receive but it didn't help.
Same with calling the can.Listener().stop()
function on a listener after printing the received message.
PCAN-view shows the hardware is working as I can see the changes happening in the screen.
Eg.
I've read the readthedocs several times but I'm afraid I missed something. Do I need to flush the inputbuffer? I only saw info on a function to flush the output buffer.
So, I decided to implement it differently with a buffer and notifier. I also disabled the auto-update feature of the units so they don't spam the bus.
I came up with this class to use the bus.
import can
import logging
def _get_message(msg):
return msg
class PCANBus(object):
RX_SDO = 0x600
TX_SDO = 0x580
RX_PDO = 0x200
TX_PDO = 0x180
id_unit_a = [120, 121, 122, 123]
id_unit_b = [124, 125, 126, 127]
def __init__(self):
logging.info("Initializing CANbus")
self.bus = can.Bus(channel="PCAN_USBBUS1", bustype="pcan")
self.buffer = can.BufferedReader()
self.notifier = can.Notifier(self.bus, [_get_message, self.buffer])
def send_message(self, message):
try:
self.bus.send(message)
return True
except can.CanError:
logging.error("message not sent!")
return False
def read_input(self, id):
msg = can.Message(arbitration_id=self.RX_PDO + id,
data=[0x00],
is_extended_id=False)
self.send_message(msg)
return self.buffer.get_message()
def flush_buffer(self):
msg = self.buffer.get_message()
while (msg is not None):
msg = self.buffer.get_message()
def cleanup(self):
self.notifier.stop()
self.bus.shutdown()
def disable_update(self):
for i in [50, 51, 52, 53]:
msg = can.Message(arbitration_id=0x600 + i,
data=[0x23, 0xEA, 0x5F, 0x00, 0x00, 0x00, 0x00, 0x00],
is_extended_id=False)
self.send_message(msg)
And can be used as:
pcan = PCANBus()
msg = Message(arbitration_id=pcan.RX_PDO + 50, is_extended_id=False, data=[0x4F, 0x00])
pcan.send_message(msg)
ret = pcan.read_input(0x78)
pcan.cleanup()
'ret' holds the return message