""" """
import os
from datetime import datetime
from threading import Event, Lock, Thread
from time import sleep
from typing import Optional
from pi_weather.api.models import LatestSTCC4SensorReading
from pi_weather.codes import SENSOR_NOT_ENABLED
from pi_weather.config.config_handler import Config
from pi_weather.core.sensors.stcc4.co2_ranges import get_air_quality_index
from pi_weather.core.sensors.stcc4.constants import SENSOR_SAMPLING_PERIOD
from pi_weather.core.sensors.stcc4.handler import STCC4I2C
from pi_weather.utils.logs import get_logger
from pi_weather.utils.metrics import (
STCC4_AIR_QUALITY_GAUGE_METRIC,
STCC4_CO2_CONCENTRATION_GAUGE_METRIC,
STCC4_HUMIDITY_GAUGE_METRIC,
STCC4_SENSOR_LAST_UPDATE_ET_GAUGE_METRIC,
STCC4_SENSOR_STATUS_GAUGE_METRIC,
STCC4_SENSOR_UPDATES_COUNTER_METRIC,
STCC4_SENSOR_UPDATES_TIME_INTERVAL_ET_GAUGE_METRIC,
STCC4_TEMPERATURE_GAUGE_METRIC,
)
logger = get_logger(__name__)
SENSOR_DATA_ACQUISITION_MODE: str = os.environ.get(
"SENSOR_DATA_ACQUISITION_MODE", default="polling"
)
[docs]
class STCC4Service:
"""STCC4 sensor service class"""
[docs]
def __init__(self) -> None:
"""
Constructor function (init)
"""
if not self._is_sensor_enabled():
logger.fatal(
"STCC4 is not enabled by user. Set up env var PI_WEATHER_STCC4_ENABLED=1"
)
logger.fatal("Forcing program exit")
exit(SENSOR_NOT_ENABLED)
# Sensor (I2C)
self.sensor: Optional[STCC4I2C] = None
# Object to store latest sensor data read
self.latest_reading: Optional[LatestSTCC4SensorReading] = None
# Multi-threading element for thread-safety
self.lock: Lock = Lock()
self.stop_event: Event = Event()
# Thread to run polling procedure into a separate one
self.thread: Optional[Thread] = None
# Timestamp of the latest update (polling call)
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 STCC4 sensor is enabled, False otherwise
:rtype: bool
"""
return 1 == int(os.environ.get("PI_WEATHER_STCC4_ENABLED", default=0))
[docs]
def get_latest(self) -> Optional[LatestSTCC4SensorReading]:
"""
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,
co2: float,
t: float,
h: float,
sensor_status: int,
et: float,
air_quality: int,
) -> None:
"""
Function to update Prometheus metrics
:param co2: CO2 concentration
:type co2: float
:param t: Temperature
:type t: float
:param h: Humidity
:type h: float
:param sensor_status: Sensor status
:type sensor_status: int
:param et: Event time generated
:type et: float
:param air_quality: Air quality index
:type air_quality: int
:return: Nothing
:rtype: None
"""
logger.info("Updating Prometheus metrics")
STCC4_CO2_CONCENTRATION_GAUGE_METRIC.set(value=co2)
STCC4_TEMPERATURE_GAUGE_METRIC.set(value=t)
STCC4_HUMIDITY_GAUGE_METRIC.set(value=h)
STCC4_AIR_QUALITY_GAUGE_METRIC.set(value=air_quality)
STCC4_SENSOR_STATUS_GAUGE_METRIC.set(value=sensor_status)
STCC4_SENSOR_UPDATES_COUNTER_METRIC.inc(amount=1)
STCC4_SENSOR_LAST_UPDATE_ET_GAUGE_METRIC.set(value=et)
# Time interval between 2 latest consecutive update
STCC4_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_et = et
[docs]
def _save_latest_reading(
self,
co2: float,
t: float,
h: float,
sensor_status: int,
et: float,
air_quality: int,
) -> None:
"""
Function to record/save the latest sensor reading
:param co2: CO2 concentration
:type co2: float
:param t: Temperature
:type t: float
:param h: Humidity
:type h: float
:param sensor_status: Sensor status
:type sensor_status: int
:param et: Event time generated
:type et: float
:param air_quality: Air quality index
:type air_quality: int
:return: Nothing
:rtype: None
"""
with self.lock:
self.latest_reading = LatestSTCC4SensorReading(
co2_concentration=co2,
temperature=t,
humidity=h,
sensor_status=sensor_status,
event_ts=et,
air_quality=air_quality,
)
[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
results = self.sensor.get_measurement()
if results:
t = float(results["temperature"])
h = float(results["humidity"])
co2 = float(results["co2_concentration"])
sensor_status = int(results["sensor_status"])
# Generating event time
event_time: float = datetime.now().timestamp()
# Getting air quality
air_quality: int = get_air_quality_index(co2_concentration=co2)
# Recording latest reading
self._save_latest_reading(
co2=co2,
t=t,
h=h,
sensor_status=sensor_status,
et=event_time,
air_quality=air_quality,
)
# Updating Prometheus metrics
self._update_metrics(
co2=co2,
t=t,
h=h,
sensor_status=sensor_status,
et=event_time,
air_quality=air_quality,
)
# Logging new data
self._log_new_data(
co2=co2,
t=t,
h=h,
sensor_status=sensor_status,
et=event_time,
air_quality=air_quality,
)
# 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,
co2: float,
t: float,
h: float,
sensor_status: int,
et: float,
air_quality: int,
) -> None:
"""
Function to correctly log the new data read (regardless the data acquisition mode
:param co2: CO2 concentration
:type co2: float
:param t: Temperature
:type t: float
:param h: Humidity
:type h: float
:param sensor_status: Sensor status
:type sensor_status: int
:param et: Event time generated
:type et: float
:param air_quality: Air quality index
:type air_quality: int
:return: Nothing
:rtype: None
"""
logger.info(
f"CO2 := {co2} ppm \t- AQ := {air_quality} \t- T := {t} °C \t- h := {h} % \t- sensor_status := {sensor_status} \t- et := {et}"
)
# 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=SENSOR_SAMPLING_PERIOD)
logger.info("Stopping polling loop")
[docs]
def start(self) -> None:
"""
Function to start the sensor data capture
:return: Nothing
:rtype: None
"""
logger.info("Starting STCC4 sensor service...")
# Defining sensor object
self.sensor = STCC4I2C(
i2c_addr=Config().get("sensors", "stcc4", "i2c_address"), # 0x64,
bus=Config().get("sensors", "stcc4", "i2c_bus"), # 3
)
if self.sensor is None:
exit(1)
# Starting sensor init
self.sensor.wakeup()
sleep(0.02)
p_id = self.sensor.get_product_id()
logger.info(f"Product ID: {p_id}")
if SENSOR_DATA_ACQUISITION_MODE == "polling":
# Starting sensor config
self.sensor.set_pressure_compensation(pressure=946)
self.sensor.set_rht_compensation(temperature=22, humidity=78)
self.sensor.start_measurement()
sleep(2)
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 STCC4 sensor service...")
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.sensor:
try:
logger.warning("Stopping continuous measurements on STCC4 sensor")
self.sensor.stop_measurement()
except Exception as exc_:
logger.error(
f"An error occurred while trying to stop measurements: {exc_}"
)