Bleak Notify does not work with QThread?

[ImmanuelT8]

Hello, i have an ESP an send 2 array with 100 values to an python app and then print them and also safe it in an excel file via pandas. In this script it work:

ESP: https://github.com/ImmanuelT8/Smart-Health-8/blob/main/BLE/Meilen/Empfange_2_Variablen_via_array_from_ESP_safe_in_exel.ino
Python App: https://github.com/ImmanuelT8/Smart-Health-8/blob/main/BLE/Meilen/Empfange_2_Variablen_via_array_from_ESP_safe_in_exel.py

But if i want it via QThread it does not work. I have 2 functions, first one to update Parameter on ESP after click the button update, this work. Second the same function as in the .py above after click Start, but this does not work. Here the code: https://pastebin.com/Gk9LkTeG

I get: "print("read data start")" and then "Gerät nicht gefunden" Any ideas? Here the function:

class Worker_Daten_Array(QThread):
    completed = pyqtSignal()
    received = pyqtSignal(str)

    def __init__(self, address):
        super().__init__()
        self.address = address

    async def handle_uuid1_notify(self, sender, data):
        spo2_values = numpy.frombuffer(data, dtype=numpy.uint32)
        print("Red: {0}".format(spo2_values))

        for spo2_value in spo2_values:
            table.add_row([spo2_value, ""])
            spo2_list.append(spo2_value)

    async def handle_uuid2_notify(self, sender, data):
            ir_values = numpy.frombuffer(data, dtype=numpy.uint32)
            print("IR: {0}".format(ir_values))
            for ir_value in ir_values:
            
                if ir_value:
                    table.add_row(["", ir_value])
                    ir_list.append(ir_value)

                if len(spo2_list) == 100 and len(ir_list) == 100:
                    global df, first_run

                    new_data_available = True

                    if first_run:
                        df = pd.DataFrame({'Red': spo2_list, 'IR': ir_list})
                        df.to_excel(filename, index=False)
                    else:
                        new_data = pd.DataFrame({'Red': spo2_list, 'IR': ir_list})
                        df = pd.concat([df, new_data], ignore_index=True)
                        df.to_excel(filename, index=False)

                        print(df.head())

                    spo2_list.clear()
                    ir_list.clear()
    async def read_data(self):
                    UUID1 = "beb5483e-36e1-4688-b7f5-ea07361b26a8"
                    UUID2 = "beb5483e-36e1-4688-b7f5-ea07361b26a9"
                    print("read data start")

                    async with BleakClient(self.address) as client:
                        print ("asnc bleak läuft")
                        services = await client.get_services()
                        for service in services:
                            characteristics = service.characteristics
                            for characteristic in characteristics:
                                if characteristic.uuid == UUID1:
                                    await client.start_notify(characteristic.handle, self.handle_uuid1_notify)
                                elif characteristic.uuid == UUID2:
                                    await client.start_notify(characteristic.handle, self.handle_uuid2_notify)
                        while True:
                            await asyncio.sleep(1)


    def run(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

  

        async def do_work():

             try:

                def print_table():
                    print(table)

                self.received.emit("Suche Gerät")  # Text "Suche Geräte" anzeigen
                print("Suche Gerät")

               
                


                #def run_read_data(self):
                    #asyncio.run(read_data(self))

                # async def run_read_data(self):
                #    await read_data(self)


                client = bleak.BleakClient(mac_address)
                await self.connect_client(client)  # Verbindung herstellen
                print("vor read")
               # run_read_data(self)
                await self.read_data()


       

             except Exception as e:
                 Start.setEnabled(True)
                 print("Nicht verbunden")
                 Main_Connect_Text.setStyleSheet("color: red;")
                 self.received.emit("Gerät nicht gefunden")  # Text "Gerät nicht gefunden" anzeigen




        loop.run_until_complete(do_work())

    async def connect_client(self, client):
        print("connect client gestartet")
        await client.connect()

    async def disconnect_client(self, client):
        await client.disconnect()

[dlech]

This doesn't look like a full program so it is hard to see exactly what it going on. However, you certainly can't call asyncio stuff from different threads. The asyncio run loop and all calls to Bleak APIs need to be on a single thread.

[ImmanuelT8]

Thanks for the answer. I've posted most of it as a pastebin and github link above, the rest is just a qt designer file. I could also email them if that were possible, otherwise that would be a bit much? I have 2 Q threads, one where parameters are updated and one where files are received and saved, but the threads are not running at the same time, it would only be a problem if they were running at the same time, or is that wrong? Is it generally only allowed to have one thread? Or is it about all asnc functions being in the thread where they were used? Thank you for the feedback!

[dlech]

It would be best to only have one asyncio run loop that never stops until the application exits.

[ImmanuelT8]

Like this?

class WorkerDatenArray(QThread):
    completed = pyqtSignal()
    received = pyqtSignal(str)

    def __init__(self, address):
        super().__init__()
        self.address = address

    async def handle_uuid1_notify(self, sender, data):
        spo2_values = numpy.frombuffer(data, dtype=numpy.uint32)
        print("Red: {0}".format(spo2_values))

        for spo2_value in spo2_values:
            table.add_row([spo2_value, ""])
            spo2_list.append(spo2_value)

    async def handle_uuid2_notify(self, sender, data):
        ir_values = numpy.frombuffer(data, dtype=numpy.uint32)
        print("IR: {0}".format(ir_values))
        
        for ir_value in ir_values:
            if ir_value:
                table.add_row(["", ir_value])
                ir_list.append(ir_value)

            if len(spo2_list) == 100 and len(ir_list) == 100:
                global df, first_run

                new_data_available = True

                if first_run:
                    df = pd.DataFrame({'Red': spo2_list, 'IR': ir_list})
                    df.to_excel(filename, index=False)
                else:
                    new_data = pd.DataFrame({'Red': spo2_list, 'IR': ir_list})
                    df = pd.concat([df, new_data], ignore_index=True)
                    df.to_excel(filename, index=False)

                    print(df.head())

                spo2_list.clear()
                ir_list.clear()

    async def read_data(self):
        UUID1 = "beb5483e-36e1-4688-b7f5-ea07361b26a8"
        UUID2 = "beb5483e-36e1-4688-b7f5-ea07361b26a9"
        print("read data start")

        async with bleak.BleakClient(self.address) as client:
            print("async bleak läuft")
            services = await client.get_services()
            for service in services:
                characteristics = service.characteristics
                for characteristic in characteristics:
                    if characteristic.uuid == UUID1:
                        await client.start_notify(characteristic.handle, self.handle_uuid1_notify)
                    elif characteristic.uuid == UUID2:
                        await client.start_notify(characteristic.handle, self.handle_uuid2_notify)
            while True:
                await asyncio.sleep(1)

    async def connect_client(self, client):
        print("connect client gestartet")
        await client.connect()

    async def disconnect_client(self, client):
        await client.disconnect()

    async def do_work(self):
        try:
            def print_table():
                print(table)

            self.received.emit("Suche Gerät")  # Text "Suche Geräte" anzeigen
            print("Suche Gerät")

            client = bleak.BleakClient(mac_address)
            await self.connect_client(client)  # Verbindung herstellen
            print("vor read")
            await self.read_data()

        except Exception as e:
            Start.setEnabled(True)
            print("Nicht verbunden")
            Main_Connect_Text.setStyleSheet("color: red;")
            self.received.emit("Gerät nicht gefunden")  # Text "Gerät nicht gefunden" anzeigen

    def run(self):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(self.do_work())
        loop.close()
        self.completed.emit()
        

But i have still the same problem.

[ImmanuelT8]

import pandas as pd

import asyncio
import threading
from bleak import BleakClient
import numpy
from prettytable import PrettyTable


from datetime import datetime

address = "CC:50:E3:9C:15:02"
MODEL_NBR_UUID = "beb5483e-36e1-4688-b7f5-ea07361b26a8"

table = PrettyTable()
table.field_names = ["Wert 1", "Wert 2"]

now = datetime.now()
dt_string = now.strftime("%d-%m-%Y_%H-%M-%S")  # Variable mit Datum und Uhrzeit erzeugen

filename = f"{dt_string}.xlsx"  # Datei mit Name und Datum erzeugen

# Ein neues Sheet erstellen und es zum Writer hinzufügen
df = pd.DataFrame(columns=["Red", "IR"])

spo2_list = []
ir_list = []

first_run = True

def print_table():
    print(table)


class BLEThread(threading.Thread):
    def __init__(self, address):
        threading.Thread.__init__(self)
        self.address = address

    async def handle_uuid1_notify(self, sender, data):
        spo2_values = numpy.frombuffer(data, dtype=numpy.uint32)
        print("Red: {0}".format(spo2_values))
        for spo2_value in spo2_values:
            table.add_row([spo2_value, ""])
            # ws.append([spo2_value, ""])  # Zeile hinzufügen
            spo2_list.append(spo2_value)  # füge den empfangenen Wert der Liste hinzu

    async def handle_uuid2_notify(self, sender, data):

        ir_values = numpy.frombuffer(data, dtype=numpy.uint32)
        print("IR: {0}".format(ir_values))
        for ir_value in ir_values:
            if ir_value:
                table.add_row(["", ir_value])
                # ws.append(["", ir_value])  # Zeile hinzufügen
                ir_list.append(ir_value)  # füge den empfangenen Wert der Liste hinzu

                # wenn alle Werte empfangen wurden, erstelle einen Pandas DataFrame und gib ihn aus
            if len(spo2_list) == 100 and len(ir_list) == 100:
                global df

                new_data_available = True


                # DataFrame mit vorhandenen Daten
                if first_run == True:

                    df = pd.DataFrame({'Red': spo2_list, 'IR': ir_list})
                    df.to_excel(filename, index=False)


                if first_run == False and new_data_available == True:

                # Neue Daten
                    new_data = pd.DataFrame({'Red': spo2_list, 'IR': ir_list})

                # Concatenation von alten und neuen Daten
                    df = pd.concat([df, new_data], ignore_index=True)

                    df.to_excel(filename, index=False)


                print(df.head())

                spo2_list.clear()  # Die Liste leeren, um sie mit neuen Werten zu füllen
                ir_list.clear()



    async def read_data(self):
        UUID1 = "beb5483e-36e1-4688-b7f5-ea07361b26a8"
        UUID2 = "beb5483e-36e1-4688-b7f5-ea07361b26a9"

        async with BleakClient(self.address) as client:
            services = await client.get_services()
            for service in services:
                characteristics = service.characteristics
                for characteristic in characteristics:
                    if characteristic.uuid == UUID1:
                        await client.start_notify(characteristic.handle, self.handle_uuid1_notify)
                    elif characteristic.uuid == UUID2:
                        await client.start_notify(characteristic.handle, self.handle_uuid2_notify)
            while True:
                await asyncio.sleep(1)

    def run(self):
        asyncio.run(self.read_data())

def start_threads():
    ble_thread = BLEThread(address)
    ble_thread.start()

if __name__ == '__main__':
    start_threads()

while first_run :
    #print("Dieser Code wird immer ausgeführt")
    first_run = False  # Flag auf False setzen, um den Code beim nächsten Durchlauf nicht erneut auszuführen

This works fine.