我尝试在urwid中对一些操作进行排序我制作了一个定时器，它在后台运行并与主进程通信像这样：
from multiprocessing import Process, Pipe
import time
import urwid
def show_or_exit(key):
if key in ('q', 'Q'):
raise urwid.ExitMainLoop()
class midiloop(urwid.Frame):
def __init__(self):
self.message = urwid.Text('Press Space', align='center')
self.filler = urwid.Filler(self.message, "middle")
super().__init__(urwid.Frame(self.filler))
def keypress(self, size, key):
if key == " ":
self.seq()
else:
return key
def timer(self,conn):
x = 0
while True:
if (conn.poll() == False):
pass
else:
z = conn.recv()
if (z == "kill"):
return()
conn.send(x)
x+=1
time.sleep(0.05)
def seq(self):
self.parent_conn, self.child_conn = Pipe()
self.p = Process(target=self.timer, args=(self.child_conn,))
self.p.start()
while True:
if (self.parent_conn.poll(None)):
self.y = self.parent_conn.recv()
self.message.set_text(str(self.y))
loop.draw_screen()
if ( self.y > 100 ):
self.parent_conn.send("kill")
self.message.set_text("Press Space")
return()
if __name__ == '__main__':
midiloop = midiloop()
loop = urwid.MainLoop(midiloop, unhandled_input=show_or_exit, handle_mouse=True)
loop.run()
问题是我用阻塞urwid主循环，而True：因此，任何人都可以给我一个解决方案，让我在密钥Q到达循环结束之前退出程序，例如，更广泛地说，与urwid交互并与子进程通信
将多处理和urwid结合起来似乎相当复杂。
由于您使用的是计时器，并且您的类名为midiloop，我猜您可能想要实现一个迷你序列器。
一种可能的实现方式是使用异步循环而不是urwid的MainLoop，并使用loop.call_later()函数调度事件。我过去用这种方法实现了一个简单的鼓机，使用urwid绘制序列器，使用asyncio调度播放事件，并使用simpleaudio进行播放。你可以在这里看到它的代码：https://github.com/eliasdorneles/kickit
如果您仍然想用多处理实现通信，我认为最好使用urwid.AsyncioEventLoop和aiopipe助手进行双工通信。
恐怕这不是很简单。然而，我确实花了一天时间编写了这个Urwid前端，它启动、停止并与子流程通信。
import os
import sys
from multiprocessing import Process, Pipe, Event
from collections import deque
import urwid
class suppress_stdout_stderr(object):
"""
Supresses the stdout and stderr by piping them to dev null...
The same place I send bad faith replies to my tweets
"""
def __enter__(self):
self.outnull_file = open(os.devnull, 'w')
self.errnull_file = open(os.devnull, 'w')
self.old_stdout_fileno_undup = sys.stdout.fileno()
self.old_stderr_fileno_undup = sys.stderr.fileno()
self.old_stdout_fileno = os.dup(sys.stdout.fileno())
self.old_stderr_fileno = os.dup(sys.stderr.fileno())
self.old_stdout = sys.stdout
self.old_stderr = sys.stderr
os.dup2(self.outnull_file.fileno(), self.old_stdout_fileno_undup)
os.dup2(self.errnull_file.fileno(), self.old_stderr_fileno_undup)
sys.stdout = self.outnull_file
sys.stderr = self.errnull_file
return self
def __exit__(self, *_):
sys.stdout = self.old_stdout
sys.stderr = self.old_stderr
os.dup2(self.old_stdout_fileno, self.old_stdout_fileno_undup)
os.dup2(self.old_stderr_fileno, self.old_stderr_fileno_undup)
os.close(self.old_stdout_fileno)
os.close(self.old_stderr_fileno)
self.outnull_file.close()
self.errnull_file.close()
def subprocess_main(transmit, stop_process):
with suppress_stdout_stderr():
import time
yup = ['yuuuup', 'yuuuuup', 'yeaup', 'yeoop']
nope = ['noooooooe', 'noooope', 'nope', 'nope']
mesg = 0
i = 0
while True:
i = i % len(yup)
if transmit.poll():
mesg = transmit.recv()
if mesg == 'Yup':
transmit.send(yup[i])
if mesg == 'Nope':
transmit.send(nope[i])
if stop_process.wait(0):
break
i += 1
time.sleep(2)
class SubProcess:
def __init__(self, main):
"""
Handles forking, stopping and communication with a subprocess
:param main: subprocess method to run method signature is
def main(transmit, stop_process):
transmit: is a multiprocess Pipe to send data to parent process
stop_process: is multiprocess Event to set when you want the process to exit
"""
self.main = main
self.recv, self.transmit = None, None
self.stop_process = None
self.proc = None
def fork(self):
"""
Forks and starts the subprocess
"""
self.recv, self.transmit = Pipe(duplex=True)
self.stop_process = Event()
self.proc = Process(target=self.main, args=(self.transmit, self.stop_process))
self.proc.start()
def write_pipe(self, item):
self.recv.send(item)
def read_pipe(self):
"""
Reads data sent by the process into a list and returns it
:return:
"""
item = []
if self.recv is not None:
try:
while self.recv.poll():
item += [self.recv.recv()]
except:
pass
return item
def stop(self):
"""
Sets the event to tell the process to exit.
note: this is co-operative multi-tasking, the process must respect the flag or this won't work!
"""
self.stop_process.set()
self.proc.join()
class UrwidFrontend:
def __init__(self, subprocess_main):
"""
Urwid frontend to control the subprocess and display it's output
"""
self.title = 'Urwid Frontend Demo'
self.choices = 'Start Subprocess|Quit'.split('|')
self.response = None
self.item = deque(maxlen=10)
self.event_loop = urwid.SelectEventLoop()
# start the heartbeat
self.event_loop.alarm(0, self.heartbeat)
self.main = urwid.Padding(self.main_menu(), left=2, right=2)
self.top = urwid.Overlay(self.main, urwid.SolidFill(u'N{MEDIUM SHADE}'),
align='center', width=('relative', 60),
valign='middle', height=('relative', 60),
min_width=20, min_height=9)
self.loop = urwid.MainLoop(self.top, palette=[('reversed', 'standout', ''), ], event_loop=self.event_loop)
self.subprocess = SubProcess(subprocess_main)
def exit_program(self, button):
raise urwid.ExitMainLoop()
def main_menu(self):
body = [urwid.Text(self.title), urwid.Divider()]
for c in self.choices:
button = urwid.Button(c)
urwid.connect_signal(button, 'click', self.handle_button, c)
body.append(urwid.AttrMap(button, None, focus_map='reversed'))
return urwid.ListBox(urwid.SimpleFocusListWalker(body))
def subproc_menu(self):
self.response = urwid.Text('Waiting ...')
body = [self.response, urwid.Divider()]
choices = ['Yup', 'Nope', 'Stop Subprocess']
for c in choices:
button = urwid.Button(c)
urwid.connect_signal(button, 'click', self.handle_button, c)
body.append(urwid.AttrMap(button, None, focus_map='reversed'))
listbox = urwid.ListBox(urwid.SimpleFocusListWalker(body))
return listbox
def update_subproc_menu(self, text):
self.response.set_text(text)
def handle_button(self, button, choice):
if choice == 'Start Subprocess':
self.main.original_widget = self.subproc_menu()
self.subprocess.fork()
self.item = deque(maxlen=10)
if choice == 'Stop Subprocess':
self.subprocess.stop()
self.main.original_widget = self.main_menu()
if choice == 'Quit':
self.exit_program(button)
if choice == 'Yup':
self.subprocess.write_pipe('Yup')
if choice == 'Nope':
self.subprocess.write_pipe('Nope')
def heartbeat(self):
"""
heartbeat that runs 24 times per second
"""
# read from the process
self.item.append(self.subprocess.read_pipe())
# display it
if self.response is not None:
self.update_subproc_menu(['Subprocess startedn', f'{self.item}n', ])
self.loop.draw_screen()
# set the next beat
self.event_loop.alarm(1 / 24, self.heartbeat)
def run(self):
self.loop.run()
if __name__ == "__main__":
app = UrwidFrontend(subprocess_main)
app.run()