如何在urwid中侦听文件更改?

Question

我想远程控制一个使用urwid作为用户界面的python应用程序。

我的想法是创建一个文件，将其名称作为命令行参数传递给应用程序，每当我写入文件时，应用程序都会从该文件读取。

Urwid的事件循环有一个方法watch_file(fd, callback)。这个方法描述为"当fd有一些数据要读取时调用callback()。"这听起来正是我想要的，但它会导致无限循环。回调会尽可能频繁地执行，尽管文件是空的。即使我删除了文件，回调仍然被调用。

#!/usr/bin/env python3
import urwid
import atexit
def onkeypress(key, size=None):
if key == 'q':
raise urwid.ExitMainLoop()
text.set_text(key)
def onfilechange():
text.set_text(cmdfile.read())
# clear file so that I don't read already executed commands again
# and don't run into an infinite loop - but I am doing that anyway
with open(cmdfile.name, 'w') as f:
pass
cmdfile = open('/tmp/cmd', 'rt')
atexit.register(cmdfile.close)
text = urwid.Text("hello world")
filler = urwid.Filler(text)
loop = urwid.MainLoop(filler, unhandled_input=onkeypress)
loop.watch_file(cmdfile, onfilechange)
if __name__ == '__main__':
loop.run()

(我最初的想法是打开文件仅用于读取，而不是保持它一直打开，但fd必须是一个文件对象，而不是路径。)

Urwid提供了几个不同的事件循环。默认情况下，使用SelectEventLoop。GLibEventLoop有相同的行为，它运行到一个无限循环。相反，AsyncioEventLoop抛出"不允许的操作";例外。twisstedeventloop和TornadoEventLoop需要安装额外的软件。

我考虑过使用独立的看门狗库，但似乎从另一个线程访问用户界面需要写一个新的循环，参见这个堆栈溢出问题。这个问题的答案建议轮询而不是我宁愿避免。

如果urwid专门提供了一个监视文件的方法，我不相信它在任何实现中都不起作用。那么我做错了什么呢?

在python/urwid应用程序中我如何对文件更改做出反应?

---

编辑:我尝试过使用命名管道(并删除代码以清除文件)，但在视觉上它具有相同的行为:应用程序不启动。但是，听起来有一个很大的区别:直到我写入文件，它才进入无限循环。在我写入文件之前，回调没有被调用，但应用程序也没有启动，它只是什么也不做。写入文件后，它的行为与上面描述的常规文件相同。

Answer 1

我发现了以下工作:在另一个线程中读取命名管道，保护队列中的每一行，并在UI线程中轮询，看看是否有东西在队列中。

使用mkfifo /tmp/mypipe创建命名管道。然后用echo >>/tmp/mypipe "some text"写

#!/usr/bin/env python3
import os
import threading
import queue
import urwid

class App:
POLL_TIME_S = .5
def __init__(self):
self.text = urwid.Text("hello world")
self.filler = urwid.Filler(self.text)
self.loop = urwid.MainLoop(self.filler, unhandled_input=self.onkeypress)

def watch_pipe(self, path):
self._cmd_pipe = path
self.queue = queue.Queue()
threading.Thread(target=self._read_pipe_thread, args=(path,)).start()
self.loop.set_alarm_in(0, self._poll_queue)
def _read_pipe_thread(self, path):
while self._cmd_pipe:
with open(path, 'rt') as pipe:
for ln in pipe:
self.queue.put(ln)
self.queue.put("!! EOF !!")
def _poll_queue(self, loop, args):
while not self.queue.empty():
ln = self.queue.get()
self.text.set_text(ln)
self.loop.set_alarm_in(self.POLL_TIME_S, self._poll_queue)
def close(self):
path = self._cmd_pipe
# stop reading
self._cmd_pipe = None
with open(path, 'wt') as pipe:
pipe.write("")
os.remove(path)

def run(self):
self.loop.run()

def onkeypress(self, key, size=None):
if key == 'q':
raise urwid.ExitMainLoop()
self.text.set_text(key)

if __name__ == '__main__':
a = App()
a.watch_pipe('/tmp/mypipe')
a.run()
a.close()