utils.py

"""
Utility Functions

Helper functions for formatting, data export, and visualization configuration.
"""

import json
import csv
from datetime import datetime
from typing import Dict, List, Any
import pandas as pd


def format_timestamp(timestamp: float, format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
    """
    Format a Unix timestamp as a human-readable string.
    
    Args:
        timestamp: Unix timestamp
        format_str: strftime format string
        
    Returns:
        Formatted date/time string
    """
    return datetime.fromtimestamp(timestamp).strftime(format_str)


def format_currency(amount: float, decimals: int = 2, symbol: str = "") -> str:
    """
    Format a number as currency.
    
    Args:
        amount: Amount to format
        decimals: Number of decimal places
        symbol: Currency symbol (e.g., "$", "€")
        
    Returns:
        Formatted currency string
    """
    formatted = f"{amount:,.{decimals}f}"
    return f"{symbol}{formatted}" if symbol else formatted


def format_percentage(value: float, decimals: int = 1) -> str:
    """
    Format a decimal as a percentage.
    
    Args:
        value: Decimal value (e.g., 0.25 for 25%)
        decimals: Number of decimal places
        
    Returns:
        Formatted percentage string
    """
    return f"{value * 100:.{decimals}f}%"


def get_color_scheme() -> Dict[str, str]:
    """
    Get the default color scheme for visualizations.
    
    Returns:
        Dictionary mapping element types to colors
    """
    return {
        # Node types
        'agent': '#FF6B6B',
        'shard': '#4ECDC4',
        'asset': '#FFE66D',
        
        # Status colors
        'active': '#00FF00',
        'inactive': '#888888',
        'pending': '#FFA500',
        'completed': '#00FF00',
        'failed': '#FF0000',
        
        # Chain colors
        'ethereum': '#627EEA',
        'bsc': '#F3BA2F',
        'polygon': '#8247E5',
        
        # Shard types
        'compute': '#FF6B6B',
        'storage': '#4ECDC4',
        'ai': '#A8E6CF',
        
        # General
        'primary': '#3498db',
        'secondary': '#2ecc71',
        'accent': '#e74c3c',
        'background': '#f5f5f5'
    }


def get_plotly_template() -> Dict[str, Any]:
    """
    Get default Plotly template configuration.
    
    Returns:
        Dictionary with Plotly layout settings
    """
    return {
        'plot_bgcolor': 'rgba(240,240,240,0.9)',
        'paper_bgcolor': 'white',
        'font': {'family': 'Arial, sans-serif', 'size': 12},
        'title': {'font': {'size': 16, 'color': '#333'}},
        'showlegend': True,
        'hovermode': 'closest'
    }


def export_to_json(data: Any, filename: str) -> str:
    """
    Export data to JSON file.
    
    Args:
        data: Data to export (must be JSON serializable)
        filename: Output filename
        
    Returns:
        Success message
    """
    try:
        with open(filename, 'w') as f:
            json.dump(data, f, indent=2, default=str)
        return f"Data exported to {filename}"
    except Exception as e:
        return f"Export failed: {str(e)}"


def export_to_csv(data: List[Dict], filename: str) -> str:
    """
    Export list of dictionaries to CSV file.
    
    Args:
        data: List of dictionaries with consistent keys
        filename: Output filename
        
    Returns:
        Success message
    """
    try:
        if not data:
            return "No data to export"
        
        keys = data[0].keys()
        
        with open(filename, 'w', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=keys)
            writer.writeheader()
            writer.writerows(data)
        
        return f"Data exported to {filename}"
    except Exception as e:
        return f"Export failed: {str(e)}"


def validate_amount(amount: float, min_value: float = 0.0, 
                   max_value: float = float('inf')) -> tuple[bool, str]:
    """
    Validate a numeric amount.
    
    Args:
        amount: Amount to validate
        min_value: Minimum allowed value
        max_value: Maximum allowed value
        
    Returns:
        Tuple of (is_valid, error_message)
    """
    if not isinstance(amount, (int, float)):
        return False, "Amount must be a number"
    
    if amount < min_value:
        return False, f"Amount must be at least {min_value}"
    
    if amount > max_value:
        return False, f"Amount must not exceed {max_value}"
    
    return True, ""


def truncate_string(s: str, max_length: int = 20, suffix: str = "...") -> str:
    """
    Truncate a string to maximum length.
    
    Args:
        s: String to truncate
        max_length: Maximum length (including suffix)
        suffix: Suffix to add if truncated
        
    Returns:
        Truncated string
    """
    if len(s) <= max_length:
        return s
    
    return s[:max_length - len(suffix)] + suffix


def aggregate_by_key(data: List[Dict], key: str, value_key: str, 
                     agg_func: str = 'sum') -> Dict:
    """
    Aggregate data by a key.
    
    Args:
        data: List of dictionaries
        key: Key to group by
        value_key: Key containing values to aggregate
        agg_func: Aggregation function ('sum', 'avg', 'count', 'min', 'max')
        
    Returns:
        Dictionary mapping key values to aggregated values
    """
    from collections import defaultdict
    
    groups = defaultdict(list)
    
    for item in data:
        if key in item and value_key in item:
            groups[item[key]].append(item[value_key])
    
    result = {}
    
    for k, values in groups.items():
        if agg_func == 'sum':
            result[k] = sum(values)
        elif agg_func == 'avg':
            result[k] = sum(values) / len(values) if values else 0
        elif agg_func == 'count':
            result[k] = len(values)
        elif agg_func == 'min':
            result[k] = min(values) if values else 0
        elif agg_func == 'max':
            result[k] = max(values) if values else 0
    
    return result


def create_dataframe(data: List[Dict]) -> pd.DataFrame:
    """
    Create a pandas DataFrame from list of dictionaries.
    
    Args:
        data: List of dictionaries
        
    Returns:
        pandas DataFrame
    """
    if not data:
        return pd.DataFrame()
    
    return pd.DataFrame(data)


def format_dict_for_display(d: Dict, indent: int = 2) -> str:
    """
    Format a dictionary for readable display.
    
    Args:
        d: Dictionary to format
        indent: Indentation spaces
        
    Returns:
        Formatted string
    """
    lines = []
    for key, value in d.items():
        # Format key
        key_str = str(key).replace('_', ' ').title()
        
        # Format value
        if isinstance(value, float):
            value_str = f"{value:.2f}"
        elif isinstance(value, dict):
            value_str = json.dumps(value, indent=indent)
        else:
            value_str = str(value)
        
        lines.append(f"{' ' * indent}{key_str}: {value_str}")
    
    return '\n'.join(lines)


def calculate_growth_rate(old_value: float, new_value: float) -> float:
    """
    Calculate percentage growth rate.
    
    Args:
        old_value: Previous value
        new_value: Current value
        
    Returns:
        Growth rate as decimal (e.g., 0.25 for 25% growth)
    """
    if old_value == 0:
        return 0.0 if new_value == 0 else float('inf')
    
    return (new_value - old_value) / old_value


def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
    """
    Safely divide two numbers, returning default if denominator is zero.
    
    Args:
        numerator: Numerator
        denominator: Denominator
        default: Value to return if denominator is zero
        
    Returns:
        Result of division or default
    """
    return numerator / denominator if denominator != 0 else default


def create_summary_stats(values: List[float]) -> Dict[str, float]:
    """
    Calculate summary statistics for a list of values.
    
    Args:
        values: List of numeric values
        
    Returns:
        Dictionary with summary statistics
    """
    if not values:
        return {
            'count': 0,
            'sum': 0.0,
            'mean': 0.0,
            'min': 0.0,
            'max': 0.0,
            'std': 0.0
        }
    
    import statistics
    
    return {
        'count': len(values),
        'sum': sum(values),
        'mean': statistics.mean(values),
        'min': min(values),
        'max': max(values),
        'std': statistics.stdev(values) if len(values) > 1 else 0.0
    }


if __name__ == "__main__":
    # Demo usage
    import time
    
    print("=== Utility Functions Demo ===\n")
    
    # Timestamp formatting
    now = time.time()
    print(f"Timestamp: {format_timestamp(now)}")
    
    # Currency formatting
    print(f"Currency: {format_currency(1234.567, symbol='$')}")
    
    # Percentage formatting
    print(f"Percentage: {format_percentage(0.847)}")
    
    # Colors
    colors = get_color_scheme()
    print(f"\nColor scheme has {len(colors)} colors")
    print(f"Agent color: {colors['agent']}")
    
    # Validation
    is_valid, error = validate_amount(50.0, min_value=0, max_value=100)
    print(f"\nValidation: {is_valid} (error: '{error}')")
    
    # Aggregation
    data = [
        {'type': 'A', 'value': 10},
        {'type': 'B', 'value': 20},
        {'type': 'A', 'value': 15},
        {'type': 'B', 'value': 25}
    ]
    
    aggregated = aggregate_by_key(data, 'type', 'value', 'sum')
    print(f"\nAggregated by type: {aggregated}")
    
    # Summary stats
    values = [10, 20, 30, 40, 50]
    stats = create_summary_stats(values)
    print(f"\nSummary statistics:")
    print(format_dict_for_display(stats))