Tools and Utilities

def create_folder(dir_path):
    """
    Create a folder in the given path if it doesn't exist yet.
    
    Uses os.makedirs with exist_ok=True to safely create directory hierarchies
    without raising errors if directories already exist.
    
    Args:
        dir_path (str): Path to directory to create
    """

def title_with_actions(task):
    """
    Return task name with task actions for display.
    
    Formats task display to show both the task name and its actions,
    or shows group task information if no actions are present.
    
    Args:
        task: Task object with name, actions, and task_dep attributes
        
    Returns:
        str: Formatted string showing task name and actions
    """

def run_once(task, values):
    """
    Execute task just once (uptodate checker).
    
    Used when user manually manages a dependency and wants the task
    to run only on first execution, regardless of file changes.
    
    Args:
        task: Task object to configure
        values (dict): Saved values from previous executions
        
    Returns:
        bool: True if task should be considered up-to-date
    """

class config_changed:
    """
    Check if passed configuration was modified (uptodate checker).
    
    Monitors configuration data (strings or dictionaries) and determines
    if the configuration has changed since the last successful run.
    """
    
    def __init__(self, config, encoder=None):
        """
        Initialize configuration checker.
        
        Args:
            config (str|dict): Configuration data to monitor
            encoder (json.JSONEncoder, optional): Custom JSON encoder for non-default values
        """
    
    def __call__(self, task, values):
        """
        Check if configuration is unchanged.
        
        Args:
            task: Task object being checked
            values (dict): Saved values from previous executions
            
        Returns:
            bool: True if configuration is unchanged (task up-to-date)
        """
    
    def configure_task(self, task):
        """Configure task to save configuration digest"""

class timeout:
    """
    Add timeout to task (uptodate checker).
    
    Task is considered not up-to-date if more time has elapsed since
    last execution than the specified timeout limit.
    """
    
    def __init__(self, timeout_limit):
        """
        Initialize timeout checker.
        
        Args:
            timeout_limit (datetime.timedelta|int): Timeout in seconds or timedelta object
            
        Raises:
            Exception: If timeout_limit is not timedelta or int
        """
    
    def __call__(self, task, values):
        """
        Check if task has timed out.
        
        Args:
            task: Task object being checked
            values (dict): Saved values from previous executions
            
        Returns:
            bool: True if task has not timed out (still up-to-date)
        """

class check_timestamp_unchanged:
    """
    Check if timestamp of a file/directory is unchanged since last run (uptodate checker).
    
    Monitors file timestamps and considers task up-to-date if the specified
    timestamp has not changed since last successful execution.
    """
    
    def __init__(self, file_name, time='mtime', cmp_op=operator.eq):
        """
        Initialize timestamp checker.
        
        Args:
            file_name (str): Path to file/directory to monitor
            time (str): Timestamp type - 'atime'/'access', 'ctime'/'status', 'mtime'/'modify'
            cmp_op (callable): Comparison function (prev_time, current_time) -> bool
            
        Raises:
            ValueError: If invalid time value is specified
        """
    
    def __call__(self, task, values):
        """
        Check if file timestamp is unchanged.
        
        Args:
            task: Task object being checked
            values (dict): Saved values from previous executions
            
        Returns:
            bool: True if timestamp is unchanged (task up-to-date)
            
        Raises:
            OSError: If cannot stat the specified file
        """

def set_trace():
    """
    Start debugger, ensuring stdout shows pdb output.
    
    Sets up Python debugger (pdb) with proper stdin/stdout configuration
    for debugging doit tasks. Output is not restored after debugging.
    
    Note: This is a development utility and output streams are not restored.
    """

def load_ipython_extension(ip=None):
    """
    Define a %doit magic function for IPython that discovers and executes tasks
    from interactive variables (global namespace).
    
    Creates a magic function that allows running doit commands directly within
    IPython sessions, using the current namespace as the source for task definitions.
    
    Args:
        ip (IPython instance, optional): IPython instance, auto-detected if None
        
    Raises:
        ImportError: If not running within an IPython environment
        
    Usage in IPython:
        %load_ext doit
        %doit list        # List tasks from current namespace
        %doit             # Run tasks from current namespace
        %doit --help      # Show help
    """

register_doit_as_IPython_magic = load_ipython_extension
# Alternative alias for registering IPython extension

from .task import result_dep  # Imported for backward compatibility

from doit.tools import config_changed
import json

def task_deploy():
    """Deploy with configuration change detection"""
    
    config = {
        'server': 'production.example.com',
        'port': 443,
        'ssl': True,
        'workers': 4
    }
    
    return {
        'actions': ['deploy.sh --config deployment.json'],
        'uptodate': [config_changed(config)],
        'verbosity': 2
    }

# Task will only run if config dictionary changes

from doit.tools import timeout
import datetime

def task_backup():
    """Run backup every 6 hours"""
    return {
        'actions': ['backup_script.sh'],
        'uptodate': [timeout(datetime.timedelta(hours=6))],
        'verbosity': 2
    }

# Task runs again if more than 6 hours have passed

from doit.tools import check_timestamp_unchanged

def task_process_config():
    """Process configuration file"""
    return {
        'actions': ['process_config.py config.yaml'],
        'uptodate': [check_timestamp_unchanged('config.yaml')],
        'targets': ['processed_config.json']
    }

# Task runs only if config.yaml timestamp changes

# In IPython/Jupyter notebook cell:

def task_analyze():
    """Analyze data in current notebook"""
    return {
        'actions': ['python analyze_data.py'],
        'file_dep': ['data.csv']
    }

def task_plot():
    """Generate plots"""
    return {
        'actions': ['python generate_plots.py'],
        'file_dep': ['processed_data.json'],
        'targets': ['plots.png']
    }

# Load the extension
%load_ext doit

# List available tasks from notebook namespace
%doit list

# Run tasks
%doit analyze
%doit plot

from doit.tools import create_folder, set_trace, config_changed

def task_setup_dev():
    """Setup development environment"""
    
    dev_config = {
        'debug': True,
        'log_level': 'DEBUG',
        'db_name': 'myapp_dev'
    }
    
    def setup_action():
        create_folder('logs')
        create_folder('tmp')
        create_folder('data/dev')
        
        # Uncomment for debugging
        # set_trace()
        
        print("Development environment ready")
    
    return {
        'actions': [setup_action],
        'uptodate': [config_changed(dev_config)],
        'verbosity': 2
    }