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>
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""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()
|