114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
import logging
|
|
import os
|
|
import signal
|
|
import sys
|
|
|
|
import bme680
|
|
from prometheus_client import Gauge, start_http_server
|
|
from pyhap.accessory import Accessory, Bridge
|
|
from pyhap.accessory_driver import AccessoryDriver
|
|
from pyhap.const import CATEGORY_SENSOR
|
|
|
|
|
|
def fail(message: str):
|
|
"""
|
|
Fail crashes the program and logs the message.
|
|
"""
|
|
logging.critical(message)
|
|
sys.exit(1)
|
|
|
|
|
|
class Bme680Sensor(Accessory):
|
|
"""Implementation of a mock temperature sensor accessory."""
|
|
|
|
category = CATEGORY_SENSOR # This is for the icon in the iOS Home app.
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
"""Here, we just store a reference to the current temperature characteristic and
|
|
add a method that will be executed every time its value changes.
|
|
"""
|
|
# If overriding this method, be sure to call the super's implementation first.
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.sensor = bme680.BME680(bme680.I2C_ADDR_PRIMARY)
|
|
|
|
self.sensor.set_humidity_oversample(bme680.OS_2X)
|
|
self.sensor.set_pressure_oversample(bme680.OS_4X)
|
|
self.sensor.set_temperature_oversample(bme680.OS_8X)
|
|
self.sensor.set_filter(bme680.FILTER_SIZE_3)
|
|
|
|
if os.getenv("ENABLE_GAS_MEASUREMENT", "false") == "true":
|
|
self.sensor.set_gas_status(bme680.ENABLE_GAS_MEAS)
|
|
self.sensor.set_gas_heater_temperature(320)
|
|
self.sensor.set_gas_heater_duration(150)
|
|
self.sensor.select_gas_heater_profile(0)
|
|
|
|
self._temperature_metric = Gauge(
|
|
"bme680_temperature_celsius", "The temperature measurement in celsius."
|
|
)
|
|
self._humidity_metric = Gauge(
|
|
"bme680_humidity_rh", "The humidity measurement in relative humidity."
|
|
)
|
|
self._pressure_metric = Gauge(
|
|
"bme680_pressure_hpa", "The pressure measurement in hectopascals."
|
|
)
|
|
self._gas_resistance_metric = Gauge(
|
|
"bme860_gas_resistance_ohm", "The gas resistance measurement in ohms."
|
|
)
|
|
|
|
# Add the services that this Accessory will support with add_preload_service here
|
|
temp_service = self.add_preload_service("TemperatureSensor")
|
|
humidity_service = self.add_preload_service("HumiditySensor")
|
|
|
|
self.temp_value = temp_service.get_characteristic("CurrentTemperature")
|
|
self.humidity_value = humidity_service.get_characteristic(
|
|
"CurrentRelativeHumidity"
|
|
)
|
|
|
|
def _run(self):
|
|
if self.sensor.get_sensor_data():
|
|
self.temp_value.set_value(self.sensor.data.temperature)
|
|
self.humidity_value.set_value(self.sensor.data.humidity)
|
|
|
|
# Update prometheus metrics.
|
|
self._temperature_metric.set(self.sensor.data.temperature)
|
|
self._humidity_metric.set(self.sensor.data.humidity)
|
|
self._pressure_metric.set(self.sensor.data.pressure)
|
|
|
|
if self.sensor.data.heat_stable:
|
|
self._gas_resistance_metric.set(self.sensor.data.gas_resistance)
|
|
|
|
@Accessory.run_at_interval(120)
|
|
def run(self):
|
|
"""
|
|
This function runs the accessory. It polls for data and publishes it at the given interval.
|
|
"""
|
|
try:
|
|
self._run()
|
|
except IOError as e:
|
|
# This happens from time to time, best we stop and let systemd restart us.
|
|
fail(f"Failed due to IOError: {e}")
|
|
|
|
def stop(self):
|
|
"""We override this method to clean up any resources or perform final actions, as
|
|
this is called by the AccessoryDriver when the Accessory is being stopped.
|
|
"""
|
|
print("Stopping accessory.")
|
|
|
|
|
|
def get_bridge(accessory_driver):
|
|
bridge = Bridge(accessory_driver, "Bridge")
|
|
bridge.add_accessory(Bme680Sensor(accessory_driver, "Sensor"))
|
|
return bridge
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Prometheus' metrics server
|
|
start_http_server(8000)
|
|
# Python HAP
|
|
driver = AccessoryDriver(
|
|
port=51826, persist_file="/home/pi/bme680-homekit/sensors/accessory.state"
|
|
)
|
|
driver.add_accessory(accessory=get_bridge(driver))
|
|
signal.signal(signal.SIGTERM, driver.signal_handler)
|
|
driver.start()
|