# Christian Fiot 2020 # VOC sensor, power requirement 48mA max # PID, power < 85 mW at 3.2 V (26mA), < 300 mW transient for 200ms (93mA) # # # Usage: # from SGP30 import SGP30 # # Voc = SGP30(0x58, traceDebug=False) # Voc.begin() # returns True if chip found, this command can take up to 20sec # Voc.get_air_quality() # returns co2 ppm, voc ppb # # # Credit https://github.com/pimoroni/sgp30-python/blob/master/library/sgp30/__init__.py # # 2.01 calculate_dv(temp, rh) temp in degrees C, relative humidity in % -> return dv gr/m3 # 2.01 set_humidity(dv) set dv gr/m3 # 2.01 Voc.read() returns co2 ppm, voc ppb # # 2.02 save / restore voc and co2 baseline # import struct import time import utime # 2.00 import math # 2.01 from trap import * # 2.2.9b from machine import I2C # 2.2.18 class SGP30: def __init__(self, funcIamAlive, iicadr=0x58, i2c=None, sda='P22', scl='P21', traceDebug=False, sdAccess=False): # 2.2.9b self.funcIamAlive = funcIamAlive # 2.2.20 self.trace = traceDebug self.adr = iicadr self.sdAccess = sdAccess # 2.2.9b self.available = False # cf3/8 self.total_voc = 0 # ppb self.equivalent_co2 = 0 # ppm self.dv = 0 # 2.01 compensation humidity gr/m3 self.tvoc_base = 0 # 2.02 self.co2_base = 0 # 2.02 # I2C bus declaration is extenal # MCP are connected on I2C bus #1 SDA-P22 SCL-P21 # I2C bus #2 SDA-P18 SCL-P17 # if i2c is not None: # 2.2.18 self.i2c1 = i2c else: # from machine import I2C self.i2c1 = I2C(0, mode=I2C.MASTER, pins=(sda, scl)) """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) } def begin(self): try: self.get_unique_id() self.start_measurement() print('Detected SGP30 - VOC Sensor Adr:' + str(self.adr), self.sdAccess) self.available = True return True except Exception as e: # 2.2.9b scratch = 'SGP30 - VOC Sensor (Adr:' + str(self.adr) + ') not found - ' # 2.2.9b print('SGP30 - VOC Sensor (Adr:' + str(self.adr) + ') not found.') file_errLog(0, scratch + str(e), self.sdAccess) # 2.2.9b # 2.2.18 print('Begin I2C1 scan ...') # 2.2.18 print(self.i2c1.scan()) self.available = False return False 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: # 2.2.21 # raise ValueError("{} requires {} parameters. {} supplied!".format( # command_name, # param_len, # len(parameters) # )) # scratch = "SGP30 - Invalid command" # 2.2.21 # file_errLog(0, scratch, self.sdAccess) # 2.2.21 raise Exception("SGP30 - Invalid command") # 2.2.21 parameters_out = [cmd] # no crc for adr or cmd # add/compute crc for each paramter for i in range(len(parameters)): parameters_out.append(parameters[i]) # 1word parameters_out.append(self.calculate_crc(parameters[i])) # 1crc if self.trace: # print("parameters_out: " + str(parameters_out)) print("parameters_out: $", end = "") # + str(self.CoeffData) ) for i in parameters_out: print(" %X," % i, end = "") print("") # take 16bit parameters_out makes 8bit data_out paket data_out = struct.pack('>H' + ('HB' * param_len), *parameters_out) # (1W + 1B) * response_len # 2.00 msg_w = self._i2c_msg.write(self._i2c_addr, data_out) # 2.00 self._i2c_dev.i2c_rdwr(msg_w) if self.trace: # print("data_out: " + str(data_out)) print("data_out: $", end = "") for i in data_out: print(" %X," % i, end = "") print("") # send i2c commnand self.i2c1.writeto(self.adr, data_out) # 2.00 utime.sleep_ms(250) # 2.00 # 2.00 time.sleep(0.025) # Suitable for all commands except 'measure_test' # if a response is expected ftech it if response_len > 0: # Each parameter is a word (2 bytes) followed by a CRC (1 byte) # 2.00 msg_r = self._i2c_msg.read(self._i2c_addr, response_len * 3) # 2.00 self._i2c_dev.i2c_rdwr(msg_r) msg_r = self.i2c1.readfrom(self.adr, response_len * 3) # 2.00 if self.trace: print("msg_r: $", end = "") for i in msg_r: print(" %X," % i, end = "") print("") # 2.00 buf = msg_r.buf[0:response_len * 3] # 2.00 response = struct.unpack('>' + ('HB' * response_len), buf) # self.scratch = msg_r # 2.00 for debug response = struct.unpack('>' + ('HB' * response_len), msg_r) # (1W + 1B) * response_len if self.trace: # print("data_out: " + str(data_out)) print("response: $", end = "") for i in response: print(" %X," % i, end = "") print("") verified = [] for i in range(response_len): offset = i * 2 value, crc = response[offset:offset + 2] if crc != self.calculate_crc(value): # 2.2.21 # raise RuntimeError("Invalid CRC in response from SGP30: {:02x} != {:02x}", # crc, # self.calculate_crc(value), # buf) # scratch = "SGP30 - Invalid CRC in response" # 2.2.21 # file_errLog(0, scratch, self.sdAccess) # 2.2.21 raise Exception("SGP30 - Invalid CRC in response") # 2.2.21 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 soft_reset(self): # 2.05 self.i2c1.writeto(0x00, bytearray([0x06])) print("VOC reset - sleep mode") def get_unique_id(self): # CF result = self.command('get_serial_id') # 2.00 return result[0] << 32 | result[1] << 16 | result[0] # return result # 2.00 for debug return result[0] << 32 | result[1] << 16 | result[2] def get_feature_set_version(self): # CF result = self.command('get_feature_set_version') # 2.00 [0] # 2.00 return (result & 0xf000) >> 12, result & 0x00ff return (result[0] & 0xf000) >> 12, result[0] & 0x00ff # should be: product type 0, version n def start_measurement(self): # 2.00 , 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') # 2.00 First readings are disregarded print("Starting VOC sensor, please wait.", end = "") testsamples = 0 eco2 = 400 tvoc = 0 while ((eco2 == 400 and tvoc == 0) and (testsamples < 20)): print(".", end = "") eco2, tvoc = self.command('measure_air_quality') self.funcIamAlive() # 2.2.20 IamAlive() time.sleep(1) # .0) testsamples += 1 print(". %d" % testsamples) print("Equivalent C02: {: 5d} (ppm) Total VOC: {: 5d} (ppb)" .format(eco2, tvoc) ) def get_air_quality(self): # CF """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. """ self.equivalent_co2, self.total_voc = self.command('measure_air_quality') return self.equivalent_co2, self.total_voc def read(self): # 2.01 for standardisation with other library self.equivalent_co2, self.total_voc = self.command('measure_air_quality') return self.equivalent_co2, self.total_voc def get_baseline(self, trace = ""): """Get the current baseline setting. Returns an instance of SGP30Reading with the properties equivalent_co2 and total_voc. """ if self.available == False: # cf3/8 if trace != "": print(trace + "SGP30 not found") return 0, 0 # 2.02 self.equivalent_co2, self.total_voc = self.command('get_baseline') self.co2_base, self.tvoc_base = self.command('get_baseline') # 2.02 if trace != "": print(trace + " - Read baseline VOC: " + str(self.tvoc_base) + " CO2:" + str(self.co2_base)) return self.co2_base, self.tvoc_base # 2.02 self.equivalent_co2, self.total_voc def set_baseline(self, tvoc, eco2, trace = ""): self.command('set_baseline', [tvoc, eco2]) if trace != "": print(trace + " - Write baseline VOC: " + str(tvoc) + " CO2:" + str(eco2) ) # 2.01 Humidity Compensation - datasheet page10 def calculate_dv(self, temp = 25.0, rh = 50.0): # temp degrees C, rh % self.dv = 216.7 * ( ( (rh / 100.0) * 6.112 * math.exp((17.62 * temp) / (243.12 + temp) )) / (273.15 + temp) ) print("Temperature %1.2f C - RH %1.2f pct - %1.3f g/m3" % (temp, rh, self.dv)) return self.dv # 2.01 # 1gr = 256 # dv = x # x = (dv * 256) / 1 def set_humidity(self, dv): # dv in gr/m3 use calculate_dv() to compute rh to dv U1 = int(dv * 256) # unsigned 16bit print("set_humidity: %1.3f g/m3" % (dv)) self.command('set_humidity', [U1])