feat: allow reading and writing of device state

Author: feeph

feeph/i2c/burst_handler.py

@@ -12,6 +12,10 @@
 with feeph.i2c.Burst(i2c_bus=i2c_bus, i2c_adr=0x4C) as bh:
     value = bh.read_register(register)
     bh.write_register(register, value + 1)
+
+with feeph.i2c.Burst(i2c_bus=i2c_bus, i2c_adr=0x70) as bh:
+    value = bh.get_state()
+    bh.set_state(value + 1)
 ```
 """
 
@@ -38,6 +42,17 @@ def __init__(self, i2c_bus: busio.I2C, i2c_adr: int):
         else:
             raise ValueError(f"Provided I²C address {i2c_adr} is out of range! (allowed range: 0 ≤ x ≤ 255)")
 
+    # fundamentally there isn't actually much difference between accesses
+    # to a device's register or internal state
+    #
+    #    read from register        write to register
+    # 1. write one byte            write one byte          (device address)
+    # 2. write one byte            write one byte          (register address)
+    # 2. read one or more bytes    write one or more bytes (register value)
+    #
+    # step 2. is skipped when accessing the internal state,
+    # step 1. and 3. are kept
+
     def read_register(self, register: int, byte_count: int = 1, max_tries: int = 5) -> int:
         """
         read a single register from I²C device identified by `i2c_adr` and
@@ -97,6 +112,62 @@ def write_register(self, register: int, value: int, byte_count: int = 1, max_tri
         else:
             raise RuntimeError(f"Unable to read register 0x{register:02X} after {cur_try} attempts. Giving up.")
 
+    def get_state(self, byte_count: int = 1, max_tries: int = 5) -> int:
+        """
+        get current state of I²C device identified by `i2c_adr` and
+        return its contents as an integer value
+        - may raise a RuntimeError if it was not possible to acquire
+            the bus within allowed time
+        - may raise a RuntimeError if there were too many errors
+        """
+        _validate_inputs(register=0, value=0, byte_count=byte_count, max_tries=max_tries)
+        if byte_count > 1:
+            LH.warning("Multi byte reads are not implemented yet! Returning a single byte instead.")
+            byte_count = 1
+        for cur_try in range(1, 1 + max_tries):
+            try:
+                buf = bytearray(byte_count)
+                # TODO properly handle multi byte reads
+                self._i2c_bus.readfrom_into(address=self._i2c_adr, buffer=buf)
+                return buf[0]
+            except OSError as e:
+                # [Errno 121] Remote I/O error
+                LH.warning("[%s] Failed to read state (%i/%i): %s",  __name__, cur_try, max_tries, e)
+                time.sleep(0.001)
+            except RuntimeError as e:
+                LH.warning("[%s] Unable to read state (%i/%i): %s", __name__, cur_try, max_tries, e)
+                time.sleep(0.001)
+        else:
+            raise RuntimeError(f"Unable to read state after {cur_try} attempts. Giving up.")
+
+    def set_state(self, value: int, byte_count: int = 1, max_tries: int = 3):
+        """
+        set current state of I²C device identified by `i2c_adr`
+        - may raise a RuntimeError if it was not possible to acquire
+            the bus within allowed time
+        - may raise a RuntimeError if there were too many errors
+        """
+        _validate_inputs(register=0, value=value, byte_count=byte_count, max_tries=max_tries)
+        if byte_count > 1:
+            LH.warning("Multi byte writes are not implemented yet! Returning a single byte instead.")
+            byte_count = 1
+        for cur_try in range(1, 1 + max_tries):
+            try:
+                buf = bytearray(byte_count)
+                buf[0] = value & 0xFF
+                # TODO properly handle multi byte reads
+                self._i2c_bus.writeto(address=self._i2c_adr, buffer=buf)
+                return
+            except OSError as e:
+                # [Errno 121] Remote I/O error
+                LH.warning("[%s] Failed to write state (%i/%i): %s",  __name__, cur_try, max_tries, e)
+                time.sleep(0.1)
+            except RuntimeError as e:
+                LH.warning("[%s] Unable to write state 0x%02X (%i/%i): %s", __name__, cur_try, max_tries, e)
+                time.sleep(0.1)
+        else:
+            raise RuntimeError(f"Unable to write state after {cur_try} attempts. Giving up.")
+
 
 class BurstHandler:
     """

feeph/i2c/emulation.py

@@ -43,16 +43,31 @@ def try_lock(self) -> bool:
     def unlock(self):
         pass
 
-    def readinto(self, buffer, *, start: int = 0, end: int | None = None):
-        # provided to ensure we will never call `I2C.readinto()`
-        raise RuntimeError("EmulatedI2cBus.readinto() is not implemented")
-
-    def writeto(self, address, buffer, *, start=0, end=None):
+    def readfrom_into(self, address, buffer, *, start=0, end=None, stop=True):
+        """
+        read data from device
+        """
         i2c_device_address  = address
-        i2c_device_register = buffer[0]
-        self._state[i2c_device_address][i2c_device_register] = buffer[1]
+        buffer[0] = self._state[i2c_device_address]
+
+    def writeto(self, address: int, buffer: bytearray, *, start=0, end=None):
+        """
+        write data to device or to device register
+        """
+        if len(buffer) == 1:
+            # device status
+            i2c_device_address  = address
+            self._state[i2c_device_address] = buffer[0]
+        else:
+            # device register
+            i2c_device_address  = address
+            i2c_device_register = buffer[0]
+            self._state[i2c_device_address][i2c_device_register] = buffer[1]
 
     def writeto_then_readfrom(self, address: int, buffer_out: bytearray, buffer_in: bytearray, *, out_start=0, out_end=None, in_start=0, in_end=None, stop=False):
+        """
+        read data from a device register
+        """
         i2c_device_address  = address
         i2c_device_register = buffer_out[0]
         buffer_in[0] = self._state[i2c_device_address][i2c_device_register]

tests/test_burst_handler.py

@@ -132,6 +132,35 @@ def test_mixed_access(self):
         self.assertEqual(computed_r, expected_r)
         self.assertEqual(computed_w, expected_w)
 
+    # ---------------------------------------------------------------------
+
+    def test_get_state(self):
+        state = {
+            0x70: 0x01,
+        }
+        i2c_bus = sut.EmulatedI2C(state=state)
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x70) as bh:
+            computed = bh.get_state()
+        expected = 0x01
+        # -----------------------------------------------------------------
+        self.assertEqual(computed, expected)
+
+    def test_set_state(self):
+        state = {
+            0x70: 0x00,
+        }
+        i2c_bus = sut.EmulatedI2C(state=state)
+        # -----------------------------------------------------------------
+        with sut.BurstHandler(i2c_bus=i2c_bus, i2c_adr=0x70) as bh:
+            bh.set_state(value=0x01)
+        computed = i2c_bus._state[0x70]
+        expected = 0x01
+        # -----------------------------------------------------------------
+        self.assertEqual(computed, expected)
+
+    # ---------------------------------------------------------------------
+
     def test_no_timeout(self):
         state = {
             0x4C: {