TaskLog(Progress Meter)

task_log.py

# TaskLog by maxrt101

from dataclasses import dataclass
from enum import Enum
import threading, queue, time

class State(Enum):
    SUCCESS = 0
    FAILED = 1
    STARTING = 2
    IN_PROGRESS = 3

@dataclass(order=True)
class Task:
    msg: str
    state: State = State.STARTING
    stage: int = 0

class TaskLog:
    PROGRESS_STAGES = ('-', '\\', '|', '/')

    def __init__(self, timeout: float = 0.5, run: bool = True):
        self.timeout = timeout
        self.task = None
        self.event = threading.Event()
        self.running = False
        self.thread = None
        if run:
            self.run()

    def new_task(self, msg: str):
        if self.event.is_set():
            self._update()
        self.task = Task(msg)
        self.event.set()

    def start(self):
        self.task.state = State.IN_PROGRESS

    def success(self):
        self.task.state = State.SUCCESS
    
    def fail(self): # msg?
        self.current_task.state = State.FAILED

    def __enter__(self):
        if not self.running:
            self.run()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.stop()

    def _update(self):
        state_str = ''
        if self.task.state == State.STARTING:
            state_str = '.'
        elif self.task.state == State.IN_PROGRESS:
            state_str = TaskLog.PROGRESS_STAGES[self.task.stage % len(TaskLog.PROGRESS_STAGES)]
            self.task.stage += 1
        elif self.task.state == State.SUCCESS:
            state_str = '+'
            self.event.clear()
        elif self.task.state == State.FAILED:
            state_str = '-'
            self.event.clear()
        
        print(f'[{state_str}] {self.task.msg}', end='\r')
        if not self.event.is_set():
            print('')

    def _update_loop(self):
        while (self.running):
            self.event.wait()
            self._update()
            time.sleep(self.timeout)
        print('')

    def run(self):
        self.thread = threading.Thread(target=TaskLog._update_loop, args=[self])
        self.running = True
        self.thread.start()

    def stop(self):
        time.sleep(self.timeout)
        self.running = False
        self.thread.join()

# Example:

# def task(tlog):
#     time.sleep(1)
#     tlog.start()
#     time.sleep(5)
#     tlog.success()

# with TaskLog() as tlog:
#     tlog.new_task('Executing Task')
#     task(tlog)
#     tlog.new_task('Preparing Results')
#     task(tlog)