from typing import Tuple from httpx import AsyncClient from loguru import logger from app.services.platform import DataPlatformService, InfoCode from app.services.transfer import Duoduo class ACATFUSupplyAirTemperatureController: """ Supply air temperature setting logic version 1 by Wenyan. """ def __init__(self, current_supply_air_temp: float, hot_rate: float, cold_rate: float): self.current_air_temp = current_supply_air_temp self.hot_rate = hot_rate self.cold_rate = cold_rate def get_next_set(self, is_just_booted: bool) -> float: try: cold_hot_ratio = self.cold_rate / self.hot_rate except ZeroDivisionError: cold_hot_ratio = 99 if is_just_booted: if cold_hot_ratio < 0.5: next_temperature_set = 8.0 elif cold_hot_ratio < 0.9: next_temperature_set = 11.0 elif cold_hot_ratio < 1.1: next_temperature_set = 14.0 elif cold_hot_ratio < 1.5: next_temperature_set = 17.0 elif cold_hot_ratio >= 1.5: next_temperature_set = 20.0 else: # If cold hot ratio is nan. next_temperature_set = 20.0 else: if cold_hot_ratio < 0.5: diff = -3 elif cold_hot_ratio < 0.9: diff = -2 elif cold_hot_ratio < 1.1: diff = 0 elif cold_hot_ratio < 1.5: diff = 2 elif cold_hot_ratio >= 1.5: diff = 3 else: # If cold hot ratio is nan. diff = 0 next_temperature_set = self.current_air_temp + diff next_temperature_set = max(8.0, min(20.0, next_temperature_set)) return next_temperature_set async def fetch_params(project_id: str, device_id: str) -> Tuple[float, float, float, bool]: async with AsyncClient() as client: platform = DataPlatformService(client, project_id) duoduo = Duoduo(client, project_id) current_supply_air_temperature = await platform.get_realtime_supply_air_temperature(device_id) running_status_duration = await platform.get_duration(InfoCode.running_status, device_id, 15 * 60) hot_rate, cold_rate = await duoduo.query_fill_rate_by_device(device_id) is_just_booted = False if running_status_duration[-1]['value'] == 1.0: for item in running_status_duration[::-1]: if item['value'] == 0.0: is_just_booted = True break return current_supply_air_temperature, hot_rate, cold_rate, is_just_booted @logger.catch() async def get_next_acatfu_supply_air_temperature_set(project_id: str, device_id: str) -> float: current_supply_air_temperature, hot_rate, cold_rate, is_just_booted = await fetch_params(project_id, device_id) controller = ACATFUSupplyAirTemperatureController( current_supply_air_temperature, hot_rate, cold_rate ) next_temperature_set = controller.get_next_set(is_just_booted) logger.debug(next_temperature_set) return next_temperature_set