1
Description:

I'm trying to create a title screen on startup that loops through pings to APIs or sites and shows the user what succeeded and failed.

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()

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
Expected/actual outcome

I expect the bigtext to come up, then for the syscheck widget to say pinging google.com... and add [ FAILED ] or [ OK ] on completion. Instead, all of these pings are done before the title screen is ever shown, then everything is shown with the status messages already completed. This is very likely an issue that will seem trivial once solved - I'm just a simple synchronousfolk passing through. If I were to take a wild stab I'd guess it has something to do with either or both of await proc.wait() in syscheck.py or run_until_complete in title.py.

1 Answer 1

1

Figured it out, super simple. You don't wanna loop.run_until_complete, you want a

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())

as defined here. Oops!

Sign up to request clarification or add additional context in comments.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.