freq_set.py 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222
  1. # -*- coding: utf-8 -*-
  2. from typing import List, Tuple
  3. from loguru import logger
  4. import numpy as np
  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. Fan freq setting logic for zhonghai.
  12. """
  13. def __init__(self, current_freq: float, hot_rate: float, cold_rate: float):
  14. self.current_freq = current_freq
  15. self.hot_rate = hot_rate
  16. self.cold_rate = cold_rate
  17. def get_next_set(self) -> float:
  18. try:
  19. cold_hot_ratio = self.cold_rate / self.hot_rate
  20. except ZeroDivisionError:
  21. cold_hot_ratio = 99
  22. if cold_hot_ratio < 0.5:
  23. diff = 3
  24. elif cold_hot_ratio < 0.9:
  25. diff = 2
  26. elif cold_hot_ratio < 1.1:
  27. diff = 0
  28. elif cold_hot_ratio < 1.5:
  29. diff = -2
  30. elif cold_hot_ratio >= 1.5:
  31. diff = -3
  32. else: # If cold hot rate is NAN.
  33. diff = 0
  34. next_freq_set = self.current_freq + diff
  35. next_freq_set = max(30.0, min(45.0, next_freq_set))
  36. return next_freq_set
  37. class ACATFUFanFreqControllerV2:
  38. """
  39. Logic writen by Wuxu.
  40. """
  41. def __init__(
  42. self,
  43. freq: float,
  44. fresh_air_temperature: float,
  45. season: Season,
  46. ) -> None:
  47. self._freq = freq
  48. self._fresh_air_temperature = fresh_air_temperature
  49. self._season = season
  50. def get_next_set(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  51. if self._season == Season.transition:
  52. next_freq_set = self.get_transition_logic(spaces_params, on_flag)
  53. else:
  54. next_freq_set = self.get_cooling_logic(spaces_params, on_flag)
  55. next_freq_set = max(20.0, next_freq_set)
  56. next_freq_set = min(50.0, next_freq_set)
  57. return next_freq_set
  58. def get_transition_logic(
  59. self, spaces_params: List[SpaceATFU], on_flag: bool
  60. ) -> float:
  61. temp_avg, co2_avg = self.get_avg(spaces_params)
  62. if on_flag:
  63. if self._fresh_air_temperature <= 16.0:
  64. freq_set = 20.0
  65. elif self._fresh_air_temperature <= 18.0:
  66. if temp_avg <= 23.0:
  67. freq_set = 20.0
  68. else:
  69. freq_set = 25.0
  70. elif self._fresh_air_temperature <= 21.0:
  71. if temp_avg <= 23.0:
  72. freq_set = 25.0
  73. elif temp_avg <= 25.5:
  74. freq_set = 30.0
  75. else:
  76. freq_set = 35.0
  77. elif self._fresh_air_temperature <= 24.0:
  78. if temp_avg <= 23.0:
  79. freq_set = 30.0
  80. elif temp_avg <= 25.5:
  81. freq_set = 35.0
  82. else:
  83. freq_set = 40.0
  84. else:
  85. if temp_avg <= 23.0:
  86. freq_set = 35.0
  87. elif temp_avg <= 25.5:
  88. freq_set = 30.0
  89. else:
  90. freq_set = 25.0
  91. if co2_avg > 750.0:
  92. freq_set = max(freq_set, 25.0)
  93. else:
  94. if self._fresh_air_temperature <= 18.0:
  95. if temp_avg <= 23.0:
  96. freq_set = 20.0
  97. elif temp_avg <= 25.5:
  98. freq_set = self._freq - 2.0
  99. else:
  100. freq_set = self._freq + 3.0
  101. elif self._fresh_air_temperature <= 23.0:
  102. if temp_avg <= 23.0:
  103. freq_set = self._freq - 2.0
  104. elif temp_avg <= 25.5:
  105. freq_set = self._freq
  106. else:
  107. freq_set = self._freq + 2.0
  108. elif self._fresh_air_temperature <= 25.0:
  109. if temp_avg <= 23.0:
  110. freq_set = self._freq + 2.0
  111. elif temp_avg <= 25.5:
  112. freq_set = self._freq
  113. else:
  114. freq_set = self._freq + 1.0
  115. else:
  116. if temp_avg <= 23.0:
  117. freq_set = self._freq + 5.0
  118. elif temp_avg <= 25.5:
  119. freq_set = self._freq - 2.0
  120. else:
  121. freq_set = 20.0
  122. return freq_set
  123. def get_cooling_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  124. _, co2_avg = self.get_avg(spaces_params)
  125. if on_flag:
  126. freq_set = 30.0
  127. else:
  128. high_co2_ratio = self.get_high_co2_ratio(spaces_params)
  129. if high_co2_ratio == 0.0:
  130. if co2_avg < 600.0:
  131. freq_set = self._freq - 2.0
  132. else:
  133. freq_set = self._freq
  134. freq_set = max(25.0, freq_set)
  135. elif high_co2_ratio < 0.05:
  136. freq_set = self._freq + 2.0
  137. freq_set = min(35.0, freq_set)
  138. elif high_co2_ratio < 0.1:
  139. freq_set = self._freq + 3.0
  140. freq_set = min(40.0, freq_set)
  141. elif high_co2_ratio < 0.2:
  142. freq_set = self._freq + 5.0
  143. else:
  144. freq_set = 50.0
  145. return freq_set
  146. def get_heating_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  147. # The same with cooling logic.
  148. return self._freq
  149. @staticmethod
  150. def get_avg(spaces_params: List[SpaceATFU]) -> Tuple[float, float]:
  151. valid_temperature_list, valid_co2_list = list(), list()
  152. for space in spaces_params:
  153. if 0.0 < space.realtime_temperature < 40.0:
  154. valid_temperature_list.append(space.realtime_temperature)
  155. if 0.0 < space.realtime_co2 < 5000.0:
  156. valid_co2_list.append(space.realtime_co2)
  157. if valid_temperature_list:
  158. temp_avg = np.mean(valid_temperature_list)
  159. else:
  160. temp_avg = np.NAN
  161. if valid_co2_list:
  162. co2_avg = np.mean(valid_co2_list)
  163. else:
  164. co2_avg = np.NAN
  165. return temp_avg, co2_avg
  166. @staticmethod
  167. def get_high_co2_ratio(spaces_params: List[SpaceATFU]) -> float:
  168. valid_co2_count, high_co2_count = 0, 0
  169. for space in spaces_params:
  170. if 0.0 < space.realtime_co2 < 5000.0:
  171. valid_co2_count += 1
  172. if space.realtime_co2 > 900.0:
  173. high_co2_count += 1
  174. if valid_co2_count == 0:
  175. ratio = 0.0
  176. else:
  177. ratio = high_co2_count / valid_co2_count
  178. return ratio
  179. @logger.catch()
  180. async def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
  181. controller = ACATFUFanFreqControllerV2(
  182. params.freq,
  183. params.fresh_air_temperature,
  184. params.season,
  185. )
  186. on_flag = is_off_to_on(params.running_status_list)
  187. spaces = [SpaceATFU(**sp.dict()) for sp in params.spaces]
  188. freq_set = controller.get_next_set(spaces, on_flag)
  189. return freq_set