Getting user input while a timer is counting and updating the console

Question

I have a count-up/count-down timer library, and I've written some demo code that kicks off an instance of its main class, and updates the console as it counts down. When the timer expires, it displays a message to that effect. The demo code is pretty rudimentary and non-interactive, however, and I would like to improve it.

I would like to add the ability to pause/resume/reset the timer by pressing keys on the keyboard. The user would press the spacebar to pause/resume, and another key (maybe "r") to reset the timer. The console would be continually updated and display the current "remaining" time, including freezing the time when the timer is paused.

What I am struggling with is what approach would be best to use here. I have some limited experience with threads, but no experience with async. For now, I am not interested in using a full blown TUI, either, even though that might give the most satisfying results...I am treating this as a learning experience.

My first inclination was to use threads, one each for the user input and the timer countdown / console update tasks; but how do I get messages from the console when it receives user input (e.g. "pause") to the other task so that the time pauses?

I want to do this in the most Pythonic way I can - any suggestions?

Answer 1

I ended up using async and learning a bit in the process. Here's the code. And BTW it is a lot lighter weight than my threaded verssion. My Macbook pro fans spin up to max speed when I run the threaded version. But I can barely hear them at all with the async version.

import sys
import asyncio
from count_timer import CountTimer
from blessed import Terminal


def count():
    if counter.remaining > 10:
        print(
            term.bold
            + term.green
            + term.move_x(0)
            + term.move_up
            + term.clear_eol
            + str(round(counter.remaining, 3))
        )
    elif counter.remaining > 5:
        print(
            term.bold
            + term.yellow2
            + term.move_x(0)
            + term.move_up
            + term.clear_eol
            + str(round(counter.remaining, 3))
        )
    elif counter.remaining > 0:
        print(
            term.bold
            + term.red
            + term.move_x(0)
            + term.move_up
            + term.clear_eol
            + str(round(counter.remaining, 3))
        )
    else:
        print(
            term.bold
            + term.magenta
            + term.move_x(0)
            + term.move_up
            + term.clear_eol
            + "TIME'S UP!"
        )


def kb_input():
    if counter.remaining <= 0:
        return
    with term.cbreak():
        key = term.inkey(timeout=0.01).lower()
        if key:
            if key == "q":
                print(
                    term.bold
                    + term.magenta
                    + term.move_x(0)
                    + term.move_up
                    + term.clear_eol
                    + "Quitting..."
                )
                sys.exit()
            elif key == "r":
                counter.reset(duration=float(duration))
                counter.start()
            elif key == " ":
                counter.pause() if counter.running else counter.resume()


async def main():
    global counter
    global term
    global duration

    duration = input("Enter countdown timer duration: ")
    counter = CountTimer(duration=float(duration))
    counter.start()
    term = Terminal()

    def _run_executor_count():
        count()

    def _run_executor_kb_input():
        kb_input()

    while counter.remaining > 0:
        await asyncio.get_event_loop().run_in_executor(None, _run_executor_count)
        await asyncio.get_event_loop().run_in_executor(None, _run_executor_kb_input)

    await asyncio.get_event_loop().run_in_executor(None, _run_executor_count)


def async_main_entry():
    asyncio.get_event_loop().run_until_complete(main())


if __name__ == "__main__":
    async_main_entry()