文本 (Python TUI) - 启用 long-运行，外部异步功能

Question

Textual 被认为是消费事件的前端（例如，使用 Redis 的 PUBSUB 来消费和显示传入的事件和数据）。

下面是通用代码，尝试无限期地执行后台异步任务运行，同时保留文本的所有功能：

#!/usr/bin/env python

import asyncio
from datetime import datetime

from rich.align import Align
from textual.app import App
from textual.widget import Widget


class AsyncWidget(Widget):

    counter = 0

    async def async_functionality(self):
        while True:
            await asyncio.sleep(0.2)  # Mock async functionality
            self.counter += 1
            self.refresh()  # This is required for ongoing refresh
            self.app.refresh()  # Also required for ongoing refresh, unclear why, but commenting-out breaks live refresh.

    async def on_mount(self):
        await self.async_functionality()

    def render(self) -> Align:
        now = datetime.strftime(datetime.now(), "%H:%M:%S.%f")[:-5]
        text = f"{now}\nCounter: {self.counter}"
        return Align.center(text, vertical="middle")


class AsyncApp(App):
    async def on_load(self) -> None:
        await self.bind("escape", "quit", "Quit")

    async def on_mount(self) -> None:
        await self.view.dock(AsyncWidget())


AsyncApp.run(title="AsyncApp", log="async_app.log")

运行应用程序并立即按下绑定的 esc 键优雅地终止程序。这是生成的日志（另外，TUI 小部件上的计数器每 0.2 秒明显增加）：

# WITH ASYNC FUNCTIONALITY, NO MOUSE EVENT

driver=<class 'textual.drivers.linux_driver.LinuxDriver'>
Load() >>> AsyncApp(title='Textual')
Mount() >>> AsyncApp(title='AsyncApp')
Mount() >>> DockView(name='DockView#1')
Mount() >>> AsyncWidget(name='AsyncWidget#1')
view.forwarded Key(key='escape')
Key(key='escape') >>> AsyncApp(title='AsyncApp')
ACTION AsyncApp(title='AsyncApp') quit
CLOSED AsyncApp(title='AsyncApp')
PROCESS END

但是-鼠标点击小部件的主体（触发 set_focus 事件）以某种方式 [b] 锁定进一步的关键功能（即 esc 不起作用，view.forwarded如下面的日志所示，密钥不会触发）。而且，必须用ctrl-c来终止：

# WITH ASYNC FUNCTIONALITY, MOUSE EVENT BREAKS TUI

driver=<class 'textual.drivers.linux_driver.LinuxDriver'>
Load() >>> AsyncApp(title='Textual')
Mount() >>> AsyncApp(title='AsyncApp')
Mount() >>> DockView(name='DockView#1')
Mount() >>> AsyncWidget(name='AsyncWidget#1')
set_focus AsyncWidget(name='AsyncWidget#1') <--- mouse clicked anywhere on widget body
Key(key='ctrl+c') >>> AsyncApp(title='AsyncApp') <-- only way to terminate
ACTION AsyncApp(title='AsyncApp') quit
CLOSED AsyncApp(title='AsyncApp')
PROCESS END

此问题显然与 long-运行ning 异步功能有关。从小部件的 on_mount 指令中注释掉 await self.async_functionality() 显示文本的预期行为（鼠标单击 down/up 事件触发，esc 退出有效）：

# WORKING EXAMPLE, W/O ASYNC FUNCTIONALITY

driver=<class 'textual.drivers.linux_driver.LinuxDriver'>
Load() >>> AsyncApp(title='Textual')
Mount() >>> AsyncApp(title='AsyncApp')
Mount() >>> DockView(name='DockView#1')
Mount() >>> AsyncWidget(name='AsyncWidget#1')
set_focus AsyncWidget(name='AsyncWidget#1')
MouseDown(x=56, y=18, button=1) >>> AsyncWidget(name='AsyncWidget#1')
MouseUp(x=56, y=18, button=1) >>> AsyncWidget(name='AsyncWidget#1')
Click(x=56, y=18, button=1) >>> AsyncWidget(name='AsyncWidget#1')
Key(key='escape') >>> AsyncApp(title='AsyncApp')
ACTION AsyncApp(title='AsyncApp') quit
CLOSED AsyncApp(title='AsyncApp')
PROCESS END

任何关于如何在与 TUI 交互时实现此长期运行宁异步功能的建议都很棒。

Answer 1

文本小部件有一个按顺序处理事件的内部消息队列。您的 on_mount 处理程序正在处理这些事件之一，但因为它是一个无限循环，所以您正在阻止处理更多事件。

如果你想在后台处理一些东西，你需要创建一个新的异步任务。请注意，您不能等待该任务，因为这也会阻止处理程序返回。

有关任务的更多信息，请参阅 asyncio docs。

文本 (Python TUI) - 启用 long-运行，外部异步功能

Textual (Python TUI) - Enabling long-running, external asyncio functionality

python

python-asyncio

rich