modules/utils.py

# UTILITIES MODULE
import os
import json
import logging
import random
import time
import functools
import traceback
from datetime import datetime
from pathlib import Path

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

# Base paths - adjusted for Hugging Face
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA_DIR = os.path.join(BASE_DIR, "data")
CONFIG_DIR = os.path.join(BASE_DIR, "config")
LOGS_DIR = os.path.join(BASE_DIR, "logs")
STATIC_DIR = os.path.join(BASE_DIR, "static")
RECORDINGS_DIR = os.path.join(STATIC_DIR, "recordings")
TONES_DIR = os.path.join(STATIC_DIR, "tones")

# Check for Hugging Face environment
if os.environ.get('SPACE_ID'):
    # Use persistent storage on Hugging Face
    logger.info("Running on Hugging Face - using persistent storage")
    os.makedirs("/persistent", exist_ok=True)
    DATA_DIR = "/persistent/data"
    CONFIG_DIR = "/persistent/config"
    LOGS_DIR = "/persistent/logs"

# System state singleton
class SystemState:
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(SystemState, cls).__new__(cls)
            cls._instance.online = True
            cls._instance.demo_mode = True
            cls._instance.debug_mode = False
            cls._instance._load_state()
        return cls._instance
    
    def _load_state(self):
        """Load system state from file"""
        state_file = os.path.join(CONFIG_DIR, "system_state.json")
        try:
            if os.path.exists(state_file):
                with open(state_file, 'r') as f:
                    state = json.load(f)
                    self.online = state.get("online", True)
                    self.demo_mode = state.get("demo_mode", True)
                    self.debug_mode = state.get("debug_mode", False)
                    logger.info(f"Loaded system state: online={self.online}, demo_mode={self.demo_mode}")
        except Exception as e:
            logger.error(f"Error loading system state: {str(e)}")
    
    def _save_state(self):
        """Save system state to file"""
        state_file = os.path.join(CONFIG_DIR, "system_state.json")
        try:
            os.makedirs(os.path.dirname(state_file), exist_ok=True)
            state = {
                "online": self.online,
                "demo_mode": self.demo_mode,
                "debug_mode": self.debug_mode,
                "last_backup": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
            with open(state_file, 'w') as f:
                json.dump(state, f, indent=2)
                logger.info("System state saved")
        except Exception as e:
            logger.error(f"Error saving system state: {str(e)}")
    
    def set_system_online(self, value):
        """Set system online/offline"""
        self.online = bool(value)
        self._save_state()
    
    def set_demo_mode(self, value):
        """Set demo mode"""
        self.demo_mode = bool(value)
        self._save_state()
    
    def set_debug_mode(self, value):
        """Set debug mode"""
        self.debug_mode = bool(value)
        self._save_state()

# Error handling system
class ErrorManager:
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(ErrorManager, cls).__new__(cls)
            cls._instance._errors = []
            cls._instance._error_callbacks = []
            cls._instance._error_listeners = []
        return cls._instance
    
    def add_error(self, error_message, error_type="Error", module_name=None, details=None):
        """Add an error to the error list and trigger callbacks"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        error_id = generate_id('err_')
        
        error_obj = {
            "id": error_id,
            "timestamp": timestamp,
            "message": error_message,
            "type": error_type,
            "module": module_name,
            "details": details
        }
        
        self._errors.append(error_obj)
        # Limit stored errors to most recent 100
        if len(self._errors) > 100:
            self._errors.pop(0)
        
        # Notify all callbacks
        for callback in self._error_callbacks:
            try:
                callback(error_obj)
            except Exception as e:
                logger.error(f"Error in error callback: {str(e)}")
        
        # Notify all listeners
        for listener in self._error_listeners:
            try:
                listener(error_obj)
            except Exception as e:
                logger.error(f"Error in error listener: {str(e)}")
        
        # Log the error
        logger.error(f"{error_type} in {module_name or 'unknown'}: {error_message}")
        
        return error_obj
    
    def get_errors(self, limit=10):
        """Get recent errors"""
        return self._errors[-limit:] if self._errors else []
    
    def clear_errors(self):
        """Clear all errors"""
        self._errors = []
    
    def register_callback(self, callback):
        """Register a callback function that will be called when a new error occurs"""
        if callback not in self._error_callbacks:
            self._error_callbacks.append(callback)
    
    def unregister_callback(self, callback):
        """Unregister a callback function"""
        if callback in self._error_callbacks:
            self._error_callbacks.remove(callback)
    
    def add_listener(self, listener):
        """Add a listener function that will be updated with errors"""
        if listener not in self._error_listeners:
            self._error_listeners.append(listener)
            # Immediately call with most recent error if available
            if self._errors:
                listener(self._errors[-1])
    
    def remove_listener(self, listener):
        """Remove a listener function"""
        if listener in self._error_listeners:
            self._error_listeners.remove(listener)

# Utility functions

def generate_id(prefix='id_'):
    """Generate a unique ID"""
    timestamp = int(time.time() * 1000)
    random_part = random.randint(1000, 9999)
    return f"{prefix}{timestamp}{random_part}"

def handle_errors(func):
    """Decorator for handling exceptions in functions"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # Get module name from function
            module_name = func.__module__
            # Get the error details
            error_msg = str(e)
            error_type = type(e).__name__
            # Add traceback details
            details = traceback.format_exc()
            
            # Report the error
            error_manager = ErrorManager()
            error_manager.add_error(
                error_message=error_msg,
                error_type=error_type,
                module_name=module_name,
                details=details
            )
            
            # Re-raise the exception if in debug mode
            if SystemState().debug_mode:
                raise
            
            # Return error message to display in UI
            return f"❌ Error: {error_msg}"
    return wrapper

def initialize_directories():
    """Initialize all required directories for the application"""
    directories = [
        DATA_DIR,
        CONFIG_DIR,
        LOGS_DIR,
        STATIC_DIR,
        RECORDINGS_DIR,
        TONES_DIR,
        os.path.join(STATIC_DIR, "samples"),
        os.path.join(STATIC_DIR, "models")
    ]
    
    # Create all directories
    for directory in directories:
        try:
            Path(directory).mkdir(parents=True, exist_ok=True)
            logger.info(f"Created directory: {directory}")
        except Exception as e:
            logger.error(f"Error creating directory {directory}: {str(e)}")
        
    # Create default configuration files if they don't exist
    system_state_file = os.path.join(CONFIG_DIR, "system_state.json")
    if not os.path.exists(system_state_file):
        default_state = {
            "online": True,
            "demo_mode": True,
            "debug_mode": False,
            "last_backup": None
        }
        try:
            with open(system_state_file, 'w') as f:
                json.dump(default_state, f, indent=2)
            logger.info(f"Created default system state file: {system_state_file}")
        except Exception as e:
            logger.error(f"Error creating system state file: {str(e)}")
    
    voice_settings_file = os.path.join(CONFIG_DIR, "voice_settings.json")
    if not os.path.exists(voice_settings_file):
        default_settings = {
            "default_model": "system_female",
            "speech_rate": 1.0,
            "volume": 0.8,
            "models": [
                {
                    "id": "system_female",
                    "name": "System Female Voice",
                    "type": "system",
                    "config": {
                        "gender": "female",
                        "accent": "american"
                    }
                },
                {
                    "id": "system_male",
                    "name": "System Male Voice",
                    "type": "system",
                    "config": {
                        "gender": "male",
                        "accent": "american"
                    }
                }
            ]
        }
        try:
            with open(voice_settings_file, 'w') as f:
                json.dump(default_settings, f, indent=2)
            logger.info(f"Created default voice settings file: {voice_settings_file}")
        except Exception as e:
            logger.error(f"Error creating voice settings file: {str(e)}")