diff --git a/config.yaml.example b/config.yaml.example index 910dcb6..2eb12dc 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -286,6 +286,15 @@ sensors: # i2c_address: 0x70 # bus_number: 1 + # Example BME280 temperature/pressure/humidity sensor (commented out by default) + # - type: bme280 + # name: ambient + # enabled: true + # auto_install_packages: false + # settings: + # i2c_address: 0x76 # 0x77 when SDO is tied high + # bus_number: 1 # I2C bus number (1 for Raspberry Pi default) + # Example Waveshare UPS D Hat # - type: waveshare_ups_d # name: battery diff --git a/docs/adding_sensors.md b/docs/adding_sensors.md index 0afcd05..3758988 100644 --- a/docs/adding_sensors.md +++ b/docs/adding_sensors.md @@ -169,3 +169,4 @@ def test_my_sensor_loads_and_reads(): | `hardware_stats` | `repeater/sensors/hardware_stats.py` | Host CPU / memory / disk / network (via `psutil`) | | `ina219` | `repeater/sensors/ina219.py` | INA219 I²C current/voltage/power monitor | | `ens210` | `repeater/sensors/ens210.py` | ENS210 I²C relative humidity and temperature sensor | +| `bme280` | `repeater/sensors/bme280.py` | BME280 I²C temperature, humidity and barometric pressure sensor | diff --git a/repeater/sensors/bme280.py b/repeater/sensors/bme280.py new file mode 100644 index 0000000..504e69e --- /dev/null +++ b/repeater/sensors/bme280.py @@ -0,0 +1,198 @@ +""" +BME280 temperature, humidity and barometric pressure sensor plug-in. + +Requires: pip install smbus2 + +Config example: + - type: bme280 + name: "ambient" + enabled: true + auto_install_packages: false + settings: + i2c_address: 0x76 # 0x77 when SDO is tied high + bus_number: 1 # I2C bus number (1 for Raspberry Pi default) + read_timeout_seconds: 1.0 # Max time to wait for the measurement (polls every 5 ms) +""" + +from __future__ import annotations + +import struct +import time +from typing import Any, Dict, Optional + +from .base import SensorBase +from .registry import SensorRegistry + +# BME280 register addresses +_REG_CALIB_LOW = 0x88 # dig_T1..dig_P9 + dig_H1, 26 bytes +_REG_CALIB_HIGH = 0xE1 # dig_H2..dig_H6, 7 bytes +_REG_ID = 0xD0 +_REG_CTRL_HUM = 0xF2 +_REG_STATUS = 0xF3 +_REG_CTRL_MEAS = 0xF4 +_REG_CONFIG = 0xF5 +_REG_DATA = 0xF7 # press(3) + temp(3) + hum(2) + +_CHIP_ID = 0x60 +_CHIP_ID_BMP280 = 0x58 # Same register map, no humidity channel + +_OVERSAMPLING_X1 = 0x01 +_MODE_FORCED = 0x01 +# osrs_t x1, osrs_p x1, forced mode +_CTRL_MEAS_FORCED = (_OVERSAMPLING_X1 << 5) | (_OVERSAMPLING_X1 << 2) | _MODE_FORCED +_STATUS_MEASURING = 0x08 + + +@SensorRegistry.register("bme280") +class BME280Sensor(SensorBase): + sensor_type = "bme280" + + def __init__(self, name: str, config: Optional[Dict[str, Any]] = None, log=None): + super().__init__(name=name, config=config, log=log) + + self.i2c_address = int(self.settings.get("i2c_address", 0x76)) + self.bus_number = int(self.settings.get("bus_number", 1)) + self._poll_interval = 0.005 # 5 ms between measurement-complete checks + self._poll_attempts = max( + 1, int(float(self.settings.get("read_timeout_seconds", 1.0)) / self._poll_interval) + ) + self._calibration: Optional[Dict[str, Any]] = None + + self.available = False + if not self.ensure_python_modules([("smbus2", "smbus2")]): + return + + try: + import smbus2 # type: ignore[import-not-found] + + self._smbus2 = smbus2 + + bus = smbus2.SMBus(self.bus_number) + try: + chip_id = bus.read_byte_data(self.i2c_address, _REG_ID) + if chip_id != _CHIP_ID: + hint = ( + " (that is a BMP280, which has no humidity channel)" + if (chip_id == _CHIP_ID_BMP280) + else "" + ) + raise RuntimeError(f"unexpected chip id 0x{chip_id:02X}{hint}") + self._calibration = self._read_calibration(bus) + finally: + bus.close() + + self.available = True + self.log.info( + "BME280 initialized (addr=0x%02X, bus=%d)", + self.i2c_address, + self.bus_number, + ) + except Exception as exc: + self.log.warning( + "BME280 init failed (addr=0x%02X, bus=%d): %s", + self.i2c_address, + self.bus_number, + exc, + ) + self.available = False + + def _read_calibration(self, bus) -> Dict[str, Any]: + """Read the factory calibration block.""" + low = bytes(bus.read_i2c_block_data(self.i2c_address, _REG_CALIB_LOW, 26)) + high = bytes(bus.read_i2c_block_data(self.i2c_address, _REG_CALIB_HIGH, 7)) + + t1, t2, t3 = struct.unpack_from("> 4) + h4 = h4 - 4096 if h4 > 2047 else h4 + h5 = h5 - 4096 if h5 > 2047 else h5 + h6 = struct.unpack_from(" tuple[int, int, int]: + """Trigger one forced-mode measurement and return raw ADC counts.""" + # ctrl_hum only takes effect on the following ctrl_meas write + bus.write_byte_data(self.i2c_address, _REG_CTRL_HUM, _OVERSAMPLING_X1) + bus.write_byte_data(self.i2c_address, _REG_CONFIG, 0x00) # IIR filter off + bus.write_byte_data(self.i2c_address, _REG_CTRL_MEAS, _CTRL_MEAS_FORCED) + + for _ in range(self._poll_attempts): + time.sleep(self._poll_interval) + if not bus.read_byte_data(self.i2c_address, _REG_STATUS) & _STATUS_MEASURING: + break + else: + raise RuntimeError( + f"measurement timed out after {self._poll_attempts * self._poll_interval:.1f}s" + ) + + d = bus.read_i2c_block_data(self.i2c_address, _REG_DATA, 8) + adc_p = (d[0] << 12) | (d[1] << 4) | (d[2] >> 4) + adc_t = (d[3] << 12) | (d[4] << 4) | (d[5] >> 4) + adc_h = (d[6] << 8) | d[7] + return adc_t, adc_p, adc_h + + def _compensate(self, adc_t: int, adc_p: int, adc_h: int) -> tuple[float, float, float]: + """Apply the datasheet floating-point compensation formulas (section 8.1).""" + calibration = self._calibration + t1, t2, t3 = calibration["t"] + p1, p2, p3, p4, p5, p6, p7, p8, p9 = calibration["p"] + h1, h2, h3, h4, h5, h6 = calibration["h"] + + var1 = (adc_t / 16384.0 - t1 / 1024.0) * t2 + var2 = ((adc_t / 131072.0 - t1 / 8192.0) ** 2) * t3 + t_fine = var1 + var2 + temperature_c = t_fine / 5120.0 + + var1 = t_fine / 2.0 - 64000.0 + var2 = var1 * var1 * p6 / 32768.0 + var2 = var2 + var1 * p5 * 2.0 + var2 = var2 / 4.0 + p4 * 65536.0 + var1 = (p3 * var1 * var1 / 524288.0 + p2 * var1) / 524288.0 + var1 = (1.0 + var1 / 32768.0) * p1 + if var1 == 0: + raise RuntimeError("pressure compensation divided by zero") + pressure = 1048576.0 - adc_p + pressure = (pressure - var2 / 4096.0) * 6250.0 / var1 + var1 = p9 * pressure * pressure / 2147483648.0 + var2 = pressure * p8 / 32768.0 + pressure_pa = pressure + (var1 + var2 + p7) / 16.0 + + humidity = t_fine - 76800.0 + humidity = (adc_h - (h4 * 64.0 + h5 / 16384.0 * humidity)) * ( + h2 / 65536.0 * (1.0 + h6 / 67108864.0 * humidity * (1.0 + h3 / 67108864.0 * humidity)) + ) + humidity = humidity * (1.0 - h1 * humidity / 524288.0) + humidity_pct = min(100.0, max(0.0, humidity)) + + return temperature_c, pressure_pa, humidity_pct + + def _read(self) -> Dict[str, Any]: + """Read temperature, pressure and humidity from the BME280.""" + if not self.available: + raise RuntimeError("BME280 device not available") + + bus = self._smbus2.SMBus(self.bus_number) + try: + temperature_c, pressure_pa, humidity_pct = self._compensate(*self._measure(bus)) + return { + "temperature_c": round(temperature_c, 2), + "humidity_pct": round(humidity_pct, 2), + "pressure_hpa": round(pressure_pa / 100.0, 2), + } + except RuntimeError: + raise + except Exception as exc: + raise RuntimeError(f"BME280 read failed: {exc}") from exc + finally: + bus.close() diff --git a/tests/test_sensors.py b/tests/test_sensors.py index 162bc9b..fa51682 100644 --- a/tests/test_sensors.py +++ b/tests/test_sensors.py @@ -8,12 +8,14 @@ from unittest.mock import MagicMock, patch import pytest from repeater.sensors import SensorBase, SensorManager, SensorRegistry +from repeater.sensors import bme280 as bme280_module from repeater.sensors import ens210 as ens210_module from repeater.sensors import ina219 as ina219_module from repeater.sensors import lafvin_ups_3s as lafvin_ups_3s_module from repeater.sensors import shtc3 as shtc3_module from repeater.sensors import waveshare_ups_d as waveshare_ups_d_module from repeater.sensors import waveshare_ups_e as waveshare_ups_e_module +from repeater.sensors.bme280 import BME280Sensor from repeater.sensors.ens210 import ENS210Sensor from repeater.sensors.ina219 import INA219Sensor from repeater.sensors.lafvin_ups_3s import LafvinUps3sSensor @@ -395,6 +397,82 @@ def test_ens210_sensor_reads_temperature_and_humidity(monkeypatch): } +# Calibration block and one forced-mode measurement captured from a real BME280 +# (chip id 0x60) indoors at ~32 degC / 1021 hPa / 38 %RH. +_BME280_CALIB_LOW = list(bytes.fromhex("B26FDE673200B38E9CD6D00B58178C00F9FFB42DE8D18813004B")) +_BME280_CALIB_HIGH = list(bytes.fromhex("6901001422031E")) +_BME280_DATA = list(bytes.fromhex("5860008870006BE2")) + + +def _make_fake_bme280_bus(chip_id=0x60, measuring_reads=0): + class _Bus: + def __init__(self, bus_number): + self.bus_number = bus_number + self._remaining_measuring = measuring_reads + + def read_byte_data(self, addr, register): + if register == bme280_module._REG_ID: + return chip_id + if register == bme280_module._REG_STATUS: + if self._remaining_measuring > 0: + self._remaining_measuring -= 1 + return bme280_module._STATUS_MEASURING + return 0x00 + raise AssertionError(f"unexpected register: {register}") + + def write_byte_data(self, addr, register, value): + return None + + def read_i2c_block_data(self, addr, register, length): + if register == bme280_module._REG_CALIB_LOW: + return list(_BME280_CALIB_LOW) + if register == bme280_module._REG_CALIB_HIGH: + return list(_BME280_CALIB_HIGH) + if register == bme280_module._REG_DATA: + return list(_BME280_DATA) + raise AssertionError(f"unexpected register: {register}") + + def close(self): + return None + + return _Bus + + +def test_bme280_sensor_reads_temperature_pressure_and_humidity(monkeypatch): + _install_fake_smbus2(monkeypatch, _make_fake_bme280_bus(measuring_reads=2)) + monkeypatch.setattr(bme280_module.time, "sleep", lambda *_args, **_kwargs: None) + + reading = BME280Sensor("ambient").read() + + assert reading["ok"] is True + assert reading["data"] == { + "temperature_c": 32.13, + "humidity_pct": 38.38, + "pressure_hpa": 1021.46, + } + + +def test_bme280_sensor_rejects_wrong_chip_id(monkeypatch): + _install_fake_smbus2(monkeypatch, _make_fake_bme280_bus(chip_id=0x58)) + + sensor = BME280Sensor("ambient") + + assert sensor.available is False + reading = sensor.read() + assert reading["ok"] is False + assert "not available" in reading["error"] + + +def test_bme280_sensor_read_times_out_when_measurement_never_completes(monkeypatch): + _install_fake_smbus2(monkeypatch, _make_fake_bme280_bus(measuring_reads=10**6)) + monkeypatch.setattr(bme280_module.time, "sleep", lambda *_args, **_kwargs: None) + + reading = BME280Sensor("ambient", {"settings": {"read_timeout_seconds": 0.05}}).read() + + assert reading["ok"] is False + assert "timed out" in reading["error"] + + def test_waveshare_ups_d_sensor_reads_battery_state(monkeypatch): class _Bus: def __init__(self, bus_number):