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.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): return 0 try: pre = model.predict([[indoor_temp, outdoor_temp]]) pre_time = pre[0] except (ValueError, IndexError): 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) 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