add Electronicx battery support

README.md

@@ -20,11 +20,13 @@ This integration allows to monitor Bluetooth Low Energy (BLE) battery management
 - 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
+- E&J Technology BMS
+    - Supervolt v1 batteries
+    - Elektronicx batteries (show up as `LT-`…)
 - Ective batteries
 - JBD BMS, Jiabaida
-  - accurat batteries
-  - Supervolt v3 batteries
+    - accurat batteries
+    - 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`…)
 - LiTime, Redodo batteries
@@ -151,7 +153,7 @@ In case you have severe troubles,
 - 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)
+> [@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), [@hacsler](https://github.com/patman15/BMS_BLE-HA/issues/103)
 
 for helping with making the integration better.
 

custom_components/bms_ble/manifest.json

@@ -46,6 +46,10 @@
       "local_name": "libatt*",
       "manufacturer_id": 21320
     },
+    {
+      "local_name": "LT-*",
+      "manufacturer_id": 33384
+    },
     {
       "local_name": "$PFLAC*",
       "service_uuid": "0000ffe0-0000-1000-8000-00805f9b34fb",

custom_components/bms_ble/plugins/ej_bms.py

@@ -1,7 +1,7 @@
 """Module to support Dummy BMS."""
 
 import asyncio
-from enum import Enum
+from enum import IntEnum
 import logging
 from typing import Any, Callable, Final
 
@@ -28,7 +28,7 @@
 BAT_TIMEOUT = 10
 
 
-class Cmd(Enum):
+class Cmd(IntEnum):
     """BMS operation codes."""
 
     RT = 0x2
@@ -38,30 +38,31 @@ class Cmd(Enum):
 class BMS(BaseBMS):
     """Dummy battery class implementation."""
 
+    _BT_MODULE_MSG: Final[bytes] = bytes([0x41, 0x54, 0x0D, 0x0A])  # AT\r\n from BLE module
+    _HEAD: Final[int] = 0x3A
+    _TAIL: Final[int] = 0x7E
     _MAX_CELLS: Final[int] = 16
     _FIELDS: Final[list[tuple[str, Cmd, int, int, Callable[[int], int | float]]]] = [
         (ATTR_CURRENT, Cmd.RT, 89, 8, lambda x: float((x >> 16) - (x & 0xFFFF)) / 100),
         (ATTR_BATTERY_LEVEL, Cmd.RT, 123, 2, lambda x: x),
         (ATTR_CYCLE_CHRG, Cmd.CAP, 15, 4, lambda x: float(x) / 10),
         (ATTR_TEMPERATURE, Cmd.RT, 97, 2, lambda x: x - 40),  # only 1st sensor relevant
-        (ATTR_CYCLES, Cmd.RT, 119, 4, lambda x: x),
+        (ATTR_CYCLES, Cmd.RT, 115, 4, lambda x: x),
     ]
 
     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()
+        self._data: bytearray = bytearray()
+        self._data_final: bytearray = bytearray()
 
     @staticmethod
     def matcher_dict_list() -> list[dict[str, Any]]:
         """Provide BluetoothMatcher definition."""
-        return [
-            {
-                "local_name": "libatt*",
-                "manufacturer_id": 21320,
-                "connectable": True,
-            }
+        return [  # Fliteboard, Electronix battery
+            {"local_name": "libatt*", "manufacturer_id": 21320, "connectable": True},
+            {"local_name": "LT-*", "manufacturer_id": 33384, "connectable": True},
         ]
 
     @staticmethod
@@ -97,39 +98,65 @@ def _calc_values() -> set[str]:
 
     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] != 0x3A or data[-1] != 0x7E:
-            LOGGER.debug("%s: Incorrect SOI or EOI: %s", self.name, data)
+        if data.startswith(BMS._BT_MODULE_MSG):
+            LOGGER.debug("%s: filtering AT cmd", self.name)
+            if len(data) == len(BMS._BT_MODULE_MSG):
+                return
+            data = data[len(BMS._BT_MODULE_MSG) :]
+
+        if data[0] == BMS._HEAD:  # check for beginning of frame
+            self._data.clear()
+
+        self._data += data
+
+        LOGGER.debug(
+            "%s: RX BLE data (%s): %s",
+            self._ble_device.name,
+            "start" if data == self._data else "cnt.",
+            data,
+        )
+
+        if self._data[0] != BMS._HEAD or (
+            self._data[-1] != BMS._TAIL and len(self._data) < int(self._data[7:11], 16)
+        ):
+            return
+
+        if self._data[-1] != BMS._TAIL:
+            LOGGER.debug("%s: incorrect EOF: %s", self.name, data)
+            self._data.clear()
             return
 
-        if len(data) != int(data[7:11], 16):
+        if len(self._data) != int(self._data[7:11], 16):
             LOGGER.debug(
-                "%s: Incorrect frame length %i != %i",
+                "%s: incorrect frame length %i != %i",
                 self.name,
-                len(data),
-                int(data[7:11], 16),
+                len(self._data),
+                int(self._data[7:11], 16),
             )
+            self._data.clear()
             return
 
-        crc: Final = BMS._crc(data[1:-3])
-        if crc != int(data[-3:-1], 16):
+        crc: Final = BMS._crc(self._data[1:-3])
+        if crc != int(self._data[-3:-1], 16):
             LOGGER.debug(
                 "%s: incorrect checksum 0x%X != 0x%X",
                 self.name,
-                int(data[-3:-1], 16),
+                int(self._data[-3:-1], 16),
                 crc,
             )
+            self._data.clear()
             return
 
         LOGGER.debug(
-            "%s: address: 0x%X, commnad 0x%X, version: 0x%X",
+            "%s: address: 0x%X, commnad 0x%X, version: 0x%X, length: 0x%X",
             self.name,
-            int(data[1:3], 16),
-            int(data[3:5], 16) & 0x7F,
-            int(data[5:7], 16),
+            int(self._data[1:3], 16),
+            int(self._data[3:5], 16) & 0x7F,
+            int(self._data[5:7], 16),
+            len(self._data)
         )
-        self._data = data
+        self._data_final = self._data.copy()
         self._data_event.set()
 
     @staticmethod
@@ -154,9 +181,15 @@ async def _async_update(self) -> BMSsample:
         for cmd in [b":000250000E03~", b":001031000E05~"]:
             await self._client.write_gatt_char(BMS.uuid_tx(), data=cmd)
             await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
-            raw_data[int(cmd[3:5], 16) & 0x7F] = self._data.copy()
+            rsp: int = int(self._data_final[3:5], 16) & 0x7F
+            raw_data[rsp] = self._data_final
+            if rsp == Cmd.RT and len(self._data_final) == 0x8C: # handle metrisun version
+                LOGGER.debug("%s: single frame protocol detected", self.name)
+                raw_data[Cmd.CAP] = bytearray(15) + self._data_final[125:]
+                break
+
 
         return {
             key: func(int(raw_data[cmd.value][idx : idx + size], 16))
             for key, cmd, idx, size, func in BMS._FIELDS
-        } | self._cell_voltages(raw_data[Cmd.RT.value])
+        } | self._cell_voltages(raw_data[Cmd.RT])

tests/test_ej_bms.py

@@ -12,6 +12,8 @@
 from .bluetooth import generate_ble_device
 from .conftest import MockBleakClient
 
+BT_FRAME_SIZE = 20
+
 
 class MockEJBleakClient(MockBleakClient):
     """Emulate a E&J technology BMS BleakClient."""
@@ -26,7 +28,8 @@ def _response(
         cmd: int = int(bytearray(data)[3:5], 16)
         if cmd == 0x02:
             return bytearray(
-                b":0082310080000101C00000880F540F3C0F510FD70F310F2C0F340F3A0FED0FED0000000000000000000000000000000248424242F0000000000000000001AB~"
+                b":0082310080000101C00000880F540F3C0F510FD70F310F2C0F340F3A0FED0FED0000000000000000"
+                b"000000000000000248424242F0000000000000000001AB~"
             )  # TODO: put numbers
         if cmd == 0x10:
             return bytearray(b":009031001E00000002000A000AD8~")  # TODO: put numbers
@@ -41,9 +44,32 @@ async def write_gatt_char(
         """Issue write command to GATT."""
         await super().write_gatt_char(char_specifier, data, response)
         assert self._notify_callback is not None
-        self._notify_callback(
-            "MockPwrcoreBleakClient", self._response(char_specifier, data)
-        )
+        self._notify_callback("MockEctiveBleakClient", bytearray(b'AT\r\n'))
+        self._notify_callback("MockEctiveBleakClient", bytearray(b'AT\r\nillegal'))
+        for notify_data in [
+            self._response(char_specifier, data)[i : i + BT_FRAME_SIZE]
+            for i in range(0, len(self._response(char_specifier, data)), BT_FRAME_SIZE)
+        ]:
+            self._notify_callback("MockEctiveBleakClient", notify_data)
+
+
+class MockEJsfBleakClient(MockEJBleakClient):
+    """Emulate a E&J technology BMS BleakClient with single frame protocol."""
+
+    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("6e400002-b5a3-f393-e0a9-e50e24dcca9e"):
+            return bytearray()
+        cmd: int = int(bytearray(data)[3:5], 16)
+        if cmd == 0x02:
+            return bytearray(
+                b":008231008C000000000000000CBF0CC00CEA0CD50000000000000000000000000000000000000000"
+                b"00000000008C000041282828F000000000000100004B044C05DC05DCB2~"
+            )  # TODO: put numbers
+        return bytearray()
 
 
 async def test_update(monkeypatch, reconnect_fixture) -> None:
@@ -92,6 +118,45 @@ async def test_update(monkeypatch, reconnect_fixture) -> None:
     await bms.disconnect()
 
 
+async def test_update_single_frame(monkeypatch, reconnect_fixture) -> None:
+    """Test E&J technology BMS data update."""
+
+    monkeypatch.setattr(
+        "custom_components.bms_ble.plugins.basebms.BleakClient",
+        MockEJsfBleakClient,
+    )
+
+    bms = BMS(
+        generate_ble_device("cc:cc:cc:cc:cc:cc", "MockBLEDevice", None, -73),
+        reconnect_fixture,
+    )
+
+    result = await bms.async_update()
+
+    assert result == {
+        "voltage": 13.118,
+        "current": 1.4,
+        "battery_level": 75,
+        "cycles": 1,
+        "cycle_charge": 110.0,
+        "cell#0": 3.263,
+        "cell#1": 3.264,
+        "cell#2": 3.306,
+        "cell#3": 3.285,
+        "delta_voltage": 0.043,
+        "temperature": 25,
+        "cycle_capacity": 1442.98,
+        "power": 18.365,
+        "battery_charging": True,
+    }
+
+    # 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=[