freq_set.py 7.0 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. elif self._season == Season.cooling:
  54. next_freq_set = self.get_cooling_logic(spaces_params, on_flag)
  55. else:
  56. next_freq_set = self.get_heating_logic(spaces_params, on_flag)
  57. next_freq_set = max(20.0, next_freq_set)
  58. next_freq_set = min(50.0, next_freq_set)
  59. return next_freq_set
  60. def get_transition_logic(
  61. self, spaces_params: List[SpaceATFU], on_flag: bool
  62. ) -> float:
  63. temp_avg, co2_avg = self.get_avg(spaces_params)
  64. if on_flag:
  65. if self._fresh_air_temperature <= 16.0:
  66. freq_set = 20.0
  67. elif self._fresh_air_temperature <= 18.0:
  68. if temp_avg <= 23.0:
  69. freq_set = 20.0
  70. else:
  71. freq_set = 25.0
  72. elif self._fresh_air_temperature <= 21.0:
  73. if temp_avg <= 23.0:
  74. freq_set = 25.0
  75. elif temp_avg <= 25.5:
  76. freq_set = 30.0
  77. else:
  78. freq_set = 35.0
  79. elif self._fresh_air_temperature <= 24.0:
  80. if temp_avg <= 23.0:
  81. freq_set = 30.0
  82. elif temp_avg <= 25.5:
  83. freq_set = 35.0
  84. else:
  85. freq_set = 40.0
  86. else:
  87. if temp_avg <= 23.0:
  88. freq_set = 35.0
  89. elif temp_avg <= 25.5:
  90. freq_set = 30.0
  91. else:
  92. freq_set = 25.0
  93. if co2_avg > 750.0:
  94. freq_set = max(freq_set, 25.0)
  95. else:
  96. if self._fresh_air_temperature <= 18.0:
  97. if temp_avg <= 23.0:
  98. freq_set = 20.0
  99. elif temp_avg <= 25.5:
  100. freq_set = self._freq - 2.0
  101. else:
  102. freq_set = self._freq + 3.0
  103. elif self._fresh_air_temperature <= 23.0:
  104. if temp_avg <= 23.0:
  105. freq_set = self._freq - 2.0
  106. elif temp_avg <= 25.5:
  107. freq_set = self._freq
  108. else:
  109. freq_set = self._freq + 2.0
  110. elif self._fresh_air_temperature <= 25.0:
  111. if temp_avg <= 23.0:
  112. freq_set = self._freq + 2.0
  113. elif temp_avg <= 25.5:
  114. freq_set = self._freq
  115. else:
  116. freq_set = self._freq + 1.0
  117. else:
  118. if temp_avg <= 23.0:
  119. freq_set = self._freq + 5.0
  120. elif temp_avg <= 25.5:
  121. freq_set = self._freq - 2.0
  122. else:
  123. freq_set = 20.0
  124. return freq_set
  125. def get_cooling_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  126. _, co2_avg = self.get_avg(spaces_params)
  127. if on_flag:
  128. freq_set = 30.0
  129. else:
  130. high_co2_ratio = self.get_high_co2_ratio(spaces_params)
  131. if high_co2_ratio == 0.0:
  132. if co2_avg < 600.0:
  133. freq_set = self._freq - 2.0
  134. else:
  135. freq_set = self._freq
  136. freq_set = max(25.0, freq_set)
  137. elif high_co2_ratio < 0.05:
  138. freq_set = self._freq + 2.0
  139. freq_set = min(35.0, freq_set)
  140. elif high_co2_ratio < 0.1:
  141. freq_set = self._freq + 3.0
  142. freq_set = min(40.0, freq_set)
  143. elif high_co2_ratio < 0.2:
  144. freq_set = self._freq + 5.0
  145. else:
  146. freq_set = 50.0
  147. return freq_set
  148. def get_heating_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  149. return self._freq
  150. @staticmethod
  151. def get_avg(spaces_params: List[SpaceATFU]) -> Tuple[float, float]:
  152. valid_temperature_list, valid_co2_list = list(), list()
  153. for space in spaces_params:
  154. if 0.0 < space.realtime_temperature < 40.0:
  155. valid_temperature_list.append(space.realtime_temperature)
  156. if 0.0 < space.realtime_co2 < 5000.0:
  157. valid_co2_list.append(space.realtime_co2)
  158. if valid_temperature_list:
  159. temp_avg = np.mean(valid_temperature_list)
  160. else:
  161. temp_avg = np.NAN
  162. if valid_co2_list:
  163. co2_avg = np.mean(valid_co2_list)
  164. else:
  165. co2_avg = np.NAN
  166. return temp_avg, co2_avg
  167. @staticmethod
  168. def get_high_co2_ratio(spaces_params: List[SpaceATFU]) -> float:
  169. valid_co2_count, high_co2_count = 0, 0
  170. for space in spaces_params:
  171. if 0.0 < space.realtime_co2 < 5000.0:
  172. valid_co2_count += 1
  173. if space.realtime_co2 > 900.0:
  174. high_co2_count += 1
  175. if valid_co2_count == 0:
  176. ratio = 0.0
  177. else:
  178. ratio = high_co2_count / valid_co2_count
  179. return ratio
  180. @logger.catch()
  181. async def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
  182. controller = ACATFUFanFreqControllerV2(
  183. params.freq,
  184. params.fresh_air_temperature,
  185. params.season,
  186. )
  187. on_flag = is_off_to_on(params.running_status_list)
  188. freq_set = controller.get_next_set(params.spaces, on_flag)
  189. return freq_set