from joblib import load from loguru import logger from sqlalchemy.orm import Session from app.api.errors.iot import MissingIOTDataError from app.core.config import settings from app.crud.model_path.early_start import model_path_early_start_dtr from app.models.domain.devices import ACATFCEarlyStartPredictionRequest from app.models.ml_models_path.early_start import EarlyStartDTRModelPath from app.schemas.season import Season class EarlyStartTimeDTRBuilder: """ Build early start time by decision tree regression. """ def __init__(self, model_path: EarlyStartDTRModelPath, season: Season): self.summer_model_path = ( f"{settings.ML_MODELS_DIR}{model_path.summer_model_path}" ) self.winter_model_path = ( f"{settings.ML_MODELS_DIR}{model_path.winter_model_path}" ) self.season = season async def get_prediction(self, indoor_temp: float, outdoor_temp: float) -> float: try: if self.season == Season.cooling: model = load(self.summer_model_path) elif self.season == Season.heating: model = load(self.winter_model_path) else: return 0 except (FileNotFoundError, IsADirectoryError) as e: logger.debug(e) return 0 try: pre = model.predict([[indoor_temp, outdoor_temp]]) pre_time = pre[0] except (ValueError, IndexError) as e: logger.debug(e) pre_time = 0 except TypeError: raise MissingIOTDataError return pre_time async def build_acatfc_early_start_prediction(params: ACATFCEarlyStartPredictionRequest, db: Session) -> float: model_path = model_path_early_start_dtr.get_path_by_device(db, params.device_id) builder = EarlyStartTimeDTRBuilder(model_path, params.season) hour = await builder.get_prediction(params.space_realtime_temperature, params.outdoor_realtime_temperature) return hour * 60