如何使用urwid和asyncio使长任务无阻塞?

我正在编写一个Python curses应用程序,它通过进程发送和接收字符串来控制外部(Linux,如果有帮助的话)进程' stdin 和 stdout 分别地接口使用 urwid . 我编写了一个类来控制外部进程,并为几个urwid组件编写了几个其他类。

我还有一个按钮,用来向外部进程发送命令。然而,进程不会立即响应,其任务通常需要几秒钟,在此期间,我希望界面不要冻结。

下面是我如何运行子进程:

def run(self, args):
    import io, fcntl, os
    from subprocess import Popen, PIPE

    # Run wpa_cli with arguments, use a thread to feed the process with an input queue
    self._pss = Popen(["wpa_cli"] + args, stdout=PIPE, stdin=PIPE)
    self.stdout = io.TextIOWrapper(self._pss.stdout, encoding="utf-8")
    self.stdin = io.TextIOWrapper(self._pss.stdin, encoding="utf-8", line_buffering=True)

    # Make the process' stdout a non-blocking file
    fd = self.stdout.fileno()
    fl = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
    ...

我必须使进程的输出流无阻塞,才能解析其输出。我不知道这对我的问题是否重要。

以下是我用来控制子进程输入和输出流的方法:

def read(self, parser=None, transform=None, sentinel='>'):
    """ Reads from the controlled process' standard output until a sentinel
    is found. Optionally execute a callable object with every line. Parsed
    lines are placed in a list, which the function returns upon exiting. """

    if not transform:
        transform = lambda str: str

    def readline():
        return transform(self.stdout.readline().strip())

    # Keep a list of (non-empty) parsed lines
    items = []
    for line in iter(readline, sentinel):
        if callable(parser):
            item = parser(line)
            if item is not None:
                items.append(item)
    return items

def send(self, command, echo=True):
    """ Sends a command to the controlled process. Action commands are
    echoed to the standard output. Argument echo controls whether or not
    they're removed by the reader function before parsing. """

    print(command, file=self.stdin)

    # Should we remove the echoed command?
    if not echo:
        self.read(sentinel=command)

我提到的按钮只是从主脚本输入函数中设置了回调。该回调应该向子进程发送命令,并在结果输出行中循环,直到找到给定的文本,在这种情况下,回调函数退出。在此之前,该过程会输出一些我需要捕捉并显示在用户界面中的有趣信息。

例如:

def button_callback():
    # This is just an illustration

    filter = re.compile('(event1|event2|...)')

    def formatter(text):
        try:
            return re.search(filter, text).group(1)
        except AttributeError:
            return text

    def parser(text):
        if text == 'event1':
            # Update the info Text accordingly
        if text == 'event2':
            # Update the info Text accordingly

    controller.send('command')
    controller.read(sentinel='beacon', parser=parser, transform=formatter)

需要注意的是:

• 这个 read() 函数旋转(我找不到其他方法),即使进程输出流是静默的,直到从(可选)解析的行中读取sentinel值,
• 直到按钮回调函数退出,urwid接口才会刷新,这会阻止 乌尔维德 刷新屏幕的主循环。

我可以用一根线,但从我读到的 乌尔维德 支架 asyncio 这就是我想要实现的。你可以叫我哑巴,因为我只是不清楚,即使在浏览之后 乌尔维德 asyncio示例和Python阅读 异步IO 文档

考虑到有修改这些方法的余地,我仍然希望保留进程控制类,即包含 读取() 和 send() 尽可能通用。

到目前为止,我所做的任何尝试都没有导致在进程繁忙时更新接口。接收进程“通知”的组件是普通的 urwid.Text() 小装置。