如何使用 urwid 和 asyncio 创建异步程序?
How can I create an asynchronous program with urwid and asyncio?
我想用aiortc建一个聊天室。首先,我想用 urwid 作为 cli 和 asyncio 构建一个模型。
urwid 部分已经 运行ning 正常,可以进行用户输入。我知道想要 运行 一个生成随机文本并作为聊天客户端在该聊天室中发短信的协程。
我已经尝试 运行 我的 urwid 函数与 mainloop 作为异步协程,但没有成功。
我不知道如何将异步函数集成到我的 urwid 主循环中。
def unhandled(key):
"""
functin to handle input
"""
global TEXT_INPUT
global lw_user_input
global lw_chatroom
global listbox_chatroom
if not isinstance(key, tuple):
if key == 'enter':
del lw_user_input[-1]
# create widegt and fill with user input
lw_chatroom.append(widget)
TEXT_INPUT = ""
listbox_chatroom.set_focus(len(lw_chatroom)-1, 'above')
elif key == 'esc':
raise urwid.ExitMainLoop()
elif key == 'backspace':
if len(lw_user_input) > 0:
user_input = lw_user_input[0].get_text()[0]
user_input = user_input[:-1]
del lw_user_input[-1]
TEXT_INPUT = user_input
lw_user_input.append(urwid.Text(TEXT_INPUT))
else:
TEXT_INPUT += key # repr(key)
if len(lw_user_input) > 0:
del lw_user_input[-1]
lw_user_input.append(urwid.Text(TEXT_INPUT))
else:
lw_user_input.append(urwid.Text(key))
def generate_output():
global lw_chatroom
global listbox_chatroom
while True:
# generate text and widgets and post with delay
lw_chatroom.append(chat_widget)
listbox_chatroom.set_focus(len(lw_chatroom)-1, 'above')
def create_cli():
# generate all widgets
uloop = urwid.MainLoop(frame, palette, screen,
unhandled_input=unhandled)
uloop.start()
if __name__ == '__main__':
create_cli()
我想 运行 generate_output() 和 unhandled(key) 异步。我不知道该怎么做。
好的,我明白了。
就这么简单:
aloop = asyncio.get_event_loop()
ev_loop = urwid.AsyncioEventLoop(loop=aloop)
loop = urwid.MainLoop(frame, palette, screen,
unhandled_input=unhandled, event_loop=ev_loop)
aloop.create_task(generate_output())
loop.run()
我想用aiortc建一个聊天室。首先,我想用 urwid 作为 cli 和 asyncio 构建一个模型。 urwid 部分已经 运行ning 正常,可以进行用户输入。我知道想要 运行 一个生成随机文本并作为聊天客户端在该聊天室中发短信的协程。
我已经尝试 运行 我的 urwid 函数与 mainloop 作为异步协程,但没有成功。 我不知道如何将异步函数集成到我的 urwid 主循环中。
def unhandled(key):
"""
functin to handle input
"""
global TEXT_INPUT
global lw_user_input
global lw_chatroom
global listbox_chatroom
if not isinstance(key, tuple):
if key == 'enter':
del lw_user_input[-1]
# create widegt and fill with user input
lw_chatroom.append(widget)
TEXT_INPUT = ""
listbox_chatroom.set_focus(len(lw_chatroom)-1, 'above')
elif key == 'esc':
raise urwid.ExitMainLoop()
elif key == 'backspace':
if len(lw_user_input) > 0:
user_input = lw_user_input[0].get_text()[0]
user_input = user_input[:-1]
del lw_user_input[-1]
TEXT_INPUT = user_input
lw_user_input.append(urwid.Text(TEXT_INPUT))
else:
TEXT_INPUT += key # repr(key)
if len(lw_user_input) > 0:
del lw_user_input[-1]
lw_user_input.append(urwid.Text(TEXT_INPUT))
else:
lw_user_input.append(urwid.Text(key))
def generate_output():
global lw_chatroom
global listbox_chatroom
while True:
# generate text and widgets and post with delay
lw_chatroom.append(chat_widget)
listbox_chatroom.set_focus(len(lw_chatroom)-1, 'above')
def create_cli():
# generate all widgets
uloop = urwid.MainLoop(frame, palette, screen,
unhandled_input=unhandled)
uloop.start()
if __name__ == '__main__':
create_cli()
我想 运行 generate_output() 和 unhandled(key) 异步。我不知道该怎么做。
好的,我明白了。
就这么简单:
aloop = asyncio.get_event_loop() ev_loop = urwid.AsyncioEventLoop(loop=aloop) loop = urwid.MainLoop(frame, palette, screen, unhandled_input=unhandled, event_loop=ev_loop) aloop.create_task(generate_output()) loop.run()