freq_set.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. # -*- coding: utf-8 -*-
  2. from typing import Tuple
  3. from httpx import AsyncClient
  4. from loguru import logger
  5. from app.models.domain.devices import ACATFUFreqSetRequest
  6. from app.services.platform import DataPlatformService, InfoCode
  7. from app.services.transfer import Duoduo
  8. class ACATFUFanFreqController:
  9. """
  10. Fan freq setting logic for zhonghai.
  11. """
  12. def __init__(self, current_freq: float, hot_rate: float, cold_rate: float):
  13. self.current_freq = current_freq
  14. self.hot_rate = hot_rate
  15. self.cold_rate = cold_rate
  16. def get_next_set(self) -> float:
  17. try:
  18. cold_hot_ratio = self.cold_rate / self.hot_rate
  19. except ZeroDivisionError:
  20. cold_hot_ratio = 99
  21. if cold_hot_ratio < 0.5:
  22. diff = 3
  23. elif cold_hot_ratio < 0.9:
  24. diff = 2
  25. elif cold_hot_ratio < 1.1:
  26. diff = 0
  27. elif cold_hot_ratio < 1.5:
  28. diff = -2
  29. elif cold_hot_ratio >= 1.5:
  30. diff = -3
  31. else: # If cold hot rate is NAN.
  32. diff = 0
  33. next_freq_set = self.current_freq + diff
  34. next_freq_set = max(30.0, min(45.0, next_freq_set))
  35. return next_freq_set
  36. async def fetch_params(project_id: str, device_id: str) -> Tuple[float, float, float]:
  37. async with AsyncClient() as client:
  38. platform = DataPlatformService(client, project_id)
  39. duoduo = Duoduo(client, project_id)
  40. current_freq = await platform.get_realtime_data(InfoCode.fan_freq, device_id)
  41. hot_rate, cold_rate = await duoduo.query_fill_rate_by_device(device_id)
  42. return current_freq, hot_rate, cold_rate
  43. @logger.catch()
  44. async def get_next_acatfu_freq_set(project_id: str, device_id: str) -> float:
  45. current_freq, hot_rate, cold_rate = await fetch_params(project_id, device_id)
  46. controller = ACATFUFanFreqController(current_freq, hot_rate, cold_rate)
  47. next_freq_set = controller.get_next_set()
  48. return next_freq_set
  49. @logger.catch()
  50. def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
  51. controller = ACATFUFanFreqController(
  52. params.freq,
  53. params.hot_ratio,
  54. params.cold_ratio
  55. )
  56. freq_set = controller.get_next_set()
  57. return freq_set