### Install atpbar using pip Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Installs the atpbar library and its latest updates using the pip package installer. This is the first step to using atpbar in your Python projects. ```bash pip install -U atpbar ``` -------------------------------- ### Display progress for a single loop with atpbar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Wraps an iterable with `atpbar` to display a progress bar during loop execution. The example demonstrates a loop with a random number of iterations, each pausing briefly. ```python n = randint(1000, 10000) for _ in atpbar(range(n)): sleep(0.001) ``` -------------------------------- ### Display progress for nested loops with atpbar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Shows how `atpbar` can manage progress bars for both outer and inner loops concurrently. Each loop can be named for clarity in the output. ```python for i in atpbar(range(4), name='Outer'): n = randint(1000, 10000) for _ in atpbar(range(n), name=f'Inner {i}'): sleep(0.001) ``` -------------------------------- ### Import necessary atpbar modules and utilities Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Imports essential components from the atpbar library along with standard Python modules for the examples. This includes `randint` for random numbers, `sleep` for pausing execution, and `atpbar` itself for progress bar functionality. ```python from random import randint from time import sleep from atpbar import atpbar ``` -------------------------------- ### Import atpbar flushing and ThreadPoolExecutor for threading Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Imports `ThreadPoolExecutor` for managing a pool of threads and `flushing` from `atpbar` to ensure progress bar updates are synchronized when using threads. ```python from concurrent.futures import ThreadPoolExecutor from atpbar import flushing ``` -------------------------------- ### Install atpbar using pip Source: https://github.com/alphatwirl/atpbar/blob/main/README.md This command installs the atpbar library using pip, the Python package installer. Ensure you have pip installed and configured correctly. ```bash pip install atpbar ``` -------------------------------- ### Execute threaded tasks with atpbar and ThreadPoolExecutor Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Demonstrates using `ThreadPoolExecutor` to run multiple instances of the `func` function concurrently. The `flushing()` context manager ensures that progress bars are properly displayed and updated for all active threads. ```python n_workers = 5 n_jobs = 10 with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in range(n_jobs): n = randint(1000, 10000) f = executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### Define a function for threaded execution with atpbar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/quick.md Defines a reusable function `func` that incorporates `atpbar` to display progress for tasks executed within threads. This function takes the number of iterations and a name as arguments. ```python def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) ``` -------------------------------- ### Display Thread Progress with atpbar and Python Threading Source: https://github.com/alphatwirl/atpbar/blob/main/docs/guide/threading-thread.md This code snippet demonstrates how to use the `atpbar` library with Python's built-in `threading` module. It creates multiple threads, each running a function that iterates through a range with a progress bar managed by `atpbar`. The `flushing` context manager ensures that all progress bars are displayed correctly. ```python from random import randint from threading import Thread from time import sleep from atpbar import atpbar, flushing def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_threads = 5 with flushing(): threads = [] for i in range(n_threads): n = randint(1000, 10000) t = Thread(target=func, args=(n, f'Thread {i}')) t.start() threads.append(t) for t in threads: t.join() ``` -------------------------------- ### Execute Tasks with ThreadPoolExecutor and Monitor with atpbar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/guide/thread-pool-executor.md This Python code snippet demonstrates how to use ThreadPoolExecutor to run multiple tasks concurrently. The atpbar library is integrated to display a progress bar for each task, providing visual feedback on their execution status. It requires the 'atpbar' and 'concurrent.futures' libraries. ```python from concurrent.futures import ThreadPoolExecutor from random import randint from time import sleep from atpbar import atpbar, flushing def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_workers = 5 n_jobs = 10 with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in range(n_jobs): n = randint(1000, 10000) f = executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### ProcessPoolExecutor with atpbar Progress Reporting (Python) Source: https://github.com/alphatwirl/atpbar/blob/main/docs/guide/process-pool-executor.md This Python code demonstrates how to use ProcessPoolExecutor for parallel execution of tasks while integrating atpbar for progress reporting. It initializes the executor with a reporter and submits multiple functions that utilize atpbar for their internal progress visualization. Dependencies include multiprocessing, concurrent.futures, random, time, and atpbar. ```python import multiprocessing from concurrent.futures import ProcessPoolExecutor from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter multiprocessing.set_start_method('fork', force=True) def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_workers = 5 n_jobs = 10 with ( flushing(), ProcessPoolExecutor( max_workers=n_workers, initializer=register_reporter, initargs=(find_reporter(),), ) as executor, ): for i in range(n_jobs): n = randint(1000, 10000) f = executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### Parallel Task Progress with multiprocessing.Pool and atpbar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/guide/multiprocessing-pool.md This Python code snippet demonstrates how to use `atpbar` to display progress bars for tasks executed in parallel using `multiprocessing.Pool`. It configures the pool to use `atpbar`'s reporter and then maps a function with progress indication across multiple jobs. ```python from multiprocessing import Pool, set_start_method from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter set_start_method('fork', force=True) def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_processes = 4 n_jobs = 10 args = [(randint(1000, 10000), f'Job {i}') for i in range(n_jobs)] with ( flushing(), Pool( n_processes, initializer=register_reporter, initargs=(find_reporter(),), ) as pool, ): pool.starmap(func, args) ``` -------------------------------- ### ProcessPoolExecutor Integration with atpbar (Python) Source: https://context7.com/alphatwirl/atpbar/llms.txt Shows how to integrate atpbar with `concurrent.futures.ProcessPoolExecutor` for multi-process tasks. It requires setting the multiprocessing start method and using `register_reporter` with `find_reporter` to manage progress reporting across processes. The `flushing()` context manager is also used. ```python import multiprocessing from concurrent.futures import ProcessPoolExecutor from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter multiprocessing.set_start_method('fork', force=True) def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_workers = 5 n_jobs = 10 with ( flushing(), ProcessPoolExecutor( max_workers=n_workers, initializer=register_reporter, initargs=(find_reporter(),), ) as executor, ): for i in range(n_jobs): n = randint(1000, 10000) executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### Manual Progress Bar Flushing with flush() in Python Source: https://context7.com/alphatwirl/atpbar/llms.txt Shows how to manually flush all active progress bars using the flush() function. This is useful when you need to ensure progress bars are updated outside of a context manager, for example, after a loop completes. ```python from atpbar import atpbar, flush from time import sleep for _ in atpbar(range(100), name='Task 1'): sleep(0.01) # Manually flush progress bars flush() ``` -------------------------------- ### Basic Progress Bar with atpbar() in Python Source: https://context7.com/alphatwirl/atpbar/llms.txt Demonstrates the basic usage of the atpbar() function to wrap an iterable and display a progress bar. It shows how to use it with a simple range and with a custom name label. The iterable must have a determinable length. ```python from random import randint from time import sleep from atpbar import atpbar # Basic usage - wrap any iterable with known length n = randint(1000, 10000) for _ in atpbar(range(n)): sleep(0.001) # Output: 51.25% :::::::::::::::::::: | 4132 / 8062 |: range(0, 8062) # With custom name label for item in atpbar(range(100), name='Processing items'): sleep(0.01) # Output: 100.00% :::::::::::::::::::::::::::::::::::::::: | 100 / 100 |: Processing items # Nested loops with multiple progress bars for i in atpbar(range(4), name='Outer'): n = randint(1000, 10000) for _ in atpbar(range(n), name=f'Inner {i}'): sleep(0.001) # Output shows both outer and inner progress bars simultaneously ``` -------------------------------- ### Multiprocessing Progress Bars with atpbar (Python) Source: https://github.com/alphatwirl/atpbar/blob/main/docs/guide/multiprocessing-process.md This snippet illustrates the use of `multiprocessing.Process` to run tasks in parallel, with each process displaying its progress using `atpbar`. It requires the `multiprocessing` and `atpbar` libraries. The `set_start_method('fork', force=True)` is used for compatibility. The `flushing()` context manager ensures all progress bars are displayed correctly before the program exits. ```python from multiprocessing import Process, set_start_method from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter set_start_method('fork', force=True) def func(n, name, reporter): register_reporter(reporter) for _ in atpbar(range(n), name=name): sleep(0.001) n_processes = 5 with flushing(): processes = [] for i in range(n_processes): n = randint(1000, 10000) p = Process(target=func, args=(n, f'Job {i}', find_reporter())) p.start() processes.append(p) for p in processes: p.join() ``` -------------------------------- ### Progress Bar Flushing with flushing() Context Manager in Python Source: https://context7.com/alphatwirl/atpbar/llms.txt Illustrates the use of the flushing() context manager to ensure all progress bars are properly flushed and updated upon exiting the context. This is crucial for threading and multiprocessing to prevent bars from disappearing prematurely. It's demonstrated with ThreadPoolExecutor. ```python from concurrent.futures import ThreadPoolExecutor from random import randint from time import sleep from atpbar import atpbar, flushing def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_workers = 5 n_jobs = 10 # flushing() ensures progress bars complete before exiting with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in range(n_jobs): n = randint(1000, 10000) executor.submit(func, n, name=f'Job {i}') # All progress bars will be properly displayed and flushed ``` -------------------------------- ### threading.Thread Integration with atpbar (Python) Source: https://context7.com/alphatwirl/atpbar/llms.txt Illustrates direct integration of atpbar with Python's `threading.Thread`. The `flushing()` context manager is employed to handle concurrent progress bars. This approach is suitable for I/O-bound tasks where threads are beneficial. ```python from random import randint from threading import Thread from time import sleep from atpbar import atpbar, flushing def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_threads = 5 with flushing(): threads = [] for i in atpbar(range(n_threads)): n = randint(1000, 10000) t = Thread(target=func, args=(n, f'Thread {i}')) t.start() threads.append(t) sleep(0.01) for t in threads: t.join() ``` -------------------------------- ### Break and Exception Handling with atpbar (Python) Source: https://context7.com/alphatwirl/atpbar/llms.txt Demonstrates how atpbar handles loops that terminate early due to `break` statements or exceptions. In both cases, the progress bar accurately reflects the last completed iteration before termination. This robustness extends to threaded functions. ```python import traceback from concurrent.futures import ThreadPoolExecutor from random import randint from time import sleep from atpbar import atpbar, flushing # Breaking out of a loop - progress bar updates correctly for i in atpbar(range(2000)): if i == 1234: break sleep(0.0001) # Progress bar shows 1234/2000 as final state # Exception handling - progress bar still updates try: for i in atpbar(range(2000)): if i == 1234: raise Exception("Stopping early") sleep(0.0001) except Exception: traceback.print_exc() # Progress bar shows 1234/2000 before exception # Break in threaded functions def func(n, name): for i in atpbar(range(n), name=name): if i == 1234: break sleep(0.0001) with flushing(), ThreadPoolExecutor(max_workers=5) as executor: for i in range(10): n = randint(3000, 10000) executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### ThreadPoolExecutor Integration with atpbar (Python) Source: https://context7.com/alphatwirl/atpbar/llms.txt Demonstrates using atpbar with `concurrent.futures.ThreadPoolExecutor` for multi-threaded tasks. The `flushing()` context manager is used to ensure progress bars are displayed correctly for concurrent jobs. Each thread's progress is tracked independently. ```python from concurrent.futures import ThreadPoolExecutor from random import randint from time import sleep from atpbar import atpbar, flushing def process_data(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_workers = 5 n_jobs = 10 with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in atpbar(range(n_jobs)): n = randint(1000, 10000) executor.submit(process_data, n, name=f'Job {i}') sleep(0.01) # Output shows simultaneous progress bars for all concurrent jobs ``` -------------------------------- ### Finding Reporter for Multiprocessing with find_reporter() in Python Source: https://context7.com/alphatwirl/atpbar/llms.txt Demonstrates how to obtain the progress reporter object from the main process using find_reporter(). This is essential for multiprocessing scenarios where child processes need to report progress back to the main process. It's used in conjunction with Pool and register_reporter. ```python from multiprocessing import Pool, set_start_method from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter set_start_method('fork', force=True) def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_processes = 4 n_jobs = 10 args = [(randint(1000, 10000), f'Job {i}') for i in range(n_jobs)] # Pass find_reporter() to Pool initializer with ( flushing(), Pool( n_processes, initializer=register_reporter, initargs=(find_reporter(),), ) as pool, ): pool.starmap(func, args) ``` -------------------------------- ### flushing() Context Manager Source: https://context7.com/alphatwirl/atpbar/llms.txt A context manager that ensures all progress bars are properly flushed and updated upon exiting the context. It's crucial for preventing progress bars from disappearing prematurely in threading or multiprocessing scenarios. ```APIDOC ## flushing() ### Description Context manager that guarantees all active progress bars are flushed and finalized upon exiting the `with` block. Essential for ensuring complete progress bar display in concurrent operations. ### Method Context Manager ### Endpoint N/A (Python context manager) ### Parameters None ### Request Example ```python from concurrent.futures import ThreadPoolExecutor from random import randint from time import sleep from atpbar import atpbar, flushing def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_workers = 5 n_jobs = 10 with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in range(n_jobs): n = randint(1000, 10000) executor.submit(func, n, name=f'Job {i}') ``` ### Response #### Success Response (200) Exits the context, ensuring all progress bars are flushed. #### Response Example (Progress bars are fully displayed and finalized upon script completion) ``` -------------------------------- ### Python: Exception Handling with ATPBar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/features/break-exception.md Illustrates ATPBar's behavior when an exception occurs within a loop. The progress bar halts at the last successfully completed iteration, and the exception is raised afterward. ```python from atpbar import atpbar from time import sleep for i in atpbar(range(2000)): if i == 1234: raise Exception sleep(0.001) ``` -------------------------------- ### Execute jobs concurrently using ThreadPoolExecutor and atpbar Source: https://github.com/alphatwirl/atpbar/blob/main/README.md This code snippet demonstrates the core functionality. It initializes a ThreadPoolExecutor with a specified number of workers and submits multiple jobs. Each job calls the 'func' function with random parameters. The 'flushing()' context manager ensures that progress bars are updated and finalized correctly upon completion of all threads. ```python n_workers = 5 n_jobs = 10 with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in range(n_jobs): n = randint(1000, 10000) executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### Registering Reporter in Subprocess with register_reporter() in Python Source: https://context7.com/alphatwirl/atpbar/llms.txt Explains how to register a reporter in a subprocess using register_reporter(). This function must be called within child processes of a multiprocessing program, passing the reporter object obtained from find_reporter() in the main process. It's shown with the Process class. ```python from multiprocessing import Process, set_start_method from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter set_start_method('fork', force=True) def func(n, name, reporter): # Register the reporter in the subprocess register_reporter(reporter) for _ in atpbar(range(n), name=name): sleep(0.001) n_processes = 5 with flushing(): processes = [] for i in atpbar(range(n_processes)): n = randint(1000, 10000) # Pass find_reporter() to each process p = Process(target=func, args=(n, f'Job {i}', find_reporter())) p.start() processes.append(p) for p in processes: p.join() ``` -------------------------------- ### Python: Nested Loops and Threading with ATPBar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/features/break-exception.md Shows how ATPBar manages progress bars in a multithreaded environment with nested loops. Each thread's progress bar correctly stops at the last complete iteration, even if other threads continue. ```python from atpbar import atpbar, flushing from time import sleep from random import randint from concurrent.futures import ThreadPoolExecutor def func(n, name): for i in atpbar(range(n), name=name): if i == 1234: break sleep(0.001) n_workers = 5 n_jobs = 10 with flushing(), ThreadPoolExecutor(max_workers=n_workers) as executor: for i in range(n_jobs): n = randint(3000, 10000) f = executor.submit(func, n, name=f'Job {i}') ``` -------------------------------- ### flush() Function Source: https://context7.com/alphatwirl/atpbar/llms.txt Manually triggers a flush of all currently active progress bars. Use this function when you need to ensure progress bars are updated outside of a context manager. ```APIDOC ## flush() ### Description Manually forces an update and flush of all progress bars that are currently active. Useful for ensuring visual updates at specific points in the code, especially when not using the `flushing()` context manager. ### Method Function Call ### Endpoint N/A (Python function) ### Parameters None ### Request Example ```python from atpbar import atpbar, flush from time import sleep for _ in atpbar(range(100), name='Task 1'): sleep(0.01) # Manually flush progress bars flush() ``` ### Response #### Success Response (200) All active progress bars are updated and flushed to the display. #### Response Example (Progress bars are updated on the console/notebook) ``` -------------------------------- ### find_reporter() Function Source: https://context7.com/alphatwirl/atpbar/llms.txt Retrieves the progress reporter object from the main process. This is essential for multiprocessing scenarios where child processes need to communicate progress back to the main process. ```APIDOC ## find_reporter() ### Description Returns the progress reporter instance associated with the main process. This function is primarily used in multiprocessing environments to obtain the reporter that child processes will use to send progress updates back. ### Method Function Call ### Endpoint N/A (Python function) ### Parameters None ### Request Example ```python from multiprocessing import Pool, set_start_method from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter set_start_method('fork', force=True) def func(n, name): for _ in atpbar(range(n), name=name): sleep(0.001) n_processes = 4 n_jobs = 10 args = [(randint(1000, 10000), f'Job {i}') for i in range(n_jobs)] with ( flushing(), Pool( n_processes, initializer=register_reporter, initargs=(find_reporter(),), ) as pool, ): pool.starmap(func, args) ``` ### Response #### Success Response (200) Returns the main process's reporter object. #### Response Example (A reporter object instance) ``` -------------------------------- ### atpbar() Function Source: https://context7.com/alphatwirl/atpbar/llms.txt The main function to wrap an iterable and display a progress bar. It returns an iterable that yields items while updating the progress bar. The input iterable must have a determinable length. ```APIDOC ## atpbar() ### Description Wraps an iterable to display a progress bar, updating in real-time as items are yielded. Requires the iterable to have a defined length. ### Method Function Call ### Endpoint N/A (Python function) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from random import randint from time import sleep from atpbar import atpbar n = randint(1000, 10000) for _ in atpbar(range(n)): sleep(0.001) for item in atpbar(range(100), name='Processing items'): sleep(0.01) # Nested loops example for i in atpbar(range(4), name='Outer'): n = randint(1000, 10000) for _ in atpbar(range(n), name=f'Inner {i}'): sleep(0.001) ``` ### Response #### Success Response (200) Returns an iterable that yields items from the original iterable. #### Response Example (Iterable yielding items with progress bar updates in the console/notebook) ``` -------------------------------- ### Python: Break Loop with ATPBar Source: https://github.com/alphatwirl/atpbar/blob/main/docs/features/break-exception.md Demonstrates how ATPBar stops at the last complete iteration when a loop is terminated using a 'break' statement. The progress bar reflects the iteration count before the break. ```python from atpbar import atpbar from time import sleep for i in atpbar(range(2000)): if i == 1234: break sleep(0.0001) ``` -------------------------------- ### register_reporter() Function Source: https://context7.com/alphatwirl/atpbar/llms.txt Registers a reporter object within a subprocess. This function must be called in child processes of a multiprocessing program, passing the reporter obtained from `find_reporter()` in the main process. ```APIDOC ## register_reporter() ### Description Registers a given reporter object within the current subprocess. This is a necessary step when using multiprocessing, allowing child processes to correctly associate with the reporter established in the main process via `find_reporter()`. ### Method Function Call ### Endpoint N/A (Python function) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **reporter** (object) - Required - The reporter object obtained from `find_reporter()` in the main process. ### Request Example ```python from multiprocessing import Process, set_start_method from random import randint from time import sleep from atpbar import atpbar, find_reporter, flushing, register_reporter set_start_method('fork', force=True) def func(n, name, reporter): register_reporter(reporter) for _ in atpbar(range(n), name=name): sleep(0.001) n_processes = 5 with flushing(): processes = [] for i in atpbar(range(n_processes)): n = randint(1000, 10000) p = Process(target=func, args=(n, f'Job {i}', find_reporter())) p.start() processes.append(p) for p in processes: p.join() ``` ### Response #### Success Response (200) Successfully registers the reporter in the subprocess. #### Response Example (No direct return value, but enables progress reporting from the subprocess) ``` -------------------------------- ### Disable Progress Bars Globally (Python) Source: https://context7.com/alphatwirl/atpbar/llms.txt Disables all progress bars globally. This function must be called in the main process before using `atpbar()` or `find_reporter()`. It takes no arguments and returns nothing. Progress bars will not be displayed after this call. ```python from atpbar import disable, atpbar from time import sleep # Disable progress bars (call before using atpbar) disable() # Progress bars will not be displayed for _ in atpbar(range(100)): sleep(0.01) ``` -------------------------------- ### Disable atpbar Progress Bars in Python Source: https://github.com/alphatwirl/atpbar/blob/main/docs/guide/disable.md The `disable()` function from the `atpbar` library stops all progress bars from being displayed. It requires no arguments and should be called before any other `atpbar` functions, such as `find_reporter()`, to ensure it takes effect. ```python from atpbar import disable disable() ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.