freq_set.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. # -*- coding: utf-8 -*-
  2. from typing import List, Tuple
  3. import numpy as np
  4. from app.api.errors.iot import MissingIOTDataError
  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. try:
  24. if self._season == Season.transition:
  25. next_freq_set = self.get_transition_logic(spaces_params, on_flag)
  26. else:
  27. next_freq_set = self.get_cooling_logic(spaces_params, on_flag)
  28. next_freq_set = self.hcho_logic(spaces_params, next_freq_set)
  29. next_freq_set = max(20.0, next_freq_set)
  30. next_freq_set = min(50.0, next_freq_set)
  31. except TypeError:
  32. raise MissingIOTDataError
  33. return next_freq_set
  34. def get_transition_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  35. temp_avg, co2_avg = self.get_avg(spaces_params)
  36. if on_flag:
  37. if self._fresh_air_temperature <= 16.0:
  38. freq_set = 20.0
  39. elif self._fresh_air_temperature <= 18.0:
  40. if temp_avg <= 23.0:
  41. freq_set = 20.0
  42. else:
  43. freq_set = 25.0
  44. elif self._fresh_air_temperature <= 21.0:
  45. if temp_avg <= 23.0:
  46. freq_set = 25.0
  47. elif temp_avg <= 25.5:
  48. freq_set = 30.0
  49. else:
  50. freq_set = 35.0
  51. elif self._fresh_air_temperature <= 24.0:
  52. if temp_avg <= 23.0:
  53. freq_set = 30.0
  54. elif temp_avg <= 25.5:
  55. freq_set = 35.0
  56. else:
  57. freq_set = 40.0
  58. else:
  59. if temp_avg <= 23.0:
  60. freq_set = 35.0
  61. elif temp_avg <= 25.5:
  62. freq_set = 30.0
  63. else:
  64. freq_set = 25.0
  65. if co2_avg > 750.0:
  66. freq_set = max(freq_set, 25.0)
  67. else:
  68. if self._fresh_air_temperature <= 18.0:
  69. if temp_avg <= 23.0:
  70. freq_set = 20.0
  71. elif temp_avg <= 25.5:
  72. freq_set = self._freq - 2.0
  73. else:
  74. freq_set = self._freq + 3.0
  75. elif self._fresh_air_temperature <= 23.0:
  76. if temp_avg <= 23.0:
  77. freq_set = self._freq - 2.0
  78. elif temp_avg <= 25.5:
  79. freq_set = self._freq
  80. else:
  81. freq_set = self._freq + 2.0
  82. elif self._fresh_air_temperature <= 25.0:
  83. if temp_avg <= 23.0:
  84. freq_set = self._freq + 2.0
  85. elif temp_avg <= 25.5:
  86. freq_set = self._freq
  87. else:
  88. freq_set = self._freq + 1.0
  89. else:
  90. if temp_avg <= 23.0:
  91. freq_set = self._freq + 5.0
  92. elif temp_avg <= 25.5:
  93. freq_set = self._freq - 2.0
  94. else:
  95. freq_set = 20.0
  96. return freq_set
  97. def get_cooling_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  98. _, co2_avg = self.get_avg(spaces_params)
  99. if on_flag:
  100. freq_set = 30.0
  101. else:
  102. high_co2_ratio = self.get_high_co2_ratio(spaces_params)
  103. if high_co2_ratio == 0.0:
  104. if co2_avg < 600.0:
  105. freq_set = self._freq - 2.0
  106. else:
  107. freq_set = self._freq
  108. freq_set = max(25.0, freq_set)
  109. elif high_co2_ratio < 0.05:
  110. freq_set = self._freq + 2.0
  111. freq_set = min(35.0, freq_set)
  112. elif high_co2_ratio < 0.1:
  113. freq_set = self._freq + 3.0
  114. freq_set = min(40.0, freq_set)
  115. elif high_co2_ratio < 0.2:
  116. freq_set = self._freq + 5.0
  117. else:
  118. freq_set = 50.0
  119. return freq_set
  120. def get_heating_logic(self, spaces_params: List[SpaceATFU], on_flag: bool) -> float:
  121. # The same with cooling logic.
  122. freq_set = self.get_cooling_logic(spaces_params, on_flag)
  123. return freq_set
  124. @staticmethod
  125. def get_avg(spaces_params: List[SpaceATFU]) -> Tuple[float, float]:
  126. valid_temperature_list, valid_co2_list = list(), list()
  127. for space in spaces_params:
  128. if space.realtime_temperature:
  129. if 0.0 < space.realtime_temperature < 40.0:
  130. valid_temperature_list.append(space.realtime_temperature)
  131. if space.realtime_co2:
  132. if 0.0 < space.realtime_co2 < 5000.0:
  133. valid_co2_list.append(space.realtime_co2)
  134. if valid_temperature_list:
  135. temp_avg = np.mean(valid_temperature_list)
  136. else:
  137. temp_avg = np.NAN
  138. if valid_co2_list:
  139. co2_avg = np.mean(valid_co2_list)
  140. else:
  141. co2_avg = np.NAN
  142. return temp_avg, co2_avg
  143. @staticmethod
  144. def get_high_co2_ratio(spaces_params: List[SpaceATFU]) -> float:
  145. valid_co2_count, high_co2_count = 0, 0
  146. for space in spaces_params:
  147. if space.realtime_co2:
  148. if 0.0 < space.realtime_co2 < 5000.0:
  149. valid_co2_count += 1
  150. if space.realtime_co2 > 900.0:
  151. high_co2_count += 1
  152. if valid_co2_count == 0:
  153. ratio = 0.0
  154. else:
  155. ratio = high_co2_count / valid_co2_count
  156. return ratio
  157. @staticmethod
  158. def hcho_logic(spaces_params: List[SpaceATFU], next_freq_set: float) -> float:
  159. diff = 0.0
  160. for space in spaces_params:
  161. if space.hcho:
  162. if space.hcho >= 0.1:
  163. diff = 5
  164. break
  165. elif space.hcho >= 0.08:
  166. diff = 3
  167. else:
  168. diff = 0
  169. return next_freq_set + diff
  170. async def build_acatfu_freq_set(params: ACATFUFreqSetRequest) -> float:
  171. controller = ACATFUFanFreqController(params.freq, params.fresh_air_temperature, params.season)
  172. on_flag = is_off_to_on(params.running_status_list)
  173. spaces = [SpaceATFU(**sp.dict()) for sp in params.spaces]
  174. freq_set = controller.get_next_set(spaces, on_flag)
  175. return freq_set