early_start.py 1.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859
  1. from joblib import load
  2. from loguru import logger
  3. from sqlalchemy.orm import Session
  4. from app.core.config import settings
  5. from app.crud.model_path.early_start import model_path_early_start_dtr
  6. from app.models.domain.devices import ACATFCEarlyStartPredictionRequest
  7. from app.models.ml_models_path.early_start import EarlyStartDTRModelPath
  8. from app.schemas.season import Season
  9. class EarlyStartTimeDTRBuilder:
  10. """
  11. Build early start time by decision tree regression.
  12. """
  13. def __init__(self, model_path: EarlyStartDTRModelPath, season: Season):
  14. self.summer_model_path = (
  15. f"{settings.ML_MODELS_DIR}{model_path.summer_model_path}"
  16. )
  17. self.winter_model_path = (
  18. f"{settings.ML_MODELS_DIR}{model_path.winter_model_path}"
  19. )
  20. self.season = season
  21. async def get_prediction(self, indoor_temp: float, outdoor_temp: float) -> float:
  22. try:
  23. if self.season == Season.cooling:
  24. model = load(self.summer_model_path)
  25. elif self.season == Season.heating:
  26. model = load(self.winter_model_path)
  27. else:
  28. return 0
  29. except (FileNotFoundError, IsADirectoryError) as e:
  30. logger.debug(e)
  31. return 0
  32. try:
  33. pre = model.predict([[indoor_temp, outdoor_temp]])
  34. pre_time = pre[0]
  35. except (ValueError, IndexError) as e:
  36. logger.debug(e)
  37. pre_time = 0
  38. return pre_time
  39. @logger.catch()
  40. async def build_acatfc_early_start_prediction(
  41. params: ACATFCEarlyStartPredictionRequest, db: Session
  42. ) -> float:
  43. model_path = model_path_early_start_dtr.get_path_by_device(db, params.device_id)
  44. builder = EarlyStartTimeDTRBuilder(model_path, params.season)
  45. hour = await builder.get_prediction(
  46. params.space_realtime_temperature, params.outdoor_realtime_temperature
  47. )
  48. return hour * 60