|
@@ -0,0 +1,179 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+
|
|
|
+
|
|
|
+from typing import List, Tuple
|
|
|
+
|
|
|
+import numpy as np
|
|
|
+from loguru import logger
|
|
|
+
|
|
|
+from app.models.domain.devices import ACATAHInstructionsRequest
|
|
|
+from app.schemas.equipment import AHU
|
|
|
+from app.schemas.season import Season
|
|
|
+from app.schemas.space import SpaceATAH
|
|
|
+from app.utils.date import get_time_str
|
|
|
+
|
|
|
+
|
|
|
+class ATAHController:
|
|
|
+ def __init__(self, ahu: AHU, spaces: List[SpaceATAH], season: Season) -> None:
|
|
|
+ self._ahu = ahu
|
|
|
+ self._spaces = spaces
|
|
|
+ self._season = season
|
|
|
+
|
|
|
+ def get_switch_set(self) -> str:
|
|
|
+ switch_set = "off"
|
|
|
+ for sp in self._spaces:
|
|
|
+ if not np.isnan(sp.temperature_target):
|
|
|
+ switch_set = "on"
|
|
|
+
|
|
|
+ return switch_set
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_return_air_temp_set(virtual_target: float) -> float:
|
|
|
+ return virtual_target
|
|
|
+
|
|
|
+ def get_supply_air_temp_set(self, next_switch_set: str, virtual_realtime: float) -> float:
|
|
|
+ if next_switch_set == "on" and not self._ahu.running_status:
|
|
|
+ if self._season == Season.cooling:
|
|
|
+ next_supply_set = 20.0
|
|
|
+ elif self._season == Season.transition:
|
|
|
+ next_supply_set = 22.0
|
|
|
+ else:
|
|
|
+ next_supply_set = 28.0
|
|
|
+ else:
|
|
|
+ next_supply_set = self._ahu.supply_air_temperature_set
|
|
|
+ diff = self._ahu.return_air_temperature - virtual_realtime
|
|
|
+ if self._season == Season.cooling:
|
|
|
+ if diff > 1.0 and self._ahu.fan_freq >= self._ahu.fan_freq_upper_limit_set:
|
|
|
+ next_supply_set -= 1.0
|
|
|
+ if diff < -1.0 and self._ahu.fan_freq <= self._ahu.fan_freq_lower_limit_set:
|
|
|
+ next_supply_set += 1.0
|
|
|
+ next_supply_set = np.max([np.min([23.0, next_supply_set]), 18.0])
|
|
|
+ logger.debug(next_supply_set)
|
|
|
+ elif self._season == Season.heating:
|
|
|
+ if diff > 1.0 and self._ahu.fan_freq <= self._ahu.fan_freq_lower_limit_set:
|
|
|
+ next_supply_set -= 1.0
|
|
|
+ if diff < 1.0 and self._ahu.fan_freq >= self._ahu.fan_freq_upper_limit_set:
|
|
|
+ next_supply_set += 1.0
|
|
|
+ next_supply_set = np.max([np.min([28.0, next_supply_set]), 22.0])
|
|
|
+
|
|
|
+ return next_supply_set
|
|
|
+
|
|
|
+ def get_freq_set(self, next_switch_set: str, virtual_target: float) -> float:
|
|
|
+ if next_switch_set == "on" and not self._ahu.running_status:
|
|
|
+ next_freq_set = 40.0
|
|
|
+ else:
|
|
|
+ diff = self._ahu.return_air_temperature - virtual_target
|
|
|
+ if self._season == Season.heating:
|
|
|
+ if diff > 1.0:
|
|
|
+ adjust = -2.0
|
|
|
+ elif diff < -1.0:
|
|
|
+ adjust = 2.0
|
|
|
+ else:
|
|
|
+ adjust = 0.0
|
|
|
+ elif self._season == Season.cooling:
|
|
|
+ if diff > 1.0:
|
|
|
+ adjust = 2.0
|
|
|
+ elif diff < 1.0:
|
|
|
+ adjust = -2.0
|
|
|
+ else:
|
|
|
+ adjust = 0.0
|
|
|
+ else:
|
|
|
+ if diff > 1.0:
|
|
|
+ if self._ahu.supply_air_temperature > self._ahu.return_air_temperature:
|
|
|
+ adjust = -2.0
|
|
|
+ elif self._ahu.supply_air_temperature < self._ahu.return_air_temperature - 1.0:
|
|
|
+ adjust = 2.0
|
|
|
+ else:
|
|
|
+ adjust = 1.0
|
|
|
+ elif diff < -1.0:
|
|
|
+ if self._ahu.supply_air_temperature_set < self._ahu.return_air_temperature:
|
|
|
+ adjust = -2.0
|
|
|
+ elif self._ahu.supply_air_temperature > self._ahu.return_air_temperature + 1:
|
|
|
+ adjust = 2.0
|
|
|
+ else:
|
|
|
+ adjust = 1.0
|
|
|
+ else:
|
|
|
+ adjust = 0.0
|
|
|
+
|
|
|
+ next_freq_set = self._ahu.fan_freq_set + adjust
|
|
|
+
|
|
|
+ next_freq_set = np.max(
|
|
|
+ [np.min([self._ahu.fan_freq_upper_limit_set, next_freq_set]), self._ahu.fan_freq_lower_limit_set])
|
|
|
+
|
|
|
+ return next_freq_set
|
|
|
+
|
|
|
+ def get_valid_spaces(self) -> List:
|
|
|
+ valid_spaces = list()
|
|
|
+ for sp in self._spaces:
|
|
|
+ if not np.isnan(sp.realtime_temperature) and not np.isnan(sp.temperature_target):
|
|
|
+ valid_spaces.append(sp)
|
|
|
+
|
|
|
+ return valid_spaces
|
|
|
+
|
|
|
+ def build_virtual_temperature(self) -> Tuple[float, float]:
|
|
|
+ valid_spaces = self.get_valid_spaces()
|
|
|
+
|
|
|
+ if not valid_spaces:
|
|
|
+ virtual_realtime, virtual_target = np.NAN, np.NAN
|
|
|
+ else:
|
|
|
+ sorted_spaces = sorted(valid_spaces, key=lambda x: x.ahu_temporary_update_time)
|
|
|
+ if sorted_spaces[-1].ahu_temporary_update_time > get_time_str(60 * 60 * 2, flag="ago"):
|
|
|
+ virtual_realtime = sorted_spaces[-1].realtime_temperature
|
|
|
+ virtual_target = sorted_spaces[-1].temperature_target
|
|
|
+ else:
|
|
|
+ virtual_realtime, virtual_target = 0.0, 0.0
|
|
|
+ total_weight = 0.0
|
|
|
+ for sp in valid_spaces:
|
|
|
+ weight = sp.ahu_default_weight
|
|
|
+ virtual_realtime += sp.realtime_temperature * weight
|
|
|
+ virtual_target += sp.temperature_target * weight
|
|
|
+ total_weight += weight
|
|
|
+
|
|
|
+ if total_weight == 0:
|
|
|
+ for sp in valid_spaces:
|
|
|
+ virtual_realtime += sp.realtime_temperature
|
|
|
+ virtual_target += sp.temperature_target
|
|
|
+ valid_spaces_length = len(valid_spaces)
|
|
|
+ virtual_realtime /= valid_spaces_length
|
|
|
+ virtual_target /= valid_spaces_length
|
|
|
+ else:
|
|
|
+ virtual_realtime /= total_weight
|
|
|
+ virtual_target /= total_weight
|
|
|
+
|
|
|
+ return virtual_realtime, virtual_target
|
|
|
+
|
|
|
+ def run(self) -> Tuple[str, float, float, float]:
|
|
|
+ virtual_realtime, virtual_target = self.build_virtual_temperature()
|
|
|
+ new_switch_set = self.get_switch_set()
|
|
|
+ if np.isnan(self._ahu.return_air_temperature_set):
|
|
|
+ new_freq_set = self.get_freq_set(new_switch_set, virtual_target)
|
|
|
+ new_return_air_temp_set = np.NAN
|
|
|
+ else:
|
|
|
+ new_return_air_temp_set = self.get_return_air_temp_set(virtual_target)
|
|
|
+ new_freq_set = np.NAN
|
|
|
+ new_supply_air_temp_set = self.get_supply_air_temp_set(new_switch_set, virtual_realtime)
|
|
|
+
|
|
|
+ return new_switch_set, new_return_air_temp_set, new_freq_set, new_supply_air_temp_set
|
|
|
+
|
|
|
+
|
|
|
+@logger.catch()
|
|
|
+async def build_acatah_instructions(params: ACATAHInstructionsRequest):
|
|
|
+ space_params = []
|
|
|
+ for sp in params.spaces:
|
|
|
+ temp_sp = SpaceATAH(**sp.dict())
|
|
|
+ temp_sp.diff = temp_sp.temperature_target - temp_sp.realtime_temperature
|
|
|
+ space_params.append(temp_sp)
|
|
|
+
|
|
|
+ ahu = AHU(**params.dict())
|
|
|
+ ahu_controller = ATAHController(ahu, space_params, params.season)
|
|
|
+ switch_set, return_air_temp_set, freq_set, supply_air_temp_set = ahu_controller.run()
|
|
|
+
|
|
|
+ instructions = {"switch_set": switch_set}
|
|
|
+ if not np.isnan(return_air_temp_set):
|
|
|
+ instructions.update({"return_air_temp_set": return_air_temp_set})
|
|
|
+ if not np.isnan(freq_set):
|
|
|
+ instructions.update({"freq_set": freq_set})
|
|
|
+ if not np.isnan(supply_air_temp_set):
|
|
|
+ instructions.update({"supply_air_temp_set": supply_air_temp_set})
|
|
|
+
|
|
|
+ return instructions
|