freq_set.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181
  1. # -*- coding: utf-8 -*-
  2. from typing import List, Tuple
  3. import numpy as np
  4. from loguru import logger
  5. from app.models.domain.devices import ACATFUFreqSetRequest
  6. from app.schemas.season import Season
  7. from app.schemas.space import SpaceATFU
  8. from app.utils.helpers import is_off_to_on
  9. class ACATFUFanFreqController:
  10. """
  11. Logic writen by Wuxu.
  12. """
  13. def __init__(
  14. self,
  15. freq: float,
  16. fresh_air_temperature: float,
  17. season: Season,
  18. ) -> None:
  19. self._freq = freq
  20. self._fresh_air_temperature = fresh_air_temperature
  21. self._season = season
  22. def get_next_set(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  23. if self._season == Season.transition:
  24. next_freq_set = self.get_transition_logic(spaces_params, on_flag)
  25. else:
  26. next_freq_set = self.get_cooling_logic(spaces_params, on_flag)
  27. next_freq_set = max(20.0, next_freq_set)
  28. next_freq_set = min(50.0, next_freq_set)
  29. return next_freq_set
  30. def get_transition_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  31. temp_avg, co2_avg = self.get_avg(spaces_params)
  32. if on_flag:
  33. if self._fresh_air_temperature <= 16.0:
  34. freq_set = 20.0
  35. elif self._fresh_air_temperature <= 18.0:
  36. if temp_avg <= 23.0:
  37. freq_set = 20.0
  38. else:
  39. freq_set = 25.0
  40. elif self._fresh_air_temperature <= 21.0:
  41. if temp_avg <= 23.0:
  42. freq_set = 25.0
  43. elif temp_avg <= 25.5:
  44. freq_set = 30.0
  45. else:
  46. freq_set = 35.0
  47. elif self._fresh_air_temperature <= 24.0:
  48. if temp_avg <= 23.0:
  49. freq_set = 30.0
  50. elif temp_avg <= 25.5:
  51. freq_set = 35.0
  52. else:
  53. freq_set = 40.0
  54. else:
  55. if temp_avg <= 23.0:
  56. freq_set = 35.0
  57. elif temp_avg <= 25.5:
  58. freq_set = 30.0
  59. else:
  60. freq_set = 25.0
  61. if co2_avg > 750.0:
  62. freq_set = max(freq_set, 25.0)
  63. else:
  64. if self._fresh_air_temperature <= 18.0:
  65. if temp_avg <= 23.0:
  66. freq_set = 20.0
  67. elif temp_avg <= 25.5:
  68. freq_set = self._freq - 2.0
  69. else:
  70. freq_set = self._freq + 3.0
  71. elif self._fresh_air_temperature <= 23.0:
  72. if temp_avg <= 23.0:
  73. freq_set = self._freq - 2.0
  74. elif temp_avg <= 25.5:
  75. freq_set = self._freq
  76. else:
  77. freq_set = self._freq + 2.0
  78. elif self._fresh_air_temperature <= 25.0:
  79. if temp_avg <= 23.0:
  80. freq_set = self._freq + 2.0
  81. elif temp_avg <= 25.5:
  82. freq_set = self._freq
  83. else:
  84. freq_set = self._freq + 1.0
  85. else:
  86. if temp_avg <= 23.0:
  87. freq_set = self._freq + 5.0
  88. elif temp_avg <= 25.5:
  89. freq_set = self._freq - 2.0
  90. else:
  91. freq_set = 20.0
  92. return freq_set
  93. def get_cooling_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  94. _, co2_avg = self.get_avg(spaces_params)
  95. if on_flag:
  96. freq_set = 30.0
  97. else:
  98. high_co2_ratio = self.get_high_co2_ratio(spaces_params)
  99. if high_co2_ratio == 0.0:
  100. if co2_avg < 600.0:
  101. freq_set = self._freq - 2.0
  102. else:
  103. freq_set = self._freq
  104. freq_set = max(25.0, freq_set)
  105. elif high_co2_ratio < 0.05:
  106. freq_set = self._freq + 2.0
  107. freq_set = min(35.0, freq_set)
  108. elif high_co2_ratio < 0.1:
  109. freq_set = self._freq + 3.0
  110. freq_set = min(40.0, freq_set)
  111. elif high_co2_ratio < 0.2:
  112. freq_set = self._freq + 5.0
  113. else:
  114. freq_set = 50.0
  115. return freq_set
  116. def get_heating_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  117. # The same with cooling logic.
  118. return self._freq
  119. @staticmethod
  120. def get_avg(spaces_params: List[SpaceATFU]) -> Tuple[float, float]:
  121. valid_temperature_list, valid_co2_list = list(), list()
  122. for space in spaces_params:
  123. if 0.0 < space.realtime_temperature < 40.0:
  124. valid_temperature_list.append(space.realtime_temperature)
  125. if 0.0 < space.realtime_co2 < 5000.0:
  126. valid_co2_list.append(space.realtime_co2)
  127. if valid_temperature_list:
  128. temp_avg = np.mean(valid_temperature_list)
  129. else:
  130. temp_avg = np.NAN
  131. if valid_co2_list:
  132. co2_avg = np.mean(valid_co2_list)
  133. else:
  134. co2_avg = np.NAN
  135. return temp_avg, co2_avg
  136. @staticmethod
  137. def get_high_co2_ratio(spaces_params: List[SpaceATFU]) -> float:
  138. valid_co2_count, high_co2_count = 0, 0
  139. for space in spaces_params:
  140. if 0.0 < space.realtime_co2 < 5000.0:
  141. valid_co2_count += 1
  142. if space.realtime_co2 > 900.0:
  143. high_co2_count += 1
  144. if valid_co2_count == 0:
  145. ratio = 0.0
  146. else:
  147. ratio = high_co2_count / valid_co2_count
  148. return ratio
  149. @logger.catch()
  150. async def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
  151. controller = ACATFUFanFreqController(params.freq, params.fresh_air_temperature, params.season)
  152. on_flag = is_off_to_on(params.running_status_list)
  153. spaces = [SpaceATFU(**sp.dict()) for sp in params.spaces]
  154. freq_set = controller.get_next_set(spaces, on_flag)
  155. return freq_set