123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657 |
- 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
|