Asyncio loop.run_until_complete 在小部件视图之前运行

Question

描述：

我正在尝试在启动时创建一个标题屏幕，该屏幕循环通过对 API 或网站的 ping 操作，并向用户显示成功和失败的内容。

title.py

class TitleDisplay:

    def __init__(self):
        self.loop = asyncio.get_event_loop()
        self.sysc = syscheck.Syscheck()
        self.statustext = ""

    palette = [
        ('body',         'white',      'black'),
        ('header',       'black',      'light gray'),
        ('title',        'white',      'black'),
        ('syscheck',     'white',      'black'),
        ('exit',         'white',      'dark cyan'),
        ]

    async def update_check(self, widget_ref):
        widget = widget_ref()
        if not widget:
            # widget is dead; the main loop must've been destroyed
            return

        statuses = self.sysc.check()
        async for status in statuses:
            # simulate `next` with async iteration
            self.statustext += status
            widget.set_text(self.statustext)
            async for other_response in self.sysc.check():
                break

    def setup_view(self):
        fonts = urwid.get_all_fonts()
        utf8 = urwid.get_encoding_mode() == "utf8"
        for name, fontcls in fonts:
            font = fontcls()
            if not (font.utf8_required and not utf8):
                if fontcls == urwid.Thin6x6Font:
                    exit_font = font

        # Create BigText
        self.bigtext = urwid.BigText(u"Title", exit_font)
        bt = SwitchingPadding(self.bigtext, 'left', None)
        bt = urwid.AttrWrap(bt, 'title')
        bt = urwid.Filler(bt, 'top', None, 7)
        bt = urwid.BoxAdapter(bt, 7)

        # Create syscheck
        cah = urwid.Text(u"Checking APIs...\n")
        self.syscheckw = urwid.Text("", wrap='any')
        self.loop.run_until_complete(self.update_check(weakref.ref(self.syscheckw)))
        ca = urwid.AttrWrap(self.syscheckw, 'syscheck')
        chars = urwid.Pile([cah, ca])
        col = urwid.Columns([('fixed', 40, chars)], 3)
        l = [bt, urwid.Divider(), col]
        w = urwid.ListBox(urwid.SimpleListWalker(l))

        # Frame
        w = urwid.AttrWrap(w, 'body')
        hdr = urwid.Text("esc to quit")
        hdr = urwid.AttrWrap(hdr, 'header')
        w = urwid.Frame(header=hdr, body=w)

        # Exit message
        exit = urwid.BigText(('exit'," Quit? "), exit_font)
        exit = urwid.Overlay(exit, w, 'center', None, 'middle', None)
        return w, exit

    def main(self):
        self.view, self.exit_view = self.setup_view()
        self.urwid_loop = urwid.MainLoop(
            self.view,
            self.palette,
            event_loop=urwid.AsyncioEventLoop(loop=self.loop),
            unhandled_input=self.unhandled_input)
        self.urwid_loop.run()

系统检查.py

class Syscheck():
    def __init__(self):
        self.param = '-n' if platform.system().lower()=='windows' else '-c'
        # list of global sites to check
        self.hostlist = [
                        "google.com", # list is ~10 items but shortened for brevity
                        ]

    async def ping(self, host):
        # Building the command. Ex: "ping -c 1 google.com"
        command = f"ping {self.param} 1 {host}"

        proc = await asyncio.create_subprocess_shell(command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        status = await proc.wait()
        return (status == 0)

    async def check(self):
        # begin with pings
        connfailed = False
        for host in self.hostlist:
            status = f"Pinging {host}... "
            yield status
            pingstat = await self.ping(host)
            if not pingstat:
                connfailed = True
            status = ("[ OK ]\n" if pingstat else "[ FAILED ]\n")
            yield status

预期/实际结果

我希望出现大文本，然后让 syscheck 小部件说出 pinging google.com... 并在完成时添加 [ FAILED ] 或 [ OK ]。取而代之的是，所有这些 ping 操作都是在标题屏幕显示之前完成的，然后所有内容都显示为已完成的状态消息。这很可能是一个一旦解决就会显得微不足道的问题——我只是一个简单的同步人。如果我要大胆尝试，我猜它与 syscheck.py 中的 await proc.wait() 或 title.py 中的 run_until_complete 中的一个或两个有关。

Answer 1

想通了，超级简单。你不想 loop.run_until_complete，你想要一个

aloop = asyncio.get_event_loop()

ev_loop = urwid.AsyncioEventLoop(loop=aloop)
loop = urwid.MainLoop(frame, palette, screen,
                      unhandled_input=unhandled, event_loop=ev_loop)
aloop.create_task(generate_output())

定义为here。糟糕！