0

我尝试为命令行编写某种渲染器，该渲染器应该能够使用 and 从另一个数据源打印数据，stdin这是 python-blessings 的改进版本。asyncioblessed

这是我到目前为止所拥有的：

import asyncio
from blessed import Terminal

@asyncio.coroutine
def render(term):
    while True:
        received = yield
        if received:
            print(term.bold + received + term.normal)

async def ping(renderer):
    while True:
        renderer.send('ping')
        await asyncio.sleep(1)

async def input_reader(term, renderer):
    while True:
        with term.cbreak():
            val = term.inkey()
            if val.is_sequence:
               renderer.send("got sequence: {0}.".format((str(val), val.name, val.code)))
            elif val:
               renderer.send("got {0}.".format(val))

async def client():
    term = Terminal()
    renderer = render(term)
    render_task = asyncio.ensure_future(renderer)
    pinger = asyncio.ensure_future(ping(renderer))
    inputter = asyncio.ensure_future(input_reader(term, renderer))
    done, pending = await asyncio.wait(
        [pinger, inputter, renderer],
        return_when=asyncio.FIRST_COMPLETED,
    )
    for task in pending:
        task.cancel()

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(client())
    asyncio.get_event_loop().run_forever()

出于学习和测试的目的，只有一个'ping'每秒发送一次的转储 ping 和另一个例程，该例程应该获取关键输入并将它们发送到我的渲染器。

但是 ping 仅在使用此代码的命令行中出现一次，并且 input_reader 可以按预期工作。当我用pong类似的东西替换 input_reader 时，ping一切都很好。

这是键入“pong”时的样子，尽管如果写“pong”需要十秒钟：

$ python async_term.py
ping
got p.
got o.
got n.
got g.
4

1 回答 1

0

似乎祝福不是为了与 asyncio 一起正常工作而构建的： inkey()是一种阻塞方法。这将阻止任何其他 couroutine。

您可以使用带有kbhit()and的循环await asyncio.sleep()来控制其他协程 - 但这不是一个干净的 asyncio 解决方案。

于 2017-04-01T13:10:42.627 回答