mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-06-13 01:34:45 +02:00
feat: add LAFVIN UPS Module 3S sensor plugin (lafvin_ups_3s)
Adds a new sensor type for the LAFVIN UPS Module 3S, an INA219-based uninterruptible power supply for Raspberry Pi. Reports pack voltage, current, power, estimated state-of-charge, and charge state for a 3S Li-ion/LiPo battery configuration (9.0–12.6 V range). The INA219 address defaults to 0x41 (as wired on the LAFVIN board) and shunt resistance is configurable for accurate current readings. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,159 @@
|
||||
"""
|
||||
LAFVIN UPS Module 3S battery monitor plug-in — INA219 at I2C 0x41.
|
||||
|
||||
Reads a 3S lithium-ion/LiPo battery pack via INA219. Reports pack voltage,
|
||||
current, power, estimated state-of-charge, and charge state.
|
||||
|
||||
The INA219 address is 0x41 by default (A0 bridged to VCC, A1 to GND on the
|
||||
LAFVIN board) but is configurable for boards wired differently.
|
||||
|
||||
Requires: pip install smbus2
|
||||
|
||||
Config example:
|
||||
- type: lafvin_ups_3s
|
||||
name: "battery"
|
||||
enabled: true
|
||||
auto_install_packages: false
|
||||
settings:
|
||||
i2c_address: 0x41 # Default LAFVIN UPS 3S address
|
||||
bus_number: 1 # I2C bus (1 for Raspberry Pi default)
|
||||
shunt_ohms: 0.1 # Shunt resistor value in ohms
|
||||
max_amps: 5.0 # Maximum expected current in amps
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from .base import SensorBase
|
||||
from .registry import SensorRegistry
|
||||
|
||||
# INA219 register addresses
|
||||
_REG_CONFIG = 0x00
|
||||
_REG_SHUNT = 0x01
|
||||
_REG_BUS = 0x02
|
||||
_REG_POWER = 0x03
|
||||
_REG_CURRENT = 0x04
|
||||
_REG_CALIBRATION = 0x05
|
||||
|
||||
# 32V range, ±320mV shunt gain, 12-bit ADC, continuous shunt+bus
|
||||
_CONFIG_VALUE = 0x399F
|
||||
|
||||
# 3S LiPo/Li-ion pack voltage thresholds (3 cells in series)
|
||||
_V_MAX = 12.6 # 4.20 V/cell × 3 — fully charged
|
||||
_V_MIN = 9.0 # 3.00 V/cell × 3 — cutoff
|
||||
|
||||
|
||||
def _pack_voltage_to_percent(v: float) -> int:
|
||||
"""Piecewise linear SoC estimate for a 3S Li-ion/LiPo pack (9.0–12.6 V)."""
|
||||
cell = v / 3.0
|
||||
if cell >= 4.20: return 100
|
||||
if cell >= 4.00: return int(85 + (cell - 4.00) / 0.20 * 15)
|
||||
if cell >= 3.80: return int(60 + (cell - 3.80) / 0.20 * 25)
|
||||
if cell >= 3.70: return int(40 + (cell - 3.70) / 0.10 * 20)
|
||||
if cell >= 3.50: return int(15 + (cell - 3.50) / 0.20 * 25)
|
||||
if cell >= 3.00: return int( (cell - 3.00) / 0.50 * 15)
|
||||
return 0
|
||||
|
||||
|
||||
@SensorRegistry.register("lafvin_ups_3s")
|
||||
class LafvinUps3sSensor(SensorBase):
|
||||
sensor_type = "lafvin_ups_3s"
|
||||
|
||||
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", 0x41)
|
||||
self.i2c_address = int(addr, 0) if isinstance(addr, str) else int(addr)
|
||||
self.bus_number = int(self.settings.get("bus_number", 1))
|
||||
self.shunt_ohms = float(self.settings.get("shunt_ohms", 0.1))
|
||||
self.max_amps = float(self.settings.get("max_amps", 5.0))
|
||||
|
||||
# INA219 calibration per datasheet
|
||||
self.current_lsb = self.max_amps / 32768.0
|
||||
cal = int(0.04096 / (self.current_lsb * self.shunt_ohms))
|
||||
self.calibration = max(1, min(cal, 0xFFFF))
|
||||
self.power_lsb = self.current_lsb * 20.0
|
||||
|
||||
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:
|
||||
self._write(bus, _REG_CONFIG, _CONFIG_VALUE)
|
||||
self._write(bus, _REG_CALIBRATION, self.calibration)
|
||||
time.sleep(0.15)
|
||||
finally:
|
||||
bus.close()
|
||||
|
||||
self.available = True
|
||||
self.log.info(
|
||||
"LAFVIN UPS 3S INA219 initialized (addr=0x%02X, bus=%d, shunt=%.3fΩ)",
|
||||
self.i2c_address,
|
||||
self.bus_number,
|
||||
self.shunt_ohms,
|
||||
)
|
||||
except Exception as exc:
|
||||
self.log.warning(
|
||||
"LAFVIN UPS 3S init failed (addr=0x%02X, bus=%d): %s",
|
||||
self.i2c_address,
|
||||
self.bus_number,
|
||||
exc,
|
||||
)
|
||||
|
||||
def _write(self, bus, reg: int, val: int) -> None:
|
||||
bus.write_i2c_block_data(
|
||||
self.i2c_address, reg, [(val >> 8) & 0xFF, val & 0xFF]
|
||||
)
|
||||
|
||||
def _read_u(self, bus, reg: int) -> int:
|
||||
d = bus.read_i2c_block_data(self.i2c_address, reg, 2)
|
||||
return (d[0] << 8) | d[1]
|
||||
|
||||
def _read_s(self, bus, reg: int) -> int:
|
||||
v = self._read_u(bus, reg)
|
||||
return v - 0x10000 if v & 0x8000 else v
|
||||
|
||||
def _read(self) -> Dict[str, Any]:
|
||||
"""Read voltage, current, power, and derived battery state."""
|
||||
if not self.available:
|
||||
raise RuntimeError("LAFVIN UPS 3S not available")
|
||||
|
||||
try:
|
||||
bus = self._smbus2.SMBus(self.bus_number)
|
||||
try:
|
||||
self._write(bus, _REG_CALIBRATION, self.calibration)
|
||||
|
||||
bus_v = (self._read_u(bus, _REG_BUS) >> 3) * 4 / 1000.0
|
||||
shunt_mv = self._read_s(bus, _REG_SHUNT) * 0.01
|
||||
current_ma = self._read_s(bus, _REG_CURRENT) * self.current_lsb * 1000.0
|
||||
power_mw = self._read_u(bus, _REG_POWER) * self.power_lsb * 1000.0
|
||||
finally:
|
||||
bus.close()
|
||||
|
||||
pct = _pack_voltage_to_percent(bus_v)
|
||||
|
||||
if current_ma < -50:
|
||||
state = "charging"
|
||||
elif current_ma > 50:
|
||||
state = "discharging"
|
||||
else:
|
||||
state = "idle"
|
||||
|
||||
return {
|
||||
"bus_voltage_v": round(bus_v, 3),
|
||||
"shunt_voltage_mv": round(shunt_mv, 2),
|
||||
"current_ma": round(current_ma, 1),
|
||||
"power_mw": round(power_mw, 1),
|
||||
"battery_percent": pct,
|
||||
"charge_state": state,
|
||||
}
|
||||
except Exception as exc:
|
||||
raise RuntimeError(f"LAFVIN UPS 3S read failed: {exc}") from exc
|
||||
Reference in New Issue
Block a user