Feature/branch coverage (#49)

* full branch coverage

* full branch coverage

* detailed error message

* full branch coverage

* increased branch coverage

* fixed branch coverage for OGT

https://coverage.readthedocs.io/en/7.5.4/branch.html#generator-expressions
nedbat/coveragepy#1701

* cleaned debug messages

* added pack to test uncovered branch

* update pytest settings for branch coverage

* removed redundant debug message

* code formatting

Author: patman15

CONTRIBUTING.md

@@ -13,7 +13,7 @@ In case you have troubles, please enable the debug protocol for the integration
  6. Add an appropriate [bluetooth device matcher](https://developers.home-assistant.io/docs/creating_integration_manifest#bluetooth) to `manifest.json`. Note that this is required to match the implementation of `match_dict_list()` in the new BMS class.
  7. Test and commit the changes to the branch and create a pull request to the main repository.
 
-Note: in order to keep maintainability of this integration, pull requests are required to pass standard Home Assistant checks for integrations, python linting, and 100% line test coverage.
+Note: in order to keep maintainability of this integration, pull requests are required to pass standard Home Assistant checks for integrations, Python linting, and 100% [branch test coverage](https://coverage.readthedocs.io/en/latest/branch.html#branch).
 
 ### Any contributions you make will be under the LGPL-2.1 License
 

custom_components/bms_ble/plugins/daly_bms.py

@@ -45,11 +45,13 @@ class BMS(BaseBMS):
     CMD_INFO: Final = bytearray(b"\x00\x00\x00\x3E\xD7\xB9")
     HEAD_LEN: Final = 3
     CRC_LEN: Final = 2
-    INFO_LEN: Final = 124 + HEAD_LEN + CRC_LEN
+    MAX_CELLS: Final = 32
+    MAX_TEMP: Final = 8
+    INFO_LEN: Final = 84 + HEAD_LEN + CRC_LEN + MAX_CELLS + MAX_TEMP
 
     def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         """Intialize private BMS members."""
-        self._reconnect = reconnect
+        self._reconnect: Final[bool] = reconnect
         self._ble_device = ble_device
         assert self._ble_device.name is not None
         self._client: BleakClient | None = None
@@ -60,17 +62,9 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
             (ATTR_CURRENT, 82 + self.HEAD_LEN, lambda x: float((x - 30000) / 10)),
             (ATTR_BATTERY_LEVEL, 84 + self.HEAD_LEN, lambda x: float(x / 10)),
             (ATTR_CYCLE_CHRG, 96 + self.HEAD_LEN, lambda x: float(x / 10)),
-            (
-                KEY_TEMP_SENS,
-                100 + self.HEAD_LEN,
-                lambda x: int(x),  # pylint: disable=unnecessary-lambda
-            ),
-            (KEY_CELL_COUNT, 98 + self.HEAD_LEN, lambda x: x),
-            (
-                ATTR_CYCLES,
-                102 + self.HEAD_LEN,
-                lambda x: int(x),  # pylint: disable=unnecessary-lambda
-            ),
+            (KEY_CELL_COUNT, 98 + self.HEAD_LEN, lambda x: min(x, self.MAX_CELLS)),
+            (KEY_TEMP_SENS, 100 + self.HEAD_LEN, lambda x: min(x, self.MAX_TEMP)),
+            (ATTR_CYCLES, 102 + self.HEAD_LEN, lambda x: x),
             (ATTR_DELTA_VOLTAGE, 112 + self.HEAD_LEN, lambda x: float(x / 1000)),
         ]
 
@@ -159,52 +153,40 @@ async def async_update(self) -> BMSsample:
 
         data = {
             key: func(
-                int.from_bytes(
-                    self._data[idx : idx + 2],
-                    byteorder="big",
-                    signed=True,
-                )
+                int.from_bytes(self._data[idx : idx + 2], byteorder="big", signed=True)
             )
             for key, idx, func in self._FIELDS
         }
 
         # calculate average temperature
-        if data[KEY_TEMP_SENS] and data[KEY_TEMP_SENS] <= 8:
-            data[ATTR_TEMPERATURE] = (
-                fmean(
-                    [
-                        int.from_bytes(
-                            self._data[idx : idx + 2],
-                            byteorder="big",
-                            signed=True,
-                        )
-                        for idx in range(
-                            64 + self.HEAD_LEN,
-                            64 + self.HEAD_LEN + int(data[KEY_TEMP_SENS]) * 2,
-                            2,
-                        )
-                    ]
-                )
-                - 40
+        data[ATTR_TEMPERATURE] = (
+            fmean(
+                [
+                    int.from_bytes(
+                        self._data[idx : idx + 2], byteorder="big", signed=True
+                    )
+                    for idx in range(
+                        64 + self.HEAD_LEN,
+                        64 + self.HEAD_LEN + int(data[KEY_TEMP_SENS]) * 2,
+                        2,
+                    )
+                ]
             )
+            - 40
+        )
 
         # get cell voltages
-        if data[KEY_CELL_COUNT] and data[KEY_CELL_COUNT] <= 32:
-            data.update(
-                {
-                    f"{KEY_CELL_VOLTAGE}{idx}": float(
-                        int.from_bytes(
-                            self._data[
-                                self.HEAD_LEN + 2 * idx : self.HEAD_LEN + 2 * idx + 2
-                            ],
-                            byteorder="big",
-                            signed=True,
-                        )
-                        / 1000
-                    )
-                    for idx in range(int(data[KEY_CELL_COUNT]))
-                }
+        data |= {
+            f"{KEY_CELL_VOLTAGE}{idx}": float(
+                int.from_bytes(
+                    self._data[self.HEAD_LEN + 2 * idx : self.HEAD_LEN + 2 * idx + 2],
+                    byteorder="big",
+                    signed=True,
+                )
+                / 1000
             )
+            for idx in range(int(data[KEY_CELL_COUNT]))
+        }
 
         self.calc_values(
             data, {ATTR_CYCLE_CAP, ATTR_POWER, ATTR_BATTERY_CHARGING, ATTR_RUNTIME}

custom_components/bms_ble/plugins/jbd_bms.py

@@ -44,17 +44,20 @@ class BMS(BaseBMS):
     HEAD_CMD: Final = bytes([0xDD, 0xA5])  # read header for commands
 
     INFO_LEN: Final = 7  # minimum frame size
+    BASIC_INFO: Final = 23  # basic info data length
 
     def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         """Intialize private BMS members."""
-        self._reconnect = reconnect
+        self._reconnect: Final[bool] = reconnect
         self._ble_device = ble_device
         assert self._ble_device.name is not None
         self._client: BleakClient | None = None
-        self._data: bytearray | None = None
+        self._data: bytearray = bytearray()
         self._data_final: bytearray | None = None
         self._data_event = asyncio.Event()
-        self._FIELDS: Final[list[tuple[str, int, int, bool, Callable[[int], int | float]]]] = [
+        self._FIELDS: Final[
+            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)),
             (ATTR_CURRENT, 6, 2, True, lambda x: float(x / 100)),
@@ -97,10 +100,9 @@ def _notification_handler(self, _sender, data: bytearray) -> None:
             and (data[1] == 0x03 or data[1] == 0x04)
             and data[2] == 0x00
         ):
-            self._data = data
-        elif len(data) and self._data is not None:
-            self._data += data
+            self._data.clear()
 
+        self._data += data
         LOGGER.debug(
             "(%s) Rx BLE data (%s): %s",
             self._ble_device.name,
@@ -116,9 +118,8 @@ def _notification_handler(self, _sender, data: bytearray) -> None:
         ):
             return
 
-        frame_end: int = self.INFO_LEN + self._data[3] - 1
-
-        crc = self._crc(self._data[2 : frame_end - 2])
+        frame_end: Final[int] = self.INFO_LEN + self._data[3] - 1
+        crc: Final[int] = self._crc(self._data[2 : frame_end - 2])
         if int.from_bytes(self._data[frame_end - 2 : frame_end], "big") != crc:
             LOGGER.debug(
                 "(%s) Rx data CRC is invalid: %i != %i",
@@ -178,20 +179,15 @@ def _decode_data(self, data: bytearray) -> dict[str, int | float]:
         }
 
         # calculate average temperature
-        if result[KEY_TEMP_SENS]:
-            result[ATTR_TEMPERATURE] = (
-                fmean(
-                    [
-                        int.from_bytes(data[idx : idx + 2], byteorder="big")
-                        for idx in range(
-                            27,
-                            27 + int(result[KEY_TEMP_SENS]) * 2,
-                            2,
-                        )
-                    ]
-                )
-                - 2731
-            ) / 10
+        result[ATTR_TEMPERATURE] = (
+            fmean(
+                [
+                    int.from_bytes(data[idx : idx + 2], byteorder="big")
+                    for idx in range(27, 27 + int(result[KEY_TEMP_SENS]) * 2, 2)
+                ]
+            )
+            - 2731
+        ) / 10
 
         return result
 
@@ -211,28 +207,28 @@ async def async_update(self) -> BMSsample:
         await self._connect()
         assert self._client is not None
 
-        # query basic info
-        await self._client.write_gatt_char(UUID_TX, data=self._cmd(b"\x03"))
-        await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
-
-        if self._data_final is None:
-            return {}
-        if len(self._data_final) != self.INFO_LEN + self._data_final[3]:
-            LOGGER.debug(
-                "(%s) Wrong data length (%i): %s",
-                self._ble_device.name,
-                len(self._data_final),
-                self._data_final,
-            )
-
-        data = self._decode_data(self._data_final)
-
-        # query cell block info
-        await self._client.write_gatt_char(UUID_TX, data=self._cmd(b"\x04"))
-        await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
+        data = {}
+        for cmd, exp_len, dec_fct in [
+            (self._cmd(b"\x03"), self.BASIC_INFO, self._decode_data),
+            (self._cmd(b"\x04"), 0, self._cell_voltages),
+        ]:
+            await self._client.write_gatt_char(UUID_TX, data=cmd)
+            await asyncio.wait_for(self._wait_event(), timeout=BAT_TIMEOUT)
+
+            if self._data_final is None:
+                continue
+            if (
+                len(self._data_final) != self.INFO_LEN + self._data_final[3]
+                or len(self._data_final) < self.INFO_LEN + exp_len
+            ):
+                LOGGER.debug(
+                    "(%s) Wrong data length (%i): %s",
+                    self._ble_device.name,
+                    len(self._data_final),
+                    self._data_final,
+                )
 
-        if self._data_final is not None:
-            data.update(self._cell_voltages(self._data_final))
+            data.update(dec_fct(self._data_final))
 
         self.calc_values(
             data,

custom_components/bms_ble/plugins/jikong_bms.py

@@ -50,7 +50,7 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
         self._ble_device = ble_device
         assert self._ble_device.name is not None
         self._client: BleakClient | None = None
-        self._data: bytearray | None = None
+        self._data: bytearray = bytearray()
         self._data_final: bytearray | None = None
         self._data_event = asyncio.Event()
         self._char_write_handle: int | None = None
@@ -103,9 +103,9 @@ def _notification_handler(self, _sender, data: bytearray) -> None:
             data = data[len(self.BT_MODULE_MSG) :]
 
         if data[0 : len(self.HEAD_RSP)] == self.HEAD_RSP:
-            self._data = data
-        elif len(data) and self._data is not None:
-            self._data += data
+            self._data.clear()
+
+        self._data += data
 
         LOGGER.debug(
             "(%s) Rx BLE data (%s): %s",
@@ -171,7 +171,9 @@ async def _connect(self) -> None:
                     "(%s) Failed to detect characteristics", self._ble_device.name
                 )
                 await self._client.disconnect()
-                raise ConnectionError(f"Unable to connect to {self._ble_device.name}.")
+                raise ConnectionError(
+                    f"Failed to detect characteristics from {self._ble_device.name}."
+                )
             LOGGER.debug(
                 "(%s) Using characteristics handle #%i (notify), #%i (write)",
                 self._ble_device.name,
@@ -206,8 +208,7 @@ def _crc(self, frame: bytes) -> int:
 
     def _cmd(self, cmd: bytes, value: list[int] | None = None) -> bytes:
         """Assemble a Jikong BMS command."""
-        if value is None:
-            value = []
+        value = [] if value is None else value
         assert len(value) <= 13
         frame = bytes([*self.HEAD_CMD, cmd[0]])
         frame += bytes([len(value), *value])

custom_components/bms_ble/plugins/ogt_bms.py

@@ -93,13 +93,10 @@ def __init__(self, ble_device: BLEDevice, reconnect: bool = False) -> None:
                 23: (ATTR_CYCLES, 2, None),
             }
             # add cell voltage registers, note: need to be last!
-            self._REGISTERS.update(
-                {
-                    63
-                    - reg: (f"{KEY_CELL_VOLTAGE}{reg+1}", 2, lambda x: float(x) / 1000)
-                    for reg in range(16)
-                }
-            )
+            self._REGISTERS |= {  # pragma: no branch
+                63 - reg: (f"{KEY_CELL_VOLTAGE}{reg+1}", 2, lambda x: float(x) / 1000)
+                for reg in range(16)
+            }
             self._HEADER = "+R16"
         else:
             self._REGISTERS = {}
@@ -154,7 +151,7 @@ async def async_update(self) -> BMSsample:
         )
 
         # remove remaining runtime if battery is charging
-        if self._values.get(ATTR_RUNTIME) == 0xFFFF*60:
+        if self._values.get(ATTR_RUNTIME) == 0xFFFF * 60:
             del self._values[ATTR_RUNTIME]
 
         if self._reconnect: