From 50c31775855b78a382b47f3346433ff2d06595bf Mon Sep 17 00:00:00 2001 From: Yeongdo Park Date: Thu, 14 Jan 2021 04:58:47 +0900 Subject: [PATCH] replace PCAN API with python-can for Seeed USB to CAN Analyzer --- Main.py | 104 ++++++++++++++++++++++------------------------- PycanMessages.py | 6 +++ test_seeed.py | 23 +++++++++++ 3 files changed, 78 insertions(+), 55 deletions(-) create mode 100644 PycanMessages.py create mode 100644 test_seeed.py diff --git a/Main.py b/Main.py index 65f1d93..30a031d 100644 --- a/Main.py +++ b/Main.py @@ -1,89 +1,89 @@ -import Messages import time import datetime import argparse +from argparse import ArgumentParser + +import PycanMessages as Messages from ProcessList import process_list from WriteToFile import report -from PCANBasic import * -from PCANBasic import TPCANTimestamp, TPCANMsg -#import matplotlib.pyplot as plt -#import numpy as np -parser = argparse.ArgumentParser() +import can + +parser: ArgumentParser = argparse.ArgumentParser() +parser.add_argument("-c", "--channel", + help=('The serial device to open. ' + 'For example "/dev/ttyS1" or "/dev/ttyUSB0" on Linux ' + 'or "COM1" on Windows systems.'), + required=True) parser.add_argument("-o", "--output", help="log file name", required=True) - args = parser.parse_args() +can.rc['interface'] = 'seeedstudio' +can.rc['channel'] = args.channel +can.rc['bitrate'] = 500000 + id_response = 0x7BB id_periodic = 0x1DB -filename = args.output # input("Please Enter a File Name: ") +id_mask11 = 0x7FF -# Initialize PCAN device and set filter for incoming messages with ID id_response -PCAN = PCANBasic() -res = PCAN.Initialize(PCAN_USBBUS1, PCAN_BAUD_500K) -#PCAN.FilterMessages(PCAN_USBBUS1, id_response, id_response, PCAN_MESSAGE_STANDARD) +voltage_filters = [ + {"can_id": id_response, "can_mask": id_mask11, "extended": False}, + {"can_id": id_periodic, "can_mask": id_mask11, "extended": False}, +] + +filename = args.output # input("Please Enter a File Name: ") + +# Initialize CAN device and set filter for incoming messages with ID id_response +bus = can.Bus() +bus.set_filters(voltage_filters) # Infinite loop runs until program is closed while True: - # Filters messages so we only get cell voltage data - PCAN.Reset(PCAN_USBBUS1) - PCAN.FilterMessages(PCAN_USBBUS1, id_response, id_response, PCAN_MESSAGE_STANDARD) - PCAN.FilterMessages(PCAN_USBBUS1, id_periodic, id_periodic, PCAN_MESSAGE_STANDARD) # Create empty list for cell voltage message strings # Each entry in the list is a message containing the voltage information v = [] - # wait 10 seconds to read the cell voltages again - #time.sleep(10) # For cell voltages we need to request and read 28 additional messages. Each message contains 3 1/2 cell voltages. for i in range(0, 29): - if i == 0 : + if i == 0: # Send message to request cell voltage data - PCAN.Write(PCAN_USBBUS1, Messages.rCellVInit) + bus.send(Messages.rCellVInit) else: # Request a new messages - PCAN.Write(PCAN_USBBUS1, Messages.rCellV) + bus.send(Messages.rCellV) # Wait 18 ms time.sleep(.014) # Read the next message v_message = True while v_message: - MSG = PCAN.Read(PCAN_USBBUS1) - MSG = MSG[1] - if MSG.ID == id_response: + MSG = bus.recv() + if MSG.arbitration_id == id_response: v_message = False # For the remaining voltages, the cell voltage data is contained in bytes 1 - 7 # Concatenate byte strings - s = "".join(map("{0:08b}".format, MSG.DATA[1:8])) + s = "".join(map("{0:08b}".format, MSG.data)) # Append them to the list v.append(s) vv = process_list(v) - # Filter messages for pack voltage and current - #PCAN.Reset(PCAN_USBBUS1) - #time.sleep(.005) - #PCAN.FilterMessages(PCAN_USBBUS1, id_periodic, id_periodic, PCAN_MESSAGE_STANDARD) - - #time.sleep(.050) p_message = True while p_message: - MSG = PCAN.Read(PCAN_USBBUS1) - MSG = MSG[1] - if MSG.ID == id_periodic: + MSG = bus.recv() + if MSG.arbitration_id == id_periodic: p_message = False - voltage_byte1 = "{0:08b}".format(MSG.DATA[2]) - voltage_byte2 = "{0:08b}".format(MSG.DATA[3]) + voltage_byte1 = "{0:08b}".format(MSG.data[2]) + voltage_byte2 = "{0:08b}".format(MSG.data[3]) pack_voltage_string = voltage_byte1 + voltage_byte2 pack_voltage_string = pack_voltage_string[0:9] pack_voltage = float(int(pack_voltage_string, 2)) - current_byte1 = "{0:08b}".format(MSG.DATA[0]) - current_byte2 = "{0:08b}".format(MSG.DATA[1]) + current_byte1 = "{0:08b}".format(MSG.data[0]) + current_byte2 = "{0:08b}".format(MSG.data[1]) pack_current_string = current_byte1 + current_byte2 pack_current_string = pack_current_string[0:10] if pack_current_string[0] == 1: @@ -91,44 +91,38 @@ while True: else: pack_current = float(int(pack_current_string, 2)) - #pack_current = pack_current/2 - - #PCAN.Reset(PCAN_USBBUS1) - #time.sleep(.005) - #PCAN.FilterMessages(PCAN_USBBUS1, id_response, id_response, PCAN_MESSAGE_STANDARD) - group1_msg = [] for i in range(0, 5): if i == 0: - PCAN.Write(PCAN_USBBUS1, Messages.request_group1) + bus.send(Messages.request_group1) else: - PCAN.Write(PCAN_USBBUS1, Messages.request_additional_line) + bus.send(Messages.request_additional_line) time.sleep(.014) p_message = True while p_message: - MSG = PCAN.Read(PCAN_USBBUS1) - MSG = MSG[1] - if MSG.ID == id_response: + MSG = bus.recv() + if MSG.arbitration_id == id_response: p_message = False group1_msg.append(MSG) - health_byte1 = "{0:08b}".format(MSG.DATA[2]) - health_byte2 = "{0:08b}".format(MSG.DATA[3]) + health_byte1 = "{0:08b}".format(MSG.data[2]) + health_byte2 = "{0:08b}".format(MSG.data[3]) pack_health_string = health_byte1 + health_byte2 pack_health = float(int(pack_health_string, 2))/100 - charge_byte1 = "{0:08b}".format(MSG.DATA[5]) - charge_byte2 = "{0:08b}".format(MSG.DATA[6]) - charge_byte3 = "{0:08b}".format(MSG.DATA[7]) + charge_byte1 = "{0:08b}".format(MSG.data[5]) + charge_byte2 = "{0:08b}".format(MSG.data[6]) + charge_byte3 = "{0:08b}".format(MSG.data[7]) charge_string = charge_byte1 + charge_byte2 + charge_byte3 pack_charge = float(int(charge_string, 2))/10000 tnow = datetime.datetime.now() report_string = report(vv, pack_voltage, pack_current, pack_health, pack_charge, tnow.ctime()) - print (report_string) + print(report_string) with open(filename, "a") as f: f.write(report_string) f.write('\n') + # wait 10 seconds to read the cell voltages again time.sleep(10) diff --git a/PycanMessages.py b/PycanMessages.py new file mode 100644 index 0000000..5f60041 --- /dev/null +++ b/PycanMessages.py @@ -0,0 +1,6 @@ +from can import Message + +rCellVInit = Message(arbitration_id=0x79b, data=[0x02, 0x21, 0x02, 0xff, 0xff, 0xff, 0xff, 0xff]) +rCellV = Message(arbitration_id=0x79b, data=[0x30, 0x01, 0x00, 0xff, 0xff, 0xff, 0xff, 0xff]) +request_group1 = Message(arbitration_id=0x79b, data=[0x02, 0x21, 0x01, 0xff, 0xff, 0xff, 0xff, 0xff]) +request_additional_line = Message(arbitration_id=0x79b, data=[0x30, 0x01, 0x00, 0xff, 0xff, 0xff, 0xff, 0xff]) \ No newline at end of file diff --git a/test_seeed.py b/test_seeed.py new file mode 100644 index 0000000..0672cd0 --- /dev/null +++ b/test_seeed.py @@ -0,0 +1,23 @@ +import time +import datetime +import argparse +from argparse import ArgumentParser + +import PycanMessages as Messages +from ProcessList import process_list +from WriteToFile import report + +import can + +can.rc['interface'] = 'seeedstudio' +can.rc['channel'] = "COM3" +can.rc['operation_mode'] = "loopback" +can.rc['bitrate'] = 500000 + +bus = can.Bus() + +can.Notifier(bus, [can.Printer()]) + +for i in range(100): + bus.send(Messages.rCellVInit) + time.sleep(1) \ No newline at end of file