如何监听urwid中的文件更改？

Question

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

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

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提供了几个different event loops。默认情况下，使用SelectEventLoop。GLibEventLoop也有同样的行为，它会陷入无限循环。相反，AsyncioEventLoop会抛出一个“不允许操作”异常。TwistedEventLoop和TornadoEventLoop需要额外的软件才能安装。

我曾考虑过使用独立的watchdog库，但从另一个线程访问用户界面似乎需要编写一个新的循环，请参阅this stack overflow question。这个问题的答案是建议轮询，而我更希望避免轮询。

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

如何对python/urwid应用程序中的文件更改做出反应？

编辑:我试过使用named pipes (并删除了清除文件的代码)，但从视觉上看，它有相同的行为:应用程序无法启动。但是，听起来有一个很大的不同:它不会进入无限循环，直到我写入文件。在我写入文件之前，回调没有被调用，但是应用程序也没有启动，它什么也不做。在我写入文件后，它的行为就像上面描述的常规文件一样。

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()