File size: 4,204 Bytes
397f012 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """
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())
|