<reponame>Daryle1K/sgp30-python<gh_stars>10-100
import struct
import time


__version__ = '0.0.2'

SGP30_I2C_ADDR = 0x58


class SGP30Reading:
    __slots__ = 'equivalent_co2', 'total_voc'

    def __init__(self, eco2, tvoc):
        self.equivalent_co2 = eco2
        self.total_voc = tvoc

    def __str__(self):
        return """Air Quality:
Equivalent C02: {: 5d} (ppm)
Total VOC:      {: 5d} (ppb)
""".format(self.equivalent_co2, self.total_voc)


class SGP30:
    def __init__(self, i2c_dev=None, i2c_msg=None, i2c_addr=SGP30_I2C_ADDR):
        """Mapping table of SGP30 commands.

        Friendly-name, followed by 16-bit command,
        then the number of parameter and response words.

        Each word is two bytes followed by a third CRC
        checksum byte. So a response length of 2 would
        result in the transmission of 6 bytes total.
        """
        self.commands = {
            'init_air_quality': (0x2003, 0, 0),
            'measure_air_quality': (0x2008, 0, 2),
            'get_baseline': (0x2015, 0, 2),
            'set_baseline': (0x201e, 2, 0),
            'set_humidity': (0x2061, 1, 0),
            # 'measure_test': (0x2032, 0, 1),  # Production verification only
            'get_feature_set_version': (0x202f, 0, 1),
            'measure_raw_signals': (0x2050, 0, 2),
            'get_serial_id': (0x3682, 0, 3)
        }

        self._i2c_addr = i2c_addr
        self._i2c_dev = i2c_dev
        self._i2c_msg = i2c_msg
        if self._i2c_dev is None:
            from smbus2 import SMBus, i2c_msg
            self._i2c_msg = i2c_msg
            self._i2c_dev = SMBus(1)

    def command(self, command_name, parameters=None):
        if parameters is None:
            parameters = []
        parameters = list(parameters)
        cmd, param_len, response_len = self.commands[command_name]
        if len(parameters) != param_len:
            raise ValueError("{} requires {} parameters. {} supplied!".format(
                command_name,
                param_len,
                len(parameters)
            ))

        parameters_out = [cmd]

        for i in range(len(parameters)):
            parameters_out.append(parameters[i])
            parameters_out.append(self.calculate_crc(parameters[i]))

        data_out = struct.pack('>H' + ('HB' * param_len), *parameters_out)

        msg_w = self._i2c_msg.write(self._i2c_addr, data_out)
        self._i2c_dev.i2c_rdwr(msg_w)
        time.sleep(0.025)  # Suitable for all commands except 'measure_test'

        if response_len > 0:
            # Each parameter is a word (2 bytes) followed by a CRC (1 byte)
            msg_r = self._i2c_msg.read(self._i2c_addr, response_len * 3)
            self._i2c_dev.i2c_rdwr(msg_r)

            buf = msg_r.buf[0:response_len * 3]

            response = struct.unpack(
                '>' + ('HB' * response_len),
                buf)

            verified = []
            for i in range(response_len):
                offset = i * 2
                value, crc = response[offset:offset + 2]
                if crc != self.calculate_crc(value):
                    raise RuntimeError("Invalid CRC in response from SGP30: {:02x} != {:02x}",
                                       crc,
                                       self.calculate_crc(value),
                                       buf)
                verified.append(value)
            return verified

    def calculate_crc(self, data):
        """Calculate an 8-bit CRC from a 16-bit word

        Defined in section 6.6 of the SGP30 datasheet.

        Polynominal: 0x31 (x8 + x5 + x4 + x1)
        Initialization: 0xFF
        Reflect input/output: False
        Final XOR: 0x00

        """
        crc = 0xff  # Initialization value
        # calculates 8-Bit checksum with given polynomial
        for byte in [(data & 0xff00) >> 8, data & 0x00ff]:
            crc ^= byte
            for _ in range(8):
                if crc & 0x80:
                    crc = (crc << 1) ^ 0x31  # XOR with polynominal
                else:
                    crc <<= 1
        return crc & 0xff

    def get_unique_id(self):
        result = self.command('get_serial_id')
        return result[0] << 32 | result[1] << 16 | result[0]

    def get_feature_set_version(self):
        result = self.command('get_feature_set_version')[0]
        return (result & 0xf000) >> 12, result & 0x00ff

    def start_measurement(self, run_while_waiting=None):
        """Start air quality measurement on the SGP30.

        The first 15 readings are discarded so this command will block for 15s.

        :param run_while_waiting: Function to call for every discarded reading.

        """
        self.command('init_air_quality')
        testsamples = 0
        while True:
            # Discard the initialisation readings as per page 8/15 of the datasheet
            eco2, tvoc = self.command('measure_air_quality')
            # The first 15 readings should return as 400, 0 so abort when they change
            # Break after 20 test samples to avoid a potential infinite loop
            if eco2 != 400 or tvoc != 0 or testsamples >= 20:
                break
            if callable(run_while_waiting):
                run_while_waiting()
            time.sleep(1.0)
            testsamples += 1

    def get_air_quality(self):
        """Get an air quality measurement.

        Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc.

        This should be called at 1s intervals to ensure the dynamic baseline compensation on the SGP30 operates correctly.

        """
        eco2, tvoc = self.command('measure_air_quality')
        return SGP30Reading(eco2, tvoc)

    def get_baseline(self):
        """Get the current baseline setting.

        Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc.

        """
        eco2, tvoc = self.command('get_baseline')
        return SGP30Reading(eco2, tvoc)

    def set_baseline(self, eco2, tvoc):
        self.command('set_baseline', [tvoc, eco2])

    def __del__(self):
        self._i2c_dev.close()

import struct
import time


__version__ = '0.0.2'

SGP30_I2C_ADDR = 0x58


class SGP30Reading:
    __slots__ = 'equivalent_co2', 'total_voc'

    def __init__(self, eco2, tvoc):
        self.equivalent_co2 = eco2
        self.total_voc = tvoc

    def __str__(self):
        return """Air Quality:
Equivalent C02: {: 5d} (ppm)
Total VOC:      {: 5d} (ppb)
""".format(self.equivalent_co2, self.total_voc)


class SGP30:
    def __init__(self, i2c_dev=None, i2c_msg=None, i2c_addr=SGP30_I2C_ADDR):
        """Mapping table of SGP30 commands.

        Friendly-name, followed by 16-bit command,
        then the number of parameter and response words.

        Each word is two bytes followed by a third CRC
        checksum byte. So a response length of 2 would
        result in the transmission of 6 bytes total.
        """
        self.commands = {
            'init_air_quality': (0x2003, 0, 0),
            'measure_air_quality': (0x2008, 0, 2),
            'get_baseline': (0x2015, 0, 2),
            'set_baseline': (0x201e, 2, 0),
            'set_humidity': (0x2061, 1, 0),
            # 'measure_test': (0x2032, 0, 1),  # Production verification only
            'get_feature_set_version': (0x202f, 0, 1),
            'measure_raw_signals': (0x2050, 0, 2),
            'get_serial_id': (0x3682, 0, 3)
        }

        self._i2c_addr = i2c_addr
        self._i2c_dev = i2c_dev
        self._i2c_msg = i2c_msg
        if self._i2c_dev is None:
            from smbus2 import SMBus, i2c_msg
            self._i2c_msg = i2c_msg
            self._i2c_dev = SMBus(1)

    def command(self, command_name, parameters=None):
        if parameters is None:
            parameters = []
        parameters = list(parameters)
        cmd, param_len, response_len = self.commands[command_name]
        if len(parameters) != param_len:
            raise ValueError("{} requires {} parameters. {} supplied!".format(
                command_name,
                param_len,
                len(parameters)
            ))

        parameters_out = [cmd]

        for i in range(len(parameters)):
            parameters_out.append(parameters[i])
            parameters_out.append(self.calculate_crc(parameters[i]))

        data_out = struct.pack('>H' + ('HB' * param_len), *parameters_out)

        msg_w = self._i2c_msg.write(self._i2c_addr, data_out)
        self._i2c_dev.i2c_rdwr(msg_w)
        time.sleep(0.025)  # Suitable for all commands except 'measure_test'

        if response_len > 0:
            # Each parameter is a word (2 bytes) followed by a CRC (1 byte)
            msg_r = self._i2c_msg.read(self._i2c_addr, response_len * 3)
            self._i2c_dev.i2c_rdwr(msg_r)

            buf = msg_r.buf[0:response_len * 3]

            response = struct.unpack(
                '>' + ('HB' * response_len),
                buf)

            verified = []
            for i in range(response_len):
                offset = i * 2
                value, crc = response[offset:offset + 2]
                if crc != self.calculate_crc(value):
                    raise RuntimeError("Invalid CRC in response from SGP30: {:02x} != {:02x}",
                                       crc,
                                       self.calculate_crc(value),
                                       buf)
                verified.append(value)
            return verified

    def calculate_crc(self, data):
        """Calculate an 8-bit CRC from a 16-bit word

        Defined in section 6.6 of the SGP30 datasheet.

        Polynominal: 0x31 (x8 + x5 + x4 + x1)
        Initialization: 0xFF
        Reflect input/output: False
        Final XOR: 0x00

        """
        crc = 0xff  # Initialization value
        # calculates 8-Bit checksum with given polynomial
        for byte in [(data & 0xff00) >> 8, data & 0x00ff]:
            crc ^= byte
            for _ in range(8):
                if crc & 0x80:
                    crc = (crc << 1) ^ 0x31  # XOR with polynominal
                else:
                    crc <<= 1
        return crc & 0xff

    def get_unique_id(self):
        result = self.command('get_serial_id')
        return result[0] << 32 | result[1] << 16 | result[0]

    def get_feature_set_version(self):
        result = self.command('get_feature_set_version')[0]
        return (result & 0xf000) >> 12, result & 0x00ff

    def start_measurement(self, run_while_waiting=None):
        """Start air quality measurement on the SGP30.

        The first 15 readings are discarded so this command will block for 15s.

        :param run_while_waiting: Function to call for every discarded reading.

        """
        self.command('init_air_quality')
        testsamples = 0
        while True:
            # Discard the initialisation readings as per page 8/15 of the datasheet
            eco2, tvoc = self.command('measure_air_quality')
            # The first 15 readings should return as 400, 0 so abort when they change
            # Break after 20 test samples to avoid a potential infinite loop
            if eco2 != 400 or tvoc != 0 or testsamples >= 20:
                break
            if callable(run_while_waiting):
                run_while_waiting()
            time.sleep(1.0)
            testsamples += 1

    def get_air_quality(self):
        """Get an air quality measurement.

        Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc.

        This should be called at 1s intervals to ensure the dynamic baseline compensation on the SGP30 operates correctly.

        """
        eco2, tvoc = self.command('measure_air_quality')
        return SGP30Reading(eco2, tvoc)

    def get_baseline(self):
        """Get the current baseline setting.

        Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc.

        """
        eco2, tvoc = self.command('get_baseline')
        return SGP30Reading(eco2, tvoc)

    def set_baseline(self, eco2, tvoc):
        self.command('set_baseline', [tvoc, eco2])

    def __del__(self):
        self._i2c_dev.close()
Air Quality: Equivalent C02: {: 5d} (ppm) Total VOC:      {: 5d} (ppb) Mapping table of SGP30 commands.          Friendly-name, followed by 16-bit command,         then the number of parameter and response words.          Each word is two bytes followed by a third CRC         checksum byte. So a response length of 2 would         result in the transmission of 6 bytes total. # 'measure_test': (0x2032, 0, 1),  # Production verification only # Suitable for all commands except 'measure_test' # Each parameter is a word (2 bytes) followed by a CRC (1 byte) Calculate an 8-bit CRC from a 16-bit word          Defined in section 6.6 of the SGP30 datasheet.          Polynominal: 0x31 (x8 + x5 + x4 + x1)         Initialization: 0xFF         Reflect input/output: False         Final XOR: 0x00 # Initialization value # calculates 8-Bit checksum with given polynomial # XOR with polynominal Start air quality measurement on the SGP30.          The first 15 readings are discarded so this command will block for 15s.          :param run_while_waiting: Function to call for every discarded reading. # Discard the initialisation readings as per page 8/15 of the datasheet # The first 15 readings should return as 400, 0 so abort when they change # Break after 20 test samples to avoid a potential infinite loop Get an air quality measurement.          Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc.          This should be called at 1s intervals to ensure the dynamic baseline compensation on the SGP30 operates correctly. Get the current baseline setting.          Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc.