early_start.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172
  1. from typing import Tuple
  2. from httpx import AsyncClient
  3. from joblib import load
  4. from loguru import logger
  5. from sqlalchemy.orm import Session
  6. from app.core.config import settings
  7. from app.crud.model_path.early_start import model_path_early_start_dtr
  8. from app.services.platform import DataPlatformService
  9. from app.services.transfer import SpaceInfoService
  10. from app.services.weather import WeatherService
  11. class EarlyStartTimeDTRBuilder:
  12. """
  13. Build early start time by decision tree regression.
  14. """
  15. def __init__(self, model_path: str):
  16. self.model_path = f'{settings.ML_MODELS_DIR}{model_path}'
  17. async def get_prediction(self, indoor_temp: float, outdoor_temp: float) -> float:
  18. try:
  19. model = load(self.model_path)
  20. except (FileNotFoundError, IsADirectoryError):
  21. return 0
  22. try:
  23. pre = model.predict([[outdoor_temp, indoor_temp]])
  24. pre_time = pre[0]
  25. except (ValueError, IndexError):
  26. pre_time = 0
  27. return pre_time
  28. async def fetch_params(project_id: str, space_id: str, db: Session) -> Tuple[float, float, str]:
  29. async with AsyncClient() as client:
  30. platform = DataPlatformService(client, project_id)
  31. space_service = SpaceInfoService(client, project_id, space_id)
  32. weather_service = WeatherService(client)
  33. indoor_temp = await platform.get_realtime_temperature(space_id)
  34. weather_info = await weather_service.get_realtime_weather(project_id)
  35. outdoor_temp = weather_info.get('temperature')
  36. device_list = await space_service.get_equipment()
  37. device_id = ''
  38. for device in device_list:
  39. if device.get('category') == 'ACATFC':
  40. device_id = device.get('id')
  41. break
  42. if device_id:
  43. model_path = model_path_early_start_dtr.get_path_by_device(db, device_id)
  44. else:
  45. model_path = ''
  46. return indoor_temp, outdoor_temp, model_path
  47. @logger.catch()
  48. async def get_recommended_early_start_time(db: Session, project_id: str, space_id: str) -> float:
  49. indoor_temp, outdoor_temp, model_path = await fetch_params(project_id, space_id, db)
  50. builder = EarlyStartTimeDTRBuilder(model_path)
  51. hour = await builder.get_prediction(indoor_temp, outdoor_temp)
  52. logger.debug(f'{space_id}: indoor-{indoor_temp}, outdoor-{outdoor_temp}, prediction-{hour * 60}')
  53. return hour * 60