Receive continuous data and write characteristic problem

[NDDEVVlog]

Problem 1 :
I'm working on a BLE application in Python using the bleak library to connect to a BLE device and receive notifications for changes in characteristics. While the code receives data correctly sometimes, it seems to be interrupted randomly — sometimes it successfully receives signals, and other times it doesn't. I want to ensure that notifications are received reliably every time a characteristic changes, similar to how it's handled in Android with the onCharacteristicChanged callback.

Goal:
I want to replicate the behavior in Android, where the onCharacteristicChanged(BluetoothGatt gatt, BluetoothGattCharacteristic characteristic) callback is invoked every time the characteristic value changes. However, in my Python application, the notifications seem to be inconsistent.

Here is my code :

import asyncio
from bleak import BleakScanner, BleakClient
from threading import Thread
from BLEService import BLEService

class BLEManager:
    def __init__(self, debug_callback, device_listbox, service_listbox):
        self.loop = asyncio.new_event_loop()
        self.debug_callback = debug_callback
        self.device_listbox = device_listbox
        self.service_listbox = service_listbox
        self.devices = []
        self.client = None
        self.ble_Services = []
        self.notification_char = None  # Store the characteristic for notifications

        # Start the event loop in a separate thread
        self.loop_thread = Thread(target=self._run_event_loop, daemon=True)
        self.loop_thread.start()

    def _run_event_loop(self):
        """Run the event loop in a separate thread."""
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def log_debug(self, message):
        if self.debug_callback:
            self.debug_callback(message)

    def scan_devices(self):
        """Start scanning for BLE devices."""
        asyncio.run_coroutine_threadsafe(self._scan_devices(), self.loop)

    async def _scan_devices(self):
        """Asynchronous BLE scanning."""
        try:
            self.log_debug("Starting BLE scan...")
            devices = await BleakScanner.discover()
            self.log_debug("Scan complete.")
            self.devices = []  # Clear previous devices
            self.device_listbox.delete(0, "end")  # Clear the Listbox

            # Update the device list
            for device in devices:
                name = device.name or None  # Set to None if the name is not available
                if name:  # Only log devices with a name
                    address = device.address
                    rssi = device.metadata.get("rssi", "N/A")
                    display_text = f"{name} (Address: {address}, RSSI: {rssi})"
                    self.devices.append(device)
                    self.device_listbox.insert("end", display_text)
                else:
                    self.log_debug(f"Device with no name found (Address: {device.address}, RSSI: {device.metadata.get('rssi', 'N/A')})")

        except Exception as e:
            self.log_debug(f"Error during scan: {e}")

    def connect_device(self, address):
        """Start connecting to a BLE device."""
        asyncio.run_coroutine_threadsafe(self._connect_device(address), self.loop)

    async def _connect_device(self, address):
        """Asynchronous BLE connection."""
        try:
            self.log_debug(f"Connecting to device {address}...")
            self.client = BleakClient(address)
            await self.client.connect()

            if self.client.is_connected:
                self.log_debug(f"Connected to {address}")
                await self._fetch_services_and_characteristics()
            else:
                self.log_debug(f"Failed to connect to {address}")
        except Exception as e:
            self.log_debug(f"Error connecting to device {address}: {e}")

    async def _fetch_services_and_characteristics(self):
        """Fetch services and characteristics from the connected device."""
        try:
            self.service_listbox.delete(0, "end")  # Clear previous services
            services = await self.client.get_services()

            for service in services:
                # Create a BLEService instance for each service
                ble_service = BLEService(service.uuid, service.characteristics, self.client)
                self.ble_Services.append(ble_service)
                # Add the service to the service listbox
                self.service_listbox.insert("end", str(ble_service))

                # If you want to create a separate list for each characteristic
                for char in service.characteristics:
                    char_properties = ", ".join(char.properties)
                    self.service_listbox.insert("end", f"  Characteristic: {char.uuid} (Type: {char_properties})")

                    # Check if this characteristic supports notifications
                    if "notify" in char.properties:
                        self.notification_char = char  # Save the notification characteristic
                        await self._start_notifications(char)  # Automatically start notifications for this characteristic

        except Exception as e:
            self.log_debug(f"Error fetching services: {e}")

    async def _start_notifications(self, char):
        """Start listening to notifications for a characteristic."""
        try:
            def notification_handler(sender, data):
                """Handle incoming notifications."""
                hex_data = self.to_hex_string(data)  # Convert bytes to hex
                self.log_debug(f"Notification from {sender}: {hex_data}")
                # Process the incoming data as needed

            self.log_debug(f"Starting notifications for characteristic {char.uuid}...")
            await self.client.start_notify(char.uuid, notification_handler)
        except Exception as e:
            self.log_debug(f"Error starting notifications: {e}")

    @staticmethod
    def to_hex_string(byte_array):
        """Convert a byte array to a hexadecimal string."""
        hex_string = "0x"
        hex_string += " ".join(f"{byte:02X}" for byte in byte_array)
        return hex_string

    def disconnect_device(self):
        """Disconnect from the BLE device."""
        asyncio.run_coroutine_threadsafe(self._disconnect_device(), self.loop)

    async def _disconnect_device(self):
        """Asynchronous BLE disconnection."""
        if self.client:
            await self.client.disconnect()
            self.log_debug("Disconnected from the device.")

    def enable_notifications(self, char_uuid):
        """Enable notifications for a characteristic."""
        # Find the characteristic by UUID
        for service in self.ble_Services:
            for char in service.characteristics:
                if char.uuid == char_uuid:
                    asyncio.run_coroutine_threadsafe(self._start_notifications(char), self.loop)
                    return
        self.log_debug(f"Characteristic with UUID {char_uuid} not found.")   

Problem 2:
I tried to write Characteristic to make the circuit the different action but it seem to be not work. Can you help me adjust the code?

   @staticmethod  # Add this decorator
    def hex_string_to_byte_array(hex_string):
        """Convert a hex string to a byte array."""
        try:
            return bytes.fromhex(hex_string)
        except ValueError as e:
            raise ValueError(f"Invalid hex string: {hex_string}. Error: {e}")

    def write_to_characteristic(self):
        # Get the selected characteristic UUID
        selected_index = self.service_list.curselection()
        if not selected_index:
            self.log_debug("No characteristic selected.")
            return

        selected_line = self.service_list.get(selected_index[0])
        if "Characteristic:" not in selected_line:
            self.log_debug("Please select a characteristic to write.")
            return

        try:
            # Extract the UUID from the selected line
            char_uuid = selected_line.split("Characteristic: ")[1].split(" ")[0]
            value = self.characteristic_value_entry.get()

            # Convert the hex string to bytes
            value_bytes = self.hex_string_to_byte_array(value)

            # Run the write coroutine in the BLEManager's event loop
            asyncio.run_coroutine_threadsafe(
                self.ble_manager.ble_Services[0].write_characteristic(char_uuid, value_bytes),
                self.ble_manager.loop,
            )
            self.log_debug(f"Written '{value_bytes}' to characteristic {char_uuid}.")
        except ValueError as ve:
            self.log_debug(f"Error converting hex string to bytes: {ve}")
        except Exception as e:
            self.log_debug(f"Error writing to characteristic: {e}")

Many thanks for your help.

[dlech]

1. What OS are you using?
2. Did you try logging Bluetooth packets as describe in the troubleshooting page of the docs so see if the device really is sending notifications or not?
3. The program is incomplete, so it is impossible to tell, but it kind of looks like there could be problems with missing blocking code and async code.