common.py 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182
  1. # -*- coding: utf-8 -*-
  2. from typing import List, Tuple
  3. import numpy as np
  4. from loguru import logger
  5. from app.api.errors.iot import MissingIOTDataError
  6. from app.models.domain.devices import ACATAHInstructionsRequest
  7. from app.schemas.equipment import AHU
  8. from app.schemas.season import Season
  9. from app.schemas.space import SpaceATAH
  10. from app.utils.date import get_time_str
  11. class ATAHController:
  12. def __init__(self, ahu: AHU, spaces: List[SpaceATAH], season: Season) -> None:
  13. self._ahu = ahu
  14. self._spaces = spaces
  15. self._season = season
  16. def get_switch_set(self) -> str:
  17. switch_set = "off"
  18. for sp in self._spaces:
  19. if sp.temperature_target:
  20. switch_set = "on"
  21. return switch_set
  22. @staticmethod
  23. def get_return_air_temp_set(virtual_target: float) -> float:
  24. return virtual_target
  25. def get_supply_air_temp_set(self, next_switch_set: str, virtual_realtime: float) -> float:
  26. if next_switch_set == "on" and not self._ahu.running_status:
  27. if self._season == Season.cooling:
  28. next_supply_set = 20.0
  29. elif self._season == Season.transition:
  30. next_supply_set = 22.0
  31. else:
  32. next_supply_set = 28.0
  33. else:
  34. next_supply_set = self._ahu.supply_air_temperature_set
  35. diff = self._ahu.return_air_temperature - virtual_realtime
  36. if self._season == Season.cooling:
  37. if diff > 1.0 and self._ahu.freq >= self._ahu.fan_freq_upper_limit_set:
  38. next_supply_set -= 1.0
  39. if diff < -1.0 and self._ahu.freq <= self._ahu.fan_freq_lower_limit_set:
  40. next_supply_set += 1.0
  41. next_supply_set = np.max([np.min([23.0, next_supply_set]), 18.0])
  42. logger.debug(next_supply_set)
  43. elif self._season == Season.heating:
  44. if diff > 1.0 and self._ahu.freq <= self._ahu.fan_freq_lower_limit_set:
  45. next_supply_set -= 1.0
  46. if diff < 1.0 and self._ahu.freq >= self._ahu.fan_freq_upper_limit_set:
  47. next_supply_set += 1.0
  48. next_supply_set = np.max([np.min([28.0, next_supply_set]), 22.0])
  49. return next_supply_set
  50. def get_freq_set(self, next_switch_set: str, virtual_target: float) -> float:
  51. if next_switch_set == "on" and not self._ahu.running_status:
  52. next_freq_set = 40.0
  53. else:
  54. diff = self._ahu.return_air_temperature - virtual_target
  55. if self._season == Season.heating:
  56. if diff > 1.0:
  57. adjust = -2.0
  58. elif diff < -1.0:
  59. adjust = 2.0
  60. else:
  61. adjust = 0.0
  62. elif self._season == Season.cooling:
  63. if diff > 1.0:
  64. adjust = 2.0
  65. elif diff < 1.0:
  66. adjust = -2.0
  67. else:
  68. adjust = 0.0
  69. else:
  70. if diff > 1.0:
  71. if self._ahu.supply_air_temperature > self._ahu.return_air_temperature:
  72. adjust = -2.0
  73. elif self._ahu.supply_air_temperature < self._ahu.return_air_temperature - 1.0:
  74. adjust = 2.0
  75. else:
  76. adjust = 1.0
  77. elif diff < -1.0:
  78. if self._ahu.supply_air_temperature_set < self._ahu.return_air_temperature:
  79. adjust = -2.0
  80. elif self._ahu.supply_air_temperature > self._ahu.return_air_temperature + 1:
  81. adjust = 2.0
  82. else:
  83. adjust = 1.0
  84. else:
  85. adjust = 0.0
  86. next_freq_set = self._ahu.freq_set + adjust
  87. next_freq_set = np.max(
  88. [np.min([self._ahu.fan_freq_upper_limit_set, next_freq_set]), self._ahu.fan_freq_lower_limit_set])
  89. return next_freq_set
  90. def get_valid_spaces(self) -> List:
  91. valid_spaces = list()
  92. for sp in self._spaces:
  93. if sp.realtime_temperature and sp.temperature_target:
  94. valid_spaces.append(sp)
  95. return valid_spaces
  96. def build_virtual_temperature(self) -> Tuple[float, float]:
  97. valid_spaces = self.get_valid_spaces()
  98. if not valid_spaces:
  99. virtual_realtime, virtual_target = np.NAN, np.NAN
  100. else:
  101. sorted_spaces = sorted(valid_spaces, key=lambda x: x.ahu_temporary_update_time)
  102. if sorted_spaces[-1].ahu_temporary_update_time > get_time_str(60 * 60 * 2, flag="ago"):
  103. virtual_realtime = sorted_spaces[-1].realtime_temperature
  104. virtual_target = sorted_spaces[-1].temperature_target
  105. else:
  106. virtual_realtime, virtual_target = 0.0, 0.0
  107. total_weight = 0.0
  108. for sp in valid_spaces:
  109. weight = sp.ahu_default_weight
  110. virtual_realtime += sp.realtime_temperature * weight
  111. virtual_target += sp.temperature_target * weight
  112. total_weight += weight
  113. if total_weight == 0:
  114. for sp in valid_spaces:
  115. virtual_realtime += sp.realtime_temperature
  116. virtual_target += sp.temperature_target
  117. valid_spaces_length = len(valid_spaces)
  118. virtual_realtime /= valid_spaces_length
  119. virtual_target /= valid_spaces_length
  120. else:
  121. virtual_realtime /= total_weight
  122. virtual_target /= total_weight
  123. return virtual_realtime, virtual_target
  124. def run(self) -> Tuple[str, float, float, float]:
  125. try:
  126. virtual_realtime, virtual_target = self.build_virtual_temperature()
  127. new_switch_set = self.get_switch_set()
  128. if not self._ahu.return_air_temperature_set:
  129. new_freq_set = self.get_freq_set(new_switch_set, virtual_target)
  130. new_return_air_temp_set = np.NAN
  131. else:
  132. new_return_air_temp_set = self.get_return_air_temp_set(virtual_target)
  133. new_freq_set = np.NAN
  134. new_supply_air_temp_set = self.get_supply_air_temp_set(new_switch_set, virtual_realtime)
  135. except TypeError:
  136. raise MissingIOTDataError
  137. return new_switch_set, new_return_air_temp_set, new_freq_set, new_supply_air_temp_set
  138. async def build_acatah_instructions(params: ACATAHInstructionsRequest):
  139. space_params = []
  140. for sp in params.spaces:
  141. temp_sp = SpaceATAH(**sp.dict())
  142. temp_sp.diff = temp_sp.temperature_target - temp_sp.realtime_temperature
  143. space_params.append(temp_sp)
  144. ahu = AHU(**params.dict())
  145. ahu_controller = ATAHController(ahu, space_params, params.season)
  146. switch_set, return_air_temp_set, freq_set, supply_air_temp_set = ahu_controller.run()
  147. instructions = {"switch_set": switch_set}
  148. if not np.isnan(return_air_temp_set):
  149. instructions.update({"return_air_temp_set": return_air_temp_set})
  150. if not np.isnan(freq_set):
  151. instructions.update({"freq_set": freq_set})
  152. if not np.isnan(supply_air_temp_set):
  153. instructions.update({"supply_air_temp_set": supply_air_temp_set})
  154. return instructions