From 6e89272bce24d740368d00d60eda28a77ab95292 Mon Sep 17 00:00:00 2001 From: Zack Carlson Date: Tue, 19 May 2026 09:25:38 -0700 Subject: [PATCH] Add Waveshare UPS HAT (E) sensor plug-in The HAT (E) uses a dedicated BMS MCU (not an INA219) at I2C address 0x2D. It exposes charge state, pack voltage/current, per-cell voltages, remaining capacity in mAh, and time-to-empty/full estimates directly via I2C registers. Tested on Raspberry Pi 4B with pyMC_Repeater running pymc-battery-writer. Co-Authored-By: Claude Sonnet 4.6 --- repeater/sensors/waveshare_ups_e.py | 149 ++++++++++++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 repeater/sensors/waveshare_ups_e.py diff --git a/repeater/sensors/waveshare_ups_e.py b/repeater/sensors/waveshare_ups_e.py new file mode 100644 index 0000000..0ec3e5c --- /dev/null +++ b/repeater/sensors/waveshare_ups_e.py @@ -0,0 +1,149 @@ +""" +Waveshare UPS HAT (E) battery monitor plug-in — BMS MCU at I2C 0x2D. + +The HAT (E) uses a dedicated BMS chip (not an INA219) that exposes charge +state, pack voltage/current, per-cell voltages, remaining capacity in mAh, +and time-to-empty / time-to-full estimates directly via I2C registers. + +Requires: pip install smbus2 + +Config example: + - type: waveshare_ups_e + name: "battery" + enabled: true + auto_install_packages: false + settings: + i2c_address: 0x2D # Waveshare UPS HAT (E) fixed address + bus_number: 1 # I2C bus number (1 for Raspberry Pi default) + low_cell_mv: 3150 # Per-cell warning threshold in mV +""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .base import SensorBase +from .registry import SensorRegistry + +# Register map +_REG_STATUS = 0x02 # 1 byte — charge state flags +_REG_VBUS = 0x10 # 6 bytes — input (VBUS) voltage, current, power +_REG_BATT = 0x20 # 12 bytes — pack voltage, current, percent, mAh, time +_REG_CELLS = 0x30 # 8 bytes — four cell voltages (LE uint16 each) + +# Charge state flag bits +_FLAG_FAST_CHARGE = 0x40 +_FLAG_CHARGING = 0x80 +_FLAG_DISCHARGING = 0x20 + + +def _u16le(data: list, offset: int) -> int: + return data[offset] | (data[offset + 1] << 8) + + +@SensorRegistry.register("waveshare_ups_e") +class WaveshareUpsESensor(SensorBase): + sensor_type = "waveshare_ups_e" + + def __init__(self, name: str, config: Optional[Dict[str, Any]] = None, log=None): + super().__init__(name=name, config=config, log=log) + + addr = self.settings.get("i2c_address", 0x2D) + self.i2c_address = int(addr, 0) if isinstance(addr, str) else int(addr) + self.bus_number = int(self.settings.get("bus_number", 1)) + self.low_cell_mv = int(self.settings.get("low_cell_mv", 3150)) + + 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: + bus.read_i2c_block_data(self.i2c_address, _REG_STATUS, 1) + finally: + bus.close() + + self.available = True + self.log.info( + "Waveshare UPS HAT (E) initialized (addr=0x%02X, bus=%d)", + self.i2c_address, + self.bus_number, + ) + except Exception as exc: + self.log.warning( + "Waveshare UPS HAT (E) init failed (addr=0x%02X, bus=%d): %s", + self.i2c_address, + self.bus_number, + exc, + ) + + def _read(self) -> Dict[str, Any]: + """Read all battery state from the HAT (E) BMS.""" + if not self.available: + raise RuntimeError("Waveshare UPS HAT (E) not available") + + try: + bus = self._smbus2.SMBus(self.bus_number) + try: + status = bus.read_i2c_block_data(self.i2c_address, _REG_STATUS, 1)[0] + vb = bus.read_i2c_block_data(self.i2c_address, _REG_VBUS, 6) + bd = bus.read_i2c_block_data(self.i2c_address, _REG_BATT, 12) + cd = bus.read_i2c_block_data(self.i2c_address, _REG_CELLS, 8) + finally: + bus.close() + + # Charge state + if status & _FLAG_FAST_CHARGE: charge_state = "fast_charging" + elif status & _FLAG_CHARGING: charge_state = "charging" + elif status & _FLAG_DISCHARGING: charge_state = "discharging" + else: charge_state = "idle" + + # VBUS (input power from charger) + vbus_voltage_mv = _u16le(vb, 0) + vbus_current_ma = _u16le(vb, 2) + vbus_power_mw = _u16le(vb, 4) + + # Battery pack + batt_voltage_mv = _u16le(bd, 0) + batt_current_ma = _u16le(bd, 2) + if batt_current_ma > 0x7FFF: # signed 16-bit + batt_current_ma -= 0xFFFF + batt_percent = _u16le(bd, 4) + remaining_mah = _u16le(bd, 6) + time_remaining = _u16le(bd, 8) + time_to_full = _u16le(bd, 10) + + # Per-cell voltages (4 cells) + cells_mv = [ + _u16le(cd, 0), _u16le(cd, 2), + _u16le(cd, 4), _u16le(cd, 6), + ] + + result: Dict[str, Any] = { + "charge_state": charge_state, + "battery_voltage_mv": batt_voltage_mv, + "battery_current_ma": batt_current_ma, + "battery_percent": batt_percent, + "remaining_capacity_mah": remaining_mah, + "vbus_voltage_mv": vbus_voltage_mv, + "vbus_current_ma": vbus_current_ma, + "vbus_power_mw": vbus_power_mw, + "cell_voltages_mv": cells_mv, + "low_cell_warning": any(0 < v < self.low_cell_mv for v in cells_mv), + } + + # Only include whichever time estimate is relevant + if batt_current_ma < 0: + result["time_to_empty_min"] = time_remaining + else: + result["time_to_full_min"] = time_to_full + + return result + + except Exception as exc: + raise RuntimeError(f"Waveshare UPS HAT (E) read failed: {exc}") from exc