""" Factory para crear modelos de forecasting. Este módulo implementa el patrón Factory aplicando OCP (Open/Closed Principle) - abierto para extensión, cerrado para modificación. """ from typing import Dict, Type, List from app.domain.interfaces.forecast_model import IForecastModel from app.infrastructure.ml.chronos_model import ChronosModel from app.utils.logger import setup_logger logger = setup_logger(__name__) class ModelFactory: """ Factory para crear modelos de forecasting. Permite agregar nuevos modelos sin modificar código existente, aplicando el principio OCP (Open/Closed Principle). Ejemplo de uso: >>> model = ModelFactory.create("chronos2", model_id="amazon/chronos-2") >>> # Futuro: model = ModelFactory.create("prophet", ...) """ # Registro de modelos disponibles _models: Dict[str, Type[IForecastModel]] = { "chronos2": ChronosModel, # Futuro: Agregar sin modificar código existente # "prophet": ProphetModel, # "arima": ARIMAModel, # "custom": CustomModel, } @classmethod def create( cls, model_type: str, **kwargs ) -> IForecastModel: """ Crea una instancia de modelo de forecasting. Args: model_type: Tipo de modelo ("chronos2", "prophet", etc.) **kwargs: Parámetros específicos del modelo Returns: Instancia de IForecastModel Raises: ValueError: Si el tipo de modelo no existe Example: >>> model = ModelFactory.create( ... "chronos2", ... model_id="amazon/chronos-2", ... device_map="cpu" ... ) """ if model_type not in cls._models: available = ", ".join(cls._models.keys()) raise ValueError( f"Unknown model type: '{model_type}'. " f"Available: {available}" ) model_class = cls._models[model_type] logger.info(f"Creating model: {model_type}") try: instance = model_class(**kwargs) logger.info(f"Model created: {instance}") return instance except Exception as e: logger.error(f"Failed to create model '{model_type}': {e}") raise @classmethod def register_model( cls, name: str, model_class: Type[IForecastModel] ) -> None: """ Registra un nuevo tipo de modelo (OCP - extensión). Permite agregar nuevos modelos dinámicamente sin modificar el código de la factory. Args: name: Nombre del modelo model_class: Clase que implementa IForecastModel Raises: TypeError: Si model_class no implementa IForecastModel ValueError: Si el nombre ya está registrado Example: >>> class MyCustomModel(IForecastModel): ... pass >>> ModelFactory.register_model("custom", MyCustomModel) """ # Validar que implementa la interfaz if not issubclass(model_class, IForecastModel): raise TypeError( f"{model_class.__name__} debe implementar IForecastModel" ) # Validar que no esté duplicado if name in cls._models: raise ValueError( f"Model '{name}' ya está registrado. " f"Use un nombre diferente o llame a unregister_model primero." ) cls._models[name] = model_class logger.info(f"Registered new model: {name} -> {model_class.__name__}") @classmethod def unregister_model(cls, name: str) -> None: """ Elimina un modelo del registro. Args: name: Nombre del modelo a eliminar Raises: ValueError: Si el modelo no existe """ if name not in cls._models: raise ValueError(f"Model '{name}' no está registrado") del cls._models[name] logger.info(f"Unregistered model: {name}") @classmethod def list_available_models(cls) -> List[str]: """ Lista todos los modelos disponibles. Returns: Lista de nombres de modelos """ return list(cls._models.keys()) @classmethod def get_model_info(cls, model_type: str) -> Dict[str, str]: """ Obtiene información sobre un tipo de modelo. Args: model_type: Nombre del tipo de modelo Returns: Diccionario con información del modelo Raises: ValueError: Si el modelo no existe """ if model_type not in cls._models: raise ValueError(f"Model '{model_type}' no está registrado") model_class = cls._models[model_type] return { "name": model_type, "class": model_class.__name__, "module": model_class.__module__ }