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

如何解决Asyncio loop.run_until_complete 在小部件视图之前运行

描述：

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

标题.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',('syscheck',('exit','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','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()

syscheck.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

预期/实际结果

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

解决方法

想通了，超级简单。你不想 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。糟糕！