adafruit_sgp30.py

# SPDX-FileCopyrightText: 2017 ladyada for Adafruit Industries
#
# SPDX-License-Identifier: MIT

"""
`adafruit_sgp30`
====================================================

I2C driver for SGP30 Sensirion VoC sensor

* Author(s): ladyada

Implementation Notes
--------------------

**Hardware:**

* Adafruit `SGP30 Air Quality Sensor Breakout - VOC and eCO2
  <https://www.adafruit.com/product/3709>`_ (Product ID: 3709)

**Software and Dependencies:**

* Adafruit CircuitPython firmware for the supported boards:
  https://github.com/adafruit/circuitpython/releases
* Adafruit's Bus Device library: https://github.com/adafruit/Adafruit_CircuitPython_BusDevice
"""

import time
from math import exp

from adafruit_bus_device.i2c_device import I2CDevice
from micropython import const

try:
    from typing import List, Tuple

    from busio import I2C
except ImportError:
    pass

__version__ = "0.0.0+auto.0"
__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_SGP30.git"


_SGP30_DEFAULT_I2C_ADDR = const(0x58)
_SGP30_FEATURESETS = (0x0020, 0x0022)

_SGP30_CRC8_POLYNOMIAL = const(0x31)
_SGP30_CRC8_INIT = const(0xFF)
_SGP30_WORD_LEN = const(2)


class Adafruit_SGP30:
    """
    A driver for the SGP30 gas sensor.

    :param ~busio.I2C i2c: The I2C bus the SGP30 is connected to.
    :param int address: The I2C address of the device. Defaults to :const:`0x58`


    **Quickstart: Importing and using the SGP30 temperature sensor**

        Here is one way of importing the `Adafruit_SGP30` class so you
        can use it with the name ``sgp30``.
        First you will need to import the libraries to use the sensor

        .. code-block:: python

            import busio
            import board
            import adafruit_sgp30

        Once this is done you can define your `busio.I2C` object and define your sensor object

        .. code-block:: python

            i2c = busio.I2C(board.SCL, board.SDA, frequency=100000)
            sgp30 = adafruit_sgp30.Adafruit_SGP30(i2c)

        Now you have access to the Carbon Dioxide Equivalent baseline using the
        :attr:`baseline_eCO2` attribute and the Total Volatile Organic Compound
        baseline using the :attr:`baseline_TVOC`

        .. code-block:: python

            eCO2 = sgp30.baseline_eCO2
            TVOC = sgp30.baseline_TVOC


    """

    def __init__(self, i2c: I2C, address: int = _SGP30_DEFAULT_I2C_ADDR) -> None:
        """Initialize the sensor, get the serial # and verify that we found a proper SGP30"""
        self._device = I2CDevice(i2c, address)

        # get unique serial, its 48 bits so we store in an array
        self.serial = self._i2c_read_words_from_cmd([0x36, 0x82], 0.01, 3)
        # get featureset
        featureset = self._i2c_read_words_from_cmd([0x20, 0x2F], 0.01, 1)
        if featureset[0] not in _SGP30_FEATURESETS:
            raise RuntimeError("SGP30 Not detected")
        self.iaq_init()

    @property
    def TVOC(self) -> int:
        """Total Volatile Organic Compound in parts per billion."""
        return self.iaq_measure()[1]

    @property
    def baseline_TVOC(self) -> int:
        """Total Volatile Organic Compound baseline value"""
        return self.get_iaq_baseline()[1]

    @property
    def eCO2(self) -> int:
        """Carbon Dioxide Equivalent in parts per million"""
        return self.iaq_measure()[0]

    @property
    def baseline_eCO2(self) -> int:
        """Carbon Dioxide Equivalent baseline value"""
        return self.get_iaq_baseline()[0]

    @property
    def Ethanol(self) -> int:
        """Ethanol Raw Signal in ticks"""
        return self.raw_measure()[1]

    @property
    def H2(self) -> int:
        """H2 Raw Signal in ticks"""
        return self.raw_measure()[0]

    def iaq_init(self) -> List[int]:
        """Initialize the IAQ algorithm"""
        # name, command, signals, delay
        self._run_profile(("iaq_init", [0x20, 0x03], 0, 0.01))

    def iaq_measure(self) -> List[int]:
        """Measure the eCO2 and TVOC"""
        # name, command, signals, delay
        return self._run_profile(("iaq_measure", [0x20, 0x08], 2, 0.05))

    def raw_measure(self) -> List[int]:
        """Measure H2 and Ethanol (Raw Signals)"""
        # name, command, signals, delay
        return self._run_profile(("raw_measure", [0x20, 0x50], 2, 0.025))

    def get_iaq_baseline(self) -> List[int]:
        """Retreive the IAQ algorithm baseline for eCO2 and TVOC"""
        # name, command, signals, delay
        return self._run_profile(("iaq_get_baseline", [0x20, 0x15], 2, 0.01))

    def set_iaq_baseline(self, eCO2: int, TVOC: int) -> None:
        """Set the previously recorded IAQ algorithm baseline for eCO2 and TVOC"""
        if eCO2 == 0 and TVOC == 0:
            raise RuntimeError("Invalid baseline")
        buffer = []
        for value in [TVOC, eCO2]:
            arr = [value >> 8, value & 0xFF]
            arr.append(self._generate_crc(arr))
            buffer += arr
        self._run_profile(("iaq_set_baseline", [0x20, 0x1E] + buffer, 0, 0.01))

    def set_iaq_humidity(self, gramsPM3: float) -> None:
        """Set the humidity in g/m3 for eCO2 and TVOC compensation algorithm"""
        tmp = int(gramsPM3 * 256)
        buffer = []
        for value in [tmp]:
            arr = [value >> 8, value & 0xFF]
            arr.append(self._generate_crc(arr))
            buffer += arr
        self._run_profile(("iaq_set_humidity", [0x20, 0x61] + buffer, 0, 0.01))

    def set_iaq_relative_humidity(self, celsius: float, relative_humidity: float):
        """
        Set the humidity in g/m3 for eCo2 and TVOC compensation algorithm.
        The absolute humidity is calculated from the temperature (Celsius)
        and relative humidity (as a percentage).
        """
        numerator = ((relative_humidity / 100) * 6.112) * exp(
            (17.62 * celsius) / (243.12 + celsius)
        )
        denominator = 273.15 + celsius

        humidity_grams_pm3 = 216.7 * (numerator / denominator)
        self.set_iaq_humidity(humidity_grams_pm3)

    # Low level command functions

    def _run_profile(self, profile: Tuple[str, List[int], int, float]) -> List[int]:
        """Run an SGP 'profile' which is a named command set"""
        name, command, signals, delay = profile  # noqa: F841

        # print("\trunning profile: %s, command %s, %d, delay %0.02f" %
        #   (name, ["0x%02x" % i for i in command], signals, delay))
        return self._i2c_read_words_from_cmd(command, delay, signals)

    def _i2c_read_words_from_cmd(
        self, command: List[int], delay: float, reply_size: int
    ) -> List[int]:
        """Run an SGP command query, get a reply and CRC results if necessary"""
        with self._device:
            self._device.write(bytes(command))
            time.sleep(delay)
            if not reply_size:
                return None
            crc_result = bytearray(reply_size * (_SGP30_WORD_LEN + 1))
            self._device.readinto(crc_result)
            # print("\tRaw Read: ", crc_result)
            result = []
            for i in range(reply_size):
                word = [crc_result[3 * i], crc_result[3 * i + 1]]
                crc = crc_result[3 * i + 2]
                if self._generate_crc(word) != crc:
                    raise RuntimeError("CRC Error")
                result.append(word[0] << 8 | word[1])
            # print("\tOK Data: ", [hex(i) for i in result])
            return result

    def _generate_crc(self, data: bytearray) -> int:  # noqa: PLR6301
        """8-bit CRC algorithm for checking data"""
        crc = _SGP30_CRC8_INIT
        # calculates 8-Bit checksum with given polynomial
        for byte in data:
            crc ^= byte
            for _ in range(8):
                if crc & 0x80:
                    crc = (crc << 1) ^ _SGP30_CRC8_POLYNOMIAL
                else:
                    crc <<= 1
        return crc & 0xFF