utils/dependency_helpers.py

"""
Dependency helpers to make the model work even if some libraries are missing.
This file provides fallback implementations for missing dependencies.
"""
import os
import logging
import importlib.util
from typing import Any, Dict, Optional, Type, Callable

logger = logging.getLogger(__name__)

# Dictionary to track mock implementations
MOCK_MODULES = {}

def is_module_available(module_name: str) -> bool:
    """Check if a module is available without importing it"""
    return importlib.util.find_spec(module_name) is not None

def create_mock_emissions_tracker() -> Type:
    """Create a mock implementation of codecarbon's EmissionsTracker"""
    class MockEmissionsTracker:
        def __init__(self, *args, **kwargs):
            logger.info("Using mock EmissionsTracker")
            
        def __enter__(self):
            return self
            
        def __exit__(self, exc_type, exc_val, exc_tb):
            pass
            
        def start(self):
            return self
            
        def stop(self):
            return 0.0  # Return zero emissions
    
    return MockEmissionsTracker

def create_mock_pydantic_classes() -> Dict[str, Type]:
    """Create mock implementations of pydantic classes"""
    class MockBaseModel:
        """Mock implementation of pydantic's BaseModel"""
        def __init__(self, **kwargs):
            for key, value in kwargs.items():
                setattr(self, key, value)
                
        def dict(self) -> Dict[str, Any]:
            return {k: v for k, v in self.__dict__.items() 
                    if not k.startswith('_')}
                    
        def json(self) -> str:
            import json
            return json.dumps(self.dict())
            
    def mock_field(*args, **kwargs) -> Any:
        """Mock implementation of pydantic's Field"""
        return kwargs.get('default', None)
    
    class MockValidationError(Exception):
        """Mock implementation of pydantic's ValidationError"""
        pass
    
    mock_config_dict = dict
    
    return {
        "BaseModel": MockBaseModel,
        "Field": mock_field, 
        "ValidationError": MockValidationError,
        "ConfigDict": mock_config_dict
    }

def setup_dependency_fallbacks():
    """Setup fallbacks for all required dependencies"""
    # Handle codecarbon
    if not is_module_available("codecarbon"):
        logger.warning("codecarbon not found, using mock implementation")
        MOCK_MODULES["codecarbon"] = type("MockCodecarbon", (), {
            "EmissionsTracker": create_mock_emissions_tracker()
        })
    
    # Handle pydantic
    if not is_module_available("pydantic"):
        logger.warning("pydantic not found, using mock implementation")
        mock_classes = create_mock_pydantic_classes()
        MOCK_MODULES["pydantic"] = type("MockPydantic", (), mock_classes)
    
    # Setup service_registry fallback if needed
    if not is_module_available("service_registry"):
        from types import SimpleNamespace
        registry_obj = SimpleNamespace()
        registry_obj.register = lambda *args, **kwargs: None
        registry_obj.get = lambda *args: None
        registry_obj.has = lambda *args: False
        
        MOCK_MODULES["service_registry"] = type("MockServiceRegistry", (), {
            "registry": registry_obj,
            "MODEL": "MODEL",
            "TOKENIZER": "TOKENIZER"
        })

# Custom import hook to provide mock implementations
class DependencyImportFinder:
    def __init__(self):
        self._mock_modules = MOCK_MODULES
    
    def find_module(self, fullname, path=None):
        if fullname in self._mock_modules:
            return self
    
    def load_module(self, fullname):
        import sys
        if fullname in sys.modules:
            return sys.modules[fullname]
        
        module = self._mock_modules[fullname]
        sys.modules[fullname] = module
        return module

# Initialize the fallbacks
setup_dependency_fallbacks()

# Install the custom import hook
import sys
sys.meta_path.insert(0, DependencyImportFinder())