feature: Improve BLE async communication

bugfix: Deal with slow first initial request
feature: Resilient BLE communication after disconnect
feature: Bump Bleak for fixed disconnect handling & BLE address search on macOS

etc/dbus-serialbattery/battery.py

@@ -110,6 +110,15 @@ def test_connection(self) -> bool:
         # return false when failed, true if successful
         return False
 
+    def connection_name(self) -> str:
+        return "Serial " + self.port
+
+    def custom_name(self) -> str:
+        return "SerialBattery(" + self.type + ")"
+
+    def product_name(self) -> str:
+        return "SerialBattery(" + self.type + ")"
+
     @abstractmethod
     def get_settings(self) -> bool:
         """

etc/dbus-serialbattery/dbushelper.py

@@ -116,21 +116,21 @@ def setup_vedbus(self):
         self._dbusservice.add_path(
             "/Mgmt/ProcessVersion", "Python " + platform.python_version()
         )
-        self._dbusservice.add_path("/Mgmt/Connection", "Serial " + self.battery.port)
+        self._dbusservice.add_path("/Mgmt/Connection", self.battery.connection_name())
 
         # Create the mandatory objects
         self._dbusservice.add_path("/DeviceInstance", self.instance)
         self._dbusservice.add_path("/ProductId", 0x0)
         self._dbusservice.add_path(
-            "/ProductName", "SerialBattery(" + self.battery.type + ")"
+            "/ProductName", self.battery.product_name()
         )
         self._dbusservice.add_path(
             "/FirmwareVersion", str(DRIVER_VERSION) + DRIVER_SUBVERSION
         )
         self._dbusservice.add_path("/HardwareVersion", self.battery.hardware_version)
         self._dbusservice.add_path("/Connected", 1)
         self._dbusservice.add_path(
-            "/CustomName", "SerialBattery(" + self.battery.type + ")", writeable=True
+            "/CustomName", self.battery.custom_name(), writeable=True
         )
 
         # Create static battery info
@@ -489,7 +489,7 @@ def publish_dbus(self):
             if (
                 self.battery.capacity is not None
                 and len(TIME_TO_SOC_POINTS) > 0
-                and self.battery.time_to_soc_update <= 0
+                and self.battery.time_to_soc_update == 0
             ):
                 self.battery.time_to_soc_update = TIME_TO_SOC_LOOP_CYCLES
                 crntPrctPerSec = (
@@ -498,7 +498,7 @@ def publish_dbus(self):
 
                 for num in TIME_TO_SOC_POINTS:
                     self._dbusservice["/TimeToSoC/" + str(num)] = (
-                        self.battery.get_timetosoc(num, crntPrctPerSec)
+                        self.battery.get_timetosoc(float(num), crntPrctPerSec)
                         if self.battery.current
                         else None
                     )

etc/dbus-serialbattery/default_config.ini

@@ -114,7 +114,7 @@ TIME_TO_SOC_POINTS=100,95,90,85,75,50,25,20,10,0
 ; TIME_TO_SOC_VALUE_TYPE = 1      ; Seconds
 ; TIME_TO_SOC_VALUE_TYPE = 2      ; Time string HH:MN:SC
 ; Both Seconds and time str "<seconds> [days, HR:MN:SC]"
-TIME_TO_SOC_VALUE_TYPE = 3
+TIME_TO_SOC_VALUE_TYPE = 1
 ; Specify how many loop cycles between each TimeToSoc updates
 TIME_TO_SOC_LOOP_CYCLES = 5
 ; Include TimeToSoC points when moving away from the SoC point. [Valid values True,False]

etc/dbus-serialbattery/lltjbd.py

@@ -49,6 +49,7 @@ def __init__(self, port, baud, address):
         super(LltJbd, self).__init__(port, baud, address)
         self.protection = LltJbdProtection()
         self.type = self.BATTERYTYPE
+        self._product_name: str = ""
 
     # degree_sign = u'\N{DEGREE SIGN}'
     command_general = b"\xDD\xA5\x03\x00\xFF\xFD\x77"
@@ -67,6 +68,9 @@ def test_connection(self):
 
         return result
 
+    def product_name(self) -> str:
+        return self._product_name
+
     def get_settings(self):
         self.read_gen_data()
         self.max_battery_charge_current = MAX_BATTERY_CHARGE_CURRENT
@@ -149,7 +153,7 @@ def read_gen_data(self):
         self.capacity_remain = capacity_remain / 100
         self.capacity = capacity / 100
         self.to_cell_bits(balance, balance2)
-        self.version = float(str(version >> 4 & 0x0F) + "." + str(version & 0x0F))
+        self.hardware_version = float(str(version >> 4 & 0x0F) + "." + str(version & 0x0F))
         self.to_fet_bits(fet)
         self.to_protection_bits(protection)
         self.max_battery_voltage = MAX_CELL_VOLTAGE * self.cell_count
@@ -182,10 +186,10 @@ def read_hardware_data(self):
         if hardware_data is False:
             return False
 
-        self.hardware_version = unpack_from(
+        self._product_name = unpack_from(
             ">" + str(len(hardware_data)) + "s", hardware_data
         )[0].decode()
-        logger.debug(self.hardware_version)
+        logger.debug(self._product_name)
         return True
 
     def read_serial_data_llt(self, command):

etc/dbus-serialbattery/lltjbd_ble.py

@@ -1,10 +1,11 @@
 # -*- coding: utf-8 -*-
 import asyncio
 import atexit
+import functools
 import threading
 from typing import Union, Optional
 
-from bleak import BleakClient
+from bleak import BleakClient, BleakScanner, BLEDevice
 
 from utils import *
 from struct import *
@@ -25,36 +26,48 @@ def __init__(self, port: str, baud: Optional[int], address: Optional[str]):
         self.protection = LltJbdProtection()
         self.type = self.BATTERYTYPE
         self.main_thread = threading.current_thread()
-        self.bt_loop = asyncio.new_event_loop()
         self.data: bytearray = bytearray()
         self.run = True
-        self.bt_thread = threading.Thread(name="BLELoop", target=self.background_loop, args=(self.bt_loop, ), daemon=True)
+        self.bt_thread = threading.Thread(name="BLELoop", target=self.background_loop, daemon=True)
+        self.bt_loop: Optional[asyncio.AbstractEventLoop] = None
         self.bt_client: Optional[BleakClient] = None
+        self.device: Optional[BLEDevice] = None
+        self.response_queue: Optional[asyncio.Queue] = None
+        self.ready_event: Optional[asyncio.Event] = None
+
+    def connection_name(self) -> str:
+        return "BLE " + self.address
+
+    def custom_name(self) -> str:
+        return self.device.name
 
     def on_disconnect(self, client):
         logger.info("BLE client disconnected")
 
     async def bt_main_loop(self):
-        async with BleakClient(self.address, disconnected_callback=self.on_disconnect) as client:
-            await client.start_notify(BLE_CHARACTERISTICS_RX_UUID, self.ncallback)
-            await asyncio.sleep(1)
+        self.device = await BleakScanner.find_device_by_address(
+            self.address, cb=dict(use_bdaddr=True)
+        )
+
+        if not self.device:
+            self.run = False
+            return
+
+        async with BleakClient(self.device, disconnected_callback=self.on_disconnect) as client:
             self.bt_client = client
-            self.name
+            self.bt_loop = asyncio.get_event_loop()
+            self.response_queue = asyncio.Queue()
+            self.ready_event.set()
             while self.run and client.is_connected and self.main_thread.is_alive():
-                await asyncio.sleep(1)
-            await client.stop_notify(BLE_CHARACTERISTICS_RX_UUID)
+                await asyncio.sleep(0.1)
+        self.bt_loop = None
 
-    def background_loop(self, loop: asyncio.AbstractEventLoop):
-        asyncio.set_event_loop(loop)
+    def background_loop(self):
         while self.run and self.main_thread.is_alive():
-            loop.run_until_complete(self.bt_main_loop())
-            sleep(0.01)
-        loop.stop()
-
-    def test_connection(self):
-        if not self.address:
-            return False
+            asyncio.run(self.bt_main_loop())
 
+    async def async_test_connection(self):
+        self.ready_event = asyncio.Event()
         if not self.bt_thread.is_alive():
             self.bt_thread.start()
 
@@ -63,68 +76,62 @@ def shutdown_ble_atexit(thread):
                 thread.join()
 
             atexit.register(shutdown_ble_atexit, self.bt_thread)
-        count = 0
-        while not self.bt_client:
-            count += 1
-            sleep(0.2)
-            if count == 10:
-                return False
-        return super().test_connection()
+        try:
+            return await asyncio.wait_for(self.ready_event.wait(), timeout=5)
+        except asyncio.TimeoutError:
+            logger.error(">>> ERROR: Unable to connect with BLE device")
+            return False
 
-    def ncallback(self, sender, data: bytearray):
-        self.data.extend(data)
+    def test_connection(self):
+        if not self.address:
+            return False
 
-    async def send_command(self, command) -> Union[bytearray, bool]:
-        await self.bt_client.write_gatt_char(BLE_CHARACTERISTICS_TX_UUID, command, False)
-        await asyncio.sleep(0.5)
-        result = await self.read_bluetooth_data()
-        return result
+        if not asyncio.run(self.async_test_connection()):
+            return False
+        return super().test_connection()
 
-    async def read_bluetooth_data(
-        self
-    ) -> Union[bytearray, bool]:
-        count = 0
-        while len(self.data) < (self.LENGTH_POS + 1):
-            await asyncio.sleep(0.01)
-            count += 1
-            if count > 50:
-                break
-
-        if len(self.data) < (self.LENGTH_POS + 1):
-            if len(self.data) == 0:
-                logger.error(">>> ERROR: No reply - returning")
-            else:
-                logger.error(
-                    ">>> ERROR: No reply - returning [len:" + str(len(self.data)) + "]"
-                )
+    async def send_command(self, command) -> Union[bytearray, bool]:
+        if not self.bt_client:
+            logger.error(">>> ERROR: No BLE client connection - returning")
             return False
 
-        length = self.data[self.LENGTH_POS]
-        count = 0
-        while len(self.data) <= length + self.LENGTH_POS:
-            await asyncio.sleep(0.005)
-            count += 1
-            if count > 150:
-                logger.error(
-                    ">>> ERROR: No reply - returning [len:"
-                    + str(len(self.data))
-                    + "/"
-                    + str(length + self.LENGTH_POS)
-                    + "]"
-                )
-                return False
-
-        result = self.data
-        self.data = bytearray()
+        fut = self.bt_loop.create_future()
+
+        def rx_callback(future: asyncio.Future, data: bytearray, sender, rx: bytearray):
+            data.extend(rx)
+            if len(data) < (self.LENGTH_POS + 1):
+                return
+
+            length = data[self.LENGTH_POS]
+            if len(data) <= length + self.LENGTH_POS + 1:
+                return
+            if not future.done():
+                future.set_result(data)
+
+        rx_collector = functools.partial(rx_callback, fut, bytearray())
+        await self.bt_client.start_notify(BLE_CHARACTERISTICS_RX_UUID, rx_collector)
+        await self.bt_client.write_gatt_char(BLE_CHARACTERISTICS_TX_UUID, command, False)
+        result = await fut
+        await self.bt_client.stop_notify(BLE_CHARACTERISTICS_RX_UUID)
+
         return result
 
-    def read_serial_data_llt(self, command):
-        task = asyncio.run_coroutine_threadsafe(self.send_command(command), self.bt_loop)
+    async def async_read_serial_data_llt(self, command):
         try:
-            data = task.result(timeout=2)
-        except:
+            bt_task = asyncio.run_coroutine_threadsafe(self.send_command(command), self.bt_loop)
+            result = await asyncio.wait_for(asyncio.wrap_future(bt_task), 20)
+            return result
+        except asyncio.TimeoutError:
             logger.error(">>> ERROR: No reply - returning")
             return False
+        except Exception as e:
+            logger.error(">>> ERROR: No reply - returning", e)
+            return False
+
+    def read_serial_data_llt(self, command):
+        if not self.bt_loop:
+            return False
+        data = asyncio.run(self.async_read_serial_data_llt(command))
         if not data:
             return False
 
@@ -145,11 +152,8 @@ async def testBLE():
         logger.error(">>> ERROR: Unable to connect")
     else:
         bat.refresh_data()
-    bat.refresh_data()
-    bat.refresh_data()
-    print("Done")
 
 
 if __name__ == "__main__":
-    asyncio.run(testBLE())
+    testBLE()
 

requirements.txt

@@ -1,3 +1,3 @@
 pyserial==3.5
 minimalmodbus==2.0.1
-bleak==0.19.5
+bleak==0.20.0