pipelines/utils/cli.py

from datetime import timedelta
import itertools
import sys
import threading
import time


def loading_animation(done, text):
    for c in itertools.cycle(['|', '/', '-', '\\']):
        if done():
            sys.stdout.write('\r')
            sys.stdout.flush()
            break
        sys.stdout.write(f'\r{text} ' + c)
        sys.stdout.flush()
        time.sleep(0.1)


def start_loading_animation(done, text):
    t = threading.Thread(
        target=loading_animation, args=(lambda: done[0], text))
    t.start()
    return t


def stop_loading_animation(done, thread):
    done[0] = True
    thread.join()


def print_execution_time(start_time):
    end_time = time.time()
    execution_time = end_time - start_time

    execution_time_td = timedelta(seconds=execution_time)

    hours, remainder = divmod(execution_time_td.total_seconds(), 3600)
    minutes, remainder = divmod(remainder, 60)
    seconds, milliseconds = divmod(remainder, 1)
    milliseconds *= 1000

    return f"{int(hours):02}:{int(minutes):02}:{int(seconds):02}:{int(milliseconds):03}" # noqa