"""Definition of BMP390 sensor service"""
import os
from datetime import datetime
from threading import Event, Lock, Thread
from typing import Optional
from gpiozero import DigitalInputDevice
from gpiozero.devices import Device
from gpiozero.pins.lgpio import LGPIOFactory
from pi_weather.api.models import LatestSensorReading
from pi_weather.codes import INIT_FAILURE_EXIT_CODE, SENSOR_NOT_ENABLED
from pi_weather.config.config_handler import Config
from pi_weather.core.sensors.bmp390.addresses import BMP3XX_INT_STATUS, SLEEP_MODE
from pi_weather.core.sensors.bmp390.handler import BMP3XXI2C
from pi_weather.core.sensors.bmp390.precisions import ULTRA_PRECISION
from pi_weather.utils.logs import get_logger
from pi_weather.utils.metrics import (
BMP390_ALTITUDE_GAUGE_METRIC,
BMP390_PRESSURE_GAUGE_METRIC,
BMP390_SENSOR_LAST_UPDATE_ET_GAUGE_METRIC,
BMP390_SENSOR_LAST_UPDATE_ST_GAUGE_METRIC,
BMP390_SENSOR_UPDATES_COUNTER_METRIC,
BMP390_SENSOR_UPDATES_TIME_INTERVAL_ET_GAUGE_METRIC,
BMP390_SENSOR_UPDATES_TIME_INTERVAL_ST_GAUGE_METRIC,
BMP390_TEMPERATURE_GAUGE_METRIC,
)
logger = get_logger(__name__)
SENSOR_DATA_ACQUISITION_MODE: str = os.environ.get(
"SENSOR_DATA_ACQUISITION_MODE", default="interrupt"
)
[docs]
class BMP390Service:
"""BMP390 sensor service class"""
[docs]
def __init__(self) -> None:
"""
Constructor function (init)
"""
if not self._is_sensor_enabled():
logger.fatal(
"BMP390 is not enabled by user. Set up env var PI_WEATHER_BMP390_ENABLED=1"
)
logger.fatal("Forcing program exit")
exit(SENSOR_NOT_ENABLED)
self._INT_GPIO_PIN: int = int(
Config().get("sensors", "bmp390", "interrupt_gpio_pin")
)
self._FORCE_SLEEP_MODE: bool = bool(
Config().get("sensors", "bmp390", "forcing_sleep_mode_on_close")
)
self._POLL_INTERVAL: float = float(
Config().get("sensors", "bmp390", "polling_interval_period")
)
# Sensor (I2C)
self.sensor: Optional[BMP3XXI2C] = None
# Object to store latest sensor data read
self.latest_reading: Optional[LatestSensorReading] = None
#
self.lock: Lock = Lock()
self.stop_event: Event = Event()
# Thread to run polling procedure into a separate one
# The interrupt phase is already handled into a separate thread
self.thread: Optional[Thread] = None
# Interrupt PIN
self.interrupt_pin: Optional[DigitalInputDevice] = None
# Timestamp of the latest update (polling call or hardware interrupt)
self.last_update_ts_st: float = 0.0 # (sensor time)
self.last_update_ts_et: float = 0.0 # (event time)
[docs]
@staticmethod
def _is_sensor_enabled() -> bool:
"""
Function to check whether current sensor is enabled or disabled from the `docker-compose.yaml`
:return: True if BMP390 sensor is enabled, False otherwise
:rtype: bool
"""
return 1 == int(os.environ.get("PI_WEATHER_BMP390_ENABLED", default=0))
[docs]
def get_latest(self) -> Optional[LatestSensorReading]:
"""
Function to get the latest sensor reading
:return: Latest sensor data read by the sensor
:rtype: Optional[LatestSensorReading]
"""
with self.lock:
if self.latest_reading is None:
logger.warning("No latest sensor reading")
return self.latest_reading
[docs]
def _update_metrics(
self, p: float, t: float, a: float, st: float, et: float
) -> None:
"""
Function to update Prometheus metrics
:param p: Pressure data read
:type p: float
:param t: Temperature data read
:type t: float
:param a: Altitude data computed
:type a: float
:param st: Sensor time data read
:type st: float
:param et: Event time generated
:type et: float
:return: Nothing
:rtype: None
"""
logger.info("Updating Prometheus metrics")
BMP390_PRESSURE_GAUGE_METRIC.set(value=p)
BMP390_TEMPERATURE_GAUGE_METRIC.set(value=t)
BMP390_ALTITUDE_GAUGE_METRIC.set(value=a)
BMP390_SENSOR_UPDATES_COUNTER_METRIC.inc(amount=1)
BMP390_SENSOR_LAST_UPDATE_ST_GAUGE_METRIC.set(value=st)
BMP390_SENSOR_LAST_UPDATE_ET_GAUGE_METRIC.set(value=et)
# Time interval between 2 latest consecutive update
BMP390_SENSOR_UPDATES_TIME_INTERVAL_ST_GAUGE_METRIC.set(
value=st - self.last_update_ts_st
)
BMP390_SENSOR_UPDATES_TIME_INTERVAL_ET_GAUGE_METRIC.set(
value=et - self.last_update_ts_et
)
# Updating last update ts with latest sensor time
self.last_update_ts_st = st
self.last_update_ts_et = et
[docs]
def _save_latest_reading(
self, p: float, t: float, a: float, st: float, et: float
) -> None:
"""
Function to record/save the latest sensor reading
:param p: Pressure data read
:type p: float
:param t: Temperature data read
:type t: float
:param a: Altitude data computed
:type a: float
:param st: Sensor time data read
:type st: float
:param et: Event time generated
:type et: float
:return: Nothing
:rtype: None
"""
with self.lock:
self.latest_reading = LatestSensorReading(
pressure=p, temperature=t, altitude=a, sensor_ts=st, event_ts=et
)
[docs]
def _read_sensor_data(self) -> None:
"""
Function to read sensor data
This method will record the latest reading + update Prometheus metrics
:return: Nothing
:rtype: None
"""
if self.sensor is None:
logger.error("Sensor is not initialized or invalid")
return None
pressure: float = self.sensor.get_pressure
temperature: float = self.sensor.get_temperature
altitude: float = self.sensor.get_altitude
sensor_time: float = self.sensor.get_sensor_time
event_time: float = datetime.now().timestamp()
# Recording latest reading
self._save_latest_reading(
p=pressure, t=temperature, a=altitude, st=sensor_time, et=event_time
)
# Updating Prometheus metrics
self._update_metrics(
p=pressure, t=temperature, a=altitude, st=sensor_time, et=event_time
)
# Logging new data
self._log_new_data(
p=pressure, t=temperature, a=altitude, st=sensor_time, et=event_time
)
# logger.info(
# f"Pressure: {pressure} Pa - Temperature: {temperature} C - Altitude: {altitude} m - Sensor time: {sensor_time} - Event time: {event_time}"
# )
return None
[docs]
def _log_new_data(self, p: float, t: float, a: float, st: float, et: float) -> None:
"""
Function to correctly log the new data read (regardless the data acquisition mode
:param p: Pressure data read
:type p: float
:param t: Temperature data read
:type t: float
:param a: Altitude data computed
:type a: float
:param st: Sensor time data read
:type st: float
:param et: Event time generated
:type et: float
:return: Nothing
:rtype: None
"""
logger.info(
f"P := {p} Pa \t- T := {t} °C \t- H := {a} m \t- st := {st} \t- et := {et}"
)
[docs]
def _on_interrupt(self) -> None:
"""
Function to be executed on hardware interrupt
:return: Nothing
:rtype: None
"""
if self.sensor is None:
logger.error("Sensor is not initialized or invalid")
return None
logger.info("New interrupt")
try:
# Reading INT_STATUS register to know which interrupt has been triggered
status: int = self.sensor._read_reg(reg=BMP3XX_INT_STATUS, length=1)[0]
if status & 0x08: # Data Ready interrupt (drdy = bit 3 (2**3 := 8)
# logger.info("Data ready !")
self._read_sensor_data()
elif status & 0x02: # FIFO Full interrupt (ffull_int = bit 1 (2**1 := 2)
# logger.info("FIFO full")
pass # IMPORTANT: Not supported now
elif status & 0x01: # FIFO Watermark interrupt (fwm_int = bit 0 (2**0 := 1)
# logger.info("FIFO watermark")
pass # IMPORTANT: Not supported now
except Exception as exc_:
logger.error(
f"An error occurred while reading sensor data from interrupt: {exc_}"
)
[docs]
def _on_interrupt_end(self) -> None:
"""
Function to be executed on hardware interrupt end (when_deactivated)
:return: Nothing
:rtype: None
"""
# logger.info("On interrupt end")
...
# Polling loop (run into a separate thread)
[docs]
def _on_polling(self) -> None:
"""
Function to be executed within the separate polling thread (target func)
:return: Nothing
:rtype: None
"""
logger.info("Starting polling loop")
while not self.stop_event.is_set():
try:
self._read_sensor_data()
except Exception as exc_:
logger.info(f"An error occurred while reading sensor data: {exc_}")
self.stop_event.wait(timeout=self._POLL_INTERVAL)
logger.info("Stopping polling loop")
[docs]
def start(self) -> None:
"""
Function to start the sensor data capture
:return: Nothing
:rtype: None
"""
logger.info("Starting BMP390 sensor service...")
# Defining sensor object
self.sensor = BMP3XXI2C(
i2c_addr=Config().get("sensors", "bmp390", "i2c_address"), # 0x77,
bus=Config().get("sensors", "bmp390", "i2c_bus"), # 1,
)
# Starting sensor init
if not self.sensor.begin():
logger.fatal("Failed to initialize BMP390 sensor")
exit(INIT_FAILURE_EXIT_CODE)
# Starting sensor config
H: float = float(Config().get("sensors", "runtime", "current_sensor_altitude"))
self.sensor.set_common_sampling_mode(mode=ULTRA_PRECISION)
self.sensor.calibrated_absolute_difference(altitude=H)
# Dispatching service behavior depending on acquisition mode
if SENSOR_DATA_ACQUISITION_MODE == "interrupt":
logger.info("[Data acquisition] Enabling INTERRUPT mode...")
# Enabling interrupts
self.sensor.enable_data_ready_interrupt()
# [IMPORTANT] Forcing pin factory to lgpio backend
# Device.pin_factory = LGPIOFactory(chip=15)
Device.pin_factory = LGPIOFactory(
chip=int(Config().get("sensors", "bmp390", "lgpio_factory_chip_id"))
)
# Defining interrupt pin
self.interrupt_pin = DigitalInputDevice(
pin=self._INT_GPIO_PIN, pull_up=None, active_state=True
)
# Registering callbacks
self.interrupt_pin.when_activated = self._on_interrupt
self.interrupt_pin.when_deactivated = self._on_interrupt_end
logger.info("[Data acquisition] INTERRUPT mode enabled")
elif SENSOR_DATA_ACQUISITION_MODE == "polling":
logger.info("[Data acquisition] Enabling POLLING mode...")
# Clearing any stop event (used in the polling thread)
self.stop_event.clear()
# Defining polling thread
self.thread = Thread(
target=self._on_polling, daemon=True, name="bmp390_polling_thread"
)
self.thread.start()
logger.info("[Data acquisition] POLLING mode enabled")
[docs]
def stop(self) -> None:
"""
Function to stop the sensor data capture
:return: Nothing
:rtype: None
"""
logger.info("Stopping BMP390 sensor service...")
if (
SENSOR_DATA_ACQUISITION_MODE == "interrupt"
and self.interrupt_pin is not None
):
# Closing interrupt PIN
self.interrupt_pin.close()
self.interrupt_pin = None
if self.thread is not None and self.thread.is_alive():
# Setting up the stop event for the polling thread (event handled in the self._on_polling func)
self.stop_event.set()
# Waiting for polling thread to terminate (with timeout)
self.thread.join(timeout=3.0)
if self._FORCE_SLEEP_MODE:
if self.sensor:
try:
# Setting sleep mode
logger.warning("Forcing sleep mode on BMP390 sensor")
self.sensor.set_power_mode(mode=SLEEP_MODE)
except Exception as exc_:
logger.error(f"An error occurred while setting sleep mode: {exc_}")