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

Compare commits

base: ess-bms-work:master
Branches Tags
ess-bms-work:master
ess-bms-work:WithoutLoop
..
compare: ess-bms-work:WithoutLoop
Branches Tags
ess-bms-work:master
ess-bms-work:WithoutLoop

No commits in common. "master" and "WithoutLoop" have entirely different histories.
master ... WithoutLoo

5 changed files with 169 additions and 131 deletions
Show stats Expand all files Collapse all files

243
Main.py
View file

@ -1,89 +1,141 @@
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
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
id_mask11 = 0x7FF
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)
# Initialize PCAN device and set filter for incoming messages with ID 1979
PCAN = PCANBasic()
res = PCAN.Initialize(PCAN_USBBUS1, PCAN_BAUD_500K)
#PCAN.FilterMessages(PCAN_USBBUS1, 1979, 1979, PCAN_MESSAGE_STANDARD)
filename = input("Please Enter a File Name: ")
# 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, 1979, 1979, PCAN_MESSAGE_STANDARD)
PCAN.FilterMessages(PCAN_USBBUS1, 475, 475, 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)
# Send message to request cell voltage data
PCAN.Write(PCAN_USBBUS1, Messages.rCellVInit)
# pause
time.sleep(.014)
# Read the first message
v_message = 0
while v_message != 1:
MSG = PCAN.Read(PCAN_USBBUS1)
MSG = MSG[1]
if MSG.ID == 1979:
v_message = 1
# MSG[1] is the data we are requesting
#MSG = MSG[1]
# Bytes 4, 5, 6, and 7 store the first two cell voltages
# For each byte we need to create a binary string and make sure they are 8 bits long
s4 = "{0:b}".format(MSG.DATA[4])
while len(s4) != 8:
s4 = "0" + s4
s5 = "{0:b}".format(MSG.DATA[5])
while len(s5) != 8:
s5 = "0" + s5
s6 = "{0:b}".format(MSG.DATA[6])
while len(s6) != 8:
s6 = "0" + s6
s7 = "{0:b}".format(MSG.DATA[7])
while len(s7) != 8:
s7 = "0" + s7
# Concatenate all the strings
s = s4 + s5 + s6 + s7
# Add them to the list of voltage messages
v.append(s)
# 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:
# Send message to request cell voltage data
bus.send(Messages.rCellVInit)
else:
# Request a new messages
bus.send(Messages.rCellV)
for i in range(1, 29):
# Request a new messages
PCAN.Write(PCAN_USBBUS1, Messages.rCellV)
# Wait 18 ms
time.sleep(.014)
# Read the next message
v_message = True
while v_message:
MSG = bus.recv()
if MSG.arbitration_id == id_response:
v_message = False
v_message = 0
while v_message != 1:
MSG = PCAN.Read(PCAN_USBBUS1)
MSG = MSG[1]
if MSG.ID == 1979:
v_message = 1
# For the remaining voltages, the cell voltage data is contained in bytes 1 - 7
s1 = "{0:b}".format(int(MSG.DATA[1]))
while len(s1) != 8:
s1 = "0" + s1
s2 = "{0:b}".format(int(MSG.DATA[2]))
while len(s2) != 8:
s2 = "0" + s2
s3 = "{0:b}".format(MSG.DATA[3])
while len(s3) != 8:
s3 = "0" + s3
s4 = "{0:b}".format(MSG.DATA[4])
while len(s4) != 8:
s4 = "0" + s4
s5 = "{0:b}".format(MSG.DATA[5])
while len(s5) != 8:
s5 = "0" + s5
s6 = "{0:b}".format(MSG.DATA[6])
while len(s6) != 8:
s6 = "0" + s6
s7 = "{0:b}".format(MSG.DATA[7])
while len(s7) != 8:
s7 = "0" + s7
# Concatenate byte strings
s = "".join(map("{0:08b}".format, MSG.data))
s = s1 + s2 + s3 + s4 + s5 + s6 + s7
# Append them to the list
v.append(s)
vv = process_list(v)
for i in range(0, 48):
print("Cell " + str(i+1) + " = %.3f" % vv[i] + "\t\t\tCell " + str(i + 49) + " = %.3f" % vv[i+48])
with open(filename, "a") as f:
f.write("\nCell " + str(i+1) + " = %.3f" % vv[i] + "\t\t\tCell " + str(i + 49) + " = %.3f" % vv[i+48])
print("Cell Voltage Sum = %.2f" % sum(vv))
with open(filename, "a") as f:
f.write("\nCell Voltage Sum = %.2f" % sum(vv))
p_message = True
while p_message:
MSG = bus.recv()
if MSG.arbitration_id == id_periodic:
p_message = False
# Filter messages for pack voltage and current
#PCAN.Reset(PCAN_USBBUS1)
#time.sleep(.005)
#PCAN.FilterMessages(PCAN_USBBUS1, 475, 475, PCAN_MESSAGE_STANDARD)
voltage_byte1 = "{0:08b}".format(MSG.data[2])
voltage_byte2 = "{0:08b}".format(MSG.data[3])
#time.sleep(.050)
p_message = 0
while p_message != 1:
MSG = PCAN.Read(PCAN_USBBUS1)
MSG = MSG[1]
if MSG.ID == 475:
p_message = 1
voltage_byte1 = "{0:b}".format(MSG.DATA[2])
while len(voltage_byte1) != 8:
voltage_byte1 = "0" + voltage_byte1
voltage_byte2 = "{0:b}".format(MSG.DATA[3])
while len(voltage_byte2) != 8:
voltage_byte2 = "0" + voltage_byte2
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:b}".format(MSG.DATA[0])
while len(current_byte1) != 8:
current_byte1 = "0" + current_byte1
current_byte2 = "{0:b}".format(MSG.DATA[1])
while len(current_byte2) != 8:
current_byte2 = "0" + current_byte2
pack_current_string = current_byte1 + current_byte2
pack_current_string = pack_current_string[0:10]
if pack_current_string[0] == 1:
@ -91,38 +143,61 @@ while True:
else:
pack_current = float(int(pack_current_string, 2))
group1_msg = []
for i in range(0, 5):
if i == 0:
bus.send(Messages.request_group1)
else:
bus.send(Messages.request_additional_line)
time.sleep(.014)
p_message = True
while p_message:
MSG = bus.recv()
if MSG.arbitration_id == id_response:
p_message = False
group1_msg.append(MSG)
#pack_current = pack_current/2
health_byte1 = "{0:08b}".format(MSG.data[2])
health_byte2 = "{0:08b}".format(MSG.data[3])
print("Pack Voltage = %.2f V" % pack_voltage)
with open(filename, "a") as f:
f.write("\nPack Voltage = %.2f V" % pack_voltage)
print("Pack Current = %.2f A" % pack_current)
with open(filename, "a") as f:
f.write("\nPack Current = %.2f A" % pack_current)
#PCAN.Reset(PCAN_USBBUS1)
#time.sleep(.005)
#PCAN.FilterMessages(PCAN_USBBUS1, 1979, 1979, PCAN_MESSAGE_STANDARD)
PCAN.Write(PCAN_USBBUS1, Messages.request_group1)
time.sleep(.014)
p_message = 0
while p_message != 1:
MSG = PCAN.Read(PCAN_USBBUS1)
MSG = MSG[1]
if MSG.ID == 1979:
p_message = 1
for i in range(0, 4):
PCAN.Write(PCAN_USBBUS1, Messages.request_additional_line)
time.sleep(.014)
p_message = 0
while p_message != 1:
MSG = PCAN.Read(PCAN_USBBUS1)
MSG = MSG[1]
if MSG.ID == 1979:
p_message = 1
health_byte1 = "{0:b}".format(MSG.DATA[2])
health_byte2 = "{0:b}".format(MSG.DATA[3])
while len(health_byte1) != 8:
health_byte1 = "0" + health_byte1
while len(health_byte2) != 8:
health_byte2 = "0" + health_byte2
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:b}".format(MSG.DATA[5])
charge_byte2 = "{0:b}".format(MSG.DATA[6])
charge_byte3 = "{0:b}".format(MSG.DATA[7])
while len(charge_byte1) != 8:
charge_byte1 = "0" + charge_byte1
while len(charge_byte2) != 8:
charge_byte2 = "0" + charge_byte2
while len(charge_byte3) != 8:
charge_byte3 = "0" + charge_byte3
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("State of Health = %.2f %%" % pack_health)
with open(filename, "a") as f:
f.write(report_string)
f.write('\n')
# wait 10 seconds to read the cell voltages again
f.write("\nState of Health = %.2f %%" % pack_health)
print("State of Charge = %.2f %%" % pack_charge)
with open(filename, "a") as f:
f.write("\nState of Charge = %.2f %%" % pack_charge)
time.sleep(10)

16
ProcessList.py
View file

@ -1,17 +1,21 @@
# This is a function to process the list of cell voltage data
def process_list(v):
# Instantiate an empty string to store all of the voltage data
# Creating one long string alleviates the problem of having 1/2 a voltage on each message
# Discard first 3 bytes of the first message since it has only 2 voltages
v_string = "".join(v[0:29])[3*8:]
v_string = ""
# Iterate through the 29 entries of the list
for i in range(0, 29):
# Create a temporary variable and store the next string
temp = v[i]
# Concatenate the temporary string to the end of the long string
v_string = v_string + temp
# Create an empty list to store the voltage of each cell
voltages = [0.0] * 96
voltages = [0.0] * 97
for i in range(0, 96):
# Each cell has a 16 bit value. Extract the next 16 bits and convert it into a float
voltages[i] = float(int(v_string[(i*16):(i*16+16)], 2))
# Divide by 1000 to get the final value
voltages[i] = voltages[i]/1000.
voltages[i] = voltages[i]/1000
# Print the voltage of the cell to the console
#print("Cell " + str(i+1) + " = %.3f" % voltages[i])
return voltages

6
PycanMessages.py
View file

@ -1,6 +0,0 @@
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])

12
WriteToFile.py
View file

@ -1,12 +0,0 @@
def report (vv, pack_voltage, pack_current, pack_health, pack_charge, timestamp):
return (
"\n".join([
"Cell {:2d} = {:.3f}\t\t\tCell {:2d} = {:.3f}".format((i+1), vv[i], (i+49), vv[i+48])
for i in range(48)]) +
"\nCell Voltage Sum = {:.2f}".format(sum(vv)) +
"\nPack Voltage = {:.2f} V" .format(pack_voltage) +
"\nPack Current = {:.2f} A" .format(pack_current) +
"\nState of Health = {:.2f} %" .format(pack_health) +
"\nState of Charge = {:.2f} %" .format(pack_charge) +
"\nTime = {}" .format(timestamp) +
"\n")

23
test_seeed.py
View file

@ -1,23 +0,0 @@
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)