X-Git-Url: https://git.ayoreis.com/Adafruit_Blinka-hackapet.git/blobdiff_plain/a79e37b2729f77faf00545062fc4a8ded1a6eb97..f73c450270e34245d0f5c1a62fbca103bfec99ba:/src/adafruit_blinka/microcontroller/nova/i2c.py diff --git a/src/adafruit_blinka/microcontroller/nova/i2c.py b/src/adafruit_blinka/microcontroller/nova/i2c.py index e5f0beb..5c1ecb1 100644 --- a/src/adafruit_blinka/microcontroller/nova/i2c.py +++ b/src/adafruit_blinka/microcontroller/nova/i2c.py @@ -5,6 +5,7 @@ from adafruit_blinka.microcontroller.nova import Connection class I2C: """Custom I2C Class for Binho Nova""" + WHR_PAYLOAD_MAX_LENGTH = 1024 def __init__(self, *, frequency=400000): self._nova = Connection.getInstance() @@ -13,6 +14,16 @@ class I2C: self._nova.setPullUpStateI2C(0, "EN") self._nova.setClockI2C(0, frequency) + self._novaCMDVer = "0" + if hasattr(self._nova, "getCommandVer"): + response = self._nova.getCommandVer().split(" ") + if response[0] != "-NG": + self._novaCMDVer = response[1] + + def __del__(self): + """Close Nova on delete""" + self._nova.close() + def scan(self): """Perform an I2C Device Scan""" scanResults = [] @@ -30,21 +41,40 @@ class I2C: def writeto(self, address, buffer, *, start=0, end=None, stop=True): """Write data from the buffer to an address""" end = end if end else len(buffer) + readBytes = 0 + if int(self._novaCMDVer) >= 1: + chunks, rest = divmod(end - start, self.WHR_PAYLOAD_MAX_LENGTH) + for i in range(chunks): + chunk_start = start + i * self.WHR_PAYLOAD_MAX_LENGTH + chunk_end = chunk_start + self.WHR_PAYLOAD_MAX_LENGTH + self._nova.writeToReadFromI2C(0, + address<<1, + stop, + readBytes, + chunk_end-chunk_start, + buffer[chunk_start:chunk_end]) + if rest: + self._nova.writeToReadFromI2C(0, + address<<1, + stop, + readBytes, + rest, + buffer[-1*rest:]) + else: + self._nova.startI2C(0, address<<1) - self._nova.startI2C(0, address << 1) - - for i in range(start, end): - self._nova.writeByteI2C(0, buffer[i]) + for i in range(start, end): + self._nova.writeByteI2C(0, buffer[i]) - if stop: - self._nova.endI2C(0) - else: - self._nova.endI2C(0, True) + if stop: + self._nova.endI2C(0) + else: + self._nova.endI2C(0, True) + # pylint: disable=unused-argument def readfrom_into(self, address, buffer, *, start=0, end=None, stop=True): """Read data from an address and into the buffer""" end = end if end else len(buffer) - result = self._nova.readBytesI2C(0, address << 1, len(buffer[start:end])) if result != "-NG": @@ -57,45 +87,89 @@ class I2C: "Received error response from Binho Nova, result = " + result ) - # pylint: disable=unused-argument def writeto_then_readfrom( - self, - address, - buffer_out, - buffer_in, - *, - out_start=0, - out_end=None, - in_start=0, - in_end=None, - stop=False + self, + address, + buffer_out, + buffer_in, + *, + out_start=0, + out_end=None, + in_start=0, + in_end=None, + stop=False ): """Write data from buffer_out to an address and then read data from an address and into buffer_in """ out_end = out_end if out_end else len(buffer_out) in_end = in_end if in_end else len(buffer_in) + if int(self._novaCMDVer) >= 1: + totalIn = in_end - in_start + totalOut = out_end - out_start + totalBytes = totalIn + if totalOut > totalIn: + totalBytes = totalOut + chunks, rest = divmod(totalBytes, self.WHR_PAYLOAD_MAX_LENGTH) + if rest > 0: + chunks += 1 + + for i in range(chunks): + # calculate the number of bytes to be written and read + numInBytes = self.WHR_PAYLOAD_MAX_LENGTH + if totalIn < self.WHR_PAYLOAD_MAX_LENGTH: + numInBytes = totalIn + numOutBytes = self.WHR_PAYLOAD_MAX_LENGTH + if totalOut < self.WHR_PAYLOAD_MAX_LENGTH: + numOutBytes = totalOut + + # setup the buffer out chunk offset + buffer = '0' + if numOutBytes > 0: + chunk_start = out_start + i * self.WHR_PAYLOAD_MAX_LENGTH + chunk_end = chunk_start + numOutBytes + buffer = buffer_out[chunk_start:chunk_end] + + result = self._nova.writeToReadFromI2C(0, + address<<1, + stop, + numInBytes, + numOutBytes, + buffer) + totalIn -= numInBytes + totalOut -= numOutBytes + + if result != "-NG": + if numInBytes: + resp = result.split(" ") + resp = resp[2] + + for j in range(numInBytes): + buffer_in[in_start+i*self.WHR_PAYLOAD_MAX_LENGTH+j] = \ + int(resp[j*2]+resp[j*2+1], 16) + else: + raise RuntimeError("Received error response from Binho Nova, result = " + \ + result) + else: + self._nova.startI2C(0, address<<1) - self._nova.startI2C(0, address << 1) - - for i in range(out_start, out_end): - self._nova.writeByteI2C(0, buffer_out[i]) - - self._nova.endI2C(0, True) - - result = self._nova.readBytesI2C( - 0, address << 1, len(buffer_in[in_start:in_end]) - ) + for i in range(out_start, out_end): + self._nova.writeByteI2C(0, buffer_out[i]) - if result != "-NG": - resp = result.split(" ") + if stop: + self._nova.endI2C(0) + else: + self._nova.endI2C(0, True) - for i in range(len(buffer_in[in_start:in_end])): - buffer_in[in_start + i] = int(resp[2 + i]) - else: - raise RuntimeError( - "Received error response from Binho Nova, result = " + result - ) + result = self._nova.readBytesI2C(0, address<<1, len(buffer_in[in_start:in_end])) + if result != "-NG": + resp = result.split(" ") + for i in range(len(buffer_in[in_start:in_end])): + buffer_in[in_start+i] = int(resp[2+i]) + else: + raise RuntimeError( + "Received error response from Binho Nova, result = " + result + ) # pylint: enable=unused-argument