123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596 |
- from typing import Tuple
- from httpx import AsyncClient
- from joblib import load
- from loguru import logger
- from sqlalchemy.orm import Session
- 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.services.platform import DataPlatformService
- from app.services.transfer import SpaceInfoService
- from app.services.weather import WeatherService
- class EarlyStartTimeDTRBuilder:
- """
- Build early start time by decision tree regression.
- """
- def __init__(self, model_path: str):
- self.model_path = f"{settings.ML_MODELS_DIR}{model_path}"
- async def get_prediction(self, indoor_temp: float, outdoor_temp: float) -> float:
- try:
- model = load(self.model_path)
- 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
- return pre_time
- async def fetch_params(
- project_id: str, space_id: str, db: Session
- ) -> Tuple[float, float, str]:
- async with AsyncClient() as client:
- platform = DataPlatformService(client, project_id)
- space_service = SpaceInfoService(client, project_id, space_id)
- weather_service = WeatherService(client)
- indoor_temp = await platform.get_realtime_temperature(space_id)
- weather_info = await weather_service.get_realtime_weather(project_id)
- outdoor_temp = weather_info.get("temperature")
- device_list = await space_service.get_equipment()
- device_id = ""
- for device in device_list:
- if device.get("category") == "ACATFC":
- device_id = device.get("id")
- break
- if device_id:
- model_path = model_path_early_start_dtr.get_path_by_device(db, device_id)
- model_path = model_path.model_path
- else:
- model_path = ""
- return indoor_temp, outdoor_temp, model_path
- @logger.catch()
- async def get_recommended_early_start_time(
- db: Session, project_id: str, space_id: str
- ) -> float:
- indoor_temp, outdoor_temp, model_path = await fetch_params(project_id, space_id, db)
- builder = EarlyStartTimeDTRBuilder(model_path)
- hour = await builder.get_prediction(indoor_temp, outdoor_temp)
- logger.debug(
- f"{space_id}: indoor-{indoor_temp}, outdoor-{outdoor_temp}, prediction-{hour * 60}"
- )
- return hour * 60
- @logger.catch()
- 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.model_path)
- hour = await builder.get_prediction(
- params.space_realtime_temperature, params.outdoor_realtime_temperature
- )
- return hour * 60
|