import importlib
from motor.config.endpoint import EndpointConfig
from motor.engine_server.core.config import IConfig
from motor.common.logger import get_logger
logger = get_logger(__name__)
class ConfigFactory:
_ENGINE_CONFIG_MAP: dict[str, str] = {
"vllm": "motor.engine_server.core.vllm.vllm_config.VLLMConfig",
"sglang": "motor.engine_server.core.sglang.sglang_config.SGLangConfig",
}
def __init__(self, endpoint_config: EndpointConfig):
self.endpoint_config = endpoint_config
def parse(self) -> IConfig:
engine_type = self.endpoint_config.engine_type
config_class_path = self._ENGINE_CONFIG_MAP.get(engine_type)
if not config_class_path:
supported_types = list(self._ENGINE_CONFIG_MAP.keys())
raise ValueError(
f"Unsupported engine type: {engine_type}. "
f"Supported types are: {supported_types}."
)
try:
module_path, class_name = config_class_path.rsplit('.', 1)
module = importlib.import_module(module_path)
config_class = getattr(module, class_name)
config_instance = config_class(endpoint_config=self.endpoint_config)
config_instance.initialize()
config_instance.convert()
config_instance.validate()
return config_instance
except (ImportError, AttributeError) as e:
raise ValueError(f"Failed to load config class for {engine_type}") from e