How do I make an asynchronous progress spinner in Python?

Question

This is the code for the progress spinner:

import sys
import time

def spinning_cursor():
    while True:
        for cursor in '|/-\\':
            yield cursor

spinner = spinning_cursor()
for _ in range(50):
    sys.stdout.write(spinner.next())
    sys.stdout.flush()
    time.sleep(10)
    sys.stdout.write('\b')

Output

python2.7 test.py
|

It is spinning very slowly since the loop sleeps for 10 seconds...

How do I keep rotating the spinner while the process is sleeping?

Answer 1

Here's the asyncio example from Fluent Python (copied here as an allowed use of the book content) which covers this exact program. It uses async functions (aka coroutines) and the asyncio built-in Python library to display a spinner until a separate coroutine waits 3 seconds and returns a value (the integer 42). You can read more about asyncio from Chapter 19 in the book.

# spinner_await.py

# credits: Example by Luciano Ramalho inspired by
# Michele Simionato's multiprocessing example in the python-list:
# https://mail.python.org/pipermail/python-list/2009-February/538048.html

# BEGIN SPINNER_AWAIT
import asyncio
import itertools
import sys


async def spin(msg):  # <1>
    write, flush = sys.stdout.write, sys.stdout.flush
    for char in itertools.cycle('|/-\\'):
        status = char + ' ' + msg
        write(status)
        flush()
        write('\x08' * len(status))
        try:
            await asyncio.sleep(.1)  # <3>
        except asyncio.CancelledError:  # <4>
            break
    write(' ' * len(status) + '\x08' * len(status))


async def slow_function():  # <5>
    # pretend waiting a long time for I/O
    await asyncio.sleep(3)  # <6>
    return 42


async def supervisor():  # <7>
    spinner = asyncio.ensure_future(spin('thinking!'))  # <8>
    print('spinner object:', spinner)  # <9>
    result = await slow_function()  # <10>
    spinner.cancel()  # <11>
    return result


def main():
    loop = asyncio.get_event_loop()  # <12>
    result = loop.run_until_complete(supervisor())  # <13>
    loop.close()
    print('Answer:', result)


if __name__ == '__main__':
    main()
# END SPINNER_AWAIT

Answer 2

You could sleep in smaller steps until you reach 10 seconds:

import sys, time

def spinning_cursor():
    while True:
        for cursor in '|/-\\':
            yield cursor

spinner = spinning_cursor()
end_time = time.time() + 10
while time.time() < end_time:
    sys.stdout.write(spinner.next())
    sys.stdout.flush()
    time.sleep(0.2) # adjust this to change the speed
    sys.stdout.write('\b')

But this will block your main thread, so it will only be useful if you want to wait for 10 seconds without doing anything else in your Python program (e.g., waiting for some external process to complete).

If you want to run other Python code while the spinner is spinning, you will need two threads -- one for the spinner, one for the main work. You could set that up like this:

import sys, time, threading

def spin_cursor():
    while True:
        for cursor in '|/-\\':
            sys.stdout.write(cursor)
            sys.stdout.flush()
            time.sleep(0.1) # adjust this to change the speed
            sys.stdout.write('\b')
            if done:
                return

# start the spinner in a separate thread
done = False
spin_thread = threading.Thread(target=spin_cursor)
spin_thread.start()

# do some more work in the main thread, or just sleep:
time.sleep(10)

# tell the spinner to stop, and wait for it to do so;
# this will clear the last cursor before the program moves on
done = True
spin_thread.join()

# continue with other tasks
sys.stdout.write("all done\n")

Answer 3

You'll have to create a separate thread. The example below roughly shows how this can be done. However, this is just a simple example.

import sys
import time

import threading


class SpinnerThread(threading.Thread):

    def __init__(self):
        super().__init__(target=self._spin)
        self._stopevent = threading.Event()

    def stop(self):
        self._stopevent.set()

    def _spin(self):

        while not self._stopevent.isSet():
            for t in '|/-\\':
                sys.stdout.write(t)
                sys.stdout.flush()
                time.sleep(0.1)
                sys.stdout.write('\b')


def long_task():
    for i in range(10):
        time.sleep(1)
        print('Tick {:d}'.format(i))


def main():

    task = threading.Thread(target=long_task)
    task.start()

    spinner_thread = SpinnerThread()
    spinner_thread.start()

    task.join()
    spinner_thread.stop()


if __name__ == '__main__':
    main()

Answer 4

Spawn two threads, A and B. Thread A runs cmd to completion. Thread B displays the spinning cursor and waits for thread A to exit, which will happen when cmd completes. At that point, thread B clears the spinning cursor and then exit.

Or use an existing library instead of re-inventing the wheel. Consider the progressbar library. You'll want the RotatingMarker progress indicator.