Create bme680.py #253

Closed
cristianku wants to merge 7 commits from master into master
4 changed files with 191 additions and 10 deletions

View File

@@ -0,0 +1,130 @@
import bme680
import time
try:
sensor = bme680.BME680(bme680.I2C_ADDR_PRIMARY)
except IOError:
sensor = bme680.BME680(bme680.I2C_ADDR_SECONDARY)
# These calibration data can safely be commented
# out, if desired.
# These oversampling settings can be tweaked to
# change the balance between accuracy and noise in
# the data.
hum_weighting = float(0.25) # so hum effect is 25% of the total air quality score
MarcelloTheArcane commented 2021-08-26 10:05:09 +01:00 (Migrated from github.com)
Review

Why do you use float(0.25) as opposed to just 0.25? Python will recognise a decimal as a float I think.

You can use 40.0 for hum_reference too, to make sure it is cast as a float.

Why do you use `float(0.25)` as opposed to just `0.25`? Python will recognise a decimal as a float I think. You can use `40.0` for `hum_reference` too, to make sure it is cast as a float.
gas_weighting = float(0.75) # so gas effect is 75% of the total air quality score
sensor.set_humidity_oversample(bme680.OS_2X)
sensor.set_pressure_oversample(bme680.OS_2X)
sensor.set_temperature_oversample(bme680.OS_2X)
sensor.set_filter(bme680.FILTER_SIZE_3)
sensor.set_gas_heater_temperature(320)
sensor.set_gas_heater_duration(150)
sensor.select_gas_heater_profile(0)
sensor.set_gas_status(bme680.ENABLE_GAS_MEAS)
gas_reference = float(250000)
hum_reference = float(40)
getgasreference_count = int(0)
def GetGasReference(gas_reference):
# Now run the sensor for a burn-in period, then use combination of relative humidity and gas resistance to estimate indoor air quality as a percentage.
# print("Getting a new gas reference value")
readings = int(10)
while True:
sensor.get_sensor_data()
if sensor.data.heat_stable:
for i in range(1, readings): # // read gas for 10 x 0.150mS = 1.5secs
sensor.get_sensor_data()
gas_reference = gas_reference + sensor.data.gas_resistance
# print(sensor.data.gas_resistance)
gas_reference = gas_reference / readings
return
# for i in range(1, readings): # // read gas for 10 x 0.150mS = 1.5secs
# gas_reference = gas_reference + sensor.data.gas_resistance
# print(sensor.data.gas_resistance)
# gas_reference = gas_reference / readings
def CalculateIAQ(score):
IAQ_text = "Air quality is "
score = float((100 - score) * 5)
if score >= 301:
IAQ_text = IAQ_text + "Hazardous"
elif score >= 201 and score <= 300:
IAQ_text = IAQ_text + "Very Unhealthy"
elif score >= 176 and score <= 200:
IAQ_text = IAQ_text + "Unhealthy"
elif score >= 151 and score <= 175:
IAQ_text = IAQ_text + "Unhealthy for Sensitive Groups"
elif score >= 51 and score <= 150:
IAQ_text = IAQ_text + "Moderate"
elif score >= 00 and score <= 50:
IAQ_text = IAQ_text + "Good"
return IAQ_text
#Calculate humidity contribution to IAQ index
current_humidity = float(sensor.data.humidity)
if (current_humidity >= 38 and current_humidity <= 42):
hum_score = float(0.25*100) # Humidity +/-5% around optimum
else:
#sub-optimal
if (current_humidity < 38):
hum_score = float(0.25/hum_reference*current_humidity*100)
else:
hum_score = ((-0.25/(100-hum_reference)*current_humidity)+0.416666)*100
#Calculate gas contribution to IAQ index
gas_lower_limit = float(5000) # Bad air quality limit
gas_upper_limit = float(50000) # Good air quality limit
if (gas_reference > gas_upper_limit):
gas_reference = gas_upper_limit #gas_reference = 250000 see above
if (gas_reference < gas_lower_limit):
gas_reference = gas_lower_limit
gas_score = float((0.75/(gas_upper_limit-gas_lower_limit)*gas_reference -(gas_lower_limit*(0.75/(gas_upper_limit-gas_lower_limit))))*100)
# print('gas_upper_limit--------')
# print(gas_upper_limit)
#
# print('gas_lower_limit--------')
# print(gas_lower_limit)
#
#
# print('gas score---------')
# print(gas_score)
#Combine results for the final IAQ index value (0-100% where 100% is good quality air)
air_quality_score = float(hum_score + gas_score)
# print('Air Quality = {0:.2f} % derived from 25% of Humidity reading and 75% of Gas reading - 100% is good quality air'.format( air_quality_score))
# print('Humidity element was : {0:.2f} of 0.25'.format( hum_score/100))
# print(' Gas element was : {0:.2f} of 0.25'.format( gas_score/100))
# if (sensor.data.gas_resistance < 120000):
# print('Poor air qulity *****')
GetGasReference(gas_reference)
# print(CalculateIAQ(air_quality_score))
# print("------------------------------------------------")
# time.sleep(2)
print('{0:0.1f}'.format(air_quality_score))

View File

@@ -342,11 +342,11 @@ class EnclosurePlugin(octoprint.plugin.StartupPlugin, octoprint.plugin.TemplateP
try:
sensor_data = []
for sensor in list(filter(lambda item: item['input_type'] == 'temperature_sensor', self.rpi_inputs)):
temp, hum = self.get_sensor_data(sensor)
temp, hum, airquality = self.get_sensor_data(sensor)
if self._settings.get(["debug_temperature_log"]) is True:
self._logger.debug("Sensor %s Temperature: %s humidity %s", sensor['label'], temp, hum)
if temp is not None and hum is not None:
sensor_data.append(dict(index_id=sensor['index_id'], temperature=temp, humidity=hum))
self._logger.debug("Sensor %s Temperature: %s humidity %s Airquality %s", sensor['label'], temp, hum, airquality)
if temp is not None and hum is not None and airquality is not None:
sensor_data.append(dict(index_id=sensor['index_id'], temperature=temp, humidity=hum, airquality=airquality))
self.temperature_sensor_data = sensor_data
self.handle_temp_hum_control()
self.handle_temperature_events()
@@ -495,33 +495,43 @@ class EnclosurePlugin(octoprint.plugin.StartupPlugin, octoprint.plugin.TemplateP
def get_sensor_data(self, sensor):
try:
if self.development_mode:
temp, hum = self.read_dummy_temp()
temp, hum, airquality = self.read_dummy_temp()
else:
if sensor['temp_sensor_type'] in ["11", "22", "2302"]:
temp, hum = self.read_dht_temp(sensor['temp_sensor_type'], sensor['gpio_pin'])
elif sensor['temp_sensor_type'] == "18b20":
temp = self.read_18b20_temp(sensor['ds18b20_serial'])
hum = 0
airquality = 0
elif sensor['temp_sensor_type'] == "bme280":
temp, hum = self.read_bme280_temp(sensor['temp_sensor_address'])
airquality = 0
elif sensor['temp_sensor_type'] == "bme680":
temp, hum, airquality = self.read_bme680_temp(sensor['temp_sensor_address'])
elif sensor['temp_sensor_type'] == "si7021":
temp, hum = self.read_si7021_temp(sensor['temp_sensor_address'])
airquality = 0
elif sensor['temp_sensor_type'] == "tmp102":
temp = self.read_tmp102_temp(sensor['temp_sensor_address'])
hum = 0
airquality = 0
elif sensor['temp_sensor_type'] == "max31855":
temp = self.read_max31855_temp(sensor['temp_sensor_address'])
hum = 0
airquality = 0
else:
self._logger.info("temp_sensor_type no match")
temp = None
hum = None
if temp != -1 and hum != -1:
airquality = None
if temp != -1 and hum != -1 and airquality != -1:
temp = round(self.to_float(temp), 1) if not sensor['use_fahrenheit'] else round(
self.to_float(temp) * 1.8 + 32, 1)
hum = round(self.to_float(hum), 1)
return temp, hum
return None, None
airquality = round(self.to_float(airquality), 1)
return temp, hum, airquality
return None, None, None
except Exception as ex:
self.log_error(ex)
@@ -554,7 +564,7 @@ class EnclosurePlugin(octoprint.plugin.StartupPlugin, octoprint.plugin.TemplateP
self.dummy_value = return_value
return return_value, return_value
return return_value, return_value, return_value
def read_dht_temp(self, sensor, pin):
try:
@@ -598,6 +608,28 @@ class EnclosurePlugin(octoprint.plugin.StartupPlugin, octoprint.plugin.TemplateP
self.log_error(ex)
return (0, 0)
def read_bme680_temp(self, address):
try:
script = os.path.dirname(os.path.realpath(__file__)) + "/BME680.py "
if self._settings.get(["use_sudo"]):
sudo_str = "sudo "
else:
sudo_str = ""
cmd = sudo_str + "python " + script + str(address)
if self._settings.get(["debug_temperature_log"]) is True:
self._logger.debug("Temperature BME680 cmd: %s", cmd)
stdout = (Popen(cmd, shell=True, stdout=PIPE).stdout).read()
if self._settings.get(["debug_temperature_log"]) is True:
self._logger.debug("BME680 result: %s", stdout)
temp, hum, airq = stdout.split("|")
return (self.to_float(temp.strip()), self.to_float(hum.strip()), self.to_float(airq.strip()))
except Exception as ex:
self._logger.info(
"Failed to execute python scripts, try disabling use SUDO on advanced section of the plugin.")
self.log_error(ex)
return (0, 0)
def read_si7021_temp(self, address):
try:
script = os.path.dirname(os.path.realpath(__file__)) + "/SI7021.py "
@@ -1556,4 +1588,4 @@ def __plugin_load__():
__plugin_hooks__ = {
"octoprint.comm.protocol.gcode.queuing" : __plugin_implementation__.hook_gcode_queuing,
"octoprint.plugin.softwareupdate.check_config": __plugin_implementation__.get_update_information
}
}

View File

@@ -0,0 +1,18 @@
import bme680
try:
sensor = bme680.BME680(bme680.I2C_ADDR_PRIMARY)
except IOError:
sensor = bme680.BME680(bme680.I2C_ADDR_SECONDARY)
# These oversampling settings can be tweaked to
# change the balance between accuracy and noise in
# the data.
sensor.set_humidity_oversample(bme680.OS_2X)
sensor.set_pressure_oversample(bme680.OS_4X)
sensor.set_temperature_oversample(bme680.OS_8X)
sensor.set_filter(bme680.FILTER_SIZE_3)
if sensor.get_sensor_data():
print('{0:0.1f} | {1:0.1f}'.format(sensor.data.temperature, sensor.data.humidity))

View File

@@ -536,6 +536,7 @@
<option value="18b20">DS18B20</option>
<option value="si7021">SI7021</option>
<option value="bme280">BME280</option>
<option value="bme680">BME680</option>
<option value="tmp102">TMP102</option>
<option value="max31855">MAX31855</option>
</select>