remove connection tracking (#37)

* remove connection tracking, fixed unused parameter
* fixed exception type
* code cleanup
* removed logging

custom_components/bms_ble/plugins/basebms.py

@@ -74,8 +74,7 @@ def calc_values(cls, data: BMSsample, values: set[str]):
         """
 
         def can_calc(value: str, using: frozenset[str]) -> bool:
-            """Check that value to add does not exists, is requested and
-            the necessary parameters are available."""
+            """Check that value to add does not exists, is requested and the necessary parameters are available."""
             return (value in values) and (value not in data) and using.issubset(data)
 
         # calculate cycle capacity from voltage and cycle charge

custom_components/bms_ble/plugins/daly_bms.py

@@ -55,7 +55,6 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         self._client: BleakClient | None = None
         self._data: bytearray | None = None
         self._data_event = asyncio.Event()
-        self._connected = False  # flag to indicate active BLE connection
         self._FIELDS: list[tuple[str, int, Callable[[int], int | float]]] = [
             (ATTR_VOLTAGE, 80 + self.HEAD_LEN, lambda x: float(x / 10)),
             (ATTR_CURRENT, 82 + self.HEAD_LEN, lambda x: float((x - 30000) / 10)),
@@ -95,13 +94,12 @@ async def _wait_event(self) -> None:
         await self._data_event.wait()
         self._data_event.clear()
 
-    def _on_disconnect(self, client: BleakClient) -> None:
+    def _on_disconnect(self, _client: BleakClient) -> None:
         """Disconnect callback function."""
 
-        LOGGER.debug("Disconnected from BMS (%s)", client.address)
-        self._connected = False
+        LOGGER.debug("Disconnected from BMS (%s)", self._ble_device.name)
 
-    def _notification_handler(self, sender, data: bytearray) -> None:
+    def _notification_handler(self, _sender, data: bytearray) -> None:
         LOGGER.debug("Received BLE data: %s", data)
 
         if (
@@ -124,7 +122,7 @@ def _notification_handler(self, sender, data: bytearray) -> None:
     async def _connect(self) -> None:
         """Connect to the BMS and setup notification if not connected."""
 
-        if not self._connected:
+        if self._client is None or not self._client.is_connected:
             LOGGER.debug("Connecting BMS (%s)", self._ble_device.name)
             self._client = BleakClient(
                 self._ble_device,
@@ -133,23 +131,20 @@ async def _connect(self) -> None:
             )
             await self._client.connect()
             await self._client.start_notify(UUID_RX, self._notification_handler)
-            self._connected = True
         else:
             LOGGER.debug("BMS %s already connected", self._ble_device.name)
 
     async def disconnect(self) -> None:
         """Disconnect the BMS and includes stoping notifications."""
 
-        if self._client and self._connected:
+        if self._client and self._client.is_connected:
             LOGGER.debug("Disconnecting BMS (%s)", self._ble_device.name)
             try:
                 self._data_event.clear()
                 await self._client.disconnect()
             except BleakError:
                 LOGGER.warning("Disconnect failed!")
 
-        self._client = None
-
     async def async_update(self) -> BMSsample:
         """Update battery status information."""
         await self._connect()

custom_components/bms_ble/plugins/jbd_bms.py

@@ -54,7 +54,6 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         self._data: bytearray | None = None
         self._data_final: bytearray | None = None
         self._data_event = asyncio.Event()
-        self._connected = False  # flag to indicate active BLE connection
         self._FIELDS: list[tuple[str, int, int, bool, Callable[[int], int | float]]] = [
             (KEY_TEMP_SENS, 26, 1, False, lambda x: x),
             (ATTR_VOLTAGE, 4, 2, False, lambda x: float(x / 100)),
@@ -83,13 +82,12 @@ async def _wait_event(self) -> None:
         await self._data_event.wait()
         self._data_event.clear()
 
-    def _on_disconnect(self, client: BleakClient) -> None:
+    def _on_disconnect(self, _client: BleakClient) -> None:
         """Disconnect callback function."""
 
         LOGGER.debug("Disconnected from BMS (%s)", self._ble_device.name)
-        self._connected = False
 
-    def _notification_handler(self, sender, data: bytearray) -> None:
+    def _notification_handler(self, _sender, data: bytearray) -> None:
         if self._data_event.is_set():
             return
 
@@ -137,7 +135,7 @@ def _notification_handler(self, sender, data: bytearray) -> None:
     async def _connect(self) -> None:
         """Connect to the BMS and setup notification if not connected."""
 
-        if not self._connected:
+        if self._client is None or not self._client.is_connected:
             LOGGER.debug("Connecting BMS (%s)", self._ble_device.name)
             self._client = BleakClient(
                 self._ble_device,
@@ -146,23 +144,20 @@ async def _connect(self) -> None:
             )
             await self._client.connect()
             await self._client.start_notify(UUID_RX, self._notification_handler)
-            self._connected = True
         else:
             LOGGER.debug("BMS %s already connected", self._ble_device.name)
 
     async def disconnect(self) -> None:
         """Disconnect the BMS and includes stoping notifications."""
 
-        if self._client and self._connected:
+        if self._client and self._client.is_connected:
             LOGGER.debug("Disconnecting BMS (%s)", self._ble_device.name)
             try:
                 self._data_event.clear()
                 await self._client.disconnect()
             except BleakError:
                 LOGGER.warning("Disconnect failed!")
 
-        self._client = None
-
     def _crc(self, frame: bytes) -> int:
         """Calculate JBD frame CRC."""
         return 0x10000 - sum(frame)

custom_components/bms_ble/plugins/jikong_bms.py

@@ -53,7 +53,6 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         self._data: bytearray | None = None
         self._data_final: bytearray | None = None
         self._data_event = asyncio.Event()
-        self._connected = False  # flag to indicate active BLE connection
         self._char_write_handle: int | None = None
         self._FIELDS: list[tuple[str, int, int, bool, Callable[[int], int | float]]] = [
             (KEY_CELL_COUNT, 70, 4, False, lambda x: x.bit_count()),
@@ -87,13 +86,12 @@ async def _wait_event(self) -> None:
         await self._data_event.wait()
         self._data_event.clear()
 
-    def _on_disconnect(self, client: BleakClient) -> None:
+    def _on_disconnect(self, _client: BleakClient) -> None:
         """Disconnect callback function."""
 
         LOGGER.debug("Disconnected from BMS (%s)", self._ble_device.name)
-        self._connected = False
 
-    def _notification_handler(self, sender, data: bytearray) -> None:
+    def _notification_handler(self, _sender, data: bytearray) -> None:
         """Retrieve BMS data update."""
 
         if data[0 : len(self.BT_MODULE_MSG)] == self.BT_MODULE_MSG:
@@ -139,7 +137,7 @@ def _notification_handler(self, sender, data: bytearray) -> None:
     async def _connect(self) -> None:
         """Connect to the BMS and setup notification if not connected."""
 
-        if not self._connected:
+        if self._client is None or not self._client.is_connected:
             LOGGER.debug("Connecting BMS (%s)", self._ble_device.name)
             self._client = BleakClient(
                 self._ble_device,
@@ -171,7 +169,7 @@ async def _connect(self) -> None:
                     "(%s) Failed to detect characteristics", self._ble_device.name
                 )
                 await self._client.disconnect()
-                return
+                raise ConnectionError(f"Unable to connect to {self._ble_device.name}.")
             LOGGER.debug(
                 "(%s) Using characteristics handle #%i (notify), #%i (write)",
                 self._ble_device.name,
@@ -186,24 +184,20 @@ async def _connect(self) -> None:
             await self._client.write_gatt_char(
                 self._char_write_handle or 0, data=self._cmd(b"\x97")
             )
-
-            self._connected = True
         else:
             LOGGER.debug("BMS %s already connected", self._ble_device.name)
 
     async def disconnect(self) -> None:
         """Disconnect the BMS and includes stoping notifications."""
 
-        if self._client and self._connected:
+        if self._client and self._client.is_connected:
             LOGGER.debug("Disconnecting BMS (%s)", self._ble_device.name)
             try:
                 self._data_event.clear()
                 await self._client.disconnect()
             except BleakError:
                 LOGGER.warning("Disconnect failed!")
 
-        self._client = None
-
     def _crc(self, frame: bytes) -> int:
         """Calculate Jikong frame CRC."""
         return sum(frame) & 0xFF
@@ -244,13 +238,9 @@ def _decode_data(self, data: bytearray) -> BMSsample:
 
     async def async_update(self) -> BMSsample:
         """Update battery status information."""
+
         await self._connect()
         assert self._client is not None
-        if not self._connected:
-            LOGGER.debug(
-                "Update request, but device (%s) not connected", self._ble_device.name
-            )
-            return {}
 
         if not self._data_event.is_set():
             # request cell info (only if data is not constantly published)

custom_components/bms_ble/plugins/ogt_bms.py

@@ -2,7 +2,6 @@
 
 import asyncio
 from collections.abc import Callable
-from enum import IntEnum
 import logging
 from typing import Any
 
@@ -38,25 +37,21 @@
 UUID_SERVICE = normalize_uuid_str("fff0")
 
 
-class REG(IntEnum):
-    """Field list for _REGISTER constant."""
-
-    NAME = 0
-    LEN = 1
-    FCT = 2
-
-
 class BMS(BaseBMS):
     """Offgridtec LiFePO4 Smart Pro type A and type B battery class implementation."""
 
+    IDX_NAME = 0
+    IDX_LEN = 1
+    IDX_FCT = 2
+
     def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         """Intialize private BMS members."""
         self._reconnect = reconnect
         self._ble_device = ble_device
-        assert self._ble_device.name is not None
         self._client: BleakClient | None = None
         self._data_event = asyncio.Event()
-        self._connected = False  # flag to indicate active BLE connection
+
+        assert self._ble_device.name is not None
         self._type = self._ble_device.name[9]
         self._key = sum(
             CRYPT_SEQ[int(c, 16)] for c in (f"{int(self._ble_device.name[10:]):0>4X}")
@@ -97,7 +92,7 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
                 18: (ATTR_RUNTIME, 2, lambda x: int(x * 60)),
                 23: (ATTR_CYCLES, 2, None),
             }
-            # add cell voltage registers, note need to be last!
+            # add cell voltage registers, note: need to be last!
             self._REGISTERS.update(
                 {
                     63
@@ -139,14 +134,17 @@ async def async_update(self) -> BMSsample:
         """Update battery status information."""
 
         await self._connect()
+        assert self._client is not None
 
         self._values = {}
         for key in list(self._REGISTERS):
             await self._read(key)
             try:
                 await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
             except TimeoutError:
-                LOGGER.debug("Reading %s timed out", self._REGISTERS[key][REG.NAME])
+                LOGGER.debug(
+                    "Reading %s timed out", self._REGISTERS[key][self.IDX_NAME]
+                )
             if key > 48 and f"{KEY_CELL_VOLTAGE}{64-key}" not in self._values:
                 break
 
@@ -160,11 +158,10 @@ async def async_update(self) -> BMSsample:
             await self.disconnect()
         return self._values
 
-    def _on_disconnect(self, client: BleakClient) -> None:
-        """Disconnect callback for Bleak."""
+    def _on_disconnect(self, _client: BleakClient) -> None:
+        """Disconnect callback function."""
 
-        LOGGER.debug("Disconnected from BMS (%s)", client.address)
-        self._connected = False
+        LOGGER.debug("Disconnected from BMS (%s)", self._ble_device.name)
 
     def _notification_handler(self, sender, data: bytearray) -> None:
         LOGGER.debug("Received BLE data: %s", data)
@@ -173,7 +170,7 @@ def _notification_handler(self, sender, data: bytearray) -> None:
 
         # check that descrambled message is valid and from the right characteristic
         if valid and sender.uuid == UUID_RX:
-            name, length, func = self._REGISTERS[reg]
+            name, _length, func = self._REGISTERS[reg]
             value = func(nat_value) if func else nat_value
             LOGGER.debug(
                 "Decoded data: reg: %s (#%i), raw: %i, value: %f",
@@ -190,7 +187,7 @@ def _notification_handler(self, sender, data: bytearray) -> None:
     async def _connect(self) -> None:
         """Connect to the BMS and setup notification if not connected."""
 
-        if not self._connected:
+        if self._client is None or not self._client.is_connected:
             LOGGER.debug("Connecting BMS (%s)", self._ble_device.name)
             self._client = BleakClient(
                 self._ble_device,
@@ -199,23 +196,20 @@ async def _connect(self) -> None:
             )
             await self._client.connect()
             await self._client.start_notify(UUID_RX, self._notification_handler)
-            self._connected = True
         else:
             LOGGER.debug("BMS %s already connected", self._ble_device.name)
 
     async def disconnect(self) -> None:
         """Disconnect the BMS, includes stoping notifications."""
 
-        if self._client and self._connected:
+        if self._client and self._client.is_connected:
             LOGGER.debug("Disconnecting BMS (%s)", self._ble_device.name)
             try:
                 self._data_event.clear()
                 await self._client.disconnect()
             except BleakError:
                 LOGGER.warning("Disconnect failed!")
 
-        self._client = None
-
     def _ogt_response(self, resp: bytearray) -> tuple[bool, int, int]:
         """Descramble a response from the BMS."""
 
@@ -236,7 +230,9 @@ def _ogt_response(self, resp: bytearray) -> tuple[bool, int, int]:
     def _ogt_command(self, command: int) -> bytes:
         """Put together an scambled query to the BMS."""
 
-        cmd = f"{self._HEADER}{command:0>2X}{self._REGISTERS[command][REG.LEN]:0>2X}"
+        cmd = (
+            f"{self._HEADER}{command:0>2X}{self._REGISTERS[command][self.IDX_LEN]:0>2X}"
+        )
         LOGGER.debug("command: %s", cmd)
 
         return bytearray(ord(cmd[i]) ^ self._key for i in range(len(cmd)))