QA: Increase test coverage. Fix bugs.

as7343/__init__.py

@@ -1,8 +1,7 @@
 """Library for the AS7343 Visible Light Spectral Sensor."""
-import struct
 import time
 
-from i2cdevice import BitField, Device, Register, _int_to_bytes
+from i2cdevice import BitField, Device, Register
 from i2cdevice.adapter import Adapter, LookupAdapter, U16ByteSwapAdapter
 
 __version__ = '0.0.1'
@@ -96,14 +95,6 @@ def _encode(self, value):
  return int((value - 4) / 2) & 0x7f
 
 
-class FloatAdapter(Adapter):
- """Convert a 4 byte register set into a float."""
-
- def _decode(self, value):
- b = _int_to_bytes(value, 4)
- return struct.unpack('>f', bytearray(b))[0]
-
-
 class ResultCycle:
  """Store a single AS7343 result cycle."""
  def __init__(self, vis_tl, vis_br, astatus):
@@ -201,7 +192,7 @@ def __iter__(self): # noqa D107
 
 class AS7343:
  def __init__(self, i2c_dev=None):
- self._as7343 = Device(0x39, bit_width=8, registers=(
+ self._as7343 = Device(0x39, i2c_dev=i2c_dev, bit_width=8, registers=(
  # BANK 1
  Register('AUXID', 0x58, fields=(
  BitField('AUXID', 0b00001111), # Auxiliary Identification (0b0000)
@@ -523,35 +514,31 @@ def set_channels(self, channel_count):
  self._read_cycles = int(channel_count / 6)
  self._as7343.set('CFG20', auto_SMUX=channel_count)
 
- def get_data(self):
- results = list(self.read_fifo())
- if len(results) == self._read_cycles * 7:
- if self._read_cycles == 3:
- return (
- dict(ResultCycle1(*results[0:7])),
- dict(ResultCycle2(*results[7:14])),
- dict(ResultCycle3(*results[14:21])),
- )
- elif self._read_cycles == 2:
- return (
- dict(ResultCycle1(*results[0:7])),
- dict(ResultCycle2(*results[7:14]))
- )
- elif self._read_cycles == 1:
- return (
- dict(ResultCycle1(*results[0:7])),
- )
- else:
- # Uh oh?
- return None
-
- return results
-
- return None
-
- def read_fifo(self):
+ def get_data(self, timeout=5.0):
+ results = list(self.read_fifo(timeout=timeout))
+
+ if self._read_cycles == 3:
+ return (
+ dict(ResultCycle1(*results[0:7])),
+ dict(ResultCycle2(*results[7:14])),
+ dict(ResultCycle3(*results[14:21])),
+ )
+ elif self._read_cycles == 2:
+ return (
+ dict(ResultCycle1(*results[0:7])),
+ dict(ResultCycle2(*results[7:14]))
+ )
+ elif self._read_cycles == 1:
+ return (
+ dict(ResultCycle1(*results[0:7])),
+ )
+
+ def read_fifo(self, timeout=5.0):
+ t_start = time.time()
  while self._as7343.get('FIFO_LVL').FIFO_LVL < self._read_cycles * 7:
  time.sleep(0.001)
+ if time.time() - t_start > timeout:
+ raise TimeoutError(f"Timeout waiting for {self._read_cycles * 7} entries in FIFO.")
 
  while self._as7343.get('FIFO_LVL').FIFO_LVL > 0:
  result = self._as7343.get('FDATA').FDATA

tests/conftest.py

@@ -7,14 +7,21 @@
 
 class SMBusFakeDevice(MockSMBus):
  def __init__(self, i2c_bus):
- MockSMBus.__init__(self, i2c_bus)
+ MockSMBus.__init__(self, i2c_bus, default_registers={
+ 0x58: 0x08, # Fake aux ID
+ 0x59: 0x07, # Fake rev ID
+ 0x5A: 0b10000001 # Fake ID (part number?)
+ })
 
- self.regs = [0 for _ in range(255)]
+ def write_i2c_block_data(self, i2c_address, register, values):
+ self.regs[register:register + len(values)] = values
 
- # Virtual registers, these contain the data actually used
- self.regs[0x58] = 0x08 # Fake aux ID
- self.regs[0x59] = 0x07 # Fake rev ID
- self.regs[0x5A] = 0b10000001 # Fake ID (part number?)
+ def read_i2c_block_data(self, i2c_address, register, length):
+ # Catch reads from FDATA and decriment FIFO_LVL
+ if register == 0xFE and self.regs[0xFD] > 0:
+ self.regs[0xFD] -= 1
+
+ return self.regs[register:register + length]
 
 
 @pytest.fixture(scope='function', autouse=False)

tests/test_features.py

@@ -31,6 +31,9 @@ def test_set_integration_time(smbus):
  assert round(as7343._as7343.ASTEP.get_ASTEP(), 1) == 99999.4
  assert as7343._as7343.ATIME.get_ATIME() == 1 # Repeat twice
 
+ # Values greater than 46639948.8 are out of range
+ with pytest.raises(ValueError):
+ as7343.set_integration_time(46639948.8 + 1)
 
 def test_set_gain(smbus):
  """Test the set_gain method against various values."""
@@ -103,3 +106,53 @@ def test_soft_reset(smbus):
 
  as7343.soft_reset()
  assert as7343._as7343.CONTROL.get_SW_RESET() == 1
+
+
+def test_agc_gain(smbus):
+ from as7343 import AS7343
+ as7343 = AS7343()
+
+ assert as7343._as7343.AGC_GAIN_MAX.get_AGC_FD_GAIN_MAX() == 0.5
+
+ # AGC_FD_GAIN_MAX is the upper nibble
+ as7343._as7343._i2c.regs[0xD7] = 10 << 4
+ assert as7343._as7343.AGC_GAIN_MAX.get_AGC_FD_GAIN_MAX() == 2048
+
+ as7343._as7343.AGC_GAIN_MAX.set_AGC_FD_GAIN_MAX(1024)
+ assert (as7343._as7343._i2c.regs[0xD7] >> 4) == 9
+
+ as7343._as7343.AGC_GAIN_MAX.set_AGC_FD_GAIN_MAX(0.5)
+ assert (as7343._as7343._i2c.regs[0xD7] >> 4) == 0
+
+
+def test_set_channels(smbus):
+ from as7343 import AS7343
+ as7343 = AS7343()
+
+ as7343.set_channels(6)
+ as7343.set_channels(12)
+ as7343.set_channels(18)
+
+ with pytest.raises(ValueError):
+ as7343.set_channels(17)
+
+
+def test_get_data_timeout(smbus):
+ from as7343 import AS7343
+ as7343 = AS7343()
+
+ as7343.set_channels(6)
+
+ with pytest.raises(TimeoutError):
+ _ = as7343.get_data(timeout=0.5)
+
+
+def test_get_data(smbus):
+ from as7343 import AS7343
+ as7343 = AS7343()
+
+ for num_channels in (6, 12, 18):
+ as7343.set_channels(num_channels)
+ # Set the FIFO level, must be >= read_cycles * 7 or will timeout
+ as7343._as7343._i2c.regs[0xFD] = as7343._read_cycles * 7
+ _ = as7343.get_data()

tests/test_setup.py

@@ -1,4 +1,6 @@
 # noqa D100
+import pytest
+
 
 def test_fw_info(smbus):
  """Test against fake device information stored in hardware mock."""
@@ -10,3 +12,13 @@ def test_fw_info(smbus):
  assert auxid == 0x08
  assert revid == 0x07
  assert id == PART_ID
+
+
+def test_fw_info_fail(smbus):
+ """Test part ID check fails with RuntimeError."""
+ from as7343 import AS7343
+ i2c_dev = smbus.SMBus(1)
+ i2c_dev.regs[0x5A] = 0b11111111 # Wrong part ID
+
+ with pytest.raises(RuntimeError):
+ _ = AS7343(i2c_dev=i2c_dev)