support Redodo battery bms

README.md

@@ -17,14 +17,16 @@ This integration allows to monitor Bluetooth Low Energy (BLE) battery management
 - Readout of individual cell voltages to be able to judge battery health
 
 ### Supported Devices
-- Offgridtec LiFePo4 Smart Pro: type A & B (show up as `SmartBat-A`… or `SmartBat-B`…)
 - CBT Power BMS (Creabest batteries)
 - D-powercore BMS (show up as `DXB-`…), Fliteboard batteries (show up as `TBA-`…)
 - Daly BMS (show up as `DL-`…)
 - E&J Technology BMS, Supervolt v1 batteries
-- JK BMS, Jikong, (HW version >=11 required)
 - JBD BMS, Jiabaida, Supervolt v3 batteries
+- JK BMS, Jikong, (HW version >=11 required)
+- Offgridtec LiFePo4 Smart Pro: type A & B (show up as `SmartBat-A`… or `SmartBat-B`…)
+- Redodo batteries
 - Seplos v3 (show up as `SP0`… or `SP1`…)
+- Supervolt batteries (JBD BMS)
 
 New device types can be easily added via the plugin architecture of this integration. See the [contribution guidelines](CONTRIBUTING.md) for details.
 
@@ -143,7 +145,7 @@ Once pairing is done, the integration should automatically detect the BMS.
 
 
 ## Thanks to
-> [@gkathan](https://github.com/patman15/BMS_BLE-HA/issues/2), [@downset](https://github.com/patman15/BMS_BLE-HA/issues/19), [@gerritb](https://github.com/patman15/BMS_BLE-HA/issues/22), [@Goaheadz](https://github.com/patman15/BMS_BLE-HA/issues/24), [@alros100, @majonessyltetoy](https://github.com/patman15/BMS_BLE-HA/issues/52), [@snipah, @Gruni22](https://github.com/patman15/BMS_BLE-HA/issues/59), [@BikeAtor, @Karatzie](https://github.com/patman15/BMS_BLE-HA/issues/57)
+> [@gkathan](https://github.com/patman15/BMS_BLE-HA/issues/2), [@downset](https://github.com/patman15/BMS_BLE-HA/issues/19), [@gerritb](https://github.com/patman15/BMS_BLE-HA/issues/22), [@Goaheadz](https://github.com/patman15/BMS_BLE-HA/issues/24), [@alros100, @majonessyltetoy](https://github.com/patman15/BMS_BLE-HA/issues/52), [@snipah, @Gruni22](https://github.com/patman15/BMS_BLE-HA/issues/59), [@azisto](https://github.com/patman15/BMS_BLE-HA/issues/78), [@BikeAtor, @Karatzie](https://github.com/patman15/BMS_BLE-HA/issues/57)
 
 for helping with making the integration better.
 

custom_components/bms_ble/const.py

@@ -18,6 +18,7 @@
     "jbd_bms",
     "jikong_bms",
     "ogt_bms",
+    "redodo_bms",
     "seplos_bms",
 ]  # available BMS types
 DOMAIN: Final = "bms_ble"

custom_components/bms_ble/manifest.json

@@ -38,6 +38,10 @@
     {
       "service_uuid": "0000ffb0-0000-1000-8000-00805f9b34fb"
     },
+    {
+      "service_uuid": "0000ffe0-0000-1000-8000-00805f9b34fb",
+      "manufacturer_id": 22618
+    },
     {
       "local_name": "libatt*",
       "manufacturer_id": 21320
@@ -51,5 +55,5 @@
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/patman15/BMS_BLE-HA/issues",
   "requirements": [],
-  "version": "1.8.0"
+  "version": "1.9.0"
 }

custom_components/bms_ble/plugins/redodo_bms.py

@@ -0,0 +1,152 @@
+"""Module to support Dummy BMS."""
+
+import asyncio
+import logging
+from typing import Any, Callable, Final
+
+from bleak.backends.device import BLEDevice
+from bleak.uuids import normalize_uuid_str
+
+from custom_components.bms_ble.const import (
+    ATTR_BATTERY_CHARGING,
+    ATTR_BATTERY_LEVEL,
+    ATTR_CURRENT,
+    ATTR_CYCLE_CAP,
+    ATTR_CYCLE_CHRG,
+    ATTR_CYCLES,
+    ATTR_DELTA_VOLTAGE,
+    ATTR_POWER,
+    ATTR_RUNTIME,
+    ATTR_TEMPERATURE,
+    ATTR_VOLTAGE,
+    KEY_CELL_VOLTAGE,
+)
+
+from .basebms import BaseBMS, BMSsample
+
+LOGGER = logging.getLogger(__name__)
+BAT_TIMEOUT = 10
+
+
+class BMS(BaseBMS):
+    """Dummy battery class implementation."""
+
+    CRC_POS: Final = -1  # last byte
+    HEAD_LEN: Final = 3
+    MAX_CELLS: Final = 16
+
+    def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
+        """Initialize BMS."""
+        LOGGER.debug("%s init(), BT address: %s", self.device_id(), ble_device.address)
+        super().__init__(LOGGER, self._notification_handler, ble_device, reconnect)
+
+        self._data: bytearray = bytearray()
+        self._FIELDS: Final[
+            list[tuple[str, int, int, bool, Callable[[int], int | float]]]
+        ] = [
+            (ATTR_VOLTAGE, 12, 2, False, lambda x: float(x / 1000)),
+            (ATTR_CURRENT, 48, 4, True, lambda x: float(x / 1000)),
+            (ATTR_TEMPERATURE, 56, 2, False, lambda x: x),
+            (ATTR_BATTERY_LEVEL, 90, 2, False, lambda x: x),
+            (ATTR_CYCLE_CHRG, 62, 2, False, lambda x: float(x / 100)),
+            (ATTR_CYCLES, 96, 4, False, lambda x: x),
+        ]
+
+    @staticmethod
+    def matcher_dict_list() -> list[dict[str, Any]]:
+        """Provide BluetoothMatcher definition."""
+        return [
+            {
+                "service_uuid": BMS.uuid_services()[0],
+                "manufacturer_id": 0x585A,
+                "connectable": True,
+            }
+        ]
+
+    @staticmethod
+    def device_info() -> dict[str, str]:
+        """Return device information for the battery management system."""
+        return {"manufacturer": "Redodo", "model": "Bluetooth battery"}
+
+    @staticmethod
+    def uuid_services() -> list[str]:
+        """Return list of 128-bit UUIDs of services required by BMS."""
+        return [normalize_uuid_str("ffe0")]  # change service UUID here!
+
+    @staticmethod
+    def uuid_rx() -> str:
+        """Return 16-bit UUID of characteristic that provides notification/read property."""
+        return "ffe1"
+
+    @staticmethod
+    def uuid_tx() -> str:
+        """Return 16-bit UUID of characteristic that provides write property."""
+        return "ffe2"
+
+    @staticmethod
+    def _calc_values() -> set[str]:
+        return {
+            ATTR_BATTERY_CHARGING,
+            ATTR_DELTA_VOLTAGE,
+            ATTR_CYCLE_CAP,
+            ATTR_POWER,
+            ATTR_RUNTIME,
+        }  # calculate further values from BMS provided set ones
+
+    def _notification_handler(self, _sender, data: bytearray) -> None:
+        """Handle the RX characteristics notify event (new data arrives)."""
+        LOGGER.debug("%s: Received BLE data: %s", self.name, data.hex(" "))
+
+        if len(data) < 3 or not data.startswith(b"\x00\x00"):
+            LOGGER.debug("%s: incorrect SOF.")
+            return
+
+        if len(data) != data[2] + self.HEAD_LEN + 1:  # add header length and CRC
+            LOGGER.debug("(%s) incorrect frame length (%i)", self.name, len(data))
+            return
+
+        crc = self._crc(data[: self.CRC_POS])
+        if crc != data[self.CRC_POS]:
+            LOGGER.debug(
+                "(%s) Rx data CRC is invalid: 0x%x != 0x%x",
+                self.name,
+                data[len(data) + self.CRC_POS],
+                crc,
+            )
+            return
+
+        self._data = data
+        self._data_event.set()
+
+    def _crc(self, frame: bytes) -> int:
+        """Calculate frame CRC."""
+        return sum(frame) & 0xFF
+
+    def _cell_voltages(self, data: bytearray, cells: int) -> dict[str, float]:
+        """Return cell voltages from status message."""
+        return {
+            f"{KEY_CELL_VOLTAGE}{idx}": int.from_bytes(
+                data[16 + 2 * idx : 16 + 2 * idx + 2],
+                byteorder="little",
+                signed=False,
+            )
+            / 1000
+            for idx in range(cells)
+            if int.from_bytes(data[16 + 2 * idx : 16 + 2 * idx + 2], byteorder="little")
+        }
+
+    async def _async_update(self) -> BMSsample:
+        """Update battery status information."""
+        await self._client.write_gatt_char(
+            BMS.uuid_tx(), data=b"\x00\x00\x04\x01\x13\x55\xaa\x17"
+        )
+        await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
+
+        return {
+            key: func(
+                int.from_bytes(
+                    self._data[idx : idx + size], byteorder="little", signed=sign
+                )
+            )
+            for key, idx, size, sign, func in self._FIELDS
+        } | self._cell_voltages(self._data, self.MAX_CELLS)

tests/test_const.py

@@ -7,4 +7,4 @@ async def test_critical_constants() -> None:
     """Test general constants are not altered for debugging."""
 
     assert UPDATE_INTERVAL == 30  # ensure that update interval is 30 seconds
-    assert len(BMS_TYPES) == 8  # check number of BMS types
+    assert len(BMS_TYPES) == 9  # check number of BMS types

tests/test_redodo_bms.py

@@ -0,0 +1,155 @@
+"""Test the E&J technology BMS implementation."""
+
+import pytest
+from collections.abc import Buffer
+from uuid import UUID
+
+from bleak.backends.characteristic import BleakGATTCharacteristic
+from bleak.uuids import normalize_uuid_str
+from custom_components.bms_ble.plugins.redodo_bms import BMS
+
+from .bluetooth import generate_ble_device
+from .conftest import MockBleakClient
+
+
+class MockRedodoBleakClient(MockBleakClient):
+    """Emulate a Redodo BMS BleakClient."""
+
+    def _response(
+        self, char_specifier: BleakGATTCharacteristic | int | str | UUID, data: Buffer
+    ) -> bytearray:
+        if isinstance(char_specifier, str) and normalize_uuid_str(
+            char_specifier
+        ) != normalize_uuid_str("ffe2"):
+            return bytearray()
+        cmd: int = bytearray(data)[4]
+        if cmd == 0x13:
+            return bytearray(
+                b"\x00\x00\x65\x01\x93\x55\xaa\x00\x46\x66\x00\x00\xbc\x67\x00\x00\xf5\x0c\xf7\x0c"
+                b"\xfc\x0c\xfb\x0c\xf8\x0c\xf2\x0c\xfa\x0c\xf5\x0c\x00\x00\x00\x00\x00\x00\x00\x00"
+                b"\x00\x00\x00\x00\x00\x00\x00\x00\x65\xfa\xff\xff\x17\x00\x16\x00\x17\x00\x00\x00"
+                b"\x00\x00\xe9\x1a\x04\x29\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x41\x00\x64\x00\x00\x00\x03\x00\x00\x00"
+                b"\x5f\x01\x00\x00\xbc"
+            )  # TODO: put numbers
+        return bytearray()
+
+    async def write_gatt_char(
+        self,
+        char_specifier: BleakGATTCharacteristic | int | str | UUID,
+        data: Buffer,
+        response: bool = None,  # type: ignore[implicit-optional] # noqa: RUF013 # same as upstream
+    ) -> None:
+        """Issue write command to GATT."""
+        await super().write_gatt_char(char_specifier, data, response)
+        assert self._notify_callback is not None
+        self._notify_callback(
+            "MockRedodoBleakClient", self._response(char_specifier, data)
+        )
+
+
+async def test_update(monkeypatch, reconnect_fixture) -> None:
+    """Test Redodo technology BMS data update."""
+
+    monkeypatch.setattr(
+        "custom_components.bms_ble.plugins.basebms.BleakClient",
+        MockRedodoBleakClient,
+    )
+
+    bms = BMS(
+        generate_ble_device("cc:cc:cc:cc:cc:cc", "MockBLEDevice", None, -73),
+        reconnect_fixture,
+    )
+
+    result = await bms.async_update()
+
+    assert result == {
+        "voltage": 26.556,
+        "current": -1.435,
+        "cell#0": 3.317,
+        "cell#1": 3.319,
+        "cell#2": 3.324,
+        "cell#3": 3.323,
+        "cell#4": 3.320,
+        "cell#5": 3.314,
+        "cell#6": 3.322,
+        "cell#7": 3.317,
+        "delta_voltage": 0.01,
+        "power": -38.108,
+        "battery_charging": False,
+        "battery_level": 65,
+        "cycle_charge": 68.89,
+        "cycle_capacity": 1829.443,
+        "runtime": 172825,
+        "temperature": 23,
+        "cycles": 3,
+    }
+
+    # query again to check already connected state
+    result = await bms.async_update()
+    assert bms._client.is_connected is not reconnect_fixture  # noqa: SLF001
+
+    await bms.disconnect()
+
+
+@pytest.fixture(
+    name="wrong_response",
+    params=[
+        bytearray(
+            b"\x00\x00\x65\x01\x93\x55\xaa\x00\x46\x66\x00\x00\xbc\x67\x00\x00\xf5\x0c\xf7\x0c"
+            b"\xfc\x0c\xfb\x0c\xf8\x0c\xf2\x0c\xfa\x0c\xf5\x0c\x00\x00\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x65\xfa\xff\xff\x17\x00\x16\x00\x17\x00\x00\x00"
+            b"\x00\x00\xe9\x1a\x04\x29\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x41\x00\x64\x00\x00\x00\x03\x00\x00\x00"
+            b"\x5f\x01\x00\x00\xff"  # wrong CRC
+        ),
+        bytearray(
+            b"\x00\x01\x65\x01\x93\x55\xaa\x00\x46\x66\x00\x00\xbc\x67\x00\x00\xf5\x0c\xf7\x0c"
+            b"\xfc\x0c\xfb\x0c\xf8\x0c\xf2\x0c\xfa\x0c\xf5\x0c\x00\x00\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x65\xfa\xff\xff\x17\x00\x16\x00\x17\x00\x00\x00"
+            b"\x00\x00\xe9\x1a\x04\x29\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x41\x00\x64\x00\x00\x00\x03\x00\x00\x00"
+            b"\x5f\x01\x00\x00\xbc"  # wrong SOF
+        ),
+        bytearray(
+            b"\x00\x00\x65\x01\x93\x55\xaa\x00\x46\x66\x00\x00\xbc\x67\x00\x00\xf5\x0c\xf7\x0c"
+            b"\xfc\x0c\xfb\x0c\xf8\x0c\xf2\x0c\xfa\x0c\xf5\x0c\x00\x00\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x65\xfa\xff\xff\x17\x00\x16\x00\x17\x00\x00\x00"
+            b"\x00\x00\xe9\x1a\x04\x29\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x41\x00\x64\x00\x00\x00\x03\x00\x00\x00"
+            b"\x5f\x01\x00\xbc"  # wrong length
+        ),
+        bytearray(b"\x00"),  # much too short
+    ],
+)
+def response(request):
+    """Return all possible BMS variants."""
+    return request.param
+
+
+async def test_invalid_response(monkeypatch, wrong_response) -> None:
+    """Test data up date with BMS returning invalid data."""
+
+    monkeypatch.setattr(
+        "custom_components.bms_ble.plugins.redodo_bms.BAT_TIMEOUT",
+        0.1,
+    )
+
+    monkeypatch.setattr(
+        "tests.test_redodo_bms.MockRedodoBleakClient._response",
+        lambda _s, _c_, d: wrong_response,
+    )
+
+    monkeypatch.setattr(
+        "custom_components.bms_ble.plugins.basebms.BleakClient",
+        MockRedodoBleakClient,
+    )
+
+    bms = BMS(generate_ble_device("cc:cc:cc:cc:cc:cc", "MockBLEDevice", None, -73))
+
+    result = {}
+    with pytest.raises(TimeoutError):
+        result = await bms.async_update()
+
+    assert not result
+    await bms.disconnect()