# 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)}")