Python Parallel File Processing

2023-12-07 17:54 UTC gpt-4 Open in ChatGPT ↗

To implement a progress bar for each file being processed in parallel in Python, you can use the tqdm library, which is widely used for showing progress bars in loops and multiprocessing environments. Here’s a general structure of how you can achieve this:

Install tqdm: If you haven’t already, you need to install the tqdm library. You can do this using pip:
Terminal window
```
pip install tqdm
```
Set Up Your Parallel Processing: You can use the concurrent.futures module for parallel processing. Assume you have a list of files to process and a function process_file that processes each file.
Integrate tqdm for Progress Tracking: You can wrap your files list with tqdm to display the progress bar.

Here’s a sample code snippet to illustrate this:

import concurrent.futures
from tqdm import tqdm

def process_file(file):
    # Your file processing logic here
    pass

files = ['file1.txt', 'file2.txt', 'file3.txt']  # List of your files

# Using ThreadPoolExecutor for parallel processing
with concurrent.futures.ThreadPoolExecutor() as executor:
    # Wrapping the 'executor.map' with 'tqdm' for the progress bar
    results = list(tqdm(executor.map(process_file, files), total=len(files)))

In this example:

process_file is a function that you would define to process each file.
files is the list of files you want to process.
concurrent.futures.ThreadPoolExecutor is used for parallel processing.
tqdm is used to show the progress bar. It wraps around the iterable (files in this case) and automatically updates the progress bar as each file is processed.

This setup will display a progress bar in the console, giving you a visual indication of how many files have been processed and how many are left. Each time a file is processed, the progress bar will update accordingly.

Note: The actual implementation might vary depending on the specifics of your file processing function and the structure of your program.

Is this code ok? refactor where needed

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:
        result_dict = manager.dict()
        processes = []

        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []

        for filepath in filtered_files:
            # Get the number of lines in the file for the progress bar
            num_lines = sum(1 for _ in open(filepath))
            file_description = f'Processing {os.path.basename(filepath)}'

            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict, None))
            process.start()
            processes.append(process)

        for process,progress_bar in processes:
            process.join()

The code you provided aims to use multiprocessing to process files and then presumably write some results to an Excel file. However, there are a few issues and areas for improvement:

Multiprocessing with Shared Dictionary: You’re using a manager to create a shared dictionary (result_dict), which is fine for multiprocessing. However, be aware that accessing a manager dictionary can be slower than regular dictionaries due to the overhead of inter-process communication.
Opening Files in a Loop: You are opening files just to count the number of lines (num_lines = sum(1 for _ in open(filepath))). This is not only inefficient (especially for large files) but also leaves the file handles open, which can lead to resource leaks.
Progress Bar Handling: The code for handling the progress bar is missing. You mentioned wanting to show a progress bar for each file, but the current code doesn’t implement this.
Excel Writer: You are using pd.ExcelWriter context manager, but there’s no code showing how you write to this Excel file. Ensure that the writing process happens after all multiprocessing tasks are completed.
Process Joining Loop: The loop where you join processes is incorrect. You’re trying to unpack processes as if each element is a tuple of (process, progress_bar), but processes only contains multiprocessing.Process objects.
Handling Exceptions: There’s no exception handling, which can be crucial in a multiprocessing setup to ensure that all processes are properly terminated in case of an error.

Here’s a refactored version of your code:

import multiprocessing
import os
import pandas as pd

def process_file(filepath, result_dict):
    # Add your file processing logic here
    pass

filtered_files = [...]  # Your list of file paths
excel_file_path = "path/to/your/excelfile.xlsx"

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:
        result_dict = manager.dict()

        processes = []
        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []
            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict))
            process.start()
            processes.append(process)

        for process in processes:
            process.join()

        # Write results to Excel file here after all processes have completed
        # for example: result_dict['file1.txt'] might be a DataFrame
        # result_dict['file1.txt'].to_excel(writer, sheet_name='file1')

# You might want to add more logic here to handle writing to the Excel file
# based on the structure of your `result_dict`

Note:

Ensure that the process_file function is properly defined and writes the results to the result_dict.
The writing to the Excel file should happen after all processes have joined to ensure all data is available.
This code doesn’t include progress bar logic, as implementing it in a multiprocessing environment can be complex and might require a different approach (like using multiprocessing.Queue to send progress updates to the main process).

Adding progress bars for each file in a multiprocessing context is a bit complex because each process runs independently, and standard output (where the progress bar would normally display) is not shared across processes.

To implement this, you can use a multiprocessing.Queue to send progress updates from each process to the main process. The main process can then update the progress bars accordingly. Here’s an example of how you might implement this:

Install Required Libraries: You’ll need tqdm for the progress bars. If you haven’t already installed it:
Terminal window
```
pip install tqdm
```
Refactor the Code: Modify the process_file function to send progress updates to a queue. Then, in the main process, update the progress bars based on these updates.

Here’s an example implementation:

import multiprocessing
import os
import pandas as pd
from tqdm import tqdm

def process_file(filepath, result_dict, progress_queue):
    # Example file processing logic
    with open(filepath, 'r') as file:
        lines = file.readlines()
        for i, line in enumerate(lines):
            # Your processing logic here

            # Send progress update
            progress_queue.put((filepath, i + 1, len(lines)))

# Create a queue for progress updates
progress_queue = multiprocessing.Queue()

filtered_files = [...]  # Your list of file paths
excel_file_path = "path/to/your/excelfile.xlsx"

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:
        result_dict = manager.dict()

        processes = []
        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []
            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict, progress_queue))
            process.start()
            processes.append(process)

        # Set up progress bars
        progress_bars = {filepath: tqdm(total=sum(1 for _ in open(filepath)), desc=f"Processing {os.path.basename(filepath)}") for filepath in filtered_files}

        # Update progress bars based on queue messages
        completed_files = set()
        while len(completed_files) < len(filtered_files):
            filepath, current, total = progress_queue.get()
            progress_bars[filepath].update(current - progress_bars[filepath].n)  # Update the progress bar
            if current == total:
                completed_files.add(filepath)

        for process in processes:
            process.join()

        # Close all progress bars
        for pb in progress_bars.values():
            pb.close()

        # Write results to Excel file here

# Rest of your code...

In this implementation:

Each process_file sends updates to the progress_queue with the current progress.
The main process listens to the progress_queue and updates the respective progress bars.
Once a file is fully processed, its progress bar is marked complete, and once all files are processed, the program proceeds to join all processes.

This approach ensures that you get real-time progress updates for each file being processed in parallel. However, keep in mind that this is a simplified example. Depending on the complexity and nature of the file processing, you might need to adjust the code, particularly the process_file function.

The tqdm-multiprocessing package is specifically designed to integrate tqdm progress bars with Python’s multiprocessing, offering a more seamless and straightforward way to add progress bars to multiprocessing tasks. Using this package, you can display progress bars for each process without the need to manually handle queues or inter-process communication for progress updates.

Here’s how you can use tqdm-multiprocessing in your code:

Install tqdm-multiprocessing: First, install the package using pip:
Terminal window
```
pip install tqdm-multiprocessing
```

Modify Your Code: Incorporate tqdm-multiprocessing into your multiprocessing setup. Here’s an example based on your provided code:

import multiprocessing
import os
import pandas as pd
from tqdm_multiprocessing import TqdmPoolExecutor

def process_file(filepath, result_dict):
    # Your file processing logic here
    pass

filtered_files = [...]  # Your list of file paths
excel_file_path = "path/to/your/excelfile.xlsx"

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:
        result_dict = manager.dict()

        # Using TqdmPoolExecutor for parallel processing with progress bars
        with TqdmPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
            futures = [executor.submit(process_file, filepath, result_dict) for filepath in filtered_files]

            # Wait for all futures to complete
            for future in futures:
                future.result()

        # Write results to Excel file here after all processes have completed

# Rest of your code...

In this refactored code:

The TqdmPoolExecutor is used in place of ThreadPoolExecutor or ProcessPoolExecutor. It is a wrapper around these executors and automatically handles progress bars for each task.
You submit tasks to the executor, and tqdm-multiprocessing handles the progress bars for you.
The executor.submit() method is used for submitting tasks, and future.result() is called to ensure all tasks are completed before proceeding.

This approach simplifies the integration of progress bars with multiprocessing by abstracting the complexity of managing progress updates across multiple processes. It’s a convenient way to add visual feedback for long-running parallel tasks in Python.

vheck this official example, and adapt the code above

from time import sleep import multiprocessing import tqdm

import logging from tqdm_multiprocess.logger import setup_logger_tqdm logger = logging.getLogger(name)

from tqdm_multiprocess import TqdmMultiProcessPool

iterations1 = 100 iterations2 = 5 iterations3 = 2 def some_other_function(tqdm_func, global_tqdm):

total_iterations = iterations1 * iterations2 * iterations3
with tqdm_func(total=total_iterations, dynamic_ncols=True) as progress3:
    progress3.set_description("outer")
    for i in range(iterations3):
        logger.info("outer")
        total_iterations = iterations1 * iterations2
        with tqdm_func(total=total_iterations, dynamic_ncols=True) as progress2:
            progress2.set_description("middle")
            for j in range(iterations2):
                logger.info("middle")
                #for k in tqdm_func(range(iterations1), dynamic_ncols=True, desc="inner"):
                with tqdm_func(total=iterations1, dynamic_ncols=True) as progress1:
                    for j in range(iterations1):
                        # logger.info("inner") # Spam slows down tqdm too much
                        progress1.set_description("innert")
                        sleep(0.01)
                        progress1.update()
                        progress2.update()
                        progress3.update()
                        global_tqdm.update()

logger.warning(f"Warning test message. {multiprocessing.current_process().name}")
logger.error(f"Error test message. {multiprocessing.current_process().name}")

Multiprocessed

def example_multiprocessing_function(some_input, tqdm_func, global_tqdm):
logger.debug(f”Debug test message - I won’t show up in console. {multiprocessing.current_process().name}”) logger.info(f”Info test message. {multiprocessing.current_process().name}”) some_other_function(tqdm_func, global_tqdm) return True

def error_callback(result): print(“Error!”)

def done_callback(result): print(“Done. Result: ”, result)

def example(): pool = TqdmMultiProcessPool() process_count = 4 task_count = 10 initial_tasks = [(example_multiprocessing_function, (i,)) for i in range(task_count)]
total_iterations = iterations1 * iterations2 * iterations3 * task_count with tqdm.tqdm(total=total_iterations, dynamic_ncols=True) as global_progress: global_progress.set_description(“global”) results = pool.map(process_count, global_progress, initial_tasks, error_callback, done_callback) print(results)

if name == ‘main’: logfile_path = “tqdm_multiprocessing_example.log” setup_logger_tqdm(logfile_path) # Logger will write messages using tqdm.write example()

To adapt the provided example to your use case (processing a list of files and writing results to an Excel file), you’ll need to integrate the structure of the tqdm-multiprocessing example with your file processing logic. The key idea is to use the TqdmMultiProcessPool to handle the multiprocessing with integrated progress bars, similar to the example’s structure.

Here’s a refactored version of your code adapted to the tqdm-multiprocessing package:

import multiprocessing
import pandas as pd
from tqdm_multiprocess import TqdmMultiProcessPool
import tqdm
import logging
from tqdm_multiprocess.logger import setup_logger_tqdm

# Configure logging
logger = logging.getLogger(__name__)

# Function to process each file
def process_file(file_index, tqdm_func, global_tqdm):
    filepath = filtered_files[file_index]
    # Your file processing logic here
    # For example:
    for i in tqdm_func(range(100), desc=f"Processing {filepath}"):
        # Simulate some work
        pass
    global_tqdm.update()
    return True

# Error and Done callbacks
def error_callback(result):
    print("Error!")

def done_callback(result):
    print("Done. Result: ", result)

# Main function
def main():
    pool = TqdmMultiProcessPool()
    process_count = multiprocessing.cpu_count()
    filtered_files = [...]  # Your list of file paths
    task_count = len(filtered_files)
    initial_tasks = [(process_file, (i,)) for i in range(task_count)]

    with pd.ExcelWriter("path/to/excelfile.xlsx", engine='xlsxwriter') as writer:
        total_iterations = 100 * task_count  # Assuming each file processing involves 100 iterations
        with tqdm.tqdm(total=total_iterations, dynamic_ncols=True) as global_progress:
            global_progress.set_description("Global Progress")
            results = pool.map(process_count, global_progress, initial_tasks, error_callback, done_callback)

            # Write results to Excel file here after all processes have completed

if __name__ == '__main__':
    logfile_path = "tqdm_multiprocessing_example.log"
    setup_logger_tqdm(logfile_path)  # Logger will write messages using tqdm.write
    main()

In this adaptation:

The process_file function is modified to handle the processing of each file. It takes the index of the file in the filtered_files list as an input, along with the tqdm_func and global_tqdm for progress tracking.
The main function sets up the multiprocessing pool using TqdmMultiProcessPool and submits the file processing tasks.
The global_progress bar tracks the overall progress across all files.
The results of the processing can be collected in the results variable and used for further processing, such as writing to an Excel file.

Remember to replace the placeholder code in process_file with your actual file processing logic and adapt the Excel writing part according to your requirements. The tqdm-multiprocessing package handles the complexity of showing progress bars in a multiprocessing environment, making the implementation cleaner and more straightforward.

If each file has a different iteration count, you’ll need to slightly modify the implementation to handle this variability. Instead of having a fixed iteration count for each file, you will determine the iteration count dynamically based on the specific file being processed. Here’s how you can do it:

Determine Iteration Count Per File: Before processing each file, calculate the number of iterations it will require. This could depend on the file size, the number of lines, or any other metric relevant to your processing logic.
Update the process_file Function: Modify the process_file function to accept an additional argument representing the iteration count for that file.
Update Task Submission: When submitting tasks to the process pool, include the iteration count for each file as part of the task.

Here’s the adapted code:

import multiprocessing
import pandas as pd
from tqdm_multiprocess import TqdmMultiProcessPool
import tqdm
import logging
from tqdm_multiprocess.logger import setup_logger_tqdm

# Configure logging
logger = logging.getLogger(__name__)

# Function to process each file
def process_file(file_index, iterations, tqdm_func, global_tqdm):
    filepath = filtered_files[file_index]
    # Your file processing logic here
    for i in tqdm_func(range(iterations), desc=f"Processing {filepath}"):
        # Simulate some work
        pass
    global_tqdm.update(iterations)
    return True

# Error and Done callbacks
def error_callback(result):
    print("Error!")

def done_callback(result):
    print("Done. Result: ", result)

# Main function
def main():
    pool = TqdmMultiProcessPool()
    process_count = multiprocessing.cpu_count()
    filtered_files = [...]  # Your list of file paths

    # Calculate the iterations for each file
    file_iterations = [calculate_iterations_for_file(filepath) for filepath in filtered_files]

    # Prepare initial tasks with iteration counts
    initial_tasks = [(process_file, (i, file_iterations[i])) for i in range(len(filtered_files))]

    total_iterations = sum(file_iterations)
    with pd.ExcelWriter("path/to/excelfile.xlsx", engine='xlsxwriter') as writer:
        with tqdm.tqdm(total=total_iterations, dynamic_ncols=True) as global_progress:
            global_progress.set_description("Global Progress")
            results = pool.map(process_count, global_progress, initial_tasks, error_callback, done_callback)

            # Write results to Excel file here after all processes have completed

if __name__ == '__main__':
    logfile_path = "tqdm_multiprocessing_example.log"
    setup_logger_tqdm(logfile_path)
    main()

def calculate_iterations_for_file(filepath):
    # Implement logic to determine the number of iterations for each file
    # For example, it might be based on the file size or number of lines
    return number_of_iterations

In this version:

calculate_iterations_for_file(filepath) is a function you need to implement to determine the number of iterations required for each file.
Each task submitted to the pool includes the iteration count specific to that file.
The global progress bar’s total is the sum of all iterations for all files, ensuring that it accurately represents the overall progress.

This approach ensures that the progress bar reflects the actual amount of work being done per file, accommodating files that may require more or less processing time.

You’re correct to question the handling of the file path in the progress bars. In the current setup, each process_file function receives an index (file_index) and uses it to access the corresponding file path from the filtered_files list. However, since filtered_files is defined in the main process, it’s not directly accessible to the child processes in a multiprocessing setup.

To address this, you have a couple of options:

Pass the File Path Directly: Instead of passing the index, pass the file path directly to each process_file function. This way, each process knows exactly which file it’s working on.
Use a Manager List: If for some reason you must use indexes (e.g., if you’re passing other data structures by index), you can put filtered_files in a multiprocessing Manager list. This makes it accessible to all processes, though it can be less efficient due to the overhead of inter-process communication.

I’ll illustrate the first approach, as it’s more straightforward and efficient:

# Function to process each file
def process_file(filepath, iterations, tqdm_func, global_tqdm):
    # Your file processing logic here
    for i in tqdm_func(range(iterations), desc=f"Processing {os.path.basename(filepath)}"):
        # Simulate some work
        pass
    global_tqdm.update(iterations)
    return True

# Main function
def main():
    # ... [rest of your setup code] ...

    # Prepare initial tasks with file paths and iteration counts
    initial_tasks = [(process_file, (filtered_files[i], file_iterations[i])) for i in range(len(filtered_files))]

    total_iterations = sum(file_iterations)
    # ... [rest of your main function code] ...

In this modified version, the process_file function now directly receives the filepath as an argument, ensuring that each progress bar accurately reflects the file it’s processing. This approach is generally more reliable and straightforward in multiprocessing environments.

To customize the behavior and appearance of tqdm progress bars, you can use various arguments and environment settings. To prevent tqdm from printing new lines and to set the color to green, you can follow these steps:

Avoid New Lines: To ensure that tqdm does not print new lines, you can set the leave argument to True. This will make the progress bar stay on the same line.
Set the Color to Green: To change the color of the progress bar to green, you can use ANSI escape codes in the bar_format argument. The ANSI escape code for green text is \033[32m.

Here’s an example of how you can configure your tqdm progress bars:

from tqdm import tqdm
import time

for i in tqdm(range(100), desc="Processing", leave=True, bar_format="{l_bar}\033[32m{bar}\033[0m{r_bar}"):
    time.sleep(0.1)  # Simulating a task

In this example:

leave=True ensures the progress bar stays on the same line and doesn’t print new lines.
bar_format="{l_bar}\033[32m{bar}\033[0m{r_bar}" sets the format of the progress bar. The {bar} part (which represents the actual progress bar) is wrapped in \033[32m and \033[0m to set its color to green.

This configuration will result in a green progress bar that does not print new lines each time it updates. Remember that the appearance of ANSI color codes can vary depending on your console or terminal emulator.

To change the text color of the tqdm progress bar, you can use ANSI escape codes in the bar_format argument, just like for the bar color. However, in this case, you would apply the color codes to the other components of the progress bar, such as the description (desc), the left bar label (l_bar), and the right bar label (r_bar).

The ANSI escape code for green text is \033[32m, and you reset to the default color with \033[0m. Here’s how you can apply it to the text:

from tqdm import tqdm
import time

for i in tqdm(range(100),
              desc="\033[32mProcessing\033[0m",
              leave=True,
              bar_format="\033[32m{l_bar}\033[0m{bar}\033[32m{r_bar}\033[0m"):
    time.sleep(0.1)  # Simulating a task

In this updated example:

The description text Processing is colored green by wrapping it with \033[32m and \033[0m.
Both the left bar label ({l_bar}) and the right bar label ({r_bar}) are also colored green in the same way.
The {bar} part, which is the actual progress bar, remains unaffected in this example. You can apply a different color to it if you wish, as shown in the previous example.

This setup will result in the text of the progress bar being displayed in green, while keeping the progress bar itself in its default color (unless you change it as well). Remember that the appearance and support for ANSI color codes can vary depending on your terminal or console.

This is my code, and it still prints new lines

with pd.ExcelWriter(excel_file_path, engine=‘xlsxwriter’) as writer: with multiprocessing.Manager() as manager:

        result_dict = manager.dict()
        processes = []

        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []
            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict, progress_queue))
            process.start()
            processes.append(process)

        progress_bars = {fp: tqdm(total=sum(1 for _ in open(fp)),
                                  desc=f"\033[32mProcessing {os.path.basename(fp)}\033[0m",
                                  leave=True,
                                  bar_format="{l_bar}\033[32m{bar}\033[0m{r_bar}") for fp in filtered_files}

        # Update progress bars based on queue messages
        completed_files = set()

        while len(completed_files) < len(filtered_files):
            filepath, current, total = progress_queue.get()

            progress_bars[filepath].update(current - progress_bars[filepath].n)  # Update the progress bar

            if current == total:
                completed_files.add(filepath)

        for process in processes:
            process.join()

        # Close all progress bars
        for pb in progress_bars.values():
            pb.close()

The issue with new lines being printed despite setting leave=True in your tqdm progress bars may be related to how the progress bars are being updated in a multiprocessing context. When using tqdm in a multiprocessing setup, especially with a manager and processes, care must be taken with how progress updates are managed and displayed.

In your current setup, you’re creating individual progress bars for each file and updating them based on messages from a multiprocessing queue. However, the issue could be arising from how and when these updates are being processed and displayed.

Here’s a refined approach to ensure that the progress bars update correctly without printing new lines:

Use a Single Global Progress Bar: Instead of creating a separate progress bar for each file, consider using a single global progress bar. This can simplify the update mechanism and reduce issues with output formatting in the console.
Update the Global Progress Bar: Update this global progress bar based on the progress of each file.

Here’s how you could modify your code:

from tqdm import tqdm
import multiprocessing
import pandas as pd

def process_file(filepath, result_dict, progress_queue):
    # Your file processing logic here
    # Instead of updating a progress bar, put progress updates into the queue
    pass

filtered_files = [...]  # Your list of file paths
excel_file_path = "path/to/your/excelfile.xlsx"
progress_queue = multiprocessing.Queue()

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:
        result_dict = manager.dict()
        processes = []

        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []
            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict, progress_queue))
            process.start()
            processes.append(process)

        # Create a single global progress bar
        total_iterations = sum(sum(1 for _ in open(fp)) for fp in filtered_files)
        with tqdm(total=total_iterations, desc="\033[32mProcessing Files\033[0m", leave=True, bar_format="{l_bar}\033[32m{bar}\033[0m{r_bar}") as global_progress_bar:
            completed_files = set()

            while len(completed_files) < len(filtered_files):
                filepath, current, total = progress_queue.get()
                global_progress_bar.update(current - global_progress_bar.n)  # Update the global progress bar

                if current == total:
                    completed_files.add(filepath)

        for process in processes:
            process.join()

# Rest of your code...

In this version:

A single global progress bar is created, which simplifies the updating logic.
The progress bar is updated based on progress messages from the progress_queue.
Each process_file function should send updates to the progress_queue instead of directly updating a progress bar.

This approach should help in managing the progress updates more effectively and avoid issues with new lines being printed in the console.

To update the progress queue in your multiprocessing setup, you’ll modify the process_file function so that it sends progress updates to the progress_queue. These updates typically include information about the progress made in processing each file. The main process will then use these updates to adjust the global progress bar accordingly.

Here’s an example of how you can modify your process_file function and how the main process can handle these updates:

Modify the process_file Function: In this function, calculate the progress (like the number of lines processed, etc.) and send updates to the progress_queue. Each update can simply be the amount of progress made since the last update.
Handle Updates in the Main Process: In the main loop, read these updates from the progress_queue and update the global progress bar.

Here’s the adjusted code:

The `process_file` Function

def process_file(filepath, result_dict, progress_queue):
    # Example: Let's say the progress is based on the number of lines processed
    with open(filepath, 'r') as file:
        lines = file.readlines()
        for i, line in enumerate(lines):
            # Process the line (your logic here)

            # Send an update to the queue, e.g., each line processed is an update
            progress_queue.put(1)  # Send '1' as each line is processed

    # Optionally, mark completion for this file if needed
    progress_queue.put(0)  # Send '0' to indicate this file is fully processed

The Main Process

# ... [your setup code for multiprocessing and progress_queue] ...

# Create a single global progress bar
total_iterations = sum(sum(1 for _ in open(fp)) for fp in filtered_files)
with tqdm(total=total_iterations, desc="\033[32mProcessing Files\033[0m", leave=True, bar_format="{l_bar}\033[32m{bar}\033[0m{r_bar}") as global_progress_bar:
    completed_files = set()

    while len(completed_files) < len(filtered_files):
        update = progress_queue.get()
        if update == 0:
            # One file has completed processing
            completed_files.add(1)
        else:
            # Update the progress bar
            global_progress_bar.update(update)

    # Ensure all processes are finished
    for process in processes:
        process.join()

# ... [rest of your code] ...

In this setup:

Each time a line is processed (or any other unit of work in your context), process_file sends an update (1) to the progress_queue.
The main process listens to the progress_queue. When it receives an update, it increments the global progress bar. If it receives a 0, it knows that one file has been completely processed.
This approach allows for dynamic updating of the progress bar based on the actual progress of each file.

To make the entire output line green, you need to ensure that all components of the bar_format string are wrapped with the ANSI escape code for green text (\033[32m). This includes the description (desc), the bar itself, and any other text that tqdm outputs to the console.

Here’s an example of how you can specify bar_format to make the whole line green, including the percentage, the bar, the counter, the elapsed time, and the estimated remaining time:

from tqdm import tqdm
import time

# ANSI escape code for green text
green_start = "\033[32m"
green_end = "\033[0m"

# Define a custom bar format with everything colored green
bar_format = f"{green_start}{{l_bar}}{{bar}}| {{n_fmt}}/{{total_fmt}} [{{elapsed}}<{green_start}{{remaining}}{green_end}, " \
             f"{{rate_fmt}}{{postfix}}]{green_end}"

for i in tqdm(range(100),
              desc=f"{green_start}Processing Files{green_end}",
              leave=True,
              bar_format=bar_format):
    time.sleep(0.1)  # Simulate a task

In this code snippet:

The bar_format is defined to color the entire output line green. The green_start and green_end variables are used to wrap around each component that you want to color.
The desc argument is also wrapped with the green ANSI escape codes to ensure the description text is green as well.
All the parts of the bar_format that you want to color are included between the green_start and green_end codes.

When you run this in a console that supports ANSI escape codes, the entire line output by tqdm will be green. The actual appearance may vary based on the terminal emulator you are using and its support for ANSI color codes.

To ensure the entire line is green, including all the text and the progress bar itself, you need to wrap the ANSI escape codes for green text around each component in the bar_format. Each component includes l_bar, bar, r_bar, and any additional text elements that tqdm outputs.

Here’s how to modify the bar_format to ensure all components are green:

from tqdm import tqdm
import time

# Define the ANSI escape code for green text
green_color_code = "\033[32m"

# Define a custom bar format with all components wrapped in green color code
custom_bar_format = f"{green_color_code}{{l_bar}}{{bar}}| {{n_fmt}}/{{total_fmt}} [{{elapsed}}<{green_color_code}{{remaining}}{green_color_code}, " \
                    f"{{rate_fmt}}]{{postfix}}{green_color_code}"

# Using the custom bar format in tqdm
for i in tqdm(range(100),
              desc=f"Processing Files",
              leave=True,
              bar_format=custom_bar_format):
    time.sleep(0.02)  # Simulating work

# Reset the color at the end of the progress bar
print("\033[0m", end="")

In this custom_bar_format, all elements including the left bar (l_bar), the bar itself, the right bar (r_bar), elapsed time, estimated remaining time, and rate format are wrapped with the green color code. The bar will begin with green text and end with green text, making the entire line green.

At the end of the progress bar, a reset color code (\033[0m) is printed to ensure that any text printed to the console after the progress bar returns to the default color.

This should result in the entire output line of the progress bar being green when viewed in a terminal that supports ANSI color codes.

this code never ends

multiprocessing.set_start_method('spawn')

# Create a queue for progress updates
progress_queue = multiprocessing.Queue()

# Create an Excel writer object
excel_file_path = os.path.join(DIR, "VIDEO_DATA.xlsx")

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:

        result_dict = manager.dict()
        processes = []

        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []
            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict, progress_queue))
            process.start()
            processes.append(process)

        # Create single global progress bar
        total_iterations = sum(sum(1 for _ in open(fp)) for fp in filtered_files)

        with tqdm(total=total_iterations,
                  desc=f"Processing Files",
                  leave=True,
                  colour='green', bar_format=bar_format
                  ) as global_progress_bar:

            completed_files = set()

            while len(completed_files) < len(filtered_files):
                update = progress_queue.get()
                if update == 0:
                    # One file has completed processing
                    completed_files.add(1)
                else:
                    # Update the progress bar
                    global_progress_bar.update(update)

            global_progress_bar.close()

        for process in processes:
            process.join()

        new_dict = dict(result_dict)

        df = pd.DataFrame(dict([(k, pd.Series(v)) for k, v in new_dict.items()]))

        # Write the DataFrame to the Excel file
        df.to_excel(writer, sheet_name='movie_data', index=False)

The loop that waits for the processes to complete is only adding 1 to completed_files set every time a process completes, instead of adding the specific file path that has completed processing. This means that the completion condition len(completed_files) < len(filtered_files) will never be met if filtered_files contains more than one file, as the set completed_files will never have more than one element (the integer 1).

Here’s the corrected loop:

while len(completed_files) < len(filtered_files):
    filepath, update = progress_queue.get()
    if update == 0:
        # One file has completed processing, add the filepath to the completed set
        completed_files.add(filepath)
    else:
        # Update the progress bar
        global_progress_bar.update(update)

In the corrected loop, when a process completes, it should send a message to the progress_queue with the filepath and an update value of 0. This is then added to the completed_files set to track the completion.

Moreover, ensure that the process_file function sends the correct updates to the progress_queue. When a file has been processed completely, it should send its filepath and an update of 0. During processing, it should send the amount of work done since the last update (this could be 1 for each line processed, for instance). Here’s a pseudocode example for the process_file function:

def process_file(filepath, result_dict, progress_queue):
    # Your file processing logic here
    # ...
    # At the end of processing this file:
    progress_queue.put((filepath, 0))

With these changes, the main loop should be able to correctly identify when all files have been processed and exit accordingly. Make sure that each process is sending the correct updates to the queue for this to work.

I want to make all the parameters at the beginning of the script arguments with clean descriptions and default values

import multiprocessing import os.path import glob

from PIL import Image import compression_utils as cu import pandas as pd from tqdm import tqdm

import cv2

DIR = “C:/Users/GeorgeAdamon/Desktop/EXCEL_RAW_VIDEOS” ORIGINAL_RES = (128, 128) # Source Resolution. TODO: Automatically read this from the file name LEVELS = 16 # 2, 4, 16 or 256 only RES = 64 # Target Resolution. TODO: Allow for non-square resolutions GAMMA = 1.8 # Affects the brightness and contrast of the image. Higher values mean brighter images DITHER = False # To Dither or not TEST = False # To Perform a Compression/Decompression test to ensure correction FLIP = False # Apply Y-Axis flipping DECIMAL = False # This is a temporary fix until all values are 0.255

Use glob to find .txt files that don’t end with “PROCESSED.txt”

filtered_files = [file for file in glob.glob(os.path.join(DIR, ‘*.txt’)) if not file.endswith(‘PROCESSED.txt’)]

bpp = cu.calculate_bits_per_pixel(LEVELS)

def process_file(filepath: str, result_dict: dict, progress_queue: multiprocessing.Queue):

key = os.path.basename(filepath)
temp = []

with open(filepath) as qt:
    # All lines of the text file
    lines = qt.read().strip().split('\n')

    # Number of frames in the text file
    lines_to_process = len(lines)

    # Write the header
    temp.append(lines_to_process)
    temp.append(RES)
    temp.append(RES)
    temp.append(bpp)
    temp.append("DITHER" if DITHER else "NO-DITHER")

    for i in range(lines_to_process):
        raw = bytearray()

        values = lines[i].split(",")
        end = len(values)

        # How many values per pixel
        step = len(values) // (ORIGINAL_RES[0] * ORIGINAL_RES[1])

        # Convert strings to actual numbers
        for j in range(0, end, step):
            if DECIMAL:
                raw.extend([int(float(values[j + k]) * 255) for k in range(0, step)])
            else:
                raw.extend([int(float(values[j + k])) for k in range(0, step)])

        raw_bytes = bytes(raw)

        if step == 3:
            img = Image.frombuffer("RGB", ORIGINAL_RES, raw_bytes)
        elif step == 1:
            img = Image.frombuffer("L", ORIGINAL_RES, raw_bytes)
        elif step == 2:
            img = Image.frombuffer("L", ORIGINAL_RES, raw_bytes)

        if FLIP:
            img = img.transpose(Image.FLIP_TOP_BOTTOM)

        (img2, txt) = cu.process_image(img, RES, LEVELS, GAMMA, DITHER, TEST)

        temp.append(txt)

        # Send an update to the queue, e.g., each line processed is an update
        progress_queue.put((filepath, 1))  # Send '1' as each line is processed

result_dict[key] = temp
print(f"\n\nProcessed {len(result_dict[key])} lines from {filepath}\n")

# Optionally, mark completion for this file if needed
progress_queue.put((filepath, 0))  # Send '0' to indicate this file is fully processed

def main(): multiprocessing.set_start_method(‘spawn’)

# Create a queue for progress updates
progress_queue = multiprocessing.Queue()

# Create an Excel writer object
excel_file_path = os.path.join(DIR, "VIDEO_DATA.xlsx")

with pd.ExcelWriter(excel_file_path, engine='xlsxwriter') as writer:
    with multiprocessing.Manager() as manager:

        result_dict = manager.dict()
        processes = []

        for filepath in filtered_files:
            result_dict[os.path.basename(filepath)] = []
            process = multiprocessing.Process(target=process_file, args=(filepath, result_dict, progress_queue))
            process.start()
            processes.append(process)

        # Create single global progress bar
        total_iterations = sum(sum(1 for _ in open(fp)) for fp in filtered_files)

        # ANSI escape code for green text
        green_start = "\033[32m"
        green_end = "\033[0m"

        # Define a custom bar format with everything colored green
        bar_format = f"{green_start}{{l_bar}}{{bar}}| {{n_fmt}}/{{total_fmt}} [{{elapsed}}<{{remaining}}, " \
                     f"{{rate_fmt}}{{postfix}}]{green_end}"

        completed_files = set()

        with tqdm(total=total_iterations,
                  desc=f"Processing {len(filtered_files) - len(completed_files)} Files in Parallel",
                  leave=True,
                  colour='green', bar_format=bar_format
                  ) as global_progress_bar:

            while len(completed_files) < len(filtered_files):
                filepath, update = progress_queue.get()
                if update == 0:
                    # One file has completed processing, add the filepath to the completed set
                    completed_files.add(filepath)
                    global_progress_bar.desc =f"Processing {len(filtered_files) - len(completed_files)} Files in Parallel"
                else:
                    # Update the progress bar
                    global_progress_bar.update(update)

            progress_queue.close()
            global_progress_bar.close()

        for process in processes:
            process.join()

        df = pd.DataFrame(dict([(k, pd.Series(v)) for k, v in result_dict.items()]))

        # Write the DataFrame to the Excel file
        df.to_excel(writer, sheet_name='movie_data', index=False)

if name == “main”: main()

To make the parameters configurable with descriptions and default values at the beginning of your script, you can use the argparse module which is part of the Python standard library. This module provides an easy way to parse command-line arguments.

Below, I’ve refactored your script to include argparse. Each parameter is defined with a flag, help description, and default value:

import argparse
import multiprocessing
import os
import glob
from PIL import Image
import compression_utils as cu
import pandas as pd
from tqdm import tqdm

def parse_args():
    parser = argparse.ArgumentParser(description='Process some integers.')
    parser.add_argument('--dir', type=str, default="C:/Users/GeorgeAdamon/Desktop/EXCEL_RAW_VIDEOS",
                        help='Directory where the video files are located')
    parser.add_argument('--original_res', type=int, nargs=2, default=(128, 128),
                        help='Original resolution of the source videos (width height)')
    parser.add_argument('--levels', type=int, default=16,
                        help='Levels of compression (2, 4, 16, or 256 only)')
    parser.add_argument('--res', type=int, default=64,
                        help='Target resolution of the processed videos (assumes square resolution)')
    parser.add_argument('--gamma', type=float, default=1.8,
                        help='Gamma correction value. Higher values mean brighter images')
    parser.add_argument('--dither', action='store_true',
                        help='Apply dithering')
    parser.add_argument('--test', action='store_true',
                        help='Perform a compression/decompression test')
    parser.add_argument('--flip', action='store_true',
                        help='Apply Y-Axis flipping')
    parser.add_argument('--decimal', action='store_true',
                        help='Use decimal values until all values are between 0 and 255')
    return parser.parse_args()

def main(args):
    # Use glob to find .txt files that don't end with "PROCESSED.txt"
    filtered_files = [file for file in glob.glob(os.path.join(args.dir, '*.txt')) if not file.endswith('PROCESSED.txt')]
    bpp = cu.calculate_bits_per_pixel(args.levels)

    # Your existing code with modifications to use 'args' where required

if __name__ == "__main__":
    args = parse_args()
    main(args)

To run the script with command-line arguments, you would use a command like this:

python script.py --dir "C:/Path/To/Directory" --original_res 128 128 --levels 16 --res 64 --gamma 1.8 --dither --test

In this setup:

The --dir argument specifies the directory path.
The --original_res takes two integers, representing the width and height.
The --levels argument sets the levels of compression.
The --res sets the target resolution.
The --gamma adjusts the gamma correction value.
The --dither is a flag; if present, dithering is applied.
The --test is a flag; if present, a test is performed.
The --flip is a flag; if present, Y-Axis flipping is applied.
The --decimal is a flag; if present, decimal values are used.

The action='store_true' in argparse means that, if the flag is present on the command line, the respective variable is set to True. The default values are used if the arguments are not provided in the command line.

In the argparse module, when you use the action='store_true' argument, it means that if the command-line argument is present, then the corresponding variable is set to True. If the command-line argument is not present, the variable is set to False by default.

So for boolean flags like --dither, --test, --flip, and --decimal, you do not need to explicitly set a default value. They are False by default unless the flag is included in the command line, in which case they are set to True.

If you want to explicitly show that the default is False for documentation or clarity in the code, you can add default=False to the add_argument call, but it’s redundant. Here’s how you could modify one of the boolean arguments to explicitly include the default:

parser.add_argument('--dither', action='store_true', default=False,
                    help='Apply dithering (default: %(default)s)')

The %(default)s in the help string is a special argparse feature that inserts the default value into the help message. This makes it clear to the user what the default behavior is without requiring additional explanation.

The remaining boolean arguments can be similarly updated to explicitly include their default values if that is your preference.

Certainly! Using a dictionary to store default values is a clean and professional way to manage defaults in Python. This approach makes it easy to update, manage, and refer to defaults throughout the code. Here’s how you can refactor the script using a dictionary for default arguments:

import argparse
import multiprocessing
import os
import glob
from PIL import Image
import compression_utils as cu
import pandas as pd
from tqdm import tqdm

# Define default values in a dictionary
default_args = {
    'dir': "C:/Users/GeorgeAdamon/Desktop/EXCEL_RAW_VIDEOS",
    'original_res': (128, 128),
    'levels': 16,
    'res': 64,
    'gamma': 1.8,
    'dither': False,
    'test': False,
    'flip': False,
    'decimal': False,
}

def parse_args(defaults):
    parser = argparse.ArgumentParser(description='Process video files and store output in an Excel file.')
    parser.add_argument('--dir', type=str, default=defaults['dir'],
                        help='Directory where the video files are located (default: %(default)s)')
    parser.add_argument('--original_res', type=int, nargs=2, default=defaults['original_res'],
                        help='Original resolution of the source videos (width height) (default: %(default)s)')
    parser.add_argument('--levels', type=int, default=defaults['levels'],
                        help='Levels of compression (2, 4, 16, or 256 only) (default: %(default)s)')
    parser.add_argument('--res', type=int, default=defaults['res'],
                        help='Target resolution of the processed videos (assumes square resolution) (default: %(default)s)')
    parser.add_argument('--gamma', type=float, default=defaults['gamma'],
                        help='Gamma correction value. Higher values mean brighter images (default: %(default)s)')
    parser.add_argument('--dither', action='store_true', default=defaults['dither'],
                        help='Apply dithering (default: %(default)s)')
    parser.add_argument('--test', action='store_true', default=defaults['test'],
                        help='Perform a compression/decompression test (default: %(default)s)')
    parser.add_argument('--flip', action='store_true', default=defaults['flip'],
                        help='Apply Y-Axis flipping (default: %(default)s)')
    parser.add_argument('--decimal', action='store_true', default=defaults['decimal'],
                        help='Use decimal values until all values are between 0 and 255 (default: %(default)s)')
    return parser.parse_args()

def main(args):
    # Existing logic that uses 'args' to process files

if __name__ == "__main__":
    args = parse_args(default_args)
    main(args)

With this setup:

All the defaults are stored in the default_args dictionary, which makes them easy to manage and reference.
The parse_args function takes this dictionary as an input and sets up the command-line interface accordingly.
The help messages for each argument use %(default)s to insert the default value into the description automatically, so it’s clear to the user what the default is.
By using a dictionary, you also gain the flexibility to easily override or add new default values without changing the argument parsing logic.

To make these variables global, you can simply declare them as such within the main function and then set their values. However, in Python, when you assign a value to a variable outside of any function, it is already global by default.

To demonstrate proper Python conventions for global variables, they should be in all caps (which you’ve already done), and assignments should happen at the top level of your script if they are to be constants, or within a function if their values depend on the arguments or need to be computed.

Here’s how you would do it within your main function:

# ... (rest of your imports and the parse_args function)

def main():
    global DIR, ORIGINAL_RES, LEVELS, RES, GAMMA, DITHER, TEST, FLIP, DECIMAL

    args = parse_args(default_args)

    DIR = args.dir
    ORIGINAL_RES = args.original_res
    LEVELS = args.levels
    RES = args.res
    GAMMA = args.gamma
    DITHER = args.dither
    TEST = args.test
    FLIP = args.flip
    DECIMAL = args.decimal

    # ... (the rest of your main code here)

if __name__ == "__main__":
    main()

In this approach:

The global keyword is used to explicitly declare that these variables are global. This is necessary only if you are going to modify global variables within a function. If you’re just accessing (reading) them, the global keyword isn’t needed.
The args = parse_args(default_args) call and the setting of global variables are done within the main function, which is the entry point of the script.
Since these variables are global, they can be accessed from anywhere else in the script after they have been set in the main function.

However, using global variables extensively is generally discouraged in Python because it can make the code harder to understand and maintain. If possible, consider passing these as parameters to the functions that need them instead of using globals.

To define these as global constants at the top of your file, you simply assign them directly after parsing the arguments. This is a common pattern when the values are meant to be constants throughout the script’s execution. Here’s how you can adjust your script to define the arguments as global constants:

import argparse
import multiprocessing
import os
import glob
from PIL import Image
import compression_utils as cu
import pandas as pd
from tqdm import tqdm

# Define default values in a dictionary
default_args = {
    'dir': "C:/Users/GeorgeAdamon/Desktop/EXCEL_RAW_VIDEOS",
    'original_res': (128, 128),
    'levels': 16,
    'res': 64,
    'gamma': 1.8,
    'dither': False,
    'test': False,
    'flip': False,
    'decimal': False,
}

def parse_args(defaults):
    # ... [same as before]

args = parse_args(default_args)

# Define global constants
DIR = args.dir
ORIGINAL_RES = args.original_res
LEVELS = args.levels
RES = args.res
GAMMA = args.gamma
DITHER = args.dither
TEST = args.test
FLIP = args.flip
DECIMAL = args.decimal

# Rest of your script below
# ...

def main():
    # Use the global constants in your main function logic
    # ...

if __name__ == "__main__":
    main()

In this script, the args object is created outside of any function, at the top level of the script, which makes the parsed argument values available as global constants. Then, you can use DIR, ORIGINAL_RES, LEVELS, etc., throughout your script without passing them around as parameters.

This pattern is suitable when you are sure that the values will not change during the script execution, and it makes the values available across all functions and modules that might need them. However, remember that it is generally considered best practice to limit the use of globals and to pass variables to functions explicitly where possible to maintain clear modularity and avoid side effects.

To open the Excel file at the end of your script, you can use Python’s built-in os module to call the default program associated with Excel files on your system. Here’s a simple way to do it using os.startfile() which works on Windows:

import os

# ... (rest of your script)

if __name__ == "__main__":
    main()
    excel_file_path = os.path.join(DIR, "VIDEO_DATA.xlsx")
    os.startfile(excel_file_path)

Place this code at the end of your script outside of any function (as shown above). After the main processing is done, it will attempt to open the Excel file with the default associated application on your system. Note that os.startfile() is only available on Windows.

If you’re using macOS, you can use the subprocess module:

import subprocess

# ... (rest of your script)

if __name__ == "__main__":
    main()
    excel_file_path = os.path.join(DIR, "VIDEO_DATA.xlsx")
    subprocess.call(["open", excel_file_path])

For Linux, you can use subprocess as well:

import subprocess

# ... (rest of your script)

if __name__ == "__main__":
    main()
    excel_file_path = os.path.join(DIR, "VIDEO_DATA.xlsx")
    subprocess.call(["xdg-open", excel_file_path])

These commands will open the Excel file with the default program configured to open .xlsx files in their respective operating systems after your script’s main execution has finished.

Creating a platform-agnostic function to open an Excel file can be done using the webbrowser module, which is part of the standard library and is intended for launching a file in the system’s default viewer.

Here’s a function that checks the operating system and opens the file using the appropriate method:

import webbrowser
import platform
import subprocess
import os

def open_excel_file(file_path):
    try:
        if platform.system() == 'Darwin':  # macOS
            subprocess.call(['open', file_path])
        elif platform.system() == 'Windows':  # Windows
            os.startfile(file_path)
        else:  # linux variants
            subprocess.call(['xdg-open', file_path])
    except Exception as e:
        print(f"Error opening file: {e}")

# ... (rest of your script)

if __name__ == "__main__":
    main()
    excel_file_path = os.path.join(DIR, "VIDEO_DATA.xlsx")
    open_excel_file(excel_file_path)

When you call open_excel_file(excel_file_path), it will use the system’s default application for opening .xlsx files. The function open_excel_file checks the operating system and uses the appropriate command to open the file. If there’s an error (for example, if the default application is not set), it will print an error message.

This function is agnostic to the platform and should work on Windows, macOS, and Linux, covering the majority of desktop users.