使用 pysimplegui 实现 asyncIO

Question

我正在尝试在 python 中使用 pysimplegui 实现 asyncio。 在此 GUI 示例中，两个按钮（按钮 2 和按钮 3）模拟要完成的长任务。

目标：

• 即使调用的函数（通过按钮）需要一些时间才能返回结果，也能够返回 GUI 界面。

预期结果：

• 如果按钮 2 或按钮 3 或两者都被按下，它们都会继续执行其任务，用户可以返回 GUI 继续执行其他任务。

当前结果：

• 一旦按下按钮 2 或按钮 3，任务就会被阻止并继续直到结束，并且 GUI 会挂起直到结束。

import PySimpleGUI as sg
import asyncio
import time

sg.theme('Light Blue 3')
# This design pattern simulates button callbacks
# This implementation uses a simple "Dispatch Dictionary" to store events and functions

# The callback functions
async def button1():
    print('Button 1 callback')
    return 'nothing'

async def button2():
    print('Button 2 callback')
    for i in range(1,20):
        await asyncio.sleep(3)
        print(f"Button 2: {i}")
    return f"button2 end"

async def button3():
    print('Button 3 callback')
    for i in range(1,10):
        await asyncio.sleep(3)
        print(f"Button 3: {i}")
    return f"button3: end"


# Lookup dictionary that maps button to function to call
dispatch_dictionary = {'1':button1, '2':button2, '3':button3}
# Layout the design of the GUI
layout = [[sg.Text('Please click a button', auto_size_text=True)],
        [sg.Button('1'), sg.Button('2'), sg.Button('3'), sg.Quit()]]
# Show the Window to the user__TIMEOUT__
window = sg.Window('Button callback example', layout)
# Event loop. Read buttons, make callbacks
while True:
    # Read the Window
    event, values = window.read()
    if event in ('Quit', sg.WIN_CLOSED):
        break
    if event == '__TIMEOUT__':
        continue
    # Lookup event in function dictionary
    if event in dispatch_dictionary:
        func_to_call = dispatch_dictionary[event]   # get function from dispatch dictionary
        print(asyncio.run(func_to_call()))
    else:
        print('Event {} not in dispatch dictionary'.format(event))

window.close()
# All done!
sg.popup_ok('Done')

我以为我按照规则应用了async/wait。我错过了什么吗？

Answer 1

asyncio.run()

执行一个协程并阻塞直至完成。它不启动并行线程来运行协程。

您有两个选择：

1. 不要使用asyncio，使用Threading为每个长操作启动一个新线程。
2. 在程序开始时启动一个带有异步事件循环的线程，然后使用

   asyncio.run_coroutine_threadsafe()

   将协程从主 GUI 线程调度到事件循环上。

我将在这里解释选项 2.. 程序开始时的示例：

from threading import Thread

def asyncloop(loop):
    # Set loop as the active event loop for this thread
    asyncio.set_event_loop(loop)
    # We will get our tasks from the main thread so just run an empty loop    
    loop.run_forever()

# create a new loop
loop = asyncio.new_event_loop()
# Create the new thread, giving loop as argument
t = Thread(target=asyncloop, args=(loop,))
# Start the thread
t.start()

稍后在按钮事件代码中（在主线程中）：

asyncio.run_coroutine_threadsafe(func_to_call(), loop)

这将安排协程作为我们创建的线程内的并行任务运行。

Answer 2

我今天遇到了这个问题，我最终通过将整个 GUI 主循环包装到自己的任务中来解决它，并在循环中使用

await asyncio.sleep(0)

让实际任务运行

import PySimpleGUI as sg
import asyncio
import time

sg.theme('Light Blue 3')
# This design pattern simulates button callbacks
# This implementation uses a simple "Dispatch Dictionary" to store events and functions

# The callback functions
async def button1():
    print('Button 1 callback')
    return 'nothing'

async def button2():
    print('Button 2 callback')
    for i in range(1,20):
        await asyncio.sleep(3)
        print(f"Button 2: {i}")
    return f"button2 end"

async def button3():
    print('Button 3 callback')
    for i in range(1,10):
        await asyncio.sleep(3)
        print(f"Button 3: {i}")
    return f"button3: end"

async def main_window():
    # List of tasks that are active
    tasks = []
    # Lookup dictionary that maps button to function to call
    dispatch_dictionary = {'1':button1, '2':button2, '3':button3}
    # Layout the design of the GUI
    layout = [[sg.Text('Please click a button', auto_size_text=True)],
            [sg.Button('1'), sg.Button('2'), sg.Button('3'), sg.Quit()]]
    # Show the Window to the user__TIMEOUT__
    window = sg.Window('Button callback example', layout)
    # Event loop. Read buttons, make callbacks
    while True:
        # Read the Window
        event, values = window.read(timeout=100)
        if event in ('Quit', sg.WIN_CLOSED):
            break
        elif event == '__TIMEOUT__':
            pass
        # Lookup event in function dictionary
        elif event in dispatch_dictionary:
            func_to_call = dispatch_dictionary[event]   # get function from dispatch dictionary
            tasks.append(asyncio.create_task(func_to_call()))
        else:
            print('Event {} not in dispatch dictionary'.format(event))
        # Check which tasks have finished, and make a new list
        # In newer versions of python, 3.11 and high use asyncio.TaskGroup
        new_tasks = []
        for task in tasks:
            if task.done():
                print(task.result())
            else:
                new_tasks.append(task)
        tasks = new_tasks
        await asyncio.sleep(0)
    # Cancel and await all remaining tasks
    for task in tasks:
        task.cancel()
        await task
    window.close()
    # All done!
    sg.popup_ok('Done')

asyncio.run(main_window())

这是使用 python 3.7.11 和 PySimpleGUI 5.0.4 进行测试的