smbus i2c read/write more than 32 bytes - Raspberry Pi Forums

hello everyone,

i've been playing around i2c using smbus on python , need pi talking powerboard. hoping review code , see if working expect to, since powerboard expensive piece of equipment not want screw up.

also, main problem i've been having smbus apparently has 32 byte read/write limit , need send/receive structs powerboard on limit. i've seen other answers here use fcntl , io modules access device directly. wondering if there simpler way this, like:

code: select all

# bytes bytearray n in bytes:     bus.write_byte(device_address, n) 

i'm confused functions lower level-wise. if need use fcntl , io modules, pointers in right direction helpful.

code, here sample functions use control powerboard:

code: select all

# have several helper functions use convert structs <-> bytearray <-> long[]: # structs , bytearrays created using ctypes module , converted struct <-> bytearray # using memmove function. # # c_longlisttostruct(longlist): long[] -> bytearray -> struct # tobytes(int, numbytes): int -> longlist (e.g. tobytes(300, 2) -> [00000001, 00101100])  import smbus import ctypes  bus = smbus.smbus(1) power_address = 0x00 # address powerboard, don't know actual address yet  # returns eps_hk_out_t struct (62 bytes) def get_hk_out(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, [0x02]) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "eps_hk_out_t")  # sets single output on or off: # channel [1 byte]  -> voltage = (0~5), bp4 heater = 6, bp4 switch = 7 # value   [1 byte]  -> on = 1, off = 0 # delay   [2 bytes] -> [seconds] def set_single_output(channel, value, delay): 	cmd = 0x0a 	delay2bytes = tobytes(delay, 2) 	bus.write_i2c_block_data(power_address, cmd, [channel, value]+delay2bytes)  # sets voltage output channels bit mask:  # byte [8] -> [nc nc 3.3v3 3.3v2 3.3v1 5v3 5v2 5v1] def set_output(byte): 	cmd = 0x09 	bus.write_i2c_block_data(power_address, cmd, [byte]) 

here full code:

code: select all

import ctypes ctypes import * import smbus  class testingstruct(bigendianstructure):     _fields_ = [         ("field1", c_uint8),         ("field2", c_uint8),         ("field3", c_uint16)     ]  class hkparam_t(bigendianstructure): 	_fields_ = [ 		("pv", 				c_uint16*3), 	# photo-voltaic input voltage [mv] 		("pc",   			c_uint16),  	# total photo current [ma] 		("bv",    			c_uint16), 		# battery voltage [mv] 		("sc",  			c_uint16), 		# total system current [ma] 		("temp", 		   	c_int16*4), 	# temp. of boost converters (1,2,3) , onboard battery [degc] 		("batt_temp", 	   	c_int16*2), 	# external board battery temperatures [degc]; 		("latchup", 		c_uint16*6), 	# number of latch-ups on each output 5v , +3v3 channel 							 				# order[5v1 5v2 5v3 3.3v1 3.3v2 3.3v3] 							 				# transmit 5v1 first , 3.3v3 last 		("reset", 			c_uint8), 		# cause of last eps reset 		("bootcount", 		c_uint16), 		# number of eps reboots 		("sw_errors", 		c_uint16), 		# number of errors in eps software 		("ppt_mode", 		c_uint8),  		# 0 = hardware, 1 = mppt, 2 = fixed sw ppt. 		("channel_status", 	c_uint8) 		# mask of output channel status, 1=on, 0=off 											# msb - [qh qs 3.3v3 3.3v2 3.3v1 5v3 5v2 5v1] - lsb 											# qh = quadbat heater, qs = quadbat switch		 	]  class eps_hk_t(bigendianstructure): 	_fields_ = [ 		("vboost", 				c_uint16*3),	# voltage of boost converters [mv] [pv1, pv2, pv3] 		("vbatt", 				c_uint16),		# voltage of battery [mv] 		("curin", 				c_uint16*3),	# current in [ma] 		("cursun", 				c_uint16),		# current boost converters [ma] 		("cursys", 				c_uint16),		# current out of battery [ma] 		("reserved1", 			c_uint16),		# reserved future use 		("curout", 				c_uint16*6),	# current out (switchable outputs) [ma]  		("output", 				c_uint8*8),		# status of outputs** 		("output_on_delta", 	c_uint16*8), 	# time till power on** [s]  		("output_off_delta", 	c_uint16*8),	# time till power off** [s] 		("latchup", 			c_uint16*6),	# number of latch-ups 		("wdt_i2c_time_left", 	c_uint32),		# time left on i2c wdt [s]  		("wdt_gnd_time_left", 	c_uint32),		# time left on i2c wdt [s]  		("wdt_csp_pings_left", 	c_uint8*2),		# pings left on csp wdt 		("counter_wdt_i2c", 	c_uint32),		# number of wdt i2c reboots  		("counter_wdt_gnd", 	c_uint32),		# number of wdt gnd reboots  		("counter_wdt_csp", 	c_uint32*2),	# number of wdt csp reboots  		("counter_boot", 		c_uint32), 		# number of eps reboots 		("temp", 				c_int16*6),		# temperatures [degc] [0 = temp1, temp2, temp3, temp4, bp4a, bp4b] 		("bootcause", 			c_uint8),		# cause of last eps reset 		("battmode", 			c_uint8),		# mode battery [0 = initial, 1 = undervoltage, 2 = safemode, 3 = nominal, 4=full] 		("pptmode", 			c_uint8),		# mode of ppt tracker [1=mppt, 2=fixed] 		("reserved2", 			c_uint16) 	]   class eps_hk_vi_t(bigendianstructure): 	_fields_ = [ 		("vboost", 		c_uint16*3), 		# voltage of boost converters [mv] [pv1, pv2, pv3] 		("vbatt", 		c_uint16), 			# voltage of battery [mv] 		("curin", 		c_uint16*3), 		# current in [ma] 		("cursun", 		c_uint16), 			# current boost converters [ma]  		("cursys", 		c_uint16), 			# current out of battery [ma] 		("reserved1", 	c_uint16)			# reserved future use 	] 	 class eps_hk_out_t(bigendianstructure): 	_fields_ = [ 		("curout", 			c_uint16*6),	# current out (switchable outputs) [ma]  		("output", 			c_uint8*8),		# status of outputs** 		("output_on_delta", c_uint16*8), 	# time till power on** [s] 		("output_off_delta",c_uint16*8), 	# time till power off** [s] 		("latchup", 		c_uint16*6), 	# number of latch-ups 	]  class eps_hk_wdt_t(bigendianstructure): 	_fields_ = [ 		("wdt_i2c_time_left", 	c_uint32), 	# time left on i2c wdt [s]  		("wdt_gnd_time_left", 	c_uint32),	# time left on i2c wdt [s]  		("wdt_csp_pings_left", 	c_uint8*2), # pings left on csp wdt 		("counter_wdt_i2c", 	c_uint32), 	# number of wdt i2c reboots  		("counter_wdt_gnd", 	c_uint32), 	# number of wdt gnd reboots  		("counter_wdt_csp", 	c_uint32*2) # number of wdt csp reboots 	]  class eps_hk_basic_t(bigendianstructure): 	_fields_ = [ 		("counter_boot", 	c_uint32), 		# number of eps reboots 		("temp", 			c_int16*6), 	# temperatures [degc] [0 = temp1, temp2, temp3, temp4, batt0, batt1] 		("bootcause", 		c_uint8), 		# cause of last eps reset 		("battmode", 		c_uint8), 		# mode battery [0 = initial, 1 = undervoltage, 2 = safemode, 3 = nominal, 4=full] 		("pptmode", 		c_uint8), 		# mode of ppt tracker [1=mppt, 2=fixed] 		("reserved2", 		c_uint16) 	]  # struct -> byte array def c_structtobytearray(s):     bytearray = (c_byte*(sizeof(s))) ()     spoint = pointer(s)     cpoint = pointer(bytearray)     memmove(cpoint, spoint, sizeof(s))     return bytearray  # byte array -> long[] def c_bytearraytolonglist(b):     acc = []     n in b:         acc += [n]     return acc  # byte array -> struct def c_bytearraytostruct(b, s):c 	bpoint = pointer(b) 	struct = structmaker(s) 	spoint = pointer(struct) 	memmove(spoint, bpoint, sizeof(b)) 	return struct  # takes int , outputs long list int divided  # [num] number of bytes def tobytes(i, num): 	binary = bin(i)[2:] 	rem = len(binary) % 8  	binary = '0'*(8-rem)+binary 	# add zeros make 8 bit 	size = len(binary)/8 	bytes = '0'*8*(num-size)+binary # add zeros make right num of bytes  	acc = [] 	for n in range(num): 		acc += [int(bytes[8*n:8*(n+1)], 2)]  	return acc  # long[] -> byte array def c_longlisttobytearray(i): 	return (c_byte*len(i)) (*i)   # struct -> byte array -> long[] def c_structtolonglist(s): 	return c_bytearraytolonglist(c_structtobytearray(s))  # long[] -> byte array -> struct def c_longlisttostruct(i, s): 	return c_bytearraytostruct(c_longlisttobytearray(i), s)  # creates struct given string def structmaker(s): 	if s == "hkparam_t": return hkparam_t() 	if s == "eps_hk_t": return eps_hk_t() 	if s == "eps_hk_vi_t": return eps_hk_vi_t() 	if s == "eps_hk_out_t": return eps_hk_out_t() 	if s == "eps_hk_wdt_t": return eps_hk_wdt_t() 	if s == "eps_hk_basic_t": return eps_hk_basic_t() 	return testingstruct()  # ----------------------------------------------commands bus = smbus.smbus(1) power_address = 0x00 # don't know actual device address yet  # pings 1 byte value def ping(value): 	cmd = 0x01 	bus.write_i2c_block_data(power_address, cmd, [value]) 	return bus.read_i2c_block_data(power_address, cmd)  # reboot system def reboot(): 	cmd = 0x04 	bus.write_i2c_block_data(power_address, cmd, [0x80, 0x07, 0x80, 0x07])  # returns hkparam_t struct def get_hk_1(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, []) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "hkparam_t")  # returns eps_hk_t struct def get_hk_2(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, [0x00]) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "eps_hk_t")  # returns eps_hk_vi_t def get_hk_2_vi(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, [0x01]) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "eps_hk_vi_t")  # returns eps_hk_out_t def get_hk_out(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, [0x02]) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "eps_hk_out_t")  # returns eps_hk_wdt_t def get_hk_wdt(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, [0x03]) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "eps_hk_wdt_t")  # returns eps_hk_basic_t def get_hk_2_basic(): 	cmd = 0x08 	bus.write_i2c_block_data(power_address, cmd, [0x04]) 	recv = bus.read_i2c_block_data(power_address, cmd) 	return c_longlisttostruct(recv, "eps_hk_basic_t")  # sets voltage output channels bit mask:  # byte [8] -> [nc nc 3.3v3 3.3v2 3.3v1 5v3 5v2 5v1] def set_output(byte): 	cmd = 0x09 	bus.write_i2c_block_data(power_address, cmd, [byte])  # sets single output on or off: # channel [8]  -> voltage = (0~5), bp4 heater = 6, bp4 switch = 7 # value   [8]  -> on = 1, off = 0 # delay   [16] -> [seconds] def set_single_output(channel, value, delay): 	cmd = 0x0a 	delay2bytes = tobytes(delay, 2) 	bus.write_i2c_block_data(power_address, cmd, [channel, value]+delay2bytes)  # sets voltage on photovoltaic inputs in mv # takes effect when mode = 2 (set_pv_auto) def set_pv_volt(volt1, volt2, volt3): 	cmd = 0x0b 	v1 = tobytes(volt1, 2) 	v2 = tobytes(volt2, 2) 	v3 = tobytes(volt3, 2) 	bus.write_i2c_block_data(power_address, cmd, v1+v2+v3)  # sets solar cell power tracking mode: # mode = 0: hardware default power point # mode = 1: maximum power point tracking # mode = 2: fixed software powerpoint, value set set_pv_volt, default 4v def set_pv_auto(mode): 	cmd = 0x0c 	bus.write_i2c_block_data(power_address, cmd, [mode])  # cmd = 0: set heater on/off  # heater: 0 = bp4, 1= onboard, 2 = both # mode: 0 = off, 1 = on # returns long list heater modes def set_heater(command, heater, mode): 	cmd = 0x0d 	bus.write_i2c_block_data(power_address, cmd, [command, heater, mode]) 	return get_heater()  # returns long list heater modes def get_heater(): 	cmd = 0x0d 	return bus.read_i2c_block_data(power_address, cmd)  # resets boot counter , wdt counters. def reset_counters(): 	cmd = 0x0f 	bus.write_i2c_block_data(power_address, cmd, [0x42])  # resets (kicks) dedicated wdt. def reset_wdt(): 	cmd = 0x10 	bus.write_i2c_block_data(power_address, cmd, [0x78])  # use command control config system. # cmd=1: restore default config def config_cmd(command): 	cmd = 0x11 	bus.write_i2c_block_data(power_address, cmd, [command])  # returns eps_config_t structure def config_get(): 	cmd = 0x12 	return bus.read_i2c_block_data(power_address, cmd)  # takes eps_config_t struct , sets configuration def config_set(struct): 	cmd = 0x13 	longlist = c_structtolonglist(struct) 	bus.write_i2c_block_data(power_address, cmd, longlist)  # send command perform hard reset of p31, # including cycling permanent 5v , 3.3v , battery outputs. def hard_reset(): 	cmd = 0x14 	bus.write_i2c_block_data(power_address, cmd, [])  # use command control config 2 system. # cmd=1: restore default config cmd=2: confirm current config def config2_cmd(command): 	cmd =  0x15 	bus.write_i2c_block_data(power_address, cmd, [command])  # use command request p31 config 2. def config2_get(): 	cmd = 0x16 	return bus.read_i2c_block_data(power_address, cmd)  # use command send config 2 p31 # , save (remember confirm it) def config2_set(struct): 	cmd = 0x17 	longlist = c_structtolonglist(struct) 	bus.write_i2c_block_data(power_address, cmd, longlist)  # ----------------------------------------------tests # sending side teststruct = testingstruct() teststruct.field1 = 5 teststruct.field2 = 3 teststruct.field3 = 257 send = c_structtolonglist(teststruct)  # receiving side recv = c_longlisttostruct(send, "testingstruct") print recv.field1 print recv.field2 print recv.field3  longlist = [1, 5, 300] array = c_longlisttobytearray(longlist) print array[1] print bin(300) print int('00101100', 2) 

datasheet powerboard can found here: http://www.mediafire.com/file/a89n9x9xw ... 1u-1.0.pdf
functions , structs implemented on pages 11-15.

small thing confused smbus read/write functions (read_i2c_block_data/write_i2c_block_data) used take , return long lists, understanding, longs more single byte. each element in long list single byte in length or more?

in advanced, appreciated.

smbus limited 32 bytes in single transaction. i2c not have limitation.

following code shows 1 method of using i2c rather smbus in python.

code: select all

#!/usr/bin/env python  import io import fcntl import sys  # i2c_raw.py # 2016-02-26 # public domain  i2c_slave=0x0703  if sys.hexversion < 0x03000000:    def _b(x):       return x else:    def _b(x):       return x.encode('latin-1')  class i2c:     def __init__(self, device, bus):        self.fr = io.open("/dev/i2c-"+str(bus), "rb", buffering=0)       self.fw = io.open("/dev/i2c-"+str(bus), "wb", buffering=0)        # set device address        fcntl.ioctl(self.fr, i2c_slave, device)       fcntl.ioctl(self.fw, i2c_slave, device)     def write(self, data):       print(type(data))       if type(data) list:          data = bytearray(data)       elif type(data) str:          data = _b(data)       self.fw.write(data)     def read(self, count):       return self.fr.read(count)     def close(self):       self.fw.close()       self.fr.close()  if __name__ == "__main__":     import time    import i2c_raw     dev = i2c_raw.i2c(0x32, 1) # device 0x32, bus 1     dev.write([10, 1, 2, 3, ord('b'), ord('e'), ord('e'), ord('f')])    dev.write(b"\x0a\x01\x02\x03beef")    dev.write("\x0a\x01\x02\x03beef")     dev.close() 

python module supports arbitrary length read , writes.

http://abyz.co.uk/rpi/pigpio/python.htm ... ead_device
http://abyz.co.uk/rpi/pigpio/python.htm ... ite_device

raspberrypi