我正在编写一个 Python curses 应用程序,它通过分别通过进程'stdin
和发送和接收字符串来控制外部(Linux,如果有帮助的话)进程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()
函数也会旋转(我找不到其他方式), - urwid 界面不会刷新,直到按钮回调函数退出,这会阻止
urwid
' 的主循环刷新屏幕。
我可以使用线程,但从我读过的urwid
支持asyncio
来看,这就是我想要实现的。urwid
你可以称我为笨蛋,因为即使在浏览asyncio 示例和阅读 Pythonasyncio
文档之后,我也无法清楚地弄清楚如何。
考虑到这些方法中的任何一种都有更改的空间,我仍然希望保持过程控制类——即包含read()
和的类send()
——尽可能通用。
到目前为止,我没有尝试过导致界面在进程繁忙时被更新。接收进程“通知”的组件是一个普通的urwid.Text()
小部件。