Source code for herosdevices.hardware.bosch.bme

"""Module for BME sensors."""

from herosdevices.core.templates.i2c import I2CDeviceTemplate
from herosdevices.helper import log, mark_driver

try:
    from bme280 import BME280 as _BME280  # type: ignore[import-untyped]
except ModuleNotFoundError:
    _BME280 = None  # type: ignore[assignment,misc]




@mark_driver(
    state="beta",
    info="BME280 Environmental Sensor",
    product_page="https://www.bosch-sensortec.com/en/products/environmental-sensors/humidity-sensors-bme280",
    requires={"bme280": "pimoroni-bme280"},
)
class BME280(I2CDeviceTemplate):
    """BME280 Environmental Sensor via I2C."""

    def __init__(self, bus_id: int = 1) -> None:
        """Create a new BME280 representation.

        Args:
            bus_id: I2C bus id (default 1).
        """
        super().__init__(bus_id, keep_alive=True)
        self._device: _BME280 | None = None  # type: ignore[valid-type]

    def _open(self) -> None:
        """Open the I2C bus and initialise the sensor with a dummy read.

        The first reading from the BME280 after power-on is unreliable, so it
        is discarded here. Cleans up and re-raises on any error.
        """
        self.connection.open()
        try:
            self._device = _BME280(i2c_dev=self.connection.connection)  # type: ignore[misc]
            self._device.update_sensor()  # first reading is always garbage
        except Exception:
            self._close()
            raise

    def _close(self) -> None:
        """Close the I2C bus and reset the device handle.

        Safe to call even if the device was never opened or already closed.
        """
        try:
            self.connection.close()
        except Exception:  # noqa: BLE001
            log.exception("Error closing BME280 device")
        finally:
            self._device = None



    def measure(self) -> tuple[float, float, float]:
        """Return a single (temperature, pressure, humidity) reading.

        Opens the device on first call and keeps it open across calls. Closes
        the device and raises ValueError on any I2C or sensor error.

        Returns:
            Temperature in °C, pressure in hPa, humidity in %.

        Raises:
            ValueError: if the measurement fails.
        """
        try:
            if self._device is None:
                self._open()
            assert self._device is not None
            self._device.update_sensor()
            temperature = self._device.temperature
            pressure = self._device.pressure
            humidity = self._device.humidity
        except Exception:  # noqa: BLE001
            log.warning("Error reading from BME280 sensor", exc_info=True)
            self._close()
            raise ValueError from None
        else:
            return temperature, pressure, humidity


    def _observable_data(self) -> dict[str, tuple[float, str]] | None:
        try:
            temperature, pressure, humidity = self.measure()
        except ValueError:
            return None
        else:
            return {"temperature": (temperature, "degC"), "pressure": (pressure, "hPa"), "humidity": (humidity, "pct")}