我正在尝试使输出屏幕无尽循环.它需要输出来自另一个类的数据.
我现在发现的解决方案是：拥有一个带有队列属性的Printer类(用于实际输出类的测试替换器).当需要显示某些内容时,会将其附加到队列中.然后,有一个接口类-实际接口-带有自己的Printer实例.与MainLoop并行运行的线程检查队列中是否有项目,如果有,则将其输出.由于Printer的主要功能是无限循环,因此它也有自己的线程-在此测试中,它仅每隔几秒钟输出一次“ Hello”.
这是代码：
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()
它可以工作,但是对我来说似乎太混乱了,我担心它最终可能会成为某种编码恐怖.有没有更简单的方法来完成任务？
解决方法:
如果需要外部线程,请考虑以下代码.它设置一个队列,启动“发送当前时间到队列”线程,然后运行主界面.该接口会不时检查共享队列,并根据需要进行更新.
当接口退出时,外部代码会发信号通知线程要有礼貌地退出.
资源
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()
标签：multithreading,urwid,python
来源： https://codeday.me/bug/1029/1958391.html
如果觉得《python无限循环语句纵向输出字符串 python-urwid-无限循环的输出屏幕》对你有帮助，请点赞、收藏，并留下你的观点哦！
