Full protocol coverage: binary sensors, energy, hardware version

Sensors added:
- energy_stored (kWh = V × Ah / 1000) for energy dashboard

Binary sensors added (all from existing 0x03 frame, no extra BLE requests):
- Charge MOSFET / Discharge MOSFET (MOS gate status)
- Cell Balancing (any balance bit active)
- 13× protection flags: cell/pack over/under-voltage, charge/discharge
  over/under-temperature, charge/discharge over-current, short circuit,
  frontend IC error, software lock

Other:
- Hardware version string fetched once via CMD 0x05, shown in device card
- DeviceInfo centralised on coordinator (sensor + binary_sensor share it)
- CONF_ADDRESS removed from sensor.py (coordinator holds address)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-11 19:52:10 +02:00
parent e9132c33a7
commit 5d527168e2
6 changed files with 240 additions and 38 deletions
+1 -1
View File
@@ -7,7 +7,7 @@ from homeassistant.core import HomeAssistant
from .const import CONF_ADDRESS, CONF_POLL_INTERVAL, DEFAULT_POLL_INTERVAL, DOMAIN from .const import CONF_ADDRESS, CONF_POLL_INTERVAL, DEFAULT_POLL_INTERVAL, DOMAIN
from .coordinator import BmsCoordinator from .coordinator import BmsCoordinator
PLATFORMS = ["sensor"] PLATFORMS = ["sensor", "binary_sensor"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -0,0 +1,161 @@
"""Binary sensor entities for the Xiaoxiang Smart BMS integration."""
from __future__ import annotations
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import BmsCoordinator
# ---------------------------------------------------------------------------
# Entity descriptions
# ---------------------------------------------------------------------------
BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = (
# -- MOS gate status --------------------------------------------------
BinarySensorEntityDescription(
key="mos_charge_enabled",
name="Charge MOSFET",
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
icon="mdi:battery-charging",
),
BinarySensorEntityDescription(
key="mos_discharge_enabled",
name="Discharge MOSFET",
device_class=BinarySensorDeviceClass.POWER,
icon="mdi:power-plug",
),
# -- Cell balancing ---------------------------------------------------
BinarySensorEntityDescription(
key="balance_active",
name="Cell Balancing",
device_class=BinarySensorDeviceClass.RUNNING,
icon="mdi:battery-sync",
),
# -- Protection flags (True = protection triggered = problem) ---------
BinarySensorEntityDescription(
key="prot_cell_overvolt",
name="Cell Over-Voltage",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:battery-arrow-up",
),
BinarySensorEntityDescription(
key="prot_cell_undervolt",
name="Cell Under-Voltage",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:battery-arrow-down",
),
BinarySensorEntityDescription(
key="prot_pack_overvolt",
name="Pack Over-Voltage",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:battery-arrow-up",
),
BinarySensorEntityDescription(
key="prot_pack_undervolt",
name="Pack Under-Voltage",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:battery-arrow-down",
),
BinarySensorEntityDescription(
key="prot_charge_overtemp",
name="Charge Over-Temperature",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:thermometer-alert",
),
BinarySensorEntityDescription(
key="prot_charge_undertemp",
name="Charge Under-Temperature",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:thermometer-alert",
),
BinarySensorEntityDescription(
key="prot_discharge_overtemp",
name="Discharge Over-Temperature",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:thermometer-alert",
),
BinarySensorEntityDescription(
key="prot_discharge_undertemp",
name="Discharge Under-Temperature",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:thermometer-alert",
),
BinarySensorEntityDescription(
key="prot_charge_overcurrent",
name="Charge Over-Current",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:current-ac",
),
BinarySensorEntityDescription(
key="prot_discharge_overcurrent",
name="Discharge Over-Current",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:current-ac",
),
BinarySensorEntityDescription(
key="prot_short_circuit",
name="Short Circuit",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:flash-alert",
),
BinarySensorEntityDescription(
key="prot_frontend_ic_error",
name="Frontend IC Error",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:chip",
),
BinarySensorEntityDescription(
key="prot_software_lock",
name="Software Lock",
device_class=BinarySensorDeviceClass.PROBLEM,
icon="mdi:lock-alert",
),
)
# ---------------------------------------------------------------------------
# Platform setup
# ---------------------------------------------------------------------------
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
coordinator: BmsCoordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
BmsBinarySensor(coordinator, description)
for description in BINARY_SENSORS
)
# ---------------------------------------------------------------------------
# Entity class
# ---------------------------------------------------------------------------
class BmsBinarySensor(CoordinatorEntity[BmsCoordinator], BinarySensorEntity):
"""A binary sensor backed by a boolean key in coordinator.data."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: BmsCoordinator,
description: BinarySensorEntityDescription,
) -> None:
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.address}_{description.key}"
self._attr_device_info = coordinator.device_info
@property
def is_on(self) -> bool | None:
return self.coordinator.data.get(self.entity_description.key)
@@ -172,6 +172,10 @@ class BmsBluetoothHandler:
raw = struct.unpack_from(">H", p, 23 + i * 2)[0] raw = struct.unpack_from(">H", p, 23 + i * 2)[0]
temperatures.append(round((raw - 2731) / 10.0, 1)) temperatures.append(round((raw - 2731) / 10.0, 1))
balance = struct.unpack_from(">H", p, 12)[0] | struct.unpack_from(">H", p, 14)[0]
prot = struct.unpack_from(">H", p, 16)[0]
mos = p[20]
return { return {
"voltage": round(struct.unpack_from(">H", p, 0)[0] / 100.0, 2), "voltage": round(struct.unpack_from(">H", p, 0)[0] / 100.0, 2),
"current": round(struct.unpack_from(">h", p, 2)[0] / 100.0, 2), "current": round(struct.unpack_from(">h", p, 2)[0] / 100.0, 2),
@@ -181,8 +185,33 @@ class BmsBluetoothHandler:
"state_of_charge": p[19], "state_of_charge": p[19],
"cell_count": p[21], "cell_count": p[21],
"temperatures": temperatures, "temperatures": temperatures,
# MOS status
"mos_charge_enabled": bool(mos & 0x01),
"mos_discharge_enabled": bool(mos & 0x02),
# Cell balancing (any cell currently balancing)
"balance_active": balance != 0,
# Protection flags (bit per event, True = protection triggered)
"prot_cell_overvolt": bool(prot & (1 << 0)),
"prot_cell_undervolt": bool(prot & (1 << 1)),
"prot_pack_overvolt": bool(prot & (1 << 2)),
"prot_pack_undervolt": bool(prot & (1 << 3)),
"prot_charge_overtemp": bool(prot & (1 << 4)),
"prot_charge_undertemp": bool(prot & (1 << 5)),
"prot_discharge_overtemp": bool(prot & (1 << 6)),
"prot_discharge_undertemp": bool(prot & (1 << 7)),
"prot_charge_overcurrent": bool(prot & (1 << 8)),
"prot_discharge_overcurrent": bool(prot & (1 << 9)),
"prot_short_circuit": bool(prot & (1 << 10)),
"prot_frontend_ic_error": bool(prot & (1 << 11)),
"prot_software_lock": bool(prot & (1 << 12)),
} }
@staticmethod
def parse_version(frame: bytes) -> str:
"""Parse a 0x05 hardware version response frame into an ASCII string."""
p = frame[_HEADER_LEN:-_TRAILER_LEN]
return p.decode("ascii", errors="replace").strip("\x00").strip()
@staticmethod @staticmethod
def parse_cell_info(frame: bytes) -> dict: def parse_cell_info(frame: bytes) -> dict:
"""Parse a 0x04 cell voltage response frame. """Parse a 0x04 cell voltage response frame.
+2
View File
@@ -17,6 +17,7 @@ TX_CHAR_UUID = "0000ff02-0000-1000-8000-00805f9b34fb" # HA → BMS (write)
# Request frames: [0xDD, 0xA5, CMD, 0x00, CHK_HI, CHK_LO, 0x77] # Request frames: [0xDD, 0xA5, CMD, 0x00, CHK_HI, CHK_LO, 0x77]
CMD_GENERAL = bytes([0xDD, 0xA5, 0x03, 0x00, 0xFF, 0xFD, 0x77]) # pack info CMD_GENERAL = bytes([0xDD, 0xA5, 0x03, 0x00, 0xFF, 0xFD, 0x77]) # pack info
CMD_CELL = bytes([0xDD, 0xA5, 0x04, 0x00, 0xFF, 0xFC, 0x77]) # cell voltages CMD_CELL = bytes([0xDD, 0xA5, 0x04, 0x00, 0xFF, 0xFC, 0x77]) # cell voltages
CMD_VERSION = bytes([0xDD, 0xA5, 0x05, 0x00, 0xFF, 0xFB, 0x77]) # hardware version string
# Frame markers # Frame markers
FRAME_START = 0xDD FRAME_START = 0xDD
@@ -25,3 +26,4 @@ FRAME_END = 0x77
# Response command IDs (byte 1 of frame) # Response command IDs (byte 1 of frame)
CMD_ID_GENERAL = 0x03 CMD_ID_GENERAL = 0x03
CMD_ID_CELL = 0x04 CMD_ID_CELL = 0x04
CMD_ID_VERSION = 0x05
+28 -4
View File
@@ -6,10 +6,11 @@ from datetime import timedelta
from homeassistant.components.bluetooth import async_ble_device_from_address from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .bluetooth_handler import BmsBluetoothHandler from .bluetooth_handler import BmsBluetoothHandler
from .const import CMD_CELL, CMD_GENERAL, DOMAIN from .const import CMD_CELL, CMD_GENERAL, CMD_VERSION, DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -31,6 +32,20 @@ class BmsCoordinator(DataUpdateCoordinator[dict]):
) )
self.address = address self.address = address
self._handler = BmsBluetoothHandler(address) self._handler = BmsBluetoothHandler(address)
self.hw_version: str | None = None # populated on first successful poll
# ------------------------------------------------------------------
# Device info — centralised so sensor + binary_sensor share it
# ------------------------------------------------------------------
@property
def device_info(self) -> DeviceInfo:
return DeviceInfo(
identifiers={(DOMAIN, self.address)},
name="Xiaoxiang Smart BMS",
manufacturer="Xiaoxiang",
model=self.hw_version or "Smart BMS",
)
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Lifecycle # Lifecycle
@@ -39,8 +54,7 @@ class BmsCoordinator(DataUpdateCoordinator[dict]):
def _get_ble_device(self): def _get_ble_device(self):
"""Resolve the best available BLE device via HA's Bluetooth subsystem. """Resolve the best available BLE device via HA's Bluetooth subsystem.
This automatically uses ESPHome BLE proxies if they can reach the BMS Automatically uses ESPHome BLE proxies if they can reach the BMS.
and the local adapter cannot (or vice versa).
""" """
device = async_ble_device_from_address(self.hass, self.address, connectable=True) device = async_ble_device_from_address(self.hass, self.address, connectable=True)
if device is None: if device is None:
@@ -79,6 +93,14 @@ class BmsCoordinator(DataUpdateCoordinator[dict]):
cell_frame = await self._handler.request(CMD_CELL) cell_frame = await self._handler.request(CMD_CELL)
if cell_frame is None: if cell_frame is None:
raise UpdateFailed("No response to cell info request (0x04)") raise UpdateFailed("No response to cell info request (0x04)")
# Fetch hardware version string once — used in DeviceInfo model field
if self.hw_version is None:
version_frame = await self._handler.request(CMD_VERSION)
if version_frame:
self.hw_version = BmsBluetoothHandler.parse_version(version_frame)
_LOGGER.debug("BMS hardware version: %s", self.hw_version)
except UpdateFailed: except UpdateFailed:
raise raise
except Exception as exc: except Exception as exc:
@@ -89,6 +111,7 @@ class BmsCoordinator(DataUpdateCoordinator[dict]):
# Derived fields # Derived fields
data["power"] = round(data["voltage"] * data["current"], 2) data["power"] = round(data["voltage"] * data["current"], 2)
data["energy_stored"] = round(data["voltage"] * data["residual_capacity"] / 1000, 3)
if data["cell_voltages"]: if data["cell_voltages"]:
v_max = max(data["cell_voltages"]) v_max = max(data["cell_voltages"])
@@ -98,11 +121,12 @@ class BmsCoordinator(DataUpdateCoordinator[dict]):
data["cell_delta"] = None data["cell_delta"] = None
_LOGGER.debug( _LOGGER.debug(
"BMS data: %.2fV %.2fA %d%% %.2fAh %d cells", "BMS data: %.2fV %.2fA %d%% %.2fAh %.3fkWh %d cells",
data["voltage"], data["voltage"],
data["current"], data["current"],
data["state_of_charge"], data["state_of_charge"],
data["residual_capacity"], data["residual_capacity"],
data["energy_stored"],
len(data["cell_voltages"]), len(data["cell_voltages"]),
) )
return data return data
+19 -33
View File
@@ -12,19 +12,19 @@ from homeassistant.const import (
PERCENTAGE, PERCENTAGE,
UnitOfElectricCurrent, UnitOfElectricCurrent,
UnitOfElectricPotential, UnitOfElectricPotential,
UnitOfEnergy,
UnitOfPower, UnitOfPower,
UnitOfTemperature, UnitOfTemperature,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_ADDRESS, DOMAIN from .const import DOMAIN
from .coordinator import BmsCoordinator from .coordinator import BmsCoordinator
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Static sensor definitions — one entity per key in coordinator.data # Static sensor definitions — one entity per scalar key in coordinator.data
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
STATIC_SENSORS: tuple[SensorEntityDescription, ...] = ( STATIC_SENSORS: tuple[SensorEntityDescription, ...] = (
@@ -52,6 +52,15 @@ STATIC_SENSORS: tuple[SensorEntityDescription, ...] = (
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=1, suggested_display_precision=1,
), ),
SensorEntityDescription(
key="energy_stored",
name="Energy Stored",
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
device_class=SensorDeviceClass.ENERGY_STORAGE,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=3,
icon="mdi:battery-charging",
),
SensorEntityDescription( SensorEntityDescription(
key="state_of_charge", key="state_of_charge",
name="State of Charge", name="State of Charge",
@@ -103,10 +112,9 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Create all BMS sensor entities after the first successful poll.""" """Create all BMS sensor entities after the first successful poll."""
coordinator: BmsCoordinator = hass.data[DOMAIN][entry.entry_id] coordinator: BmsCoordinator = hass.data[DOMAIN][entry.entry_id]
address: str = entry.data[CONF_ADDRESS]
entities: list[SensorEntity] = [ entities: list[SensorEntity] = [
BmsStaticSensor(coordinator, description, address) BmsStaticSensor(coordinator, description)
for description in STATIC_SENSORS for description in STATIC_SENSORS
] ]
@@ -115,14 +123,12 @@ async def async_setup_entry(
entities.append( entities.append(
BmsDynamicSensor( BmsDynamicSensor(
coordinator=coordinator, coordinator=coordinator,
address=address,
unique_key=f"temperature_{i + 1}", unique_key=f"temperature_{i + 1}",
friendly_name=f"Temperature {i + 1}", friendly_name=f"Temperature {i + 1}",
data_key="temperatures", data_key="temperatures",
index=i, index=i,
unit=UnitOfTemperature.CELSIUS, unit=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE, device_class=SensorDeviceClass.TEMPERATURE,
icon=None,
) )
) )
@@ -131,33 +137,18 @@ async def async_setup_entry(
entities.append( entities.append(
BmsDynamicSensor( BmsDynamicSensor(
coordinator=coordinator, coordinator=coordinator,
address=address,
unique_key=f"cell_voltage_{i + 1}", unique_key=f"cell_voltage_{i + 1}",
friendly_name=f"Cell {i + 1} Voltage", friendly_name=f"Cell {i + 1} Voltage",
data_key="cell_voltages", data_key="cell_voltages",
index=i, index=i,
unit=UnitOfElectricPotential.VOLT, unit=UnitOfElectricPotential.VOLT,
device_class=SensorDeviceClass.VOLTAGE, device_class=SensorDeviceClass.VOLTAGE,
icon=None,
) )
) )
async_add_entities(entities) async_add_entities(entities)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _device_info(address: str) -> DeviceInfo:
return DeviceInfo(
identifiers={(DOMAIN, address)},
name="Xiaoxiang Smart BMS",
manufacturer="Xiaoxiang",
model="Smart BMS",
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Entity classes # Entity classes
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -171,12 +162,11 @@ class BmsStaticSensor(CoordinatorEntity[BmsCoordinator], SensorEntity):
self, self,
coordinator: BmsCoordinator, coordinator: BmsCoordinator,
description: SensorEntityDescription, description: SensorEntityDescription,
address: str,
) -> None: ) -> None:
super().__init__(coordinator) super().__init__(coordinator)
self.entity_description = description self.entity_description = description
self._attr_unique_id = f"{address}_{description.key}" self._attr_unique_id = f"{coordinator.address}_{description.key}"
self._attr_device_info = _device_info(address) self._attr_device_info = coordinator.device_info
@property @property
def native_value(self): def native_value(self):
@@ -184,9 +174,9 @@ class BmsStaticSensor(CoordinatorEntity[BmsCoordinator], SensorEntity):
class BmsDynamicSensor(CoordinatorEntity[BmsCoordinator], SensorEntity): class BmsDynamicSensor(CoordinatorEntity[BmsCoordinator], SensorEntity):
"""A sensor that reads an element from a list inside coordinator.data. """A sensor that reads an indexed element from a list inside coordinator.data.
Used for per-cell voltages and per-probe temperatures, whose count is Used for per-cell voltages and per-probe temperatures whose count is
only known after the first successful BMS poll. only known after the first successful BMS poll.
""" """
@@ -196,25 +186,21 @@ class BmsDynamicSensor(CoordinatorEntity[BmsCoordinator], SensorEntity):
def __init__( def __init__(
self, self,
coordinator: BmsCoordinator, coordinator: BmsCoordinator,
address: str,
unique_key: str, unique_key: str,
friendly_name: str, friendly_name: str,
data_key: str, data_key: str,
index: int, index: int,
unit: str, unit: str,
device_class: SensorDeviceClass | None, device_class: SensorDeviceClass | None,
icon: str | None,
) -> None: ) -> None:
super().__init__(coordinator) super().__init__(coordinator)
self._data_key = data_key self._data_key = data_key
self._index = index self._index = index
self._attr_unique_id = f"{address}_{unique_key}" self._attr_unique_id = f"{coordinator.address}_{unique_key}"
self._attr_name = friendly_name self._attr_name = friendly_name
self._attr_native_unit_of_measurement = unit self._attr_native_unit_of_measurement = unit
self._attr_device_class = device_class self._attr_device_class = device_class
self._attr_device_info = _device_info(address) self._attr_device_info = coordinator.device_info
if icon:
self._attr_icon = icon
if device_class == SensorDeviceClass.VOLTAGE: if device_class == SensorDeviceClass.VOLTAGE:
self._attr_suggested_display_precision = 3 self._attr_suggested_display_precision = 3
elif device_class == SensorDeviceClass.TEMPERATURE: elif device_class == SensorDeviceClass.TEMPERATURE: