Utilities and Extensions

from tqdm.contrib import tenumerate, tzip, tmap, DummyTqdmFile
from tqdm.contrib.itertools import product

def tenumerate(iterable, start=0, total=None, tqdm_class=tqdm_auto, **kwargs):
    """
    Progress-wrapped enumerate function.
    
    Equivalent to enumerate() but with tqdm progress tracking.
    
    Parameters:
    - iterable: Iterable to enumerate
    - start: Starting value for enumeration (default: 0)
    - total: Total number of items (auto-detected if None)
    - tqdm_class: tqdm class to use (default: tqdm.auto.tqdm)
    - **kwargs: Additional arguments passed to tqdm constructor
    
    Yields:
    Tuples of (index, item) with progress tracking
    """

def tzip(iter1, *iter2plus, **kwargs):
    """
    Progress-wrapped zip function.
    
    Equivalent to zip() but with tqdm progress tracking.
    
    Parameters:
    - iter1: First iterable
    - *iter2plus: Additional iterables to zip
    - **kwargs: Arguments passed to tqdm constructor
    
    Yields:
    Tuples of items from each iterable with progress tracking
    """

def tmap(function, *sequences, **kwargs):
    """
    Progress-wrapped map function.
    
    Equivalent to map() but with tqdm progress tracking.
    
    Parameters:
    - function: Function to apply to each element
    - *sequences: Sequences to map over
    - **kwargs: Arguments passed to tqdm constructor
    
    Yields:
    Results of function applied to each element with progress tracking
    """

def product(*iterables, **tqdm_kwargs):
    """
    Progress-wrapped itertools.product function.
    
    Equivalent to itertools.product() but with tqdm progress tracking.
    Automatically calculates total iterations based on input sizes.
    
    Parameters:
    - *iterables: Iterables to compute Cartesian product of
    - tqdm_class: tqdm class to use (default: tqdm.auto.tqdm)
    - **tqdm_kwargs: Additional arguments passed to tqdm constructor
    
    Yields:
    Tuples from Cartesian product with progress tracking
    """

class DummyTqdmFile(ObjectWrapper):
    """
    File-like wrapper for tqdm output redirection.
    
    Allows redirecting progress bar output to custom file objects
    while maintaining compatibility with standard file operations.
    """
    def __init__(self, wrapped): ...
    def write(self, x, nolock=False): ...
    def __del__(self): ...

from tqdm import TMonitor, TqdmSynchronisationWarning

class TMonitor(threading.Thread):
    """
    Monitoring thread for maximum interval enforcement.
    
    Ensures progress bars are updated at least every maxinterval seconds
    even if the main thread is busy with other tasks.
    """
    def __init__(self, tqdm_cls, sleep_interval=None): ...
    def exit(self): ...
    def get_instances(self): ...

class TqdmSynchronisationWarning(TqdmWarning, RuntimeWarning):
    """
    Warning for thread synchronization issues.
    
    Raised when multiple progress bars interfere with each other
    in multi-threaded environments.
    """
    pass

from tqdm.utils import (
    FormatReplace, Comparable, ObjectWrapper, SimpleTextIOWrapper,
    DisableOnWriteError, CallbackIOWrapper, envwrap, _supports_unicode,
    _is_ascii, _screen_shape_wrapper, _term_move_up, disp_len, disp_trim
)

class FormatReplace(object):
    """
    String formatting with field replacements.
    
    Provides advanced string templating with replacement patterns
    for custom progress bar formats.
    """
    def __init__(self, fmt, **fmt_vars): ...
    def format(self, **fmt_vars): ...

class Comparable(object):
    """
    Base class providing comparison operations.
    
    Enables comparison and sorting of tqdm instances based on
    position, description, or other attributes.
    """
    def __lt__(self, other): ...
    def __le__(self, other): ...
    def __eq__(self, other): ...
    def __ne__(self, other): ...
    def __gt__(self, other): ...
    def __ge__(self, other): ...

class ObjectWrapper(object):
    """
    Base wrapper class for object proxying.
    
    Provides foundation for creating wrapper objects that
    proxy method calls and attribute access to wrapped objects.
    """
    def __init__(self, wrapped): ...
    def __getattr__(self, name): ...
    def __setattr__(self, name, value): ...

class SimpleTextIOWrapper(ObjectWrapper):
    """
    Text stream wrapper with encoding handling.
    
    Wraps text I/O streams to handle encoding issues and
    provide consistent text output across different platforms.
    """
    def write(self, s): ...
    def flush(self): ...

class DisableOnWriteError(ObjectWrapper):  
    """
    Wrapper that disables on write errors.
    
    Automatically disables progress bars when write operations
    fail, preventing cascading errors in restricted environments.
    """
    def write(self, s): ...

class CallbackIOWrapper(ObjectWrapper):
    """
    I/O wrapper with callback functionality.
    
    Provides hooks for monitoring I/O operations and
    integrating with progress tracking systems.
    """
    def __init__(self, callback, stream, method="read"): ...
    def read(self, n=-1): ...
    def write(self, s): ...

def envwrap(prefix, types=None, is_method=False):
    """
    Decorator for environment variable configuration.
    
    Automatically reads configuration from environment variables
    with specified prefix and applies type conversion.
    
    Parameters:
    - prefix: Environment variable prefix (e.g., 'TQDM_')
    - types: Type conversion mapping for parameters
    - is_method: Whether decorating a method (vs function)
    
    Returns:
    Decorator function that applies environment configuration
    """

def _supports_unicode(fp):
    """
    Check if file pointer supports Unicode output.
    
    Parameters:
    - fp: File pointer to test
    
    Returns:
    Boolean indicating Unicode support
    """

def _is_ascii(s):
    """
    Check if string contains only ASCII characters.
    
    Parameters:
    - s: String to test
    
    Returns:
    Boolean indicating ASCII-only content
    """

def _screen_shape_wrapper():
    """
    Get terminal screen dimensions with fallbacks.
    
    Returns:
    Tuple of (rows, columns) or None if unavailable
    """

def _term_move_up():
    """
    Generate terminal cursor move up sequence.
    
    Returns:
    ANSI escape sequence for moving cursor up one line
    """

def disp_len(data):
    """
    Calculate display length accounting for Unicode.
    
    Parameters:
    - data: String data to measure
    
    Returns:
    Display width of string accounting for wide characters
    """

def disp_trim(data, length):
    """
    Trim string to specified display length.
    
    Parameters:
    - data: String to trim
    - length: Target display length
    
    Returns:
    Trimmed string with proper Unicode handling
    """

from tqdm.contrib.logging import tqdm_logging_redirect

@contextmanager
def tqdm_logging_redirect(*tqdm_args, **tqdm_kwargs):
    """
    Context manager for redirecting logging output during progress tracking.
    
    Temporarily redirects log messages to avoid interfering with
    progress bar display, then restores normal logging.
    
    Parameters:
    - *tqdm_args: Arguments passed to tqdm constructor
    - **tqdm_kwargs: Keyword arguments passed to tqdm constructor
    
    Yields:
    tqdm instance with logging redirection active
    """

from tqdm.contrib.slack import SlackTqdm, tqdm_slack
from tqdm.contrib.discord import DiscordTqdm, tqdm_discord
from tqdm.contrib.telegram import TelegramTqdm, tqdm_telegram

class SlackTqdm(tqdm_auto):
    """
    Progress bar that sends updates to Slack channels.
    
    Requires slack-sdk package and proper authentication setup.
    Useful for monitoring long-running processes remotely.
    """
    def __init__(self, *args, token=None, channel=None, **kwargs): ...

class DiscordTqdm(tqdm_auto):
    """
    Progress bar that sends updates to Discord channels.
    
    Requires requests package and Discord webhook URL.
    Provides real-time progress updates via Discord messages.
    """
    def __init__(self, *args, token=None, channel=None, **kwargs): ...

class TelegramTqdm(tqdm_auto):
    """
    Progress bar that sends updates to Telegram chats.
    
    Requires requests package and Telegram bot token.
    Sends progress notifications to specified chat ID.
    """
    def __init__(self, *args, token=None, chat_id=None, **kwargs): ...

# Convenience functions
def tqdm_slack(*args, **kwargs):
    """Slack-enabled tqdm with simplified setup"""

def tqdm_discord(*args, **kwargs):
    """Discord-enabled tqdm with simplified setup"""

def tqdm_telegram(*args, **kwargs):
    """Telegram-enabled tqdm with simplified setup"""

from tqdm.cli import main, cast

def main(fp=sys.stderr, argv=None):
    """
    Command-line interface for tqdm as shell pipe processor.
    
    Enables using tqdm in shell pipelines to add progress bars
    to any command that processes stdin/stdout. Supports extensive
    configuration options for progress display and data processing.
    
    CLI Options:
    - --help: Show help message
    - --version: Show version information
    - --desc: Set description text
    - --total: Set total expected iterations
    - --ncols: Set progress bar width
    - --ascii: Use ASCII characters only
    - --unit: Set unit of measurement
    - --unit_scale: Enable unit scaling (K, M, G)
    - --rate: Show iteration rate
    - --bar_format: Custom bar format string
    - --bytes: Count bytes instead of lines
    - --delim: Set delimiter character (default: newline)
    - --buf_size: Set buffer size for processing
    - --null: Discard input (no output)
    - --update: Treat input as progress updates
    - --update_to: Treat input as absolute position
    - --tee: Pass input to both stderr and stdout
    - --log: Set logging level
    - --manpath: Directory for manual pages
    - --comppath: Directory for completions
    
    Parameters:
    - fp: Output stream for progress bar (default: sys.stderr)
    - argv: Command line arguments (default: sys.argv)
    
    Returns:
    Exit code (0 for success)
    """

def cast(val, typ):
    """
    Type casting utility for CLI arguments with support for multiple types.
    
    Handles type conversion for CLI parameters including booleans,
    characters, strings, integers, floats, and union types.
    
    Parameters:
    - val: String value to cast
    - typ: Target type string (e.g., 'int', 'float', 'bool', 'str or int')
    
    Returns:
    Casted value with appropriate type
    
    Raises:
    TqdmTypeError: If type conversion fails
    """

from tqdm.contrib import tenumerate, tzip, tmap
import time

# Progress-wrapped enumerate
data = ['item1', 'item2', 'item3', 'item4', 'item5'] * 200
for i, item in tenumerate(data, desc="Enumerating"):
    time.sleep(0.001)
    # Process item with index
    
# Progress-wrapped zip
list1 = range(1000)
list2 = range(1000, 2000)
list3 = range(2000, 3000)

for a, b, c in tzip(list1, list2, list3, desc="Zipping"):
    result = a + b + c
    time.sleep(0.0001)

# Progress-wrapped map
def square(x):
    time.sleep(0.001)  # Simulate processing
    return x ** 2

numbers = range(100)
squared = list(tmap(square, numbers, desc="Mapping"))
print(f"First 10 squares: {squared[:10]}")

from tqdm.contrib import tenumerate, tzip, tmap
import itertools
import random

# Complex data processing pipeline
def process_data_pipeline():
    # Generate sample data
    raw_data = [random.randint(1, 1000) for _ in range(1000)]
    
    # Step 1: Filter and enumerate with progress
    print("Step 1: Filtering data...")
    filtered_data = []
    for i, value in tenumerate(raw_data, desc="Filtering"):
        if value % 2 == 0:  # Keep even numbers
            filtered_data.append((i, value))
    
    # Step 2: Pair with additional data
    print("Step 2: Pairing with metadata...")
    metadata = [f"meta_{i}" for i in range(len(filtered_data))]
    
    paired_data = []
    for (idx, val), meta in tzip(filtered_data, metadata, desc="Pairing"):
        paired_data.append({
            'original_index': idx,
            'value': val,
            'metadata': meta,
            'processed': False
        })
    
    # Step 3: Apply transformations
    print("Step 3: Applying transformations...")
    def transform_item(item):
        time.sleep(0.001)  # Simulate processing
        return {
            **item,
            'transformed_value': item['value'] * 2 + 10,
            'processed': True
        }
    
    transformed_data = list(tmap(
        transform_item, 
        paired_data, 
        desc="Transforming",
        total=len(paired_data)
    ))
    
    return transformed_data

# Run the pipeline
results = process_data_pipeline()
print(f"Processed {len(results)} items")

from tqdm.contrib.logging import tqdm_logging_redirect
from tqdm.auto import tqdm
import logging
import time

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def process_with_logging():
    """Example of processing with logging that doesn't interfere with progress bars"""
    
    # Without tqdm_logging_redirect, log messages would interfere with progress bar
    with tqdm_logging_redirect(total=100, desc="Processing with logging") as pbar:
        for i in range(100):
            # Simulate work
            time.sleep(0.02)
            
            # Log messages are redirected and won't interfere with progress bar
            if i % 10 == 0:
                logger.info(f"Processed {i} items")
            
            if i % 25 == 0 and i > 0:
                logger.warning(f"Checkpoint at {i} items")
            
            pbar.update(1)
    
    # After context exits, logging returns to normal
    logger.info("Processing completed successfully")

# Run example
process_with_logging()

from tqdm.utils import CallbackIOWrapper, DisableOnWriteError
from tqdm.auto import tqdm
import io
import time

def file_processing_with_progress():
    """Example of file processing with progress tracking"""
    
    # Create sample data
    sample_data = "Sample line of data\n" * 10000
    input_stream = io.StringIO(sample_data)
    output_stream = io.StringIO()
    
    # Calculate total size for progress tracking
    total_size = len(sample_data)
    
    # Create progress bar
    pbar = tqdm(total=total_size, desc="Processing file", unit="B", unit_scale=True)
    
    def update_callback(chunk_size):
        """Callback to update progress bar"""
        pbar.update(chunk_size)
    
    # Wrap input stream with callback
    wrapped_input = CallbackIOWrapper(update_callback, input_stream, method="read")
    
    # Process data with progress tracking
    chunk_size = 1024
    while True:
        chunk = wrapped_input.read(chunk_size)
        if not chunk:
            break
        
        # Simulate processing
        processed_chunk = chunk.upper()  # Simple transformation
        output_stream.write(processed_chunk)
        time.sleep(0.001)  # Simulate processing time
    
    pbar.close()
    
    # Get results
    result = output_stream.getvalue()
    print(f"Processed {len(result)} characters")
    
    return result

# Run file processing example
processed_data = file_processing_with_progress()

from tqdm.utils import DisableOnWriteError
from tqdm.auto import tqdm
import sys
import io

def resilient_output_example():
    """Example of error-resilient output handling"""
    
    # Create a stream that might fail
    class UnreliableStream:
        def __init__(self):
            self.write_count = 0
            
        def write(self, s):
            self.write_count += 1
            # Simulate occasional write failures
            if self.write_count % 50 == 0:
                raise IOError("Simulated write failure")
            return len(s)
        
        def flush(self):
            pass
    
    unreliable_stream = UnreliableStream()
    
    # Wrap with error-resilient wrapper
    safe_stream = DisableOnWriteError(unreliable_stream)
    
    # Use with tqdm - progress bar will automatically disable on write errors
    try:
        for i in tqdm(range(200), desc="Resilient output", file=safe_stream):
            time.sleep(0.01)
    except Exception as e:
        print(f"Caught exception: {e}")
    
    print("Processing completed (progress bar may have been disabled due to write errors)")

# Run resilient output example
resilient_output_example()

from tqdm.utils import FormatReplace, disp_len, disp_trim
from tqdm.auto import tqdm
import time

def custom_formatting_example():
    """Example of custom progress bar formatting"""
    
    # Custom format with replaceable fields
    custom_format = FormatReplace(
        "Processing: {percentage:3.0f}% |{bar}| {n_fmt}/{total_fmt} "
        "[{elapsed}<{remaining}, {rate_fmt}] {custom_field}"
    )
    
    # Progress with custom formatting
    pbar = tqdm(total=100, desc="Custom Format")
    
    for i in range(100):
        time.sleep(0.02)
        
        # Update progress with custom field
        custom_info = f"Step_{i//10}"
        pbar.set_postfix_str(f"Stage: {custom_info}")
        pbar.update(1)
    
    pbar.close()
    
    # Demonstrate text measurement and trimming
    long_text = "This is a very long text that might not fit in the available space"
    display_width = 30
    
    print(f"Original text length: {len(long_text)}")
    print(f"Display length: {disp_len(long_text)}")
    
    trimmed_text = disp_trim(long_text, display_width)
    print(f"Trimmed text: '{trimmed_text}'")
    print(f"Trimmed display length: {disp_len(trimmed_text)}")

# Run custom formatting example
custom_formatting_example()

from tqdm.utils import envwrap
from tqdm.auto import tqdm
import os
import time

# Example of environment-configurable function
@envwrap('MYTQDM_')
def configurable_progress(iterable, desc="Processing", **kwargs):
    """Function that can be configured via environment variables"""
    # Environment variables like MYTQDM_NCOLS, MYTQDM_ASCII, etc. 
    # will automatically override default values
    return tqdm(iterable, desc=desc, **kwargs)

def environment_config_example():
    """Example of environment-based configuration"""
    
    # Set some environment variables for testing
    os.environ['MYTQDM_NCOLS'] = '50'
    os.environ['MYTQDM_ASCII'] = 'True'
    os.environ['MYTQDM_COLOUR'] = 'green'
    
    # Use the configurable function
    data = range(100)
    for item in configurable_progress(data, desc="Env Configured"):
        time.sleep(0.01)
    
    # Clean up environment variables
    for key in ['MYTQDM_NCOLS', 'MYTQDM_ASCII', 'MYTQDM_COLOUR']:
        os.environ.pop(key, None)

# Run environment configuration example
environment_config_example()

from tqdm.contrib.slack import tqdm_slack
from tqdm.contrib.discord import tqdm_discord
from tqdm.contrib.telegram import tqdm_telegram
import time
import os

def notification_examples():
    """Examples of using external service notifications"""
    
    # Slack integration example
    if os.getenv('SLACK_TOKEN') and os.getenv('SLACK_CHANNEL'):
        print("Running with Slack notifications...")
        for i in tqdm_slack(
            range(100), 
            desc="Processing with Slack",
            token=os.getenv('SLACK_TOKEN'),
            channel=os.getenv('SLACK_CHANNEL')
        ):
            time.sleep(0.1)
    
    # Discord integration example  
    if os.getenv('DISCORD_WEBHOOK'):
        print("Running with Discord notifications...")
        for i in tqdm_discord(
            range(50),
            desc="Processing with Discord", 
            token=os.getenv('DISCORD_WEBHOOK')
        ):
            time.sleep(0.2)
    
    # Telegram integration example
    if os.getenv('TELEGRAM_TOKEN') and os.getenv('TELEGRAM_CHAT_ID'):
        print("Running with Telegram notifications...")
        for i in tqdm_telegram(
            range(75),
            desc="Processing with Telegram",
            token=os.getenv('TELEGRAM_TOKEN'),
            chat_id=os.getenv('TELEGRAM_CHAT_ID')
        ):
            time.sleep(0.15)

# Run notification examples (only if environment variables are set)
notification_examples()

import subprocess
import os

def cli_examples():
    """Examples of using tqdm CLI in shell pipelines"""
    
    # Basic pipe usage
    cmd1 = "seq 1000 | python -m tqdm --desc 'Counting' --unit 'numbers'"
    print(f"Running: {cmd1}")
    subprocess.run(cmd1, shell=True)
    
    # Byte counting with progress
    cmd2 = "cat /dev/zero | head -c 10MB | python -m tqdm --bytes --desc 'Reading bytes'"
    print(f"Running: {cmd2}")
    subprocess.run(cmd2, shell=True)
    
    # Custom formatting
    cmd3 = "seq 500 | python -m tqdm --desc 'Custom' --ncols 50 --ascii"
    print(f"Running: {cmd3}")
    subprocess.run(cmd3, shell=True)
    
    # Progress updates from external source
    cmd4 = "echo -e '10\\n20\\n30\\n40\\n50' | python -m tqdm --update --total 50"
    print(f"Running: {cmd4}")
    subprocess.run(cmd4, shell=True)

# Uncomment to run CLI examples
# cli_examples()

from tqdm.utils import _supports_unicode, _is_ascii, _screen_shape_wrapper, _term_move_up
import sys

def terminal_capabilities_demo():
    """Demonstrate terminal capability detection"""
    
    print("Terminal Capability Detection:")
    print(f"Unicode support: {_supports_unicode(sys.stderr)}")
    print(f"Screen dimensions: {_screen_shape_wrapper()}")
    
    # Test ASCII detection
    test_strings = [
        "Hello World",
        "Hello 🌍",
        "Progress: ██████",
        "Progress: ======",
    ]
    
    for test_str in test_strings:
        is_ascii = _is_ascii(test_str)
        display_length = disp_len(test_str)
        print(f"'{test_str}': ASCII={is_ascii}, Display length={display_length}")
    
    # Demonstrate cursor movement (be careful with this in production)
    print("\nDemonstrating cursor movement:")
    print("Line 1")
    print("Line 2")
    print("Line 3")
    
    # Move cursor up and overwrite
    move_sequence = _term_move_up()
    if move_sequence:
        sys.stderr.write(move_sequence)
        sys.stderr.write("Overwritten Line 3\n")
    
# Run terminal capabilities demo
terminal_capabilities_demo()

from tqdm.utils import ObjectWrapper
from tqdm.auto import tqdm

class CustomTqdmExtension(ObjectWrapper):
    """Example of creating a custom tqdm extension"""
    
    def __init__(self, wrapped_tqdm, custom_config=None):
        super().__init__(wrapped_tqdm)
        self.custom_config = custom_config or {}
        self.custom_stats = {'custom_counter': 0}
    
    def update(self, n=1):
        """Override update to add custom functionality"""
        result = super().update(n)
        self.custom_stats['custom_counter'] += n
        
        # Custom logic based on progress
        if self.custom_stats['custom_counter'] % 100 == 0:
            self.set_postfix({
                'milestone': self.custom_stats['custom_counter']
            })
        
        return result
    
    def get_custom_stats(self):
        """Provide access to custom statistics"""
        return self.custom_stats.copy()

# Usage example
def custom_extension_example():
    base_pbar = tqdm(total=1000, desc="Custom Extension")
    custom_pbar = CustomTqdmExtension(base_pbar, {'threshold': 100})
    
    for i in range(1000):
        custom_pbar.update(1)
        time.sleep(0.001)
    
    stats = custom_pbar.get_custom_stats()
    print(f"Custom stats: {stats}")
    custom_pbar.close()

# Run custom extension example
custom_extension_example()