urwid - 无限循环的输出屏幕

Question

我正在尝试用urwid创建一个简单的输出屏幕，用于显示一个无限循环中的数据。这个数据是来自另一个类的。

我现在找到的解决方案是：创建一个打印机类（这是一个测试用的替代类，用于实际输出）并给它一个队列属性。当需要显示内容时，就把内容添加到队列中。然后，还有一个接口类 - 也就是实际的界面 - 它有自己的打印机实例。一个与主循环并行运行的线程会检查队列里是否有内容，如果有，就输出这些内容。由于打印机的主要功能是一个无限循环，所以它自己也有一个线程 - 在这个测试中，它每隔几秒就输出一次“你好”。

以下是代码：

import urwid
import threading
import time

class Interface:
    palette = [
        ('body', 'white', 'black'),
        ('ext', 'white', 'dark blue'),
        ('ext_hi', 'light cyan', 'dark blue', 'bold'),
        ]

    header_text = [
        ('ext_hi', 'ESC'), ':quit        ',
        ('ext_hi', 'UP'), ',', ('ext_hi', 'DOWN'), ':scroll',
        ]

    def __init__(self):
        self.header = urwid.AttrWrap(urwid.Text(self.header_text), 'ext')
        self.flowWalker = urwid.SimpleListWalker([])
        self.body = urwid.ListBox(self.flowWalker)
        self.footer = urwid.AttrWrap(urwid.Edit("Edit:  "), 'ext')
        self.view = urwid.Frame(
            urwid.AttrWrap(self.body, 'body'),
            header = self.header,
            footer = self.footer)
        self.loop = urwid.MainLoop(self.view, self.palette, 
            unhandled_input = self.unhandled_input)
        self.printer = Printer()

    def start(self):
        t1 = threading.Thread(target = self.fill_screen)
        t1.daemon = True
        t2 = threading.Thread(target = self.printer.fill_queue)
        t2.daemon = True
        t1.start()
        t2.start()
        self.loop.run()

    def unhandled_input(self, k):
        if k == 'esc':
            raise urwid.ExitMainLoop()

    def fill_screen(self):  
        while True:
            if self.printer.queue:
                self.flowWalker.append(urwid.Text(('body', self.printer.queue.pop(0))))
                try:
                    self.loop.draw_screen()
                    self.body.set_focus(len(self.flowWalker)-1, 'above')
                except AssertionError: pass

    def to_screen(self, text):
        self.queue.append(text)


class Printer:
    def __init__(self):
        self.message = 'Hello'
        self.queue = []

    def fill_queue(self):
        while 1:
            self.queue.append(self.message)
            time.sleep(2)


if __name__ == '__main__':
    i = Interface()
    i.start()

这个方法可以工作，但我觉得看起来太乱了，我担心这可能会变成一种编码噩梦。有没有更简单的方法来完成这个任务？

Answer 1

如果你需要一个外部线程，可以看看下面的代码。它会先建立一个队列，然后启动一个“把当前时间发送到队列”的线程，接着运行主界面。这个界面会不时检查共享的队列，并根据需要进行更新。

当界面退出时，外部代码会礼貌地通知线程退出。

来源

import logging, Queue, sys, threading, time

import urwid

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)-4s %(threadName)s %(message)s", 
    datefmt="%H:%M:%S",
    filename='trace.log',
)

class Interface:
    palette = [
        ('body', 'white', 'black'),
        ('ext', 'white', 'dark blue'),
        ('ext_hi', 'light cyan', 'dark blue', 'bold'),
        ]

    header_text = [
        ('ext_hi', 'ESC'), ':quit        ',
        ('ext_hi', 'UP'), ',', ('ext_hi', 'DOWN'), ':scroll',
        ]

    def __init__(self, msg_queue):
        self.header = urwid.AttrWrap(urwid.Text(self.header_text), 'ext')
        self.flowWalker = urwid.SimpleListWalker([])
        self.body = urwid.ListBox(self.flowWalker)
        self.footer = urwid.AttrWrap(urwid.Edit("Edit:  "), 'ext')
        self.view = urwid.Frame(
            urwid.AttrWrap(self.body, 'body'),
            header = self.header,
            footer = self.footer)
        self.loop = urwid.MainLoop(self.view, self.palette, 
            unhandled_input = self.unhandled_input)
        self.msg_queue = msg_queue
        self.check_messages(self.loop, None)

    def unhandled_input(self, k):
        if k == 'esc':
            raise urwid.ExitMainLoop()

    def check_messages(self, loop, *_args):
        """add message to bottom of screen"""
        loop.set_alarm_in(
            sec=0.5,
            callback=self.check_messages,
            )
        try:
            msg = self.msg_queue.get_nowait()
        except Queue.Empty:
            return
        self.flowWalker.append(
            urwid.Text(('body', msg))
            )
        self.body.set_focus(
            len(self.flowWalker)-1, 'above'
            )

def update_time(stop_event, msg_queue):
    """send timestamp to queue every second"""
    logging.info('start')
    while not stop_event.wait(timeout=1.0):
        msg_queue.put( time.strftime('time %X') )
    logging.info('stop')

if __name__ == '__main__':

    stop_ev = threading.Event()
    message_q = Queue.Queue()

    threading.Thread(
        target=update_time, args=[stop_ev, message_q],
        name='update_time',
    ).start()

    logging.info('start')
    Interface(message_q).loop.run()
    logging.info('stop')

    # after interface exits, signal threads to exit, wait for them
    logging.info('stopping threads')

    stop_ev.set()
    for th in threading.enumerate():
        if th != threading.current_thread():
            th.join()