created class, removed a few unneccesary rows

In order to make edits to class readable i commit this mostly white
space edit. Further more it removes a few application specific or
testing lines.
This commit is contained in:
Simon Albinsson
2018-04-07 22:28:28 +00:00
parent 3faf3c621e
commit 3e10e27c85
2 changed files with 64 additions and 70 deletions

View File

@@ -10,3 +10,5 @@ It currently lacks support for many features features and is more of a crude dem
* reading raw-values * reading raw-values
If you have the AdaFruit board with built in level shifters and voltage regulator it is should work if you just plug in [SDA to pin 3, SCL to pin 5, VCC to pin 17 and GND to pin 20](https://pinout.xyz/pinout/i2c). You should then be able to find the SGP30 an address 0x58 using `i2cdetect -y 1`. If you get an error message you probbably need to enable i2c in the kernel using [`raspi-config` and reboot](https://learn.sparkfun.com/tutorials/raspberry-pi-spi-and-i2c-tutorial) If you have the AdaFruit board with built in level shifters and voltage regulator it is should work if you just plug in [SDA to pin 3, SCL to pin 5, VCC to pin 17 and GND to pin 20](https://pinout.xyz/pinout/i2c). You should then be able to find the SGP30 an address 0x58 using `i2cdetect -y 1`. If you get an error message you probbably need to enable i2c in the kernel using [`raspi-config` and reboot](https://learn.sparkfun.com/tutorials/raspberry-pi-spi-and-i2c-tutorial)
After 12 hours of run time the base-line data is saved once every hour as per the "device driver integration guide". The data is saved as json to ~/sgp_config_data.txt

View File

@@ -2,7 +2,6 @@ import smbus2
from smbus2 import SMBusWrapper, SMBus, i2c_msg from smbus2 import SMBusWrapper, SMBus, i2c_msg
from collections import namedtuple from collections import namedtuple
from functools import partial from functools import partial
#import Adafruit_PureIO.smbus as adabus
from time import sleep, asctime,time from time import sleep, asctime,time
import json import json
from copy import copy from copy import copy
@@ -15,18 +14,19 @@ BASELINE_FILENAME = os.path.expanduser("~/sgp_config_data.txt")
#print bus.write_byte_datadata(DEVICE_ADDR, 0x00, 0x01) #print bus.write_byte_datadata(DEVICE_ADDR, 0x00, 0x01)
#self._i2c_read_words_from_cmd([0x36, 0x82], 0.01, 3) #self._i2c_read_words_from_cmd([0x36, 0x82], 0.01, 3)
Sgp30Cmd = namedtuple("Sgp30Cmd",["commands","replylen","waittime"]) class Sgp30():
GET_SERIAL=Sgp30Cmd([0x36, 0x82],6,10) Sgp30Cmd = namedtuple("Sgp30Cmd",["commands","replylen","waittime"])
GET_FEATURES=Sgp30Cmd([0x20, 0x15],2,2) GET_SERIAL=Sgp30Cmd([0x36, 0x82],6,10)
IAQ_INIT=Sgp30Cmd([0x20, 0x03],0,10) GET_FEATURES=Sgp30Cmd([0x20, 0x15],2,2)
IAQ_MEASURE=Sgp30Cmd([0x20, 0x08],6,12) IAQ_INIT=Sgp30Cmd([0x20, 0x03],0,10)
IAQ_SELFTEST=Sgp30Cmd([0x20, 0x32],3,520) IAQ_MEASURE=Sgp30Cmd([0x20, 0x08],6,12)
GET_BASELINE=Sgp30Cmd([0x20, 0x15],6,120) IAQ_SELFTEST=Sgp30Cmd([0x20, 0x32],3,520)
SET_BASELINE=Sgp30Cmd([0x20, 0x1e],0,10) GET_BASELINE=Sgp30Cmd([0x20, 0x15],6,120)
SET_BASELINE=Sgp30Cmd([0x20, 0x1e],0,10)
Sgp30Answer = namedtuple("Sgp30Answer",["data","raw"]) Sgp30Answer = namedtuple("Sgp30Answer",["data","raw"])
def read_write(cmd,bus,addr=DEVICE_ADDR): def read_write(cmd,bus,addr=DEVICE_ADDR):
write = i2c_msg.write(addr,cmd.commands) write = i2c_msg.write(addr,cmd.commands)
if cmd.replylen <= 0 : if cmd.replylen <= 0 :
bus.i2c_rdwr(write) bus.i2c_rdwr(write)
@@ -39,7 +39,7 @@ def read_write(cmd,bus,addr=DEVICE_ADDR):
answer = [i<<8 | j for i,j in zip(r[0::3],r[1::3])] answer = [i<<8 | j for i,j in zip(r[0::3],r[1::3])]
return Sgp30Answer(answer,r) return Sgp30Answer(answer,r)
def try_set_baseline(): def try_set_baseline():
try: try:
with open(BASELINE_FILENAME,"w") as conf: with open(BASELINE_FILENAME,"w") as conf:
baseline=json.load(conf) baseline=json.load(conf)
@@ -54,7 +54,7 @@ def try_set_baseline():
print("Loading baseline data into sensor") print("Loading baseline data into sensor")
rw(baseline_cmd) rw(baseline_cmd)
def init_sgp(bus): def init_sgp(bus):
print("Initializing SGP30") print("Initializing SGP30")
rw(IAQ_INIT) rw(IAQ_INIT)
print("Waiting for sensor warmup") print("Waiting for sensor warmup")
@@ -62,8 +62,7 @@ def init_sgp(bus):
#print(rw(SET_BASELINE)) #print(rw(SET_BASELINE))
sleep(15) sleep(15)
def i2c_geral_call(bus): def i2c_geral_call(bus):
print("resetting bus")
bus.write_byte(0,0x06) bus.write_byte(0,0x06)
sleep(.1) sleep(.1)
@@ -73,13 +72,10 @@ def store_baseline(n):
baseline= rw(GET_BASELINE) baseline= rw(GET_BASELINE)
json.dump(baseline.raw,conf) json.dump(baseline.raw,conf)
def http_post_data(url="http://localhost/setdata"): if __name__ == "__main__":
try: main()
requests.get(url,params={"co2": co2eq,"tvoc":tvoc},timeout=.5) main():
except: with SMBusWrapper(1) as bus:
pass
with SMBusWrapper(1) as bus:
rw=partial(read_write,bus=bus) rw=partial(read_write,bus=bus)
i2c_geral_call(bus) i2c_geral_call(bus)
print("Features: %s"%repr(rw(GET_FEATURES))) print("Features: %s"%repr(rw(GET_FEATURES)))
@@ -90,10 +86,7 @@ with SMBusWrapper(1) as bus:
print(rw(IAQ_MEASURE)) print(rw(IAQ_MEASURE))
print("Testing meassure again") print("Testing meassure again")
sleep(1) sleep(1)
print(rw(IAQ_MEASURE))
sleep(1)
print("Running") print("Running")
with open("/tmp/air.txt","w") as f:
n=0 n=0
while(True): while(True):
start = time() start = time()
@@ -102,9 +95,8 @@ with SMBusWrapper(1) as bus:
res = ( "%s CO_2eq: %d ppm, TVOC: %d"%( asctime(), co2eq, tvoc )) res = ( "%s CO_2eq: %d ppm, TVOC: %d"%( asctime(), co2eq, tvoc ))
print(res) print(res)
f.write("%i %i %i \n"%(time(),co2eq,tvoc)) f.write("%i %i %i \n"%(time(),co2eq,tvoc))
http_post_data()
store_baseline(n) store_baseline(n)
elapsed = (time() - start) elapsed = (time() - start)
sleep(1 - elapsed ) sleep(1 - elapsed )
bus.close() bus.close()