# -*- coding: utf-8 -*- 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 ACATFUFanFreqController: """ Fan freq setting logic for zhonghai. """ def __init__(self, current_freq: float, hot_rate: float, cold_rate: float): self.current_freq = current_freq self.hot_rate = hot_rate self.cold_rate = cold_rate def get_next_set(self) -> float: try: cold_hot_ratio = self.cold_rate / self.hot_rate except ZeroDivisionError: cold_hot_ratio = 99 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 rate is NAN. diff = 0 next_freq_set = self.current_freq + diff next_freq_set = max(30.0, min(45.0, next_freq_set)) return next_freq_set async def fetch_params(project_id: str, device_id: str) -> Tuple[float, float, float]: async with AsyncClient() as client: platform = DataPlatformService(client, project_id) duoduo = Duoduo(client, project_id) current_freq = await platform.get_realtime_data(InfoCode.fan_freq, device_id) hot_rate, cold_rate = await duoduo.query_fill_rate_by_device(device_id) return current_freq, hot_rate, cold_rate @logger.catch() async def get_next_acatfu_freq_set(project_id: str, device_id: str) -> float: current_freq, hot_rate, cold_rate = await fetch_params(project_id, device_id) controller = ACATFUFanFreqController(current_freq, hot_rate, cold_rate) next_freq_set = controller.get_next_set() return next_freq_set