混合使用 asyncio 和 Kivy：如何同时启动 asyncio 循环和 Kivy 应用程序？

Question

迷失在异步中。

我同时在学习 Kivy 和 asyncio，并且陷入了解决运行 Kivy 和运行 asyncio 循环的问题，不管我怎么转，两者都是阻塞调用，需要按顺序执行（好吧，我希望我错了），例如

loop = asyncio.get_event_loop()
loop.call_soon(MyAsyncApp().run())
loop.run_forever()

我当前的尝试导致应用程序启动，但没有协程正在执行，例如当我单击 "Connect" 按钮时，我应该使用 loop.call_soon 开始安排和执行任务，但没有任何反应。

有人可以看看我的代码并提出解决问题的正确方法吗？

import asyncio
import random
import time
from kivy.app import App
from kivy.lang import Builder

ui = Builder.load_string('''
BoxLayout:
    orientation: 'vertical'
    GridLayout:
        rows: 2
        cols: 2
        Label:
            text: 'Status:'
            size_hint: 0.3, 1
        Label:
            id: status
            text: ''
        Label:
            text: 'Data:'
            size_hint: 0.7, 1
        Label:
            id: data
            text: ''
    BoxLayout:
        direction: 'horizontal'
        Button:
            text: 'Get Data'
            on_press: app.connect()
        Button:
            text: 'Stop Data'
            on_press: pass
''')

class MyAsyncApp(App):

    def __init__(self):
        super(self.__class__, self).__init__()

        self.x_connected = None
        self.x_widget_data = None
        self.x_widget_status = None
        self.x_loop = asyncio.get_event_loop()

    def build(self):
        return ui

    def connect(self):
        # Get widget
        self.x_widget_status = self.root.ids.status

        # Update status
        self.x_widget_status.text = 'Preparing to connect...'

        # Connect using asyncio
        # --> But the loop must be already running <---
        self.x_loop.call_soon(self.do_connect)

    async def do_connect(self):
        # Connect asynchronously

        # Get widget
        self.x_widget_data = self.root.ids.data

        # Update status
        self.x_connected = False
        self.x_widget_status.text = 'Connecting...'

        # Perform actual actions
        try:
            result = await self.feed_kivy()
            if result:
                self.x_widget_status.text = 'Service not available: ' + result
                return
        except Exception as e:
            self.x_widget_status.text = 'Error while connecting'
            return

        # Update status
        self.x_connected = True
        self.x_widget_status.text = 'Connected'

    async def feed_kivy(self):
        # Deliver fresh data at random interval

        # Some slow process to get data
        result = await asyncio.sleep(random.randint(1, 5), time.time())
        self.x_widget_data.text = result

        # Reschedule ourselves
        await self.x_loop.call_soon(self.feed_kivy())


def main():
    # If loop started here, app is never started
    loop = asyncio.get_event_loop()
    loop.call_soon(MyAsyncApp().run())
    loop.run_forever()
    loop.close()


if __name__ == '__main__':
    main()

Answer 1

My current attempt results in the application being launched, but no coroutine is being executed

它的发生是因为 MyAsyncApp().run() 阻止了执行流并且控制永远不会 returns 到 asyncio 的事件循环。这就是所有事件循环的工作方式。

与其手动尝试跨越两个循环，更简单的方法是使用现有的尝试：

https://github.com/kivy/kivy/pull/5241

此 PR 来自 Kivy 的一位开发人员，包含带有解释和用法示例的工作实现。

但是它还没有合并到 master 中：你需要用这个 PR 手动构建 Kivy。

Answer 2

我最近在 Kivy 的文档中找到 this。现在 Kivy 支持异步循环库，比如 asyncio 和 trio。在 link 你可以找到这个例子：

Asyncio 示例 ~~~~~~~~~~~~~–

import asyncio

from kivy.app import async_runTouchApp
from kivy.uix.label import Label


loop = asyncio.get_event_loop()
loop.run_until_complete(
    async_runTouchApp(Label(text='Hello, World!'), async_lib='asyncio'))
loop.close()

三重奏示例~~~~~~~~~~–

import trio

from kivy.app import async_runTouchApp
from kivy.uix.label import Label

from functools import partial

# use functools.partial() to pass keyword arguments:
async_runTouchApp_func = partial(async_runTouchApp, async_lib='trio')

trio.run(async_runTouchApp_func, Label(text='Hello, World!'))

抱歉英语不好。希望对你有帮助。

混合使用 asyncio 和 Kivy：如何同时启动 asyncio 循环和 Kivy 应用程序？

Mixing asyncio and Kivy: How to start the asyncio loop and the Kivy application at the same time?

python

kivy

python-asyncio