smallsolar/PySGP30
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleaned up sgp30.py

Browse Source
This commit is contained in:
Connor
2019-09-16 09:46:24 +10:00
parent 3f004190cd
commit 724829e585
1 changed files with 16 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
sgp30.py
View File

@@ -1,14 +1,8 @@
import smbus2 import smbus2
from smbus2 import SMBusWrapper, SMBus, i2c_msg
from collections import namedtuple from collections import namedtuple
from functools import partial import time
from time import sleep, asctime, time
from copy import copy
from .crc import CRC8 from .crc import CRC8
DEVICE_BUS = 1
class _cmds(): class _cmds():
"""container class for mapping between human readable names and the command values used by the sgp""" """container class for mapping between human readable names and the command values used by the sgp"""
SGP30Cmd = namedtuple("SGP30Cmd", ["commands", "replylen", "waittime"]) SGP30Cmd = namedtuple("SGP30Cmd", ["commands", "replylen", "waittime"])
@@ -29,12 +23,10 @@ class _cmds():
class SGP30(): class SGP30():
def __init__(self, bus, device_address=0x58, baseline=[]): def __init__(self, bus, address=0x58, baseline=[]):
self._bus = bus self.bus = bus
self._device_addr = device_address self.address = address
self._start_time = time() self.baseline = baseline
self._last_save_time = time()
self._baseline = baseline
SGP30Packet = namedtuple("SGP30Packet", ["data", "raw", "crc_ok"]) SGP30Packet = namedtuple("SGP30Packet", ["data", "raw", "crc_ok"])
@@ -44,14 +36,14 @@ class SGP30():
return(crc, a) return(crc, a)
def read_write(self, cmd): def read_write(self, cmd):
write = i2c_msg.write(self._device_addr, cmd.commands) write = smbus2.i2c_msg.write(self.address, cmd.commands)
if cmd.replylen <= 0: if cmd.replylen <= 0:
self._bus.i2c_rdwr(write) self.bus.i2c_rdwr(write)
else: else:
read = i2c_msg.read(self._device_addr, cmd.replylen) read = smbus2.i2c_msg.read(self.address, cmd.replylen)
self._bus.i2c_rdwr(write) self.bus.i2c_rdwr(write)
sleep(cmd.waittime/1000.0) time.sleep(cmd.waittime/1000.0)
self._bus.i2c_rdwr(read) self.bus.i2c_rdwr(read)
r = list(read) r = list(read)
crc_ok, a = self._raw_validate_crc(r) crc_ok, a = self._raw_validate_crc(r)
answer = [i << 8 | j for i, j in a] answer = [i << 8 | j for i, j in a]
@@ -65,9 +57,9 @@ class SGP30():
print("Ignoring baseline due to invalid CRC") print("Ignoring baseline due to invalid CRC")
def set_baseline(self): def set_baseline(self):
crc, _ = self._raw_validate_crc(self._baseline) crc, _ = self._raw_validate_crc(self.baseline)
if len(self._baseline) == 6 and crc == True: if len(self.baseline) == 6 and crc == True:
self.read_write(_cmds.new_set_baseline(self._baseline)) self.read_write(_cmds.new_set_baseline(self.baseline))
return True return True
else: else:
#print("Failed to load baseline, invalid data") #print("Failed to load baseline, invalid data")
@@ -98,24 +90,5 @@ class SGP30():
This will usually un-stick the SGP30, but might reset or otherwise This will usually un-stick the SGP30, but might reset or otherwise
affect any device on the bus. affect any device on the bus.
""" """
self._bus.write_byte(0, 0x06) self.bus.write_byte(0, 0x06)
sleep(0.1) time.sleep(0.1)
def main():
with SMBusWrapper(1) as bus:
sgp = SGP30(bus, baseline_filename=BASELINE_FILENAME+".TESTING")
print("resetting all i2c devices")
sgp.i2c_general_call()
print(sgp.read_features())
print(sgp.read_serial())
sgp.init_sgp()
for i in range(300):
print(sgp.read_measurements())
time.sleep(0.1)
sgp.store_baseline()
bus.close()
if __name__ == "__main__":
main()