You've already forked CMB18_CellMonitoringApplication
428 lines
16 KiB
Python
428 lines
16 KiB
Python
######################################################################################
|
|
# #
|
|
# Copyright (c) Infineon Technologies AG #
|
|
# All rights reserved. #
|
|
# #
|
|
######################################################################################
|
|
|
|
# low-level user interface
|
|
#
|
|
|
|
from tle9018dqk.registers import *
|
|
|
|
import serial
|
|
import serial.tools.list_ports as port_list
|
|
|
|
import time
|
|
|
|
|
|
def bitreverse(b):
|
|
"""
|
|
short lsb-msb reversal function
|
|
works with uint8
|
|
|
|
:param b: uint8 that should be converted from LSB to MSB first
|
|
:return: converted byte
|
|
"""
|
|
ret = 0
|
|
for i in range(8):
|
|
if b & (1 << i):
|
|
ret |= (1 << (7 - i))
|
|
return ret
|
|
|
|
|
|
def bytereverse(d: bytearray) -> bytearray:
|
|
"""
|
|
Apply a bitreverse on all elements of a byte array.
|
|
Note that the order of bytes in the array is not affected
|
|
|
|
:param d: original byte array
|
|
:return: byte array with reversed bitorder
|
|
"""
|
|
ret = bytearray(d)
|
|
for i in range(len(d)):
|
|
ret[i] = bitreverse(d[i])
|
|
return ret
|
|
|
|
|
|
def crc_poly(data, n: int, poly: int, crc: int = 0, xor_out: int = 0) -> int:
|
|
"""
|
|
Calculate a CRC over given input data of length n.
|
|
|
|
:param data:
|
|
:param n: length of the data array for CRC calculation
|
|
:param poly: CRC Polynomial
|
|
:param crc: initial CRC Value
|
|
:param xor_out: the result of the CRC is xor-ed with this value before the value is returned
|
|
:return: CRC Value
|
|
"""
|
|
# just for exemplary use: can be implemented with built in CRC lib
|
|
g = 1 << n | poly
|
|
for d in data:
|
|
crc ^= d << (n - 8)
|
|
for _ in range(8):
|
|
crc <<= 1
|
|
if crc & (1 << n):
|
|
crc ^= g
|
|
return crc ^ xor_out
|
|
|
|
|
|
def CRC_calc(msg):
|
|
"""
|
|
Wrapper function to calculate the CRC over a message using the TLE9018 CRC Polynomial
|
|
:param msg: Bytearray for which the CRC shall be calculated
|
|
:return: CRC Checksum based on CRC8 Message Polynomial for TLE9018
|
|
"""
|
|
# calculates 8bit CRC of TLE9012DQU payload (crc_poly = 0x1D, initial value = 0xFF, final_xor = 0xFF)
|
|
return crc_poly(msg, 8, 0x1D, 0xFF, xor_out=0xFF)
|
|
|
|
|
|
class request6:
|
|
"""
|
|
Dataclass containing a Write Frame for IsoUART
|
|
"""
|
|
|
|
def __init__(self, psync, pid, paddr, pd1, pd0, pcrc=None):
|
|
"""
|
|
Constructor Method for an isoUART write access frame
|
|
calculates CRC checksum and provides bit-reversed structure for serial submission
|
|
supports two call modes
|
|
req = request(b'\x1E\x80\x36\x08\x01')
|
|
or
|
|
req = request(0x1E,0x80,0x36,0x08,0x01)
|
|
CRC checksum will be calculated automatically
|
|
|
|
:param psync: Synchronization Frame, should be fixed as 0x1E
|
|
:param pid: Node ID of the addressed device in the daisy chain
|
|
:param paddr: Register Address for write operation
|
|
:param pd1: High databyte that shall be written
|
|
:param pd0: Low databyte that shall be written
|
|
:param pcrc: CRC checksum, usually not required as parameter and calculated by constructor. Parameter overwrites calculated value if given
|
|
"""
|
|
self.data = bytearray(6)
|
|
self._data = bytearray(6)
|
|
if isinstance(psync, bytearray):
|
|
self.data[0:5] = psync[0:5]
|
|
else:
|
|
self.data[0] = psync
|
|
self.data[1] = pid
|
|
self.data[2] = paddr
|
|
self.data[3] = pd1
|
|
self.data[4] = pd0
|
|
self.data[5] = CRC_calc(self.data[0:5]) if (pcrc == None) else (pcrc)
|
|
for i in range(6):
|
|
self._data[i] = bitreverse(self.data[i])
|
|
|
|
|
|
class request4:
|
|
"""
|
|
Dataclass containing a Read Request Frame for IsoUART
|
|
"""
|
|
def __init__(self, psync, pid, paddr, pcrc=None):
|
|
"""
|
|
Constructor Method for an isoUART read access frame
|
|
|
|
:param psync: Synchronization Frame, should be fixed as 0x1E
|
|
:param pid: Node ID of the addressed device in the daisy chain
|
|
:param paddr: Register Address for read operation
|
|
:param pcrc: CRC checksum, usually not required as parameter and calculated by constructor. Parameter overwrites calculated value if given
|
|
"""
|
|
self.data = bytearray(4)
|
|
self._data = bytearray(4)
|
|
if isinstance(psync, bytearray):
|
|
self.data[0:3] = psync[0:3]
|
|
else:
|
|
self.data[0] = psync
|
|
self.data[1] = pid
|
|
self.data[2] = paddr
|
|
self.data[3] = CRC_calc(self.data[0:3]) if (pcrc == None) else (pcrc)
|
|
for i in range(4):
|
|
self._data[i] = bitreverse(self.data[i])
|
|
|
|
|
|
class response:
|
|
"""
|
|
Dataclass for Response Frames received from devices of a connected daisy chain
|
|
"""
|
|
|
|
def __init__(self, pid, paddr, pd1, pd0, pcrc):
|
|
"""
|
|
:param pid: Node ID of the responding device (Note that Bit 7 indicates flags in GEN DIAG are set)
|
|
:param paddr: Register that is addressed in the response
|
|
:param pd1: Response High Data Byte
|
|
:param pd0: Response Low Data Byte
|
|
:param pcrc: CRC Value
|
|
"""
|
|
self.data = bytearray(5)
|
|
self._data = bytearray(5)
|
|
if isinstance(pid, bytearray):
|
|
self.data[0:5] = pid[0:5]
|
|
else:
|
|
self.data[0] = pid
|
|
self.data[1] = paddr
|
|
self.data[2] = pd1
|
|
self.data[3] = pd0
|
|
self.data[4] = pcrc
|
|
for i in range(5):
|
|
self._data[i] = self.data[i]
|
|
self.data[i] = bitreverse(self.data[i])
|
|
|
|
def close(self):
|
|
"""
|
|
Close the Serial connection
|
|
"""
|
|
# release serial port --> seems to be needed for python running on windows operating systems
|
|
self.ser.close()
|
|
|
|
def getdata(self):
|
|
"""
|
|
Convert databytes into 16 Bit value
|
|
:return: 16 bit payload data of the response frame
|
|
"""
|
|
return (self.data[2] << 8) | (self.data[1])
|
|
|
|
def crccheck(self):
|
|
"""
|
|
Check if the CRC of the response frame is valid
|
|
:return: True if CRC check was successfull
|
|
"""
|
|
return (self.data[4] == CRC_calc(self.data[0:4]))
|
|
|
|
|
|
def prompt_serialport(baudrate=2000000):
|
|
"""
|
|
Provide user with a console slection of available serial ports
|
|
:param baudrate: Baudrate of the serial port
|
|
:return: Serial Port object in opened state
|
|
"""
|
|
print("Available Serial Ports:")
|
|
ports = list(port_list.comports())
|
|
for p1 in ports:
|
|
print(" %s [%s]" % (p1.device, p1.description))
|
|
pname = input("Enter port descriptor [%s]: " % ports[-1].device)
|
|
if pname == "": pname = ports[-1].device
|
|
ret = serial.Serial(port=pname, baudrate=baudrate)
|
|
print("Serial Port [%s] successfully opened @ %iBaud" % (pname, baudrate))
|
|
return ret
|
|
|
|
|
|
class TLE9018DQK:
|
|
"""
|
|
base class to control a single or chain of TLE9018DQU
|
|
"""
|
|
|
|
def __init__(self, ser, timeout=0.1):
|
|
"""
|
|
Constructor Method for the TLE9018 base class
|
|
:param ser: Serial port object in opened state that is used for communication
|
|
:param timeout: Timeout of the serial port in seconds
|
|
"""
|
|
self.ser = ser
|
|
self.ser.timeout = timeout
|
|
self.timeout = timeout # default timeout to restore
|
|
|
|
def wake(self):
|
|
"""
|
|
Wakeup all devices in the Daisy Chain
|
|
"""
|
|
if self.ser.baudrate > 1000000:
|
|
self.ser.write(b'\xcc\xcc\xcc\xcc') # For High Baudrates, numerous frames are sent to fullfill wakeup requirement
|
|
else:
|
|
self.ser.write(b'\x55\x55') #Propper way to do this using a low baudrate
|
|
time.sleep(0.008) # wait 8ms
|
|
self.ser.flushInput() # remove serial echo
|
|
|
|
def writeRegister(self, nid, addr, d1, d0):
|
|
"""
|
|
Write a register for a single TLE9018
|
|
:param nid: Node ID for the write operation
|
|
:param addr: Register address for the write operation
|
|
:param d1: High databyte
|
|
:param d0: Low databyte
|
|
:return: [boolean, data] with boolean describing if the operation was successfull and data providing the response or an error code
|
|
"""
|
|
frame = request6(0x1e, 0x80 | nid, (addr & 0xFF), d1, d0)
|
|
self.ser.write(frame._data)
|
|
resp = self.ser.read(6) # read echo
|
|
if nid != 0xbf:
|
|
if resp != frame._data:
|
|
# print("ERR(writeRegister): Missing/corrupted echo")
|
|
return False, -1
|
|
resp = self.ser.read(1) # read reply
|
|
if resp == b'':
|
|
# print("ERR(writeRegister): Missing reply")
|
|
return False, -2
|
|
return True, resp[0]
|
|
|
|
def readRegister(self, nid, addr):
|
|
"""
|
|
Read a register for a single TLE9018
|
|
:param nid: Node ID for the read operation
|
|
:param addr: Register address for read operation
|
|
:return: [boolean, data] with boolean describing if the operation was successfull and data providing the response or an error code
|
|
"""
|
|
frame = request4(0x1e, nid, (addr & 0xFF))
|
|
self.ser.write(frame._data)
|
|
resp = self.ser.read(4)
|
|
|
|
if resp != frame._data:
|
|
# print("ERR(readRegister): Missing/corrupted echo")
|
|
return False, -1
|
|
resp = self.ser.read(5) # read response
|
|
if resp == b'':
|
|
# print("ERR(readRegister): Missing reply")
|
|
return False, -2
|
|
resp = bytereverse(resp)
|
|
if ((resp[0] & 0xBF) != nid) or (resp[1] != addr):
|
|
# print(
|
|
# "ERR(readRegister): malformed preamble (%02x%02x) expecting (%02x%02x)" % (resp[0], resp[1], nid, addr))
|
|
return False, -3
|
|
if resp[4] != CRC_calc(resp[0:4]):
|
|
# print("ERR(readRegister): CRC checksum error.")
|
|
return False, -4
|
|
return True, [resp[3], resp[2]]
|
|
|
|
def readRegisterBroadcast(self, count, addr):
|
|
"""
|
|
Broadcast read operation
|
|
:param count: Number of devices in the daisy chain
|
|
:param addr: Register address for broadcast read operation
|
|
:return: [boolean, data] with boolean describing if the operation was successfull and data providing a list of responses or an error code
|
|
"""
|
|
self.ser.flushInput()
|
|
frame = request4(0x1e, 0x3F, (addr & 0xFF))
|
|
self.ser.write(frame._data)
|
|
resp = self.ser.read(4)
|
|
|
|
if resp != frame._data:
|
|
# print("ERR(readBroadcast): Missing/corrupted echo")
|
|
return False, -1
|
|
|
|
replydata = []
|
|
|
|
for i in range(count):
|
|
resp = self.ser.read(5)
|
|
if resp == b'':
|
|
# print("ERR(readBroadcast): Missing reply")
|
|
return False, -2
|
|
resp = bytereverse(resp)
|
|
if ((resp[0] & 0xBF) != (i + 1)) or (resp[1] != addr):
|
|
# print(
|
|
# "ERR(readBroadcast): malformed preamble (%02x%02x) expecting (%02x%02x)" % (
|
|
# resp[0], resp[1], (i+1), addr))
|
|
return False, -3
|
|
if resp[4] != CRC_calc(resp[0:4]):
|
|
# print("ERR(readBroadcast): CRC checksum error.")
|
|
return False, -4
|
|
|
|
replydata.append(int.from_bytes([resp[3], resp[2]], byteorder='little'))
|
|
|
|
return True, replydata
|
|
|
|
def readMultiread(self, nid, count, timeout=0.1):
|
|
"""
|
|
perform a multiread operating for a single TLE9018
|
|
:param nid: Node ID for the read operation
|
|
:param count: Number of registers expected by the multiread operation
|
|
:param timeout: Optional timeout, different from regular serial timeout
|
|
:return: [boolean, [address, data]] with boolean describing if the operation was successfull and data providing a list of register address, data pairs or an error code
|
|
"""
|
|
self.ser.flushInput()
|
|
frame = request4(0x1e, nid, 0x31)
|
|
self.ser.write(frame._data)
|
|
resp = self.ser.read(4)
|
|
|
|
if resp != frame._data:
|
|
# print("ERR(readMultiread): Missing/corrupted echo")
|
|
return False, -1
|
|
|
|
ret = []
|
|
try:
|
|
for i in range(count):
|
|
resp = self.ser.read(5) # read response
|
|
if resp == b'':
|
|
# print("ERR(readMultiread): Missing reply")
|
|
return False, -2
|
|
resp = bytereverse(resp)
|
|
if resp[4] != CRC_calc(resp[0:4]):
|
|
# print("ERR(readMultiread): CRC checksum error.")
|
|
return False, -3
|
|
ret.append([int(resp[1]), int.from_bytes([resp[3], resp[2]], byteorder='little')])
|
|
except Exception as e:
|
|
print(e)
|
|
return False, -4
|
|
|
|
return True, ret
|
|
|
|
def reset(self):
|
|
"""
|
|
Reset all devices on the daisy chain
|
|
"""
|
|
# Send whole chain to sleep by writing bit SLEEP_REG_RESET in OP_MODE
|
|
# Return is ignored here as devices go immediately into sleep mode
|
|
ok, err = self.writeRegister(0xBF, REG['OP_MODE'], 0x01, 0x00)
|
|
|
|
|
|
def assignNodeID(self, nid):
|
|
"""
|
|
Assign a node ID to uninitialized devices, note that all devices have an uninitialized ID after sleep mode
|
|
:param nid: New node ID
|
|
:return: [boolean, data] describing if operation was successfull and error code (see writeRegister function)
|
|
"""
|
|
# set node id: addr=0x00, REG['CONFIG']=0x36, val=0x0801
|
|
# ok,err = self.writeRegister(0x00,REG['IF_CFG'],0x08,nodeid)
|
|
ok, err = self.writeRegister(0x00, REG['IF_CFG'], 0x00, nid)
|
|
return ok, err
|
|
|
|
def readICVID(self, nid):
|
|
"""
|
|
Read the IC Version and ID register
|
|
:param nid: Node ID for operation
|
|
:return: [boolean, data] describing if operation was successfull and error code (see writeRegister function)
|
|
"""
|
|
|
|
ok, data = self.readRegister(nid, REG['ICVID'])
|
|
return ok, data
|
|
|
|
def readCUSTID(self, nid):
|
|
"""
|
|
Read the Customer ID
|
|
:param nid: Node ID for operation
|
|
:return: [boolean, data] describing if operation was successfull and error code (see writeRegister function)
|
|
"""
|
|
|
|
ret = bytearray(12)
|
|
for i in range(6):
|
|
ok, data = self.readRegister(nid, REG['CUSTOMER_ID_{}'.format(i)])
|
|
ret[(2 * i):(2 * i + 1)] = data
|
|
if not ok:
|
|
return ok, data
|
|
return True, ret
|
|
|
|
def readPCVM(self, nid):
|
|
"""
|
|
Perform a Primary Cell Voltage Measurement and print the results
|
|
:param nid: Node ID for operation
|
|
:return: [boolean, data] with boolean describing if the operation was successfull and data providing a list of PCVM Value or an error code
|
|
"""
|
|
ret = []
|
|
self.writeRegister(nid, REG["MEAS_CTRL"], 0xEE, 0x65)
|
|
time.sleep(0.05)
|
|
for i in range(18):
|
|
ok, data = self.readRegister(nid, REG['PCVM_{}'.format(i)])
|
|
ret.append(int.from_bytes(data, byteorder='little'))
|
|
if not ok:
|
|
return ok, data
|
|
return True, ret
|
|
|
|
def resetWDT(self, nid, val):
|
|
"""
|
|
Reset the Watchdog Timer. This function has to be triggered periodically to prevent the TLE9018 from changing
|
|
into sleep mode
|
|
:param nid: Node ID for operation, set to 0xBF for Broadcast
|
|
:param val: New Watchdog value between 0 and 127 with 16ms LSB
|
|
:return: [boolean, data] describing if operation was successfull and error code (see writeRegister function)
|
|
"""
|
|
ok, data = self.writeRegister(nid, REG['WDOG_CNT'], 0x00, val & 0x7F)
|
|
return ok, data
|