英文:
How to update a Textual TUI within a function call?
问题
我正在使用Textual框架来创建一个简单的TUI。
任务是显示ChatGPT-prompts的结果,就像它们被实时传输一样。
问题是:我无法弄清楚如何更新应用程序,以便它显示实时传输的结果。
这是一个最简示例。我期望Counter
标签能快速显示从0到9的数字。结果是,我只在等待了一秒后才看到数字9。
import time
from textual.app import App
from textual.widgets import Header, Label
from textual.reactive import reactive
def randomgen():
for i in range(10):
time.sleep(0.1)
yield str(i)
class Counter(Label):
countervalue = reactive("Press Enter to start")
def watch_countervalue(self, countervalue):
self.update(countervalue)
class Minimal(App):
def compose(self):
yield Header()
yield Counter(id="counter")
def on_key(self, event):
if event.key == "enter":
for i in randomgen():
self.query_one("#counter").countervalue = i # pyright: ignore
if __name__ == "__main__":
app = Minimal()
app.run()
英文:
I'm using the Textual framework for a simple TUI.
The task is to display the results of ChatGPT-prompts as they are streamed.
Problem is: I cannot figure out, how to update the app, so that it shows the streamed results.
Here is a minimal example. I expect the Counter
label to display the numbers from 0 to 9 very quickly. The result is, that I only get the number 9 after waiting for a second.
import time
from textual.app import App
from textual.widgets import Header, Label
from textual.reactive import reactive
def randomgen():
for i in range(10):
time.sleep(0.1)
yield str(i)
class Counter(Label):
countervalue = reactive("Press Enter to start")
def watch_countervalue(self, countervalue):
self.update(countervalue)
class Minimal(App):
def compose(self):
yield Header()
yield Counter(id="counter")
def on_key(self, event):
if event.key == "enter":
for i in randomgen():
self.query_one("#counter").countervalue = i # pyright: ignore
if __name__ == "__main__":
app = Minimal()
app.run()
答案1
得分: 2
你正确更新了countervalue
,但你的代码中还有其他问题,这些问题阻止了它正常工作。
首先,你在使用time.sleep
,它是一个阻塞调用。阻塞调用会阻止asyncio执行其他任务。你可以将time.sleep
替换为await asyncio.sleep
。
这是你代码的改进版本:
import asyncio
from textual.app import App
from textual.widgets import Header, Label
from textual.reactive import reactive
def randomgen():
for i in range(10):
yield str(i)
class Counter(Label):
countervalue = reactive("Press Enter to start")
def watch_countervalue(self, countervalue):
self.update(countervalue)
class Minimal(App):
def compose(self):
yield Header()
yield Counter(id="counter")
async def on_key(self, event):
if event.key == "enter":
for i in randomgen():
self.query_one("#counter").countervalue = i
await asyncio.sleep(0.1)
if __name__ == "__main__":
app = Minimal()
app.run()
这样修改后,你的代码会按照你预期的方式工作。不过,要避免在消息处理程序中执行耗时操作,因为在操作完成之前,你的应用程序无法处理消息。
最好将长时间运行的任务放在worker中执行。以下是一个示例:
import asyncio
from textual.app import App
from textual import work
from textual.widgets import Header, Label
from textual.reactive import reactive
class Counter(Label):
countervalue = reactive("Press Enter to start")
def watch_countervalue(self, countervalue):
self.update(countervalue)
class Minimal(App):
def compose(self):
yield Header()
yield Counter(id="counter")
@work
async def run_counter(self) -> None:
for i in range(10):
self.query_one("#counter", Counter).countervalue = str(i)
await asyncio.sleep(0.1)
async def on_key(self, event):
if event.key == "enter":
self.run_counter()
if __name__ == "__main__":
app = Minimal()
app.run()
你还可以使用set_interval
来实现类似的功能。
英文:
You are updating countervalue
correctly, but there are other issues with your code that are preventing it from working.
The first is that you are using time.sleep
which is a blocking call. Blocking calls will prevent asyncio doing anything else. You can replace time.sleep
with await asyncio.sleep
.
import asyncio
from textual.app import App
from textual.widgets import Header, Label
from textual.reactive import reactive
def randomgen():
for i in range(10):
yield str(i)
class Counter(Label):
countervalue = reactive("Press Enter to start")
def watch_countervalue(self, countervalue):
self.update(countervalue)
class Minimal(App):
def compose(self):
yield Header()
yield Counter(id="counter")
async def on_key(self, event):
if event.key == "enter":
for i in randomgen():
self.query_one("#counter").countervalue = i # pyright: ignore
await asyncio.sleep(0.1)
if __name__ == "__main__":
app = Minimal()
app.run()
This works in the way I think you intended it to. Although you should avoid doing work that takes a while in message handlers, as your app won't be able to process messages until it completes.
It is better to have long running tasks in a worker. Here's an example:
import asyncio
from textual.app import App
from textual import work
from textual.widgets import Header, Label
from textual.reactive import reactive
class Counter(Label):
countervalue = reactive("Press Enter to start")
def watch_countervalue(self, countervalue):
self.update(countervalue)
class Minimal(App):
def compose(self):
yield Header()
yield Counter(id="counter")
@work
async def run_counter(self) -> None:
for i in range(10):
self.query_one("#counter", Counter).countervalue = str(i)
await asyncio.sleep(0.1)
async def on_key(self, event):
if event.key == "enter":
self.run_counter()
if __name__ == "__main__":
app = Minimal()
app.run()
You could also implement this with set_interval
.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论