BME680-to-prometheus.py

#!/usr/bin/env python3
import bme680
from time import time, sleep
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway

# Setup:
#   python3 -m pip install bme680 prometheus-client
#   sudo apt install --no-install-recommends prometheus prometheus-pushgateway

BURN_IN_TIME = 1800  # 30 min

def get_sensor():
    try:
        return bme680.BME680(bme680.I2C_ADDR_PRIMARY)
    except IOError:
        return bme680.BME680(bme680.I2C_ADDR_SECONDARY)


def main():
    sensor = get_sensor()

    start_time = time()
    burn_in_data = []

    sensor.set_gas_heater_temperature(320)  # "Target temperature in degrees celsius, between 200 and 400"
    sensor.set_gas_heater_duration(150)     # "Heating durations between 1 ms and 4032 ms"
    sensor.select_gas_heater_profile(0)

    # Collect gas resistance burn-in values, then use the average
    # of the last 50 values to set the upper limit for calculating
    # gas_baseline.
    print(f'Collecting gas resistance burn-in data for {1800//60} mins\n')
    while (time() - start_time) < BURN_IN_TIME and len(burn_in_data) < 100:
        if sensor.get_sensor_data() and sensor.data.heat_stable:
            gas = sensor.data.gas_resistance
            burn_in_data.append(gas)
            print(f'Gas: {gas} Ohms (data: {sensor.data})')
        sleep(5)

    gas_baseline = sum(burn_in_data[-50:]) / 50.0

    # Set the humidity baseline to 40%, an optimal indoor humidity.
    hum_baseline = 40.0

    # This sets the balance between humidity and gas reading in the
    # calculation of air_quality_score (25:75, humidity:gas)
    hum_weighting = 0.25

    print('Gas baseline: {0} Ohms, humidity baseline: {1:.2f} %RH\n'.format(gas_baseline, hum_baseline))

    registry = CollectorRegistry()

    g_gas_baseline_ohms = Gauge('sensor_baseline_gas_ohms', 'Gas baseline for this session (ohms)', registry=registry)
    g_gas_baseline_ohms.set(gas_baseline)

    g_humidity_baseline_ohms = Gauge('sensor_baseline_humidity_ohms', 'Humidity baseline for this session (ohms)', registry=registry)
    g_humidity_baseline_ohms.set(hum_baseline)

    g_gas_ohms = Gauge('sensor_gas_ohms', 'Gas (Ohms)', registry=registry)
    g_humidity_ohms = Gauge('sensor_humidity_ohms', 'Humidity (Ohms)', registry=registry)
    g_temperature_degC = Gauge('sensor_temperature_degC', 'Temperature (*C)', registry=registry)
    g_pressure_hPa = Gauge('sensor_pressure_hPa', 'Pressure (hPa)', registry=registry)
    g_heat_stable = Gauge('sensor_heat_stable', 'Heat stable', registry=registry)
    g_air_quality_score = Gauge('sensor_airquality_score', 'Air quality score', registry=registry)

    while True:
        sleep(45)
        g_heat_stable.set(sensor.data.heat_stable)
        if sensor.get_sensor_data() and sensor.data.heat_stable:
            gas = sensor.data.gas_resistance
            gas_offset = gas_baseline - gas

            hum = sensor.data.humidity
            hum_offset = hum - hum_baseline

            # Calculate hum_score as the distance from the hum_baseline.
            if hum_offset > 0:
                hum_score = (100 - hum_baseline - hum_offset)
                hum_score /= (100 - hum_baseline)
                hum_score *= (hum_weighting * 100)

            else:
                hum_score = (hum_baseline + hum_offset)
                hum_score /= hum_baseline
                hum_score *= (hum_weighting * 100)

            # Calculate gas_score as the distance from the gas_baseline.
            if gas_offset > 0:
                gas_score = (gas / gas_baseline)
                gas_score *= (100 - (hum_weighting * 100))
            else:
                gas_score = 100 - (hum_weighting * 100)

            # Calculate air_quality_score.
            air_quality_score = hum_score + gas_score
           
            g_temperature_degC.set(sensor.data.temperature)
            g_pressure_hPa.set(sensor.data.pressure)
            g_gas_ohms.set(gas)
            g_humidity_ohms.set(hum)
            g_air_quality_score.set(air_quality_score)
            push_to_gateway('localhost:9091', job='sensor', registry=registry)
            print('Gas: {0:.2f} Ohms,humidity: {1:.2f} %RH,air quality: {2:.2f}'.format(gas, hum, air_quality_score))

if __name__ == "__main__":
    main()