Using Bleak within a tkinter GUI with ASYNCIO #481
Rich-Watson0126
started this conversation in
Show and tell
Replies: 7 comments 17 replies
-
|
This is interesting. Nice to see that you managed to get it to work! I will try to see if I can rework this into a Tkinter example for the documentation rewrite in #266. Did you try to use the |
Beta Was this translation helpful? Give feedback.
3 replies
-
|
Hi. |
Beta Was this translation helpful? Give feedback.
-
|
The In your case, you are probably calling the event loop without having started a coroutine or an asynchronous task. This is an In general, to use Python's import asyncio
import tkinter as tk
async def start_window():
my_window = MyWindow(asyncio.get_event_loop())
await my_window.show()
class MyWindow(tk.Tk):
def __init__(self, loop):
self.loop = loop
self.root = tk.Tk()
....
do fancy stuff
....
async def show(self):
while True:
self.root.update()
....
maybe do other fancy stuff that you want to do regularly
....
await asyncio.sleep(0.1)
asyncio.run(start_window())I simplified this from a more complex code that I'm using, so I cannot guarantee that it will run out of the box. Here BTW, using |
Beta Was this translation helpful? Give feedback.
-
|
Thank you, Markus. |
Beta Was this translation helpful? Give feedback.
-
|
@hbldh If you are interested, I can write up a small example GUI, e.g. with a button 'Scan for BLE devices', a dropdown list which then contains the found devices and maybe two other buttons, 'Connect to selected device' and 'Disconnect'. |
Beta Was this translation helpful? Give feedback.
5 replies
-
|
I am dealing with another issue at the moment. It's about the notification handler. All the examples I have seen use a print to see the receiving answer. Maybe it is a rookie problem and there is already a method of doing it. Thank you. |
Beta Was this translation helpful? Give feedback.
-
|
If you need to consume the data outside of the notification callback, I would suggest using an |
Beta Was this translation helpful? Give feedback.
-
|
Thank you very much! |
Beta Was this translation helpful? Give feedback.
-
|
@MarkusPiotrowski just wondering if you managed to put together the example? It would be super useful :) |
Beta Was this translation helpful? Give feedback.
-
|
No, sorry, I haven't since it seemed not much demand for this. If you stay tuned, I can post a small example within a week or so. |
Beta Was this translation helpful? Give feedback.
-
|
The important thing, as it seems to me, is not to initiate the """
A small example GUI to demonstrate how
to use Tkinter in combination with async
Bleak functions.
"""
import asyncio
import tkinter as tk
import tkinter.ttk as ttk
from bleak import BleakClient, BleakError, BleakScanner
scan_result = {}
is_connected = False
def build_gui():
"""Build a simple GUI."""
# For the sake of simplicity, we use some global variables:
global main_window, device_list, device_data, message_variable
main_window = tk.Tk()
main_window.title('Tkinter/bleak asyncio Demo')
# Pressing the x-Icon on the Window calls stop_loop()
main_window.protocol("WM_DELETE_WINDOW", stop_loop)
message_variable = tk.StringVar()
# Left part of the GUI, Scan button and list of detected devices
device_frame = ttk.Labelframe(main_window, text='Devices')
device_frame.grid(padx=5, pady=5, sticky='ew')
device_iframe = tk.Frame(device_frame)
device_iframe.pack()
scrollbar = ttk.Scrollbar(device_iframe)
device_list = ttk.Treeview(
device_iframe,
height=20,
yscrollcommand=scrollbar.set,
show='tree',
)
device_list.pack(side='left', padx=5, pady=5)
device_list.bind('<<TreeviewSelect>>', device_selection)
scrollbar.configure(command=device_list.yview)
scrollbar.pack(side='right', fill='y')
scan_button = ttk.Button(
main_window,
text='Scan for BLE devices',
command=lambda: asyncio.create_task(scan()), # scan is asynchronous!
)
scan_button.grid(column=0, row=1, padx=5, pady=5)
# Right part of the GUI, Connect Button, data window and status messages
data_frame = ttk.Labelframe(main_window, text='Device Data')
data_frame.grid(padx=5, pady=5, row=0, column=1, sticky='ns')
device_data = tk.Text(data_frame, height=10, width=30)
device_data.grid(padx=5, pady=5)
message_frame = ttk.Labelframe(data_frame, text='Status')
message_frame.grid(column=0, row=1, padx=5, pady=5, sticky='ew')
message_label = ttk.Label(
message_frame, textvariable=message_variable, width=38
)
message_label.grid(row=1, padx=5, pady=5)
connect_button = ttk.Button(
main_window,
text='Connect/Disconnect to/from device',
command=lambda: asyncio.create_task(connect()),
)
connect_button.grid(column=1, row=1, padx=5, pady=5)
# Don't do: main_window.mainloop()!
# We are using the asyncio event loop in 'show' to call
# main_window.update() regularly.
async def scan():
"""Scan for unconnected Bluetooth LE devices."""
device_list.delete(*device_list.get_children())
device_data.delete('0.0', tk.END)
scan_result.clear()
try:
async with BleakScanner() as scanner:
message_variable.set('Scanning (10 secs)...')
await asyncio.sleep(10)
message_variable.set('Scanning finished.')
result = scanner.discovered_devices_and_advertisement_data
if result:
scan_result.update(result)
for key in result:
device, adv_data = result[key]
if device.name:
name = device.name
else:
name = 'No name'
device_list.insert('', 'end', text=f'{name}, {device.address}')
except (OSError, BleakError):
message_variable.set('Bluetooth not available (off or absent)')
def device_selection(event):
"""Show advertised data of selected device."""
for item in device_list.selection():
name, key = device_list.item(item, 'text').split(',')
device_data.delete('0.0', tk.END)
device_data.insert('0.0', str(scan_result[key.strip()][1]))
message_variable.set(f'Device address: {str(key)}')
async def connect():
"""Connect to or disconnect from selected/connected device."""
global is_connected
if is_connected:
message_variable.set('Trying to disconnect...')
disconnect.set()
return
# Pick the BLE device from the scan result:
for item in device_list.selection():
_, key = device_list.item(item, 'text').split(',')
device = scan_result[key.strip()][0]
name = device.name if device.name is not None else device.address
try:
message_variable.set(f'Trying to connect to {name}')
async with BleakClient(device, disconnect_callback):
message_variable.set(f'Device {name} is connected!')
is_connected = True
while not disconnect.is_set():
await asyncio.sleep(0.1)
is_connected = False
return
except (BleakError, asyncio.TimeoutError):
message_variable.set(f'Connecting to {name}\nnot successful')
is_connected = False
def disconnect_callback(client):
"""Handle disconnection.
This callback is called when the device is disconnected.
"""
message_variable.set(f'Device {client.address} has/was\ndisconnected')
async def show():
"""Handle the GUI's update method asynchronously.
Most of the time the program is waiting here and
updates the GUI regularly.
This function principally replaces the Tkinter mainloop.
"""
while not stop.is_set():
main_window.update()
await asyncio.sleep(0.1)
def stop_loop():
"""Set stop event."""
stop.set()
async def main():
"""Start the GUI."""
global stop, disconnect
stop = asyncio.Event()
disconnect = asyncio.Event()
build_gui()
await show()
main_window.destroy()
asyncio.run(main()) |
Beta Was this translation helpful? Give feedback.
1 reply
-
|
@MarkusPiotrowski Thanks for the example. It will help me a lot :) |
Beta Was this translation helpful? Give feedback.
-
|
@dlech You are free to use this for the |
Beta Was this translation helpful? Give feedback.
2 replies
-
|
Hi @MarkusPiotrowski , I still have trouble avoiding hanging. How can I know that the comms have hanged? Uploading the code and the logger: Your code + logger LOG: |
Beta Was this translation helpful? Give feedback.
-
I always start by logging Bluetooth packets and inspecting them with Wireshark. |
Beta Was this translation helpful? Give feedback.
-
|
FYI, with Bleak 0.22.0 (just released a few days ago) on Windows this is affected by #1565. So, I would recommend printing or logging the actual exception message to be able to see this. To fix it, I plan on requiring programs on Windows with a GUI to have to tell Bleak that we really know what we are doing like this: try:
from bleak.backends.winrt.util import allow_sta
# tell Bleak we are using a graphical user interface that has been properly configured to work with asyncio
allow_sta()
except ImportError:
pass |
Beta Was this translation helpful? Give feedback.
3 replies
-
|
I have updated Bleak to 0.22.0 and that did not go well. My script does not run anymore. I'm in trouble. Please help. The line: Now returns:
Changed to: And, of course, it returns:
The function I call is: To update Bleak, I used the PyCharm packets manager, but allow_sta was not found. Result: Then I used pip:
|
Beta Was this translation helpful? Give feedback.
-
|
I have downgraded Bleak to 0.21.1 and my script works again. |
Beta Was this translation helpful? Give feedback.
-
|
As mentioned in the other thread, Bleak 0.22.1 contains the new function needed to work around the issue and it requires adding some extra code to your program as mentioned in #481 (comment) and in the troubleshooting page in the docs. |
Beta Was this translation helpful? Give feedback.
-
This leaks task objects that could be garbage collected before the task is complete. |
Beta Was this translation helpful? Give feedback.
3 replies
-
|
Sorry, this was 'just' another shortcut for the example, to avoid writing a regular method that runs the ....
command=connect,
....
def connect():
asyncio.create_task(async_connect())
....
asyncio def async_connect():
do the stuffwould be OK? |
Beta Was this translation helpful? Give feedback.
-
|
This still leaks the task returned by If requiring Python >=3.11 is OK, I would use an Otherwise, the create_task() docs suggest how to keep a handle to the tasks. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for the pointer! |
Beta Was this translation helpful? Give feedback.
-
|
I recently made an asyncio Tkinter GUI (with Custom Tkinter). I found that I had to tick the asyncio event loop manually in order for everything to work as expected, especially dragging and resizing the windows. That is, do not run the GUI loop from the asyncio event loop, run the asyncio event loop from the GUI loop. This is accomplished by using the after() method to give the Tkinter loop a callback that ticks the asyncio loop. class App(ctk.CTk):
_UPDATE_INTERVAL_ms: Final = 20
def __init__(self) -> None:
super().__init__()
self._close_event: Final = asyncio.Event()
self._loop: Final = asyncio.get_event_loop()
self.after(0, self._async_update)
def _on_close(self) -> None:
logger.info("User closed the window")
self._close_event.set()
def _async_update(self) -> None:
"""Run the asyncio event loop once and reschedule."""
self._loop.call_soon(self._loop.stop)
self._loop.run_forever()
if self._close_event.is_set():
self.quit()
self.after(self._UPDATE_INTERVAL_ms, self._async_update)
def run(self) -> None:
tasks: Final = [self._loop.create_task(self._one_of_your_async_tasks_for_example())]
self.mainloop()
for task in tasks:
task.cancel()
async def f() -> None:
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
logger.error(result)
logger.error(format_tb(result.__traceback__))
self._loop.run_until_complete(f())
self.destroy()TBH, it doesn't feel great, can't really recommend Tkinter at this point, but it is working for everyone. |
Beta Was this translation helpful? Give feedback.
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
-
Have spent the last few days getting Bleak to work as a constant listener for Notification characteristics within a Tkinter GUI system using ASYNCIO and was asked to pass along what I learned getting it to work. This may not be the best solution as I am no Python expert and this is for an internal tool for MFG test/validation, plus I still have some clean-up to do. The two threads communicate using Queues.
From the main tkinter loop the design spawns a second BLE Thread. That thread does some housekeeping stuff as well but the important part is the ASYNCIO loop handling and timing:
The run_forever ASYNCIO loop must be stopped and started in order for the BLE Thread to be able to execute its time.sleep in order for the tkinter loop to have a chance to run.
The tasks are created and attached to the asyncio loop when commanded from the tkinter main (the bleThread() handles the disconnect's task cancellation when needed):
The ASYNCIO loop is stopped by the one task after it does its housekeeping:
Hope this helps.
Beta Was this translation helpful? Give feedback.
All reactions