ha: add li3_battery HA integration, replacing van-li3-battery

The li3 BMS has a persistently marginal BLE link from any single fixed vantage point on this Pi (onboard adapter, a USB dongle that turned out to be BR/EDR-only, and the Athom ESPHome BT proxy even after moving it closer). Move capture into HA proper so bluetooth.async_ble_device_from_address can pick whichever known source currently has the device, instead of hardcoding one. Protocol parsing ported verbatim from li3/van-li3-battery. van-li3-battery is disabled on the Pi; its MQTT discovery entities were cleared. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-24 11:49:44 -04:00
parent 94e6a2d811
commit d25f48727f
9 changed files with 576 additions and 0 deletions
@@ -0,0 +1,98 @@
+"""Sensor platform for the Lithionics Li3 BMS integration."""
+from __future__ import annotations
+
+from homeassistant.components.sensor import SensorDeviceClass, SensorEntity, SensorStateClass
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity import DeviceInfo, EntityCategory
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN, INFO_SENSORS, SENSORS
+from .coordinator import Li3Coordinator
+
+_DEVICE_CLASS_MAP = {
+    "voltage": SensorDeviceClass.VOLTAGE,
+    "current": SensorDeviceClass.CURRENT,
+    "battery": SensorDeviceClass.BATTERY,
+    "temperature": SensorDeviceClass.TEMPERATURE,
+    "duration": SensorDeviceClass.DURATION,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    coordinator: Li3Coordinator = hass.data[DOMAIN][entry.entry_id]
+    device_info = DeviceInfo(
+        identifiers={(DOMAIN, entry.data["address"])},
+        name=entry.data.get("name", "Li3 Battery"),
+        manufacturer="Lithionics",
+        model="Li3 BMS",
+    )
+
+    entities: list[SensorEntity] = [
+        Li3Sensor(coordinator, device_info, key, name, unit, device_class, precision, "data")
+        for key, name, unit, device_class, precision in SENSORS
+    ]
+    entities.append(
+        Li3Sensor(coordinator, device_info, "status_text", "Status", None, None, None, "data")
+    )
+    entities.extend(
+        Li3Sensor(
+            coordinator, device_info, key, name, unit, device_class, precision, "info",
+            diagnostic=True,
+        )
+        for key, name, unit, device_class, precision in INFO_SENSORS
+    )
+    async_add_entities(entities)
+
+
+class Li3Sensor(SensorEntity):
+    """A single field of the Li3 BMS, read live from the coordinator."""
+
+    _attr_should_poll = False
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: Li3Coordinator,
+        device_info: DeviceInfo,
+        key: str,
+        name: str,
+        unit: str | None,
+        device_class: str | None,
+        precision: int | None,
+        source: str,
+        diagnostic: bool = False,
+    ) -> None:
+        self._coordinator = coordinator
+        self._key = key
+        self._source = source
+        self._attr_name = name
+        self._attr_native_unit_of_measurement = unit
+        self._attr_device_class = _DEVICE_CLASS_MAP.get(device_class or "")
+        self._attr_unique_id = f"{coordinator.address}_{key}"
+        self._attr_device_info = device_info
+        if precision is not None:
+            self._attr_suggested_display_precision = precision
+            self._attr_state_class = SensorStateClass.MEASUREMENT
+        if diagnostic:
+            self._attr_entity_category = EntityCategory.DIAGNOSTIC
+
+    def _current_source(self) -> dict:
+        return self._coordinator.data if self._source == "data" else self._coordinator.info
+
+    @property
+    def available(self) -> bool:
+        return self._key in self._current_source()
+
+    @property
+    def native_value(self):
+        return self._current_source().get(self._key)
+
+    async def async_added_to_hass(self) -> None:
+        self.async_on_remove(self._coordinator.async_add_listener(self._handle_update))
+
+    @callback
+    def _handle_update(self) -> None:
+        self.async_write_ha_state()