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: model = load(self.model_path) pre_time = model.predict([[outdoor_temp, indoor_temp]]) return pre_time[0] 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 model_path = model_path_early_start_dtr.get_path_by_device(db, device_id) return indoor_temp, outdoor_temp, model_path.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) return hour * 60