ess-bms-work/nissan-leaf-reader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

replace PCAN API with python-can for Seeed USB to CAN Analyzer

Browse source
This commit is contained in:
Yeongdo Park 2021-01-14 04:58:47 +09:00
parent 2688750caf
commit 50c3177585
3 changed files with 78 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

104
Main.py
View file

@ -1,89 +1,89 @@
import Messages
import time import time
import datetime import datetime
import argparse import argparse
from argparse import ArgumentParser
import PycanMessages as Messages
from ProcessList import process_list from ProcessList import process_list
from WriteToFile import report 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) parser.add_argument("-o", "--output", help="log file name", required=True)
args = parser.parse_args() args = parser.parse_args()
can.rc['interface'] = 'seeedstudio'
can.rc['channel'] = args.channel
can.rc['bitrate'] = 500000
id_response = 0x7BB id_response = 0x7BB
id_periodic = 0x1DB 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 voltage_filters = [
PCAN = PCANBasic() {"can_id": id_response, "can_mask": id_mask11, "extended": False},
res = PCAN.Initialize(PCAN_USBBUS1, PCAN_BAUD_500K) {"can_id": id_periodic, "can_mask": id_mask11, "extended": False},
#PCAN.FilterMessages(PCAN_USBBUS1, id_response, id_response, PCAN_MESSAGE_STANDARD) ]
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 # Infinite loop runs until program is closed
while True: 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 # Create empty list for cell voltage message strings
# Each entry in the list is a message containing the voltage information # Each entry in the list is a message containing the voltage information
v = [] 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 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): for i in range(0, 29):
if i == 0 : if i == 0:
# Send message to request cell voltage data # Send message to request cell voltage data
PCAN.Write(PCAN_USBBUS1, Messages.rCellVInit) bus.send(Messages.rCellVInit)
else: else:
# Request a new messages # Request a new messages
PCAN.Write(PCAN_USBBUS1, Messages.rCellV) bus.send(Messages.rCellV)
# Wait 18 ms # Wait 18 ms
time.sleep(.014) time.sleep(.014)
# Read the next message # Read the next message
v_message = True v_message = True
while v_message: while v_message:
MSG = PCAN.Read(PCAN_USBBUS1) MSG = bus.recv()
MSG = MSG[1] if MSG.arbitration_id == id_response:
if MSG.ID == id_response:
v_message = False v_message = False
# For the remaining voltages, the cell voltage data is contained in bytes 1 - 7 # For the remaining voltages, the cell voltage data is contained in bytes 1 - 7
# Concatenate byte strings # 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 # Append them to the list
v.append(s) v.append(s)
vv = process_list(v) 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 p_message = True
while p_message: while p_message:
MSG = PCAN.Read(PCAN_USBBUS1) MSG = bus.recv()
MSG = MSG[1] if MSG.arbitration_id == id_periodic:
if MSG.ID == id_periodic:
p_message = False p_message = False
voltage_byte1 = "{0:08b}".format(MSG.DATA[2]) voltage_byte1 = "{0:08b}".format(MSG.data[2])
voltage_byte2 = "{0:08b}".format(MSG.DATA[3]) voltage_byte2 = "{0:08b}".format(MSG.data[3])
pack_voltage_string = voltage_byte1 + voltage_byte2 pack_voltage_string = voltage_byte1 + voltage_byte2
pack_voltage_string = pack_voltage_string[0:9] pack_voltage_string = pack_voltage_string[0:9]
pack_voltage = float(int(pack_voltage_string, 2)) pack_voltage = float(int(pack_voltage_string, 2))
current_byte1 = "{0:08b}".format(MSG.DATA[0]) current_byte1 = "{0:08b}".format(MSG.data[0])
current_byte2 = "{0:08b}".format(MSG.DATA[1]) current_byte2 = "{0:08b}".format(MSG.data[1])
pack_current_string = current_byte1 + current_byte2 pack_current_string = current_byte1 + current_byte2
pack_current_string = pack_current_string[0:10] pack_current_string = pack_current_string[0:10]
if pack_current_string[0] == 1: if pack_current_string[0] == 1:
@ -91,44 +91,38 @@ while True:
else: else:
pack_current = float(int(pack_current_string, 2)) 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 = [] group1_msg = []
for i in range(0, 5): for i in range(0, 5):
if i == 0: if i == 0:
PCAN.Write(PCAN_USBBUS1, Messages.request_group1) bus.send(Messages.request_group1)
else: else:
PCAN.Write(PCAN_USBBUS1, Messages.request_additional_line) bus.send(Messages.request_additional_line)
time.sleep(.014) time.sleep(.014)
p_message = True p_message = True
while p_message: while p_message:
MSG = PCAN.Read(PCAN_USBBUS1) MSG = bus.recv()
MSG = MSG[1] if MSG.arbitration_id == id_response:
if MSG.ID == id_response:
p_message = False p_message = False
group1_msg.append(MSG) group1_msg.append(MSG)
health_byte1 = "{0:08b}".format(MSG.DATA[2]) health_byte1 = "{0:08b}".format(MSG.data[2])
health_byte2 = "{0:08b}".format(MSG.DATA[3]) health_byte2 = "{0:08b}".format(MSG.data[3])
pack_health_string = health_byte1 + health_byte2 pack_health_string = health_byte1 + health_byte2
pack_health = float(int(pack_health_string, 2))/100 pack_health = float(int(pack_health_string, 2))/100
charge_byte1 = "{0:08b}".format(MSG.DATA[5]) charge_byte1 = "{0:08b}".format(MSG.data[5])
charge_byte2 = "{0:08b}".format(MSG.DATA[6]) charge_byte2 = "{0:08b}".format(MSG.data[6])
charge_byte3 = "{0:08b}".format(MSG.DATA[7]) charge_byte3 = "{0:08b}".format(MSG.data[7])
charge_string = charge_byte1 + charge_byte2 + charge_byte3 charge_string = charge_byte1 + charge_byte2 + charge_byte3
pack_charge = float(int(charge_string, 2))/10000 pack_charge = float(int(charge_string, 2))/10000
tnow = datetime.datetime.now() tnow = datetime.datetime.now()
report_string = report(vv, pack_voltage, pack_current, pack_health, pack_charge, tnow.ctime()) 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: with open(filename, "a") as f:
f.write(report_string) f.write(report_string)
f.write('\n') f.write('\n')
# wait 10 seconds to read the cell voltages again
time.sleep(10) time.sleep(10)

6
PycanMessages.py Normal file
View file

@ -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])

23
test_seeed.py Normal file
View file

@ -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)