From d25f48727f3f8b10b4d5335142a6c799d25d4f1a Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Mon, 24 Aug 2026 11:43:33 -0400 Subject: [PATCH] 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 --- ha/custom_components/li3_battery/__init__.py | 27 ++++ .../li3_battery/config_flow.py | 84 ++++++++++ ha/custom_components/li3_battery/const.py | 36 +++++ .../li3_battery/coordinator.py | 147 ++++++++++++++++++ .../li3_battery/manifest.json | 17 ++ ha/custom_components/li3_battery/parser.py | 133 ++++++++++++++++ ha/custom_components/li3_battery/sensor.py | 98 ++++++++++++ ha/custom_components/li3_battery/strings.json | 17 ++ .../li3_battery/translations/en.json | 17 ++ 9 files changed, 576 insertions(+) create mode 100644 ha/custom_components/li3_battery/__init__.py create mode 100644 ha/custom_components/li3_battery/config_flow.py create mode 100644 ha/custom_components/li3_battery/const.py create mode 100644 ha/custom_components/li3_battery/coordinator.py create mode 100644 ha/custom_components/li3_battery/manifest.json create mode 100644 ha/custom_components/li3_battery/parser.py create mode 100644 ha/custom_components/li3_battery/sensor.py create mode 100644 ha/custom_components/li3_battery/strings.json create mode 100644 ha/custom_components/li3_battery/translations/en.json diff --git a/ha/custom_components/li3_battery/__init__.py b/ha/custom_components/li3_battery/__init__.py new file mode 100644 index 0000000..116ea1a --- /dev/null +++ b/ha/custom_components/li3_battery/__init__.py @@ -0,0 +1,27 @@ +"""The Lithionics Li3 BMS integration.""" +from __future__ import annotations + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant + +from .const import DOMAIN +from .coordinator import Li3Coordinator + +PLATFORMS: list[Platform] = [Platform.SENSOR] + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + coordinator = Li3Coordinator(hass, entry.data["address"]) + await coordinator.async_start() + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + if unload_ok: + coordinator: Li3Coordinator = hass.data[DOMAIN].pop(entry.entry_id) + await coordinator.async_stop() + return unload_ok diff --git a/ha/custom_components/li3_battery/config_flow.py b/ha/custom_components/li3_battery/config_flow.py new file mode 100644 index 0000000..8eb98b8 --- /dev/null +++ b/ha/custom_components/li3_battery/config_flow.py @@ -0,0 +1,84 @@ +"""Config flow for the Lithionics Li3 BMS integration.""" +from __future__ import annotations + +from typing import Any + +import voluptuous as vol + +from homeassistant.components.bluetooth import ( + BluetoothServiceInfoBleak, + async_discovered_service_info, +) +from homeassistant.config_entries import ConfigFlow +from homeassistant.data_entry_flow import FlowResult + +from .const import DOMAIN + + +class Li3ConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for a Lithionics Li3 BMS.""" + + VERSION = 1 + + def __init__(self) -> None: + self._discovery_info: BluetoothServiceInfoBleak | None = None + self._discovered: dict[str, str] = {} + + async def async_step_bluetooth( + self, discovery_info: BluetoothServiceInfoBleak + ) -> FlowResult: + """Handle a discovered Li3 advertisement (from any Bluetooth source).""" + await self.async_set_unique_id(discovery_info.address) + self._abort_if_unique_id_configured() + self._discovery_info = discovery_info + self.context["title_placeholders"] = {"name": discovery_info.name} + return await self.async_step_bluetooth_confirm() + + async def async_step_bluetooth_confirm( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + assert self._discovery_info is not None + if user_input is not None: + return self.async_create_entry( + title=self._discovery_info.name, + data={ + "address": self._discovery_info.address, + "name": self._discovery_info.name, + }, + ) + return self.async_show_form( + step_id="bluetooth_confirm", + description_placeholders={"name": self._discovery_info.name}, + ) + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Manual entry, plus a dropdown of any Li3 already seen advertising.""" + errors: dict[str, str] = {} + if user_input is not None: + address = user_input["address"] + await self.async_set_unique_id(address, raise_on_progress=False) + self._abort_if_unique_id_configured() + name = self._discovered.get(address, "Li3 Battery") + return self.async_create_entry(title=name, data={"address": address, "name": name}) + + current_addresses = self._async_current_ids() + for info in async_discovered_service_info(self.hass, connectable=True): + if info.address in current_addresses: + continue + if info.name and info.name.startswith("Li3-"): + self._discovered[info.address] = info.name + + if not self._discovered: + return self.async_show_form( + step_id="user", + data_schema=vol.Schema({vol.Required("address"): str}), + errors=errors, + ) + + return self.async_show_form( + step_id="user", + data_schema=vol.Schema({vol.Required("address"): vol.In(self._discovered)}), + errors=errors, + ) diff --git a/ha/custom_components/li3_battery/const.py b/ha/custom_components/li3_battery/const.py new file mode 100644 index 0000000..5ccf525 --- /dev/null +++ b/ha/custom_components/li3_battery/const.py @@ -0,0 +1,36 @@ +"""Constants for the Lithionics Li3 BMS integration.""" + +DOMAIN = "li3_battery" + +# (key, name, unit, device_class, display_precision) -- precision None means a +# non-numeric value (hex code, version string, serial number): published as-is, +# no rounding, no state_class. Mirrors van-li3-battery's SENSORS list. +SENSORS = [ + ("voltage", "Pack Voltage", "V", "voltage", 2), + ("cell1_voltage", "Cell 1 Voltage", "V", "voltage", 2), + ("cell2_voltage", "Cell 2 Voltage", "V", "voltage", 2), + ("cell3_voltage", "Cell 3 Voltage", "V", "voltage", 2), + ("cell4_voltage", "Cell 4 Voltage", "V", "voltage", 2), + ("current", "Current", "A", "current", 2), + ("soc", "State of Charge", "%", "battery", 0), + ("bms_temperature", "BMS Temperature", "°F", "temperature", 1), + ("battery_temperature", "Battery Temperature", "°F", "temperature", 1), + ("remaining_capacity", "Remaining Capacity", "Ah", None, 0), + ("remaining_time", "Remaining Time", "min", "duration", 0), + ("can_charger_voltage", "CAN Charger Voltage", "V", "voltage", 1), + ("can_charger_current", "CAN Charger Current", "A", "current", 1), + ("can_charger_status", "CAN Charger Status", None, None, None), + ("can_status", "CAN Status", None, None, None), +] + +# From the one-time "$info" response -- published as diagnostic entities. +INFO_SENSORS = [ + ("total_consumed", "Lifetime Consumed", "Ah", None, 0), + ("last_fault_code", "Last Fault Code", None, None, None), + ("highest_recorded_temp", "Highest Recorded Temperature", "°F", "temperature", 0), + ("lowest_recorded_temp", "Lowest Recorded Temperature", "°F", "temperature", 0), + ("firmware_version", "Firmware Version", None, None, None), + ("aging_factor_temp", "Aging Factor (Temp)", None, None, 0), + ("aging_factor_soc", "Aging Factor (SOC)", None, None, 0), + ("serial_number", "Serial Number", None, None, None), +] diff --git a/ha/custom_components/li3_battery/coordinator.py b/ha/custom_components/li3_battery/coordinator.py new file mode 100644 index 0000000..14b59de --- /dev/null +++ b/ha/custom_components/li3_battery/coordinator.py @@ -0,0 +1,147 @@ +"""Persistent BLE connection to one Li3 BMS. + +Unlike a typical HA polling coordinator, the Li3 streams telemetry +continuously once connected, so this holds a long-lived connection instead of +connect/read/disconnect cycles. bluetooth.async_ble_device_from_address picks +whichever known Bluetooth source (the host's local adapter, or any connected +ESPHome Bluetooth proxy) currently has the device, so a single weak vantage +point no longer has to carry the whole link -- see vanlink's li3-battery +project memory for why that matters here. +""" +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable + +from bleak import BleakClient +from bleak.exc import BleakError +from bleak_retry_connector import establish_connection + +from homeassistant.components import bluetooth +from homeassistant.core import HomeAssistant, callback + +from .parser import ( + FFE1_CHAR_UUID, + build_state_payload, + parse_info_line, + parse_line, + parse_trace_line, +) + +_LOGGER = logging.getLogger(__name__) + +RETRY_DELAY_S = 8 + + +class Li3Coordinator: + """Owns the BLE connection to one Li3 BMS and fans out updates to entities.""" + + def __init__(self, hass: HomeAssistant, address: str) -> None: + self.hass = hass + self.address = address + self.data: dict = {} + self.info: dict = {} + self.available = False + self._listeners: list[Callable[[], None]] = [] + self._task: asyncio.Task | None = None + self._stopping = False + + @callback + def async_add_listener(self, update_callback: Callable[[], None]) -> Callable[[], None]: + self._listeners.append(update_callback) + + def remove_listener() -> None: + self._listeners.remove(update_callback) + + return remove_listener + + def _notify_listeners(self) -> None: + for update_callback in list(self._listeners): + update_callback() + + async def async_start(self) -> None: + self._stopping = False + self._task = self.hass.loop.create_task(self._run()) + + async def async_stop(self) -> None: + self._stopping = True + if self._task is not None: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + + async def _run(self) -> None: + while not self._stopping: + ble_device = bluetooth.async_ble_device_from_address( + self.hass, self.address, connectable=True + ) + if ble_device is None: + _LOGGER.debug( + "Li3 %s not currently visible to any Bluetooth source", self.address + ) + await asyncio.sleep(RETRY_DELAY_S) + continue + try: + await self._stream(ble_device) + except asyncio.CancelledError: + raise + except (BleakError, EOFError, TimeoutError) as err: + _LOGGER.debug("Li3 %s connection error: %s", self.address, err) + self.available = False + self._notify_listeners() + await asyncio.sleep(RETRY_DELAY_S) + + async def _stream(self, ble_device) -> None: + buf = "" + info_published = False + latest_cs: dict = {} + latest_trace: dict = {} + + def notify_handler(_sender, data: bytearray) -> None: + nonlocal buf, info_published + buf += data.decode("utf-8", errors="replace") + while "\r\n" in buf: + line, buf = buf.split("\r\n", 1) + line = line.strip() + if not line: + continue + if line.startswith("&"): + trace = parse_trace_line(line) + if trace: + latest_trace.update(trace) + if latest_cs: + self.data = build_state_payload(latest_cs, latest_trace) + self._notify_listeners() + elif line.startswith("$"): + if not info_published: + info = parse_info_line(line) + if info: + self.info = info + info_published = True + self._notify_listeners() + else: + reading = parse_line(line) + if reading: + latest_cs.update(reading) + self.data = build_state_payload(latest_cs, latest_trace) + self._notify_listeners() + + _LOGGER.debug("Connecting to Li3 %s", self.address) + client = await establish_connection(BleakClient, ble_device, ble_device.address) + try: + self.available = True + self._notify_listeners() + await client.start_notify(FFE1_CHAR_UUID, notify_handler) + await client.write_gatt_char(FFE1_CHAR_UUID, b"$traceon\r\n", response=False) + await asyncio.sleep(2) + await client.write_gatt_char(FFE1_CHAR_UUID, b"$info\r\n", response=False) + + while client.is_connected and not self._stopping: + await asyncio.sleep(1) + finally: + self.available = False + if client.is_connected: + await client.disconnect() diff --git a/ha/custom_components/li3_battery/manifest.json b/ha/custom_components/li3_battery/manifest.json new file mode 100644 index 0000000..6765753 --- /dev/null +++ b/ha/custom_components/li3_battery/manifest.json @@ -0,0 +1,17 @@ +{ + "domain": "li3_battery", + "name": "Lithionics Li3 BMS", + "codeowners": ["@aew"], + "config_flow": true, + "dependencies": ["bluetooth"], + "documentation": "https://github.com/wrede/vanlink", + "iot_class": "local_push", + "requirements": ["bleak-retry-connector>=3.0.0"], + "version": "0.1.0", + "bluetooth": [ + { + "local_name": "Li3-*", + "connectable": true + } + ] +} diff --git a/ha/custom_components/li3_battery/parser.py b/ha/custom_components/li3_battery/parser.py new file mode 100644 index 0000000..3510928 --- /dev/null +++ b/ha/custom_components/li3_battery/parser.py @@ -0,0 +1,133 @@ +"""Lithionics Li3 BMS wire protocol. + +Ported verbatim from vanlink's van-li3-battery (li3/van-li3-battery in the +vanlink repo). Protocol reverse-engineered from the com.lithionics.bms Android +app (BLEMaster / MainBmsCsParameters): classic HM-10 BLE-UART (service ffe0, +characteristic ffe1, notify+write, no pairing). On connect we send "$traceon" +then "$info"; the device then streams CRLF-terminated CSV telemetry lines +forever. +""" +from __future__ import annotations + +FFE1_CHAR_UUID = "0000ffe1-0000-1000-8000-00805f9b34fb" + +# The "status" field (main Cs telemetry line, and identically-coded but +# cumulative/latched "last_fault_code" from $info) is a 24-bit flag mask. Bit +# meanings pulled from the app's own "advanced" string-array resource +# (com.lithionics.bms base.apk, array/advanced -- dumped with aapt since the +# app's Kotlin StatusCodeTable class references stale/wrong resource IDs and +# can't be trusted). Index 0 = bit 23 (MSB) down to index 23 = bit 0 (LSB); +# blank entries are unused bits. Applied to "status" only -- last_fault_code +# is a lifetime latch (many bits accumulate over time) and isn't meaningfully +# summarized the same way. +STATUS_FLAGS = [ + "", "BMS Temp High", "Overcurrent State", "Charge OFF", "Aux Input State", + "Cell Temp Low", "Cell Temp High", "AGSR State", "Temp Sensor Fault", + "CAN Charger Fault", "CAN Charger Present", "AC Power Present", + "Contactor Flutter", "Pre-Charge Fault", "Contactor Fault", + "Contactor State", "Power Off State", "Battery Protection", "Low Voltage", + "Reserve Range", "OptoLoop Open", "NeverDie Reserve", "Charge Detected", + "High Voltage", +] + + +def decode_status(hex_code): + try: + value = int(hex_code, 16) + except (ValueError, TypeError): + return hex_code + active = [ + label for i, label in enumerate(STATUS_FLAGS) + if label and (value >> (23 - i)) & 1 + ] + return ", ".join(active) if active else "OK" + + +def parse_line(line): + """Cs-series main telemetry line (no prefix character).""" + parts = line.split(",") + try: + f0 = int(parts[0]) + except ValueError: + return None + if not (101 <= f0 <= 9999): + return None # not a Cs-series main telemetry line + try: + return { + "voltage": round(f0 * 0.01, 2), + "cell1": round(int(parts[1]) * 0.01, 2), + "cell2": round(int(parts[2]) * 0.01, 2), + "cell3": round(int(parts[3]) * 0.01, 2), + "cell4": round(int(parts[4]) * 0.01, 2), + "bms_temp_f": int(parts[5]), + "batt_temp_f": int(parts[6]), + "current_a": int(parts[7]), + "soc_pct": int(parts[8]), + "status": parts[9] if len(parts) > 9 else "?", + } + except (ValueError, IndexError): + return None + + +def parse_trace_line(line): + """'&' trace line: &,batteryId,remaining,remainingTime,canChargerVoltage, + canChargerCurrent,canChargerStatus,canStatus""" + parts = line.split(",") + try: + return { + "remaining_capacity": int(parts[2]), + "remaining_time": int(parts[3]), + "can_charger_voltage": round(int(parts[4]) * 0.1, 1), + "can_charger_current": round(int(parts[5]) * 0.1, 1), + "can_charger_status": parts[6], + "can_status": parts[7], + } + except (ValueError, IndexError): + return None + + +def parse_info_line(line): + """'$' info line (response to $info): $,totalConsumed,lastFaultCode, + highestRecordedTemp,lowestRecordedTemp,firmwareVersion,agingFactorTemp, + agingFactorSoc,serialNumber""" + parts = line.split(",") + try: + return { + "total_consumed": int(parts[1]), + "last_fault_code": parts[2], + "highest_recorded_temp": int(parts[3]), + "lowest_recorded_temp": int(parts[4]), + "firmware_version": parts[5], + "aging_factor_temp": int(parts[6]), + "aging_factor_soc": int(parts[7]), + "serial_number": parts[8], + } + except (ValueError, IndexError): + return None + + +def build_state_payload(cs_reading, trace_fields): + # A non-zero status means the rest of the Cs line's fields are unreliable + # (observed 2026-08-18: status '69' alongside e.g. current=341, soc=340, + # cell4_voltage=34013.63) -- publish only the status in that case. Trace + # ("&" line) fields come from a separate message and are published + # regardless. + status = cs_reading["status"] + if status != "000000": + payload = {"status": status} + else: + payload = { + "voltage": cs_reading["voltage"], + "cell1_voltage": cs_reading["cell1"], + "cell2_voltage": cs_reading["cell2"], + "cell3_voltage": cs_reading["cell3"], + "cell4_voltage": cs_reading["cell4"], + "current": cs_reading["current_a"], + "soc": cs_reading["soc_pct"], + "bms_temperature": cs_reading["bms_temp_f"], + "battery_temperature": cs_reading["batt_temp_f"], + "status": status, + } + payload["status_text"] = decode_status(status) + payload.update(trace_fields) + return payload diff --git a/ha/custom_components/li3_battery/sensor.py b/ha/custom_components/li3_battery/sensor.py new file mode 100644 index 0000000..c3986fe --- /dev/null +++ b/ha/custom_components/li3_battery/sensor.py @@ -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() diff --git a/ha/custom_components/li3_battery/strings.json b/ha/custom_components/li3_battery/strings.json new file mode 100644 index 0000000..b4390d9 --- /dev/null +++ b/ha/custom_components/li3_battery/strings.json @@ -0,0 +1,17 @@ +{ + "config": { + "step": { + "bluetooth_confirm": { + "description": "Add the Li3 battery `{name}`?" + }, + "user": { + "data": { + "address": "Device" + } + } + }, + "abort": { + "already_configured": "This battery is already configured" + } + } +} diff --git a/ha/custom_components/li3_battery/translations/en.json b/ha/custom_components/li3_battery/translations/en.json new file mode 100644 index 0000000..b4390d9 --- /dev/null +++ b/ha/custom_components/li3_battery/translations/en.json @@ -0,0 +1,17 @@ +{ + "config": { + "step": { + "bluetooth_confirm": { + "description": "Add the Li3 battery `{name}`?" + }, + "user": { + "data": { + "address": "Device" + } + } + }, + "abort": { + "already_configured": "This battery is already configured" + } + } +}