Initial commit: Xiaoxiang Smart BMS Home Assistant integration

- BLE GATT communication via bleak (UART-over-GATT, service 0xff00)
- Parses 0x03 (general info) and 0x04 (cell voltages) frames
- Protocol verified against official RS485/UART spec V4
- DataUpdateCoordinator with 5s poll interval (configurable 2-60s)
- Auto-reconnect on BLE disconnect
- Config flow: BT auto-discovery + manual MAC entry + options flow
- Sensors: voltage, current, power, SoC, capacity, cycles, temps, cells, cell delta
- HACS-ready (hacs.json, manifest.json)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-11 19:07:18 +02:00
commit 4a769bf50f
14 changed files with 1333 additions and 0 deletions
@@ -0,0 +1,130 @@
"""Config flow for the Xiaoxiang Smart BMS integration."""
from __future__ import annotations
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from .const import (
CONF_ADDRESS,
CONF_POLL_INTERVAL,
DEFAULT_POLL_INTERVAL,
DOMAIN,
MAX_POLL_INTERVAL,
MIN_POLL_INTERVAL,
UART_SERVICE_UUID,
)
class XiaoxiangBmsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for the Xiaoxiang Smart BMS."""
VERSION = 1
def __init__(self) -> None:
self._address: str | None = None
self._name: str | None = None
# ------------------------------------------------------------------
# Auto-discovery path (HA sees BMS advertising matching service UUID)
# ------------------------------------------------------------------
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
) -> FlowResult:
"""Handle BLE auto-discovery."""
self._address = discovery_info.address
self._name = discovery_info.name or f"Xiaoxiang BMS ({discovery_info.address})"
await self.async_set_unique_id(discovery_info.address)
self._abort_if_unique_id_configured()
self.context["title_placeholders"] = {"name": self._name}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input: dict | None = None) -> FlowResult:
"""Confirm auto-discovered device."""
if user_input is not None:
return self.async_create_entry(
title=self._name,
data={CONF_ADDRESS: self._address},
)
return self.async_show_form(
step_id="confirm",
description_placeholders={
"name": self._name,
"address": self._address,
},
)
# ------------------------------------------------------------------
# Manual path (user goes to Integrations → Add → Xiaoxiang BMS)
# ------------------------------------------------------------------
async def async_step_user(self, user_input: dict | None = None) -> FlowResult:
"""Handle manual setup. Shows discovered BMS devices as a dropdown if found."""
errors: dict[str, str] = {}
if user_input is not None:
address = user_input[CONF_ADDRESS].upper().strip()
await self.async_set_unique_id(address)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"Xiaoxiang BMS ({address})",
data={CONF_ADDRESS: address},
)
# Build a dropdown of already-discovered BMS devices (filtered by service UUID)
discovered: dict[str, str] = {
info.address: f"{info.name or 'Xiaoxiang BMS'} ({info.address})"
for info in async_discovered_service_info(self.hass)
if UART_SERVICE_UUID in (info.service_uuids or [])
}
if discovered:
schema = vol.Schema({vol.Required(CONF_ADDRESS): vol.In(discovered)})
else:
schema = vol.Schema({
vol.Required(CONF_ADDRESS): str,
})
return self.async_show_form(
step_id="user",
data_schema=schema,
errors=errors,
)
# ------------------------------------------------------------------
# Options flow (change poll interval after setup)
# ------------------------------------------------------------------
@staticmethod
@callback
def async_get_options_flow(config_entry: config_entries.ConfigEntry) -> BmsOptionsFlow:
return BmsOptionsFlow(config_entry)
class BmsOptionsFlow(config_entries.OptionsFlow):
"""Allow the user to change the poll interval after initial setup."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
self._config_entry = config_entry
async def async_step_init(self, user_input: dict | None = None) -> FlowResult:
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
current = self._config_entry.options.get(CONF_POLL_INTERVAL, DEFAULT_POLL_INTERVAL)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema({
vol.Required(CONF_POLL_INTERVAL, default=current): vol.All(
int, vol.Range(min=MIN_POLL_INTERVAL, max=MAX_POLL_INTERVAL)
),
}),
)