progress.py

import threading
import sys
import time

class Progress(threading.Thread):
    def __init__(self):

        super().__init__()

        #define global variables.
        self.total_lines = 0
        self.current_line = 0
        self.p_map = {} #percentage map
        self.is_running = False

    def start(self):
        pass

    def end(self):
        pass

    def update_current(self, current_line):
        self.current_line = current_line

    def fetch_current_percentage(self):
        # Finds current line in hashmap
        if self.current_line == 0:
            return 0
        
        return self.p_map[self.current_line]

    
    def format_output_string(self, percentage):
        #Slices string to fit the current percentage.
        percentage = self.fetch_current_percentage()
        hash_count = int((percentage / 100) * self.bar_length)

        bar = '#' * hash_count
        spaces = ' ' * (self.bar_length - hash_count)
        
        return f"[{bar}{spaces}] Progress: {percentage}%"
    

def percentage_map(total_lines):
    percentage_map = {}
    
    for i in range(1, total_lines + 1):
        percentage = int((i / total_lines) * 100)
        percentage_map[i] = percentage

    return percentage_map

def get_line_range(filepath):
    valid_lines = 0

    with open(filepath, 'r') as file:
        for i in file:

            clean_lines = i.strip()

            if not i == "" or i.startswith("#"):
                valid_lines += 1

    return valid_lines

def prepare_load_visual(kill_status):
    #make sure to pass the percentage map thread id into the parameter field
    #Creating spinning cursor for each thread while a thread is mapping lines

    #Trigger by doing kill_status = threading.Event() and add it as a thread argument

    spinner_frames = ['|', '/', '-', '\\']
    idx = 0

    while not kill_status.is_set():

        frame = spinner_frames[idx]

        sys.stdout.write(f"\r\033[K[ {frame} ] Initializing setup...")
        sys.stdout.flush()
        
        idx = (idx + 1) % len(spinner_frames)
        time.sleep(0.1)


def grab_and_display(data_thread, loading_thread, kill_flag):
    # When percentage map thread is complete, kills pre-loading spinner, and clears screen for new loading bar

    if data_thread:
        data_thread.join()
    if kill_flag:
        kill_flag.set()
    if loading_thread:
        loading_thread.join()

    sys.stdout.write("\r\033[K")
    sys.stdout.flush()

def ansi_escape_format(progress_string):
    #Use ansi escape codes to keep our processed string cleanly in one position at all times, without disrupting additional output
    sys.stdout.write(f"\r\033[K{progress_string}")
    sys.stdout.flush()