Add type hints

.github/workflows/python-package.yml

@@ -27,12 +27,14 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install flake8 codecov
-        pip install -r tests/requirements.txt
+        pip install -r requirements-dev.txt
     - name: Lint with flake8
       run: |
         # stop the build if there are Python syntax errors or undefined names
         flake8 .
+    - name: Typecheck with mypy
+      run: |
+        mypy -p ina219 --strict
     - name: Test with codecov
       run: |
         coverage run --branch --source=ina219 -m unittest discover -s tests -p 'test_*.py'

.gitignore

@@ -1,5 +1,6 @@
 *.pyc
 .coverage
+coverage.xml
 build
 dist
 pi_ina219.egg-info

.travis.yml

@@ -1,20 +1,18 @@
 language: python
 python:
-  - "2.7"
-  - "3.5"
   - "3.6"
   - "3.7"
   - "3.8"
+  - "3.9"
 # command to install dependencies
 install:
-  - pip install flake8
-  - pip install codecov
-  - pip install -r tests/requirements.txt
-# check PEP8 coding standard with flake8
+  - pip install -r requirements-dev.txt
+# check PEP8 coding standard with flake8 and typing with mypy
 before_script:
-   flake8 .
+  - flake8 .
+  - mypy -p ina219 --strict
 # command to run tests
-script: 
+script:
   - coverage run --branch --source=ina219 -m unittest discover -s tests -p 'test_*.py'
   - coverage xml
 after_success:

README.md

@@ -31,14 +31,14 @@ This library and its dependency
 can be installed from PyPI by executing:
 
 ```shell
-sudo pip3 install pi-ina219
+pip3 install pi-ina219
 ```
 
 To upgrade from a previous version installed direct from Github execute:
 
 ```shell
-sudo pip3 uninstall pi-ina219
-sudo pip3 install pi-ina219
+pip3 uninstall pi-ina219
+pip3 install pi-ina219
 ```
 
 The Adafruit library supports the I2C protocol on all versions of the
@@ -291,12 +291,15 @@ Detailed logging of device register operations can be enabled with:
     ina = INA219(SHUNT_OHMS, log_level=logging.DEBUG)
 ```
 
-## Testing
+## Development
 
-Install the library as described above, this will install all the
-dependencies required for the unit tests, as well as the library
-itself. Clone the library source from Github then execute the test suite
-from the top level directory with:
+Install development dependencies first _(recommended to use virtual environments)_. This includes
+package dependencies, testing dependencies and static analysis dependencies.
+```shell
+pip3 install -r requirements-dev.txt
+```
+
+### Testing
 
 ```shell
 python3 -m unittest discover -s tests -p 'test_*.py'
@@ -313,9 +316,16 @@ Code coverage metrics may be generated and viewed with:
 ```shell
 coverage run --branch --source=ina219 -m unittest discover -s tests -p 'test_*.py'
 coverage report -m
+coverage xml
 ```
 
-## Coding Standard
+### Coding Standard
 
 This library adheres to the _PEP8_ standard and follows the _idiomatic_
 style described in the book _Writing Idiomatic Python_ by _Jeff Knupp_.
+
+To perform local linting and type checks, run:
+```sh
+flake8 .
+mypy -p ina219 --strict
+```

ina219.py

@@ -5,7 +5,8 @@
 import logging
 import time
 from math import trunc
-import Adafruit_GPIO.I2C as I2C
+from typing import cast, List, Optional
+import Adafruit_GPIO.I2C as I2C  # type: ignore
 
 
 class INA219:
@@ -90,9 +91,10 @@ class INA219:
     # to guarantee that current overflow can always be detected.
     __CURRENT_LSB_FACTOR = 32800
 
-    def __init__(self, shunt_ohms, max_expected_amps=None,
-                 busnum=None, address=__ADDRESS,
-                 log_level=logging.ERROR):
+    def __init__(self, shunt_ohms: float,
+                 max_expected_amps: Optional[float] = None,
+                 busnum: Optional[int] = None, address: int = __ADDRESS,
+                 log_level: int = logging.ERROR) -> None:
         """Construct the class.
 
         Pass in the resistance of the shunt resistor and the maximum expected
@@ -117,11 +119,12 @@ def __init__(self, shunt_ohms, max_expected_amps=None,
         self._shunt_ohms = shunt_ohms
         self._max_expected_amps = max_expected_amps
         self._min_device_current_lsb = self._calculate_min_current_lsb()
-        self._gain = None
+        self._gain: Optional[int] = None
         self._auto_gain_enabled = False
 
-    def configure(self, voltage_range=RANGE_32V, gain=GAIN_AUTO,
-                  bus_adc=ADC_12BIT, shunt_adc=ADC_12BIT):
+    def configure(self, voltage_range: int = RANGE_32V, gain: int = GAIN_AUTO,
+                  bus_adc: int = ADC_12BIT,
+                  shunt_adc: int = ADC_12BIT) -> None:
         """Configure and calibrate how the INA219 will take measurements.
 
         Arguments:
@@ -175,87 +178,88 @@ def configure(self, voltage_range=RANGE_32V, gain=GAIN_AUTO,
             self._max_expected_amps)
         self._configure(voltage_range, self._gain, bus_adc, shunt_adc)
 
-    def voltage(self):
+    def voltage(self) -> float:
         """Return the bus voltage in volts."""
         value = self._voltage_register()
         return float(value) * self.__BUS_MILLIVOLTS_LSB / 1000
 
-    def supply_voltage(self):
+    def supply_voltage(self) -> float:
         """Return the bus supply voltage in volts.
 
         This is the sum of the bus voltage and shunt voltage. A
         DeviceRangeError exception is thrown if current overflow occurs.
         """
         return self.voltage() + (float(self.shunt_voltage()) / 1000)
 
-    def current(self):
+    def current(self) -> float:
         """Return the bus current in milliamps.
 
         A DeviceRangeError exception is thrown if current overflow occurs.
         """
         self._handle_current_overflow()
         return self._current_register() * self._current_lsb * 1000
 
-    def power(self):
+    def power(self) -> float:
         """Return the bus power consumption in milliwatts.
 
         A DeviceRangeError exception is thrown if current overflow occurs.
         """
         self._handle_current_overflow()
         return self._power_register() * self._power_lsb * 1000
 
-    def shunt_voltage(self):
+    def shunt_voltage(self) -> float:
         """Return the shunt voltage in millivolts.
 
         A DeviceRangeError exception is thrown if current overflow occurs.
         """
         self._handle_current_overflow()
         return self._shunt_voltage_register() * self.__SHUNT_MILLIVOLTS_LSB
 
-    def sleep(self):
+    def sleep(self) -> None:
         """Put the INA219 into power down mode."""
         configuration = self._read_configuration()
         self._configuration_register(configuration & 0xFFF8)
 
-    def wake(self):
+    def wake(self) -> None:
         """Wake the INA219 from power down mode."""
         configuration = self._read_configuration()
         self._configuration_register(configuration | 0x0007)
         # 40us delay to recover from powerdown (p14 of spec)
         time.sleep(0.00004)
 
-    def current_overflow(self):
+    def current_overflow(self) -> bool:
         """Return true if the sensor has detect current overflow.
 
         In this case the current and power values are invalid.
         """
         return self._has_current_overflow()
 
-    def reset(self):
+    def reset(self) -> None:
         """Reset the INA219 to its default configuration."""
         self._configuration_register(1 << self.__RST)
 
-    def is_conversion_ready(self):
+    def is_conversion_ready(self) -> bool:
         """Check if conversion of a new reading has occured."""
         cnvr = self._read_voltage_register() & self.__CNVR
         return (cnvr == self.__CNVR)
 
-    def _handle_current_overflow(self):
+    def _handle_current_overflow(self) -> None:
         if self._auto_gain_enabled:
             while self._has_current_overflow():
                 self._increase_gain()
         else:
             if self._has_current_overflow():
-                raise DeviceRangeError(self.__GAIN_VOLTS[self._gain])
+                raise DeviceRangeError(
+                    self.__GAIN_VOLTS[self._gain] if self._gain else 0.0)
 
-    def _determine_gain(self, max_expected_amps):
+    def _determine_gain(self, max_expected_amps: float) -> int:
         shunt_v = max_expected_amps * self._shunt_ohms
         if shunt_v > self.__GAIN_VOLTS[3]:
             raise ValueError(self.__RNG_ERR_MSG % max_expected_amps)
         gain = min(v for v in self.__GAIN_VOLTS if v > shunt_v)
         return self.__GAIN_VOLTS.index(gain)
 
-    def _increase_gain(self):
+    def _increase_gain(self) -> None:
         self.logger.info(self.__LOG_MSG_3)
         gain = self._read_gain()
         if gain < len(self.__GAIN_VOLTS) - 1:
@@ -270,15 +274,16 @@ def _increase_gain(self):
             self.logger.info('Device limit reach, gain cannot be increased')
             raise DeviceRangeError(self.__GAIN_VOLTS[gain], True)
 
-    def _configure(self, voltage_range, gain, bus_adc, shunt_adc):
+    def _configure(self, voltage_range: int, gain: int, bus_adc: int,
+                   shunt_adc: int) -> None:
         configuration = (
             voltage_range << self.__BRNG | gain << self.__PG0 |
             bus_adc << self.__BADC1 | shunt_adc << self.__SADC1 |
             self.__CONT_SH_BUS)
         self._configuration_register(configuration)
 
-    def _calibrate(self, bus_volts_max, shunt_volts_max,
-                   max_expected_amps=None):
+    def _calibrate(self, bus_volts_max: int, shunt_volts_max: float,
+                   max_expected_amps: Optional[float] = None) -> None:
         self.logger.info(
             self.__LOG_MSG_2 %
             (bus_volts_max, shunt_volts_max,
@@ -309,7 +314,8 @@ def _calibrate(self, bus_volts_max, shunt_volts_max,
             "calibration: 0x%04x (%d)" % (calibration, calibration))
         self._calibration_register(calibration)
 
-    def _determine_current_lsb(self, max_expected_amps, max_possible_amps):
+    def _determine_current_lsb(self, max_expected_amps: Optional[float],
+                               max_possible_amps: float) -> float:
         if max_expected_amps is not None:
             if max_expected_amps > round(max_possible_amps, 3):
                 raise ValueError(self.__AMP_ERR_MSG %
@@ -327,67 +333,68 @@ def _determine_current_lsb(self, max_expected_amps, max_possible_amps):
             current_lsb = self._min_device_current_lsb
         return current_lsb
 
-    def _configuration_register(self, register_value):
+    def _configuration_register(self, register_value: int) -> None:
         self.logger.debug("configuration: 0x%04x" % register_value)
         self.__write_register(self.__REG_CONFIG, register_value)
 
-    def _read_configuration(self):
+    def _read_configuration(self) -> int:
         return self.__read_register(self.__REG_CONFIG)
 
-    def _calculate_min_current_lsb(self):
+    def _calculate_min_current_lsb(self) -> float:
         return self.__CALIBRATION_FACTOR / \
             (self._shunt_ohms * self.__MAX_CALIBRATION_VALUE)
 
-    def _read_gain(self):
+    def _read_gain(self) -> int:
         configuration = self._read_configuration()
         gain = (configuration & 0x1800) >> self.__PG0
         self.logger.info("gain is currently: %.2fV" % self.__GAIN_VOLTS[gain])
         return gain
 
-    def _configure_gain(self, gain):
+    def _configure_gain(self, gain: int) -> None:
         configuration = self._read_configuration()
         configuration = configuration & 0xE7FF
         self._configuration_register(configuration | (gain << self.__PG0))
         self._gain = gain
         self.logger.info("gain set to: %.2fV" % self.__GAIN_VOLTS[gain])
 
-    def _calibration_register(self, register_value):
+    def _calibration_register(self, register_value: int) -> None:
         self.logger.debug("calibration: 0x%04x" % register_value)
         self.__write_register(self.__REG_CALIBRATION, register_value)
 
-    def _has_current_overflow(self):
+    def _has_current_overflow(self) -> bool:
         ovf = self._read_voltage_register() & self.__OVF
         return (ovf == 1)
 
-    def _voltage_register(self):
+    def _voltage_register(self) -> int:
         register_value = self._read_voltage_register()
         return register_value >> 3
 
-    def _read_voltage_register(self):
+    def _read_voltage_register(self) -> int:
         return self.__read_register(self.__REG_BUSVOLTAGE)
 
-    def _current_register(self):
+    def _current_register(self) -> int:
         return self.__read_register(self.__REG_CURRENT, True)
 
-    def _shunt_voltage_register(self):
+    def _shunt_voltage_register(self) -> int:
         return self.__read_register(self.__REG_SHUNTVOLTAGE, True)
 
-    def _power_register(self):
+    def _power_register(self) -> int:
         return self.__read_register(self.__REG_POWER)
 
-    def __validate_voltage_range(self, voltage_range):
+    def __validate_voltage_range(self, voltage_range: int) -> None:
         if voltage_range > len(self.__BUS_RANGE) - 1:
             raise ValueError(self.__VOLT_ERR_MSG)
 
-    def __write_register(self, register, register_value):
+    def __write_register(self, register: int, register_value: int) -> None:
         register_bytes = self.__to_bytes(register_value)
         self.logger.debug(
             "write register 0x%02x: 0x%04x 0b%s" %
             (register, register_value,
              self.__binary_as_string(register_value)))
         self._i2c.writeList(register, register_bytes)
 
-    def __read_register(self, register, negative_value_supported=False):
+    def __read_register(self, register: int,
+                        negative_value_supported: bool = False) -> int:
         if negative_value_supported:
             register_value = self._i2c.readS16BE(register)
         else:
@@ -396,15 +403,16 @@ def __read_register(self, register, negative_value_supported=False):
             "read register 0x%02x: 0x%04x 0b%s" %
             (register, register_value,
              self.__binary_as_string(register_value)))
-        return register_value
+        return cast(int, register_value)
 
-    def __to_bytes(self, register_value):
+    def __to_bytes(self, register_value: int) -> List[int]:
         return [(register_value >> 8) & 0xFF, register_value & 0xFF]
 
-    def __binary_as_string(self, register_value):
+    def __binary_as_string(self, register_value: int) -> str:
         return bin(register_value)[2:].zfill(16)
 
-    def __max_expected_amps_to_string(self, max_expected_amps):
+    def __max_expected_amps_to_string(self, max_expected_amps:
+                                      Optional[float]) -> str:
         if max_expected_amps is None:
             return ''
         else:
@@ -417,7 +425,7 @@ class DeviceRangeError(Exception):
     __DEV_RNG_ERR = ('Current out of range (overflow), '
                      'for gain %.2fV')
 
-    def __init__(self, gain_volts, device_max=False):
+    def __init__(self, gain_volts: float, device_max: bool = False) -> None:
         """Construct a DeviceRangeError."""
         msg = self.__DEV_RNG_ERR % gain_volts
         if device_max:

requirements-dev.txt

@@ -0,0 +1,5 @@
+-r ./requirements.txt
+-r ./tests/requirements.txt
+
+flake8
+mypy

requirements.txt

@@ -0,0 +1 @@
+Adafruit_GPIO==1.0.1

setup.py

@@ -1,11 +1,3 @@
-try:
-    # Try using ez_setup to install setuptools if not already installed.
-    from ez_setup import use_setuptools
-    use_setuptools()
-except ImportError:
-    # Ignore import error and assume Python 3 which already has setuptools.
-    pass
-
 from setuptools import setup
 
 DESC = ('This Python library for Raspberry Pi makes it easy to leverage the '
@@ -16,11 +8,10 @@
                'Operating System :: POSIX :: Linux',
                'License :: OSI Approved :: MIT License',
                'Intended Audience :: Developers',
-               'Programming Language :: Python :: 2.7',
-               'Programming Language :: Python :: 3.4',
-               'Programming Language :: Python :: 3.5',
                'Programming Language :: Python :: 3.6',
                'Programming Language :: Python :: 3.7',
+               'Programming Language :: Python :: 3.8',
+               'Programming Language :: Python :: 3.9',
                'Topic :: System :: Hardware :: Hardware Drivers']
 
 # Define required packages.

tests/requirements.txt

@@ -1,2 +1,3 @@
-Adafruit_GPIO==1.0.1
 mock==2.0.0
+coverage==5.1
+codecov>=2.1

tests/test_configuration.py

@@ -75,7 +75,7 @@ def test_manual_gain_no_expected_amps(self, device):
     def test_auto_gain_out_of_range(self, device):
         device.return_value = Mock()
         self.ina = INA219(0.1, 4)
-        with self.assertRaisesRegexp(ValueError, "Expected amps"):
+        with self.assertRaisesRegex(ValueError, "Expected amps"):
             self.ina.configure(self.ina.RANGE_16V, self.ina.GAIN_AUTO)
 
     def test_16v_40mv(self):
@@ -147,14 +147,14 @@ def test_32v_40mv_32_samples_64_samples(self):
         self.ina._i2c.writeList.assert_has_calls(calls)
 
     def test_invalid_voltage_range(self):
-        with self.assertRaisesRegexp(ValueError, "Invalid voltage range"):
+        with self.assertRaisesRegex(ValueError, "Invalid voltage range"):
             self.ina.configure(64, self.ina.GAIN_1_40MV)
 
     @patch('Adafruit_GPIO.I2C.get_i2c_device')
     def test_max_current_exceeded(self, device):
         device.return_value = Mock()
         ina = INA219(0.1, 0.5)
-        with self.assertRaisesRegexp(ValueError, "Expected current"):
+        with self.assertRaisesRegex(ValueError, "Expected current"):
             ina.configure(ina.RANGE_32V, ina.GAIN_1_40MV)
 
     def test_sleep(self):

tests/test_read.py

@@ -101,7 +101,7 @@ def test_current_overflow_valid(self):
     def test_current_overflow_error(self):
         self.ina.configure(self.ina.RANGE_16V, self.ina.GAIN_2_80MV)
         self.ina._i2c.readU16BE = Mock(return_value=0xfa1)
-        with self.assertRaisesRegexp(DeviceRangeError, self.GAIN_RANGE_MSG):
+        with self.assertRaisesRegex(DeviceRangeError, self.GAIN_RANGE_MSG):
             self.ina.current()
 
     def test_new_read_available(self):

tests/test_read_auto_gain.py

@@ -43,5 +43,5 @@ def test_auto_gain_out_of_range(self, device):
         self.ina._read_voltage_register = Mock(return_value=0xfa1)
         self.ina._read_configuration = Mock(return_value=0x199f)
 
-        with self.assertRaisesRegexp(DeviceRangeError, self.GAIN_RANGE_MSG):
+        with self.assertRaisesRegex(DeviceRangeError, self.GAIN_RANGE_MSG):
             self.ina.current()