# -*- coding: utf-8 -*- from operator import attrgetter import arrow import numpy as np from app.api.errors.iot import MissingIOTDataError from app.models.domain.devices import ASHPRequest from app.schemas.equipment import ASHP from app.schemas.season import Season class ASHPGroupController: def __init__(self, ashp_list: list[ASHP], outdoor_temp_list: list[float], season: Season): self._season = season self._ashp_list = ashp_list self._outdoor_temp_list = outdoor_temp_list self._interval = 0 self._is_warning = False def initialization(self) -> tuple[float, float]: if self._season == Season.cooling: if self._outdoor_temp_list: outdoor_temp_avg = np.mean(self._outdoor_temp_list) else: outdoor_temp_avg = 22.0 total_device_num = len(self._ashp_list) if outdoor_temp_avg <= 20.0: device_num = 1 out_temp_set = 10.0 interval = 45 * 60 elif outdoor_temp_avg < 24.0: device_num = total_device_num // 3 out_temp_set = 9.0 interval = 0 else: device_num = total_device_num // 3 + 1 out_temp_set = 8.5 interval = 0 if self._outdoor_temp_list: if np.max(self._outdoor_temp_list) > 34.0: device_num += 1 device_num = min(device_num, total_device_num - 1) elif self._season == Season.heating: if self._outdoor_temp_list: outdoor_temp_avg = np.mean(self._outdoor_temp_list) else: outdoor_temp_avg = -10.0 total_device_num = len(self._ashp_list) if outdoor_temp_avg >= 0.0: device_num = 1 out_temp_set = 45.0 interval = 45 * 60 elif outdoor_temp_avg > -15.0: device_num = total_device_num // 3 out_temp_set = 45.0 interval = 0 else: device_num = total_device_num // 3 + 1 out_temp_set = 50.0 interval = 0 if self._outdoor_temp_list: if np.min(self._outdoor_temp_list) < -15.0: device_num += 1 device_num = min(device_num, total_device_num - 1) else: device_num = 0 interval = 0 out_temp_set = np.NAN self._interval = interval return device_num, out_temp_set def device_num_adjustment(self) -> tuple[float, float]: out_temp_15_avg = np.mean( [np.mean(device.out_temp_15min) for device in self._ashp_list if device.running_status]) out_temp_set_15_avg = np.mean( [np.mean(device.out_temp_set_15min) for device in self._ashp_list if device.running_status]) iplr_15_avg = np.mean([np.mean(device.iplr_15min) for device in self._ashp_list if device.running_status]) out_temp_30_avg = np.mean( [np.mean(device.out_temp_30min) for device in self._ashp_list if device.running_status]) out_temp_set_30_avg = np.mean( [np.mean(device.out_temp_set_30min) for device in self._ashp_list if device.running_status]) iplr_30_avg = np.mean([np.mean(device.iplr_30min) for device in self._ashp_list if device.running_status]) device_num_diff = 0 if self._season == Season.cooling: if out_temp_15_avg < out_temp_set_15_avg - 1 and iplr_15_avg > 0.85: device_num_diff = 1 if out_temp_30_avg > out_temp_set_30_avg + 1 and iplr_30_avg <= 0.85: device_num_diff = 1 self._is_warning = True if out_temp_15_avg >= out_temp_set_15_avg and iplr_15_avg < 0.6: device_num_diff = -1 elif self._season == Season.heating: if out_temp_15_avg > out_temp_set_15_avg + 1 and iplr_15_avg > 0.85: device_num_diff = 1 if out_temp_30_avg > out_temp_set_30_avg + 1 and iplr_30_avg <= 0.85: device_num_diff = 1 self._is_warning = True if out_temp_15_avg <= out_temp_set_15_avg + 0.3 and iplr_15_avg < 0.6: device_num_diff = -1 return device_num_diff, float(out_temp_set_30_avg) def out_temp_set_adjustment(self) -> float: out_temp_set = np.NAN if self._outdoor_temp_list: current_outdoor_temp = self._outdoor_temp_list[-1] if self._season == Season.cooling: if current_outdoor_temp >= 30.0: out_temp_set = 8.0 elif current_outdoor_temp >= 28.0: out_temp_set = 8.5 elif current_outdoor_temp >= 26.0: out_temp_set = 9.0 elif current_outdoor_temp > 22.0: out_temp_set = 10.0 else: out_temp_set = 11.0 elif self._season == Season.heating: if current_outdoor_temp > 10.0: out_temp_set = 40.0 elif current_outdoor_temp >= 0.0: out_temp_set = 45.0 elif current_outdoor_temp > -10.0: out_temp_set = 50.0 else: out_temp_set = 55.0 return out_temp_set def build_up(self, diff: float, out_temp_set: float): if diff > 0: self._ashp_list = sorted(self._ashp_list, key=attrgetter("acc_run_time")) for device in self._ashp_list: if not device.running_status: device.equip_switch_set = True diff -= 1 if diff == 0: break elif diff < 0: self._ashp_list = sorted(self._ashp_list, key=attrgetter("acc_run_time"), reverse=True) for device in self._ashp_list: if device.running_status: device.equip_switch_set = False diff += 1 if diff == 0: break if not np.isnan(out_temp_set): for device in self._ashp_list: if device.equip_switch_set: device.out_temp_set = out_temp_set def turn_off_all(self): for device in self._ashp_list: device.equip_switch_set = False def antifreeze(self): pass def run(self): t = arrow.utcnow() on_time = arrow.get(self._ashp_list[0].on_time, 'HHmmss') off_time = arrow.get(self._ashp_list[0].off_time, 'HHmmss') if off_time.hour > on_time.hour: if not on_time.hour < t.hour < off_time.hour: self.turn_off_all() return else: if off_time.hour < t.hour < on_time.hour: self.turn_off_all() return if t.minute < 30: if t.hour == 1: if not np.any([device.running_status for device in self._ashp_list]): # no device is running device_num_diff, out_temp_set = self.initialization() self.build_up(device_num_diff, out_temp_set) else: device_num_diff, out_temp_set = self.device_num_adjustment() self.build_up(device_num_diff, out_temp_set) else: out_temp_set = self.out_temp_set_adjustment() self.build_up(0, out_temp_set) def get_results(self) -> tuple[list[ASHP], float, bool]: return self._ashp_list, self._interval, self._is_warning async def build_ashp_instructions(params: ASHPRequest) -> dict[str, list | float | bool]: controller = ASHPGroupController(params.device_list, params.outdoor_temp, params.season) try: controller.run() except (TypeError, IndexError): raise MissingIOTDataError device_list, interval, is_warning = controller.get_results() instructions = { "device_list": device_list, "interval": interval, "is_warning": is_warning } return instructions