RTD Sampling

src/edgepi/adc/adc_voltage.py

@@ -52,16 +52,19 @@ def _adc_voltage_to_input_voltage(v_in: float, gain: float, offset: float):
     return v_in * step_up_ratio * gain - offset
 
 
-def code_to_voltage(code: list, adc_info: ADCReadInfo, calibs: CalibParam):
+def code_to_voltage(code: list[int], adc_info: ADCReadInfo, calibs: CalibParam) -> float:
     """
     Converts ADC voltage read digital code to output voltage (voltage measured at terminal block)
 
     Args:
-        `code` (BitArray): bitstring representation of digital code retrieved from ADC voltage read
+        `code` (list[int]): code bytes retrieved from ADC voltage read
 
         `adc_info` (ADCReadInfo): data about this adc's voltage reading configuration
 
         `calibs` (CalibParam): voltage reading gain and offset calibration values
+
+    Returns:
+        `float`: voltage value (V) corresponding to `code`
     """
     code_bits = bitstring_from_list(code)
     num_bits = adc_info.num_data_bytes * 8
@@ -77,6 +80,38 @@ def code_to_voltage(code: list, adc_info: ADCReadInfo, calibs: CalibParam):
     return v_out
 
 
+def code_to_temperature(
+    code: list[int],
+    ref_resistance: float,
+    rtd_sensor_resistance: float,
+    rtd_sensor_resistance_variation: float
+    ) -> float:
+    """
+    Converts ADC voltage read digital code to temperature. Intended for use in RTD sampling.
+
+    Args:
+        `code` (list[int]): code bytes retrieved from ADC voltage read
+
+       `ref_resistance` (float): EdgePi-specific RTD reference resistance (Ohms)
+
+       `rtd_sensor_resistance` (float): RTD material-dependent resistance value (Ohms)
+
+       `rtd_sensor_resistance_variation` (float): RTD model-dependent resistance variation (Ohms/°C)
+
+    Returns:
+        `float`: temperature value (°C) corresponding to `code`
+    """
+    code_bits = bitstring_from_list(code)
+
+    # refer to https://github.com/osensa/edgepi-python-sdk/issues/159 for computation details
+    r_rtd = code_bits.uint / (2 << 30) * ref_resistance
+
+    temperature = (r_rtd - rtd_sensor_resistance) / rtd_sensor_resistance_variation
+    _logger.debug(f"computed rtd temperature = {temperature}, from code = {code_bits.uint}")
+
+    return temperature
+
+
 class CRCCheckError(Exception):
     """Raised if CRC check fails to match CRC code generated by ADS1263"""
 

src/edgepi/adc/edgepi_adc.py

@@ -35,7 +35,7 @@
     RTDModes,
     AllowedChannels,
 )
-from edgepi.adc.adc_voltage import code_to_voltage, check_crc
+from edgepi.adc.adc_voltage import code_to_voltage, check_crc, code_to_temperature
 from edgepi.gpio.edgepi_gpio import EdgePiGPIO
 from edgepi.gpio.gpio_configs import GpioConfigs, ADCPins
 from edgepi.utilities.utilities import filter_dict
@@ -188,7 +188,30 @@ class EdgePiADC(SPI):
     # keep track of ADC register map state for state caching
     __state: dict = {}
 
-    def __init__(self, enable_cache: bool = False):
+    # default RTD model-dependent hardware constants
+    RTD_SENSOR_RESISTANCE = 100 # RTD sensor resistance value (Ohms)
+    RTD_SENSOR_RESISTANCE_VARIATION = 0.385 # RTD sensor resistance variation (Ohms/°C)
+
+    # TODO: this should be part of eeprom_data. Retrieve from eeprom_data in calling
+    # functions when available
+    r_ref = 1326.20
+
+    def __init__(
+        self,
+        enable_cache: bool = False,
+        rtd_sensor_resistance: float = None,
+        rtd_sensor_resistance_variation: float = None
+        ):
+        """
+        Args:
+            `enable_cache` (bool): set to True to enable state-caching
+
+            `rtd_sensor_resistance` (float): set RTD material-dependent resistance value (Ohms)
+
+            `rtd_sensor_resistance_variation` (float): set RTD model-dependent resistance
+                variation (Ohms/°C)
+        """
+
         super().__init__(bus_num=6, dev_id=1)
         # declare instance vars before config call below
         self.enable_cache = enable_cache
@@ -207,7 +230,19 @@ def __init__(self, enable_cache: bool = False):
         # TODO: adc reference should ba a config that customer passes depending on the range of
         # voltage they are measuring. To be changed later when range config is implemented
         self.set_adc_reference(ADCReferenceSwitching.GND_SW1.value)
-        # TODO: get gain, offset, ref configs from the config module
+
+        # user updated rtd hardware constants
+        self.rtd_sensor_resistance = (
+            rtd_sensor_resistance
+            if rtd_sensor_resistance is not None
+            else EdgePiADC.RTD_SENSOR_RESISTANCE
+        )
+
+        self.rtd_sensor_resistance_variation = (
+            rtd_sensor_resistance_variation
+            if rtd_sensor_resistance_variation is not None
+            else EdgePiADC.RTD_SENSOR_RESISTANCE_VARIATION
+        )
 
     def __reapply_config(self):
         """
@@ -387,7 +422,7 @@ def __voltage_read(self, adc_num: ADCNum):
         and check bytes
 
         Returns:
-            (int): uint values of representations of voltage read data ordered as
+            int, list[int], int: uint values of representations of voltage read data ordered as
                 (status_byte, voltage_data_bytes, check_byte)
         """
         read_data = self.__read_data(adc_num, ADC_VOLTAGE_READ_LEN)
@@ -403,6 +438,8 @@ def __voltage_read(self, adc_num: ADCNum):
 
         check_code = read_data[6]
 
+        check_crc(voltage_code, check_code)
+
         return status_code, voltage_code, check_code
 
     @staticmethod
@@ -449,85 +486,133 @@ def __get_calibration_values(self, adc_calibs: dict, adc_num: ADCNum) -> CalibPa
 
         return adc_calibs.get(calib_key)
 
-    def read_voltage(self, adc_num: ADCNum):
-        """
-        Read voltage from the currently configured ADC analog input channel.
-        Use this method when ADC is configured to `CONTINUOUS` conversion mode.
-        For `PULSE` conversion mode, use `single_sample()` instead.
-
-        Returns:
-            `float`: input voltage read from ADC
-        """
-        # assert adc is in continuous mode (use ADCStatus)
+    def __continuous_time_delay(self, adc_num: ADCNum):
+        """Compute and enforce continuous conversion time delay"""
         state = self.get_state()
-        if state.adc_1.conversion_mode.code != ConvMode.CONTINUOUS:
-            raise ContinuousModeError(
-                "ADC must be in CONTINUOUS conversion mode in order to call `read_voltage`."
-            )
-
         # get continuous mode time delay and wait here (delay is needed between each conversion)
         data_rate = (
             state.adc_1.data_rate.code if adc_num == ADCNum.ADC_1 else state.adc_2.data_rate.code
         )
         delay = expected_continuous_time_delay(adc_num, data_rate.value.op_code)
-        _logger.debug(
-            (
-                f"Continuous time delay = {delay} (ms) with the following config opcodes:\n"
-                f"adc_num={adc_num},  data_rate={hex(data_rate.value.op_code)}\n"
-            )
-        )
+
         time.sleep(delay / 1000)
 
-        status_code, voltage_code, check_code = self.__voltage_read(adc_num)
+    def __check_adc_1_conv_mode(self):
+        # assert adc is in continuous mode
+        state = self.get_state()
+        if state.adc_1.conversion_mode.code != ConvMode.CONTINUOUS:
+            raise ContinuousModeError(
+                "ADC1 must be in CONTINUOUS conversion mode in order to call this method."
+            )
+
+    def read_voltage(self, adc_num: ADCNum):
+        """
+        Read voltage input to either ADC1 or ADC2, when performing single channel reading
+        or differential reading. For ADC1 reading, only use this method when ADC1 is configured
+        to `CONTINUOUS` conversion mode. For `PULSE` conversion mode, use `single_sample` instead.
+
+        Args:
+            `adc_num` (ADCNum): the ADC to be read
+
+        Returns:
+            `float`: input voltage (V) read from the indicated ADC
+        """
+        self.__check_adc_1_conv_mode()
+
+        self.__continuous_time_delay(adc_num)
+
+        status_code, voltage_code, _ = self.__voltage_read(adc_num)
 
         # log STATUS byte
         status = get_adc_status(status_code)
         _logger.debug(f"Logging STATUS byte:\n{status}")
 
-        # check CRC
-        check_crc(voltage_code, check_code)
-
         calibs = self.__get_calibration_values(self.adc_calib_params, adc_num)
 
-        # convert voltage_bits from code to voltage
-        voltage = code_to_voltage(voltage_code, adc_num.value, calibs)
+        # convert from code to voltage
+        return code_to_voltage(voltage_code, adc_num.value, calibs)
 
-        return voltage
 
-    def single_sample(self):
+    def read_rtd_temperature(self):
         """
-        Perform a single `ADC1` voltage read in `PULSE` conversion mode.
-        Do not call this method for voltage reading if ADC is configured
-        to `CONTINUOUS` conversion mode: use `read_voltage` instead.
+        Read RTD temperature continuously. Note, to obtain valid temperature values,
+        RTD mode must first be enabled by calling `rtd_mode`. ADC1 must also be configured
+        to `CONTINUOUS` conversion mode via `set_config`, before calling this method.
+
+        Returns:
+            `float`: RTD measured temperature (°C)
         """
+        self.__check_adc_1_conv_mode()
+
+        self.__continuous_time_delay(ADCNum.ADC_1)
+
+        _, voltage_code, _ = self.__voltage_read(ADCNum.ADC_1)
+
+        # TODO: get RTD calibs from eeprom once added
+        return code_to_temperature(
+            voltage_code,
+            self.r_ref,
+            self.rtd_sensor_resistance,
+            self.rtd_sensor_resistance_variation
+        )
+
+    def __enforce_pulse_mode(self):
         # assert adc is in PULSE mode (check ADCState) -> set to PULSE if not
         state = self.get_state()
-
         if state.adc_1.conversion_mode.code != ConvMode.PULSE:
             self.__config(conversion_mode=ConvMode.PULSE)
 
-        # only ADC1 can perform PULSE conversion
-        adc_num = ADCNum.ADC_1
+    def single_sample(self):
+        """
+        Trigger a single ADC1 voltage sampling event, when performing single channel reading or
+        differential reading. ADC1 must be in `PULSE` conversion mode before calling this method.
+        Do not call this method for voltage reading if ADC is configured to `CONTINUOUS`
+        conversion mode: use `read_voltage` instead.
+
+        Returns:
+            `float`: input voltage (V) read from ADC1
+        """
+        self.__enforce_pulse_mode()
 
         # send command to trigger conversion
-        self.start_conversions(adc_num)
+        self.start_conversions(ADCNum.ADC_1)
 
         # send command to read conversion data.
-        status_code, voltage_code, check_code = self.__voltage_read(adc_num)
+        status_code, voltage_code, _ = self.__voltage_read(ADCNum.ADC_1)
 
         # log STATUS byte
         status = get_adc_status(status_code)
         _logger.debug(f"Logging STATUS byte:\n{status}")
 
-        # check CRC
-        check_crc(voltage_code, check_code)
+        calibs = self.__get_calibration_values(self.adc_calib_params, ADCNum.ADC_1)
 
-        calibs = self.__get_calibration_values(self.adc_calib_params, adc_num)
+        # convert from code to voltage
+        return code_to_voltage(voltage_code, ADCNum.ADC_1.value, calibs)
 
-        # convert read_data from code to voltage
-        voltage = code_to_voltage(voltage_code, adc_num.value, calibs)
+    def single_sample_rtd(self):
+        """
+        Trigger a single RTD temperature sampling event. Note, to obtain valid temperature values,
+        RTD mode must first be enabled by calling `rtd_mode`. ADC1 must also be configured
+        to `PULSE` conversion mode via `set_config`, before calling this method.
+
+        Returns:
+            `float`: RTD measured temperature (°C)
+        """
+        self.__enforce_pulse_mode()
 
-        return voltage
+        # send command to trigger conversion
+        self.start_conversions(ADCNum.ADC_1)
+
+        # send command to read conversion data.
+        _, voltage_code, _ = self.__voltage_read(ADCNum.ADC_1)
+
+        # TODO: get RTD calibs from eeprom once added
+        return code_to_temperature(
+            voltage_code,
+            self.r_ref,
+            self.rtd_sensor_resistance,
+            self.rtd_sensor_resistance_variation
+        )
 
     def reset(self):
         """
@@ -683,11 +768,14 @@ def __validate_no_rtd_conflict(self, updates: dict):
     def rtd_mode(self, enable: bool):
         """
         Enable or disable RTD. Note, this will reconfigure ADC2 to read FLOAT input channel
-        if ADC2 is configured to read RTD pins AIN4-AIN7.
+        if ADC2 is configured to read RTD pins AIN4-AIN7. To read RTD values aftering enabling
+        RTD mode here, call either `read_voltage` or `single_sample`, depending on which
+        conversion mode ADC1 is currently configured to.
 
         Args:
             `enable` (bool): True to enable RTD, False to disable
         """
+        # TODO: enable RTD_EN switching pin
         if enable:
             # check if adc_2 is reading RTD pins, remap channels if needed
             mux_reg = self.__get_register_map()
@@ -757,8 +845,8 @@ def __config(
         Args:
             `adc_1_analog_in` (ADCChannel): input voltage channel to map to ADC1 mux_p
             `adc_2_analog_in` (ADCChannel): input voltage channel to map to ADC2 mux_p
-            'adc_1_mux_n` (ADCChannel): input voltage channel to map to ADC1 mux_n
-            'adc_2_mux_n` (ADCChannel): input voltage channel to map to ADC1 mux_n
+            `adc_1_mux_n` (ADCChannel): input voltage channel to map to ADC1 mux_n
+            `adc_2_mux_n` (ADCChannel): input voltage channel to map to ADC1 mux_n
             `adc_1_data_rate` (ADC1DataRate): ADC1 data rate in samples per second
             `adc_2_data_rate` (ADC2DataRate): ADC2 data rate in samples per second,
             `filter_mode` (FilterMode): filter mode for both ADC1 and ADC2.

src/test_edgepi/unit_tests/test_adc/test_adc_voltage.py

@@ -10,6 +10,7 @@
     CRCCheckError,
     _code_to_input_voltage,
     generate_crc_8_table,
+    code_to_temperature,
     CRC_8_ATM_GEN
 )
 from edgepi.adc.adc_crc_8_atm import CRC_8_ATM_LUT
@@ -68,3 +69,16 @@ def test_crc_8_atm_adc_1(voltage_bytes, crc_code, err):
 
 def test_generate_crc_8_table():
     assert generate_crc_8_table(CRC_8_ATM_GEN) == CRC_8_ATM_LUT
+
+
+@pytest.mark.parametrize(
+    "code, ref_resistance, temp_offset, rtd_conv_constant",
+    [
+        ([51, 16, 126, 166], 1326.20, 100, 0.385),
+        ([0x8, 0x43, 0x1C, 0x45], 1326.20, 100, 0.385),
+    ]
+)
+def test_code_to_temperature(code, ref_resistance, temp_offset, rtd_conv_constant):
+    # TODO: add check for expected value later if any values are known. No errors raised
+    # is good enough for now.
+    code_to_temperature(code, ref_resistance, temp_offset, rtd_conv_constant)