Add TDT BMS support

README.md

@@ -29,9 +29,10 @@ This integration allows to monitor Bluetooth Low Energy (BLE) battery management
     - Supervolt v3 batteries (show up as `SX1*`…)
 - JK BMS, Jikong, (HW version >=6 required)
 - Offgridtec LiFePo4 Smart Pro: type A & B (show up as `SmartBat-A`… or `SmartBat-B`…)
-- LiTime, Redodo batteries
+- LiTime, Power Queen, and Redodo batteries
 - Seplos v2 (show up as `BP0`?)
 - Seplos v3 (show up as `SP0`… or `SP1`…)
+- TDT BMS (show up as e.g., `XDZN`…)
 
 New device types can be easily added via the plugin architecture of this integration. See the [contribution guidelines](CONTRIBUTING.md) for details.
 
@@ -86,6 +87,12 @@ Installation can be done using [HACS](https://hacs.xyz/docs/use/) by [adding a c
 1. Restart Home Assistant
 1. In the HA UI go to "Configuration" -> "Integrations" click "+" and [search](https://my.home-assistant.io/redirect/config_flow_start/?domain=bms_ble) for "BLE Battery Management"
 
+## Known Issues
+
+<details><summary>Elektronicx batteries</summary>
+Bluetooth is turned off, when there is no current. Thus device will get unavailble / cannot be added.
+</details>
+
 ## FAQ
 ### My sensors show unknown/unavailable at startup!
 The polling interval is 30 seconds. So at startup it takes a few minutes to detect the battery and query the sensors. Then data will be available.
@@ -164,7 +171,7 @@ Once pairing is done, the integration should automatically detect the BMS.
 - Add further battery types on [request](https://github.com/patman15/BMS_BLE-HA/issues/new?assignees=&labels=enhancement&projects=&template=feature_request.yml)
 
 ## 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), [@azisto](https://github.com/patman15/BMS_BLE-HA/issues/78), [@BikeAtor, @Karatzie](https://github.com/patman15/BMS_BLE-HA/issues/57), [@SkeLLLa,@romanshypovskyi](https://github.com/patman15/BMS_BLE-HA/issues/90), [@riogrande75, @ebagnoli, @andreas-bulling](https://github.com/patman15/BMS_BLE-HA/issues/101), [@hacsler](https://github.com/patman15/BMS_BLE-HA/issues/103)
+> [@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), [@SkeLLLa,@romanshypovskyi](https://github.com/patman15/BMS_BLE-HA/issues/90), [@riogrande75, @ebagnoli, @andreas-bulling](https://github.com/patman15/BMS_BLE-HA/issues/101), [@goblinmaks, @andreitoma-github](https://github.com/patman15/BMS_BLE-HA/issues/102), [@hacsler](https://github.com/patman15/BMS_BLE-HA/issues/103)
 
 for helping with making the integration better.
 

custom_components/bms_ble/const.py

@@ -21,6 +21,7 @@
     "redodo_bms",
     "seplos_bms",
     "seplos_v2_bms",
+    "tdt_bms",
     "dpwrcore_bms",  # only name filter
 ]  # available BMS types
 DOMAIN: Final[str] = "bms_ble"

custom_components/bms_ble/manifest.json

@@ -68,6 +68,9 @@
       "service_uuid": "0000ffe0-0000-1000-8000-00805f9b34fb",
       "manufacturer_id": 65535
     },
+    {
+      "manufacturer_id": 54976
+    },
     {
       "local_name": "BP0?",
       "service_uuid": "0000ff00-0000-1000-8000-00805f9b34fb"      

custom_components/bms_ble/plugins/basebms.py

@@ -238,7 +238,6 @@ def crc_modbus(data: bytearray) -> int:
             crc = (crc >> 1) ^ 0xA001 if crc % 2 else (crc >> 1)
     return crc & 0xFFFF
 
-
 def crc_xmodem(data: bytearray) -> int:
     """Calculate CRC-16-CCITT XMODEM."""
     crc: int = 0x0000
@@ -248,7 +247,6 @@ def crc_xmodem(data: bytearray) -> int:
             crc = (crc << 1) ^ 0x1021 if (crc & 0x8000) else (crc << 1)
     return crc & 0xFFFF
 
-
 def crc_sum(frame: bytes) -> int:
     """Calculate frame CRC."""
     return sum(frame) & 0xFF

custom_components/bms_ble/plugins/tdt_bms.py

@@ -0,0 +1,254 @@
+"""Module to support TDT BMS."""
+
+import asyncio
+import logging
+from typing import Any, Callable, Final
+
+from bleak.backends.device import BLEDevice
+from bleak.exc import BleakError
+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_COUNT,
+    KEY_CELL_VOLTAGE,
+    KEY_TEMP_SENS,
+    KEY_TEMP_VALUE,
+)
+
+from .basebms import BaseBMS, BMSsample, crc_modbus
+
+LOGGER = logging.getLogger(__name__)
+BAT_TIMEOUT = 10
+
+
+class BMS(BaseBMS):
+    """Dummy battery class implementation."""
+
+    _UUID_CFG: Final[str] = "fffa"
+    _HEAD: Final[int] = 0x7E
+    _TAIL: Final[int] = 0x0D
+    _CMD_VER: Final[int] = 0x00
+    _RSP_VER: Final[int] = 0x00
+    _CELL_POS: Final[int] = 0x8
+    _INFO_LEN: Final[int] = 10  # minimal frame length
+    _FIELDS: Final[
+        list[tuple[str, int, int, int, bool, Callable[[int], int | float]]]
+    ] = [
+        (ATTR_VOLTAGE, 0x8C, 2, 2, False, lambda x: float(x / 100)),
+        (
+            ATTR_CURRENT,
+            0x8C,
+            0,
+            2,
+            False,
+            lambda x: float((x & 0x3FFF) / 10 * (-1 if x >> 15 else 1)),
+        ),
+        (ATTR_CYCLE_CHRG, 0x8C, 4, 2, False, lambda x: float(x / 10)),
+        (ATTR_BATTERY_LEVEL, 0x8C, 13, 1, False, lambda x: x),
+        (ATTR_CYCLES, 0x8C, 8, 2, False, lambda x: x),
+    ]
+    _CMDS: Final[list[int]] = [*list({field[1] for field in _FIELDS})]
+
+    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._exp_len: int = 0
+        self._data_final: dict[int, bytearray] = {}
+
+    @staticmethod
+    def matcher_dict_list() -> list[dict[str, Any]]:
+        """Provide BluetoothMatcher definition."""
+        return [{"manufacturer_id": 54976, "connectable": True}]
+
+    @staticmethod
+    def device_info() -> dict[str, str]:
+        """Return device information for the battery management system."""
+        return {"manufacturer": "TDT", "model": "Smart BMS"}
+
+    @staticmethod
+    def uuid_services() -> list[str]:
+        """Return list of 128-bit UUIDs of services required by BMS."""
+        return [normalize_uuid_str("fff0")]
+
+    @staticmethod
+    def uuid_rx() -> str:
+        """Return 16-bit UUID of characteristic that provides notification/read property."""
+        return "fff1"
+
+    @staticmethod
+    def uuid_tx() -> str:
+        """Return 16-bit UUID of characteristic that provides write property."""
+        return "fff2"
+
+    @staticmethod
+    def _calc_values() -> set[str]:
+        return {
+            ATTR_BATTERY_CHARGING,
+            ATTR_CYCLE_CAP,
+            ATTR_DELTA_VOLTAGE,
+            ATTR_POWER,
+            ATTR_RUNTIME,
+            ATTR_TEMPERATURE,
+        }  # calculate further values from BMS provided set ones
+
+    async def _init_characteristics(self) -> None:
+        try:
+            await self._client.write_gatt_char(BMS._UUID_CFG, data=b"HiLink")
+            if (
+                ret := int.from_bytes(await self._client.read_gatt_char(BMS._UUID_CFG))
+            ) != 0x1:
+                LOGGER.debug("%s: error initializing BMS: %X", self.name, ret)
+        except (BleakError, EOFError) as err:
+            LOGGER.debug("%s: failed to intialize BMS: %s", self.name, err)
+
+        await super()._init_characteristics()
+
+    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)
+
+        if (
+            data[0] == BMS._HEAD
+            and len(data) > BMS._INFO_LEN
+            and len(self._data) >= self._exp_len
+        ):
+            self._exp_len = BMS._INFO_LEN + int.from_bytes(data[6:8])
+            self._data = bytearray()
+
+        self._data += data
+        LOGGER.debug(
+            "%s: RX BLE data (%s): %s",
+            self._ble_device.name,
+            "start" if data == self._data else "cnt.",
+            data,
+        )
+
+        # verify that data long enough
+        if len(self._data) < self._exp_len:
+            return
+
+        if self._data[-1] != BMS._TAIL:
+            LOGGER.debug("%s: frame end incorrect: %s", self.name, self._data)
+            return
+
+        if self._data[1] != BMS._RSP_VER:
+            LOGGER.debug(
+                "%s: unknown frame version: V%.1f", self.name, self._data[1] / 10
+            )
+            return
+
+        if self._data[4]:
+            LOGGER.debug("%s: BMS reported error code: 0x%X", self.name, self._data[4])
+            return
+
+        crc = crc_modbus(self._data[:-3])
+        if int.from_bytes(self._data[-3:-1], "big") != crc:
+            LOGGER.debug(
+                "%s: RX data CRC is invalid: 0x%X != 0x%X",
+                self._ble_device.name,
+                int.from_bytes(self._data[-3:-1], "big"),
+                crc,
+            )
+            return
+        self._data_final[self._data[5]] = self._data
+        self._data_event.set()
+
+    @staticmethod
+    def _cmd(cmd: int, data: bytearray = bytearray()) -> bytearray:
+        """Assemble a TDT BMS command."""
+        assert cmd in (0x8C, 0x8D, 0x92)  # allow only read commands
+        frame = bytearray(
+            [BMS._HEAD, BMS._CMD_VER, 0x1, 0x3, 0x0, cmd]
+        )  # fixed version
+        frame += len(data).to_bytes(2, "big", signed=False) + data
+        frame += bytearray(int.to_bytes(crc_modbus(frame), 2, byteorder="big"))
+        frame += bytearray([BMS._TAIL])
+        LOGGER.debug("TX cmd: %s", frame.hex(" "))  # TODO: remove
+        return frame
+
+    @staticmethod
+    def _decode_data(data: dict[int, bytearray], offs: int) -> dict[str, int | float]:
+        return {
+            key: func(
+                int.from_bytes(
+                    data[cmd][idx + offs : idx + offs + size],
+                    byteorder="big",
+                    signed=sign,
+                )
+            )
+            for key, cmd, idx, size, sign, func in BMS._FIELDS
+        }
+
+    @staticmethod
+    def _cell_voltages(data: bytearray) -> dict[str, float]:
+        return {
+            f"{KEY_CELL_VOLTAGE}{idx}": float(
+                int.from_bytes(
+                    data[BMS._CELL_POS + 1 + idx * 2 : BMS._CELL_POS + 1 + idx * 2 + 2],
+                    byteorder="big",
+                    signed=False,
+                )
+            )
+            / 1000
+            for idx in range(data[BMS._CELL_POS])
+        }
+
+    @staticmethod
+    def _temp_sensors(data: bytearray, sensors: int, offs: int) -> dict[str, float]:
+        return {
+            f"{KEY_TEMP_VALUE}{idx}": (
+                int.from_bytes(
+                    data[offs + idx * 2 : offs + (idx + 1) * 2],
+                    byteorder="big",
+                    signed=False,
+                )
+                - 2731.5
+            )
+            / 10
+            for idx in range(sensors)
+            if int.from_bytes(
+                data[offs + idx * 2 : offs + (idx + 1) * 2],
+                byteorder="big",
+                signed=False,
+            )
+        }
+
+    async def _async_update(self) -> BMSsample:
+        """Update battery status information."""
+
+        for cmd in BMS._CMDS:
+            await self._client.write_gatt_char(BMS.uuid_tx(), data=BMS._cmd(cmd))
+            await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
+
+        result: BMSsample = {KEY_CELL_COUNT: int(self._data_final[0x8C][BMS._CELL_POS])}
+        result[KEY_TEMP_SENS] = int(
+            self._data_final[0x8C][BMS._CELL_POS + int(result[KEY_CELL_COUNT]) * 2 + 1]
+        )
+
+        result |= BMS._cell_voltages(self._data_final[0x8C])
+        result |= BMS._temp_sensors(
+            self._data_final[0x8C],
+            int(result[KEY_TEMP_SENS]),
+            BMS._CELL_POS + int(result[KEY_CELL_COUNT]) * 2 + 2,
+        )
+        result |= BMS._decode_data(
+            self._data_final,
+            BMS._CELL_POS + int(result[KEY_CELL_COUNT] + result[KEY_TEMP_SENS]) * 2 + 2,
+        )
+
+        self._data_final.clear()
+
+        return result

tests/conftest.py

@@ -309,10 +309,20 @@ async def write_gatt_char(
     ) -> None:
         """Mock write GATT characteristics."""
         LOGGER.debug(
-            "MockBleakClient write_gatt_char for %s, data: %s", char_specifier, data
+            "MockBleakClient write_gatt_char %s, data: %s", char_specifier, data
         )
         assert self._connected, "write_gatt_char called, but client not connected."
 
+    async def read_gatt_char(
+        self,
+        char_specifier: BleakGATTCharacteristic | int | str | UUID,
+        **kwargs,
+    ) -> bytearray:
+        """Mock write GATT characteristics."""
+        LOGGER.debug("MockBleakClient read_gatt_char %s", char_specifier)
+        assert self._connected, "read_gatt_char called, but client not connected."
+        return bytearray()
+
     async def disconnect(self) -> bool:
         """Mock disconnect."""
         assert self._connected, "Disconnect called, but client not connected."

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) == 11  # check number of BMS types
+    assert len(BMS_TYPES) == 12  # check number of BMS types