# -*- coding: utf-8 -*- from typing import List, Tuple import arrow import numpy as np from app.schemas.equipment import ASHP class ASHPGroupController: def __init__(self, ashp_list: List[ASHP], outdoor_temp_list: List[float]): self._ashp_list = ashp_list self._outdoor_temp_list = outdoor_temp_list self._is_warning = False def initialization(self) -> Tuple[float, float, float]: 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) return device_num, out_temp_set, interval def device_num_adjustment(self) -> float: out_temp_15_avg = np.mean([np.mean(device.out_temp_15min) for device in self._ashp_list]) out_temp_set_15_avg = np.mean([np.mean(device.out_temp_set_15min) for device in self._ashp_list]) iplr_15_avg = np.mean([np.mean(device.iplr_15min) for device in self._ashp_list]) out_temp_30_avg = np.mean([np.mean(device.out_temp_30min) for device in self._ashp_list]) out_temp_set_30_avg = np.mean([np.mean(device.out_temp_set_30min) for device in self._ashp_list]) iplr_30_avg = np.mean([np.mean(device.iplr_30min) for device in self._ashp_list]) device_num_diff = 0 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 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 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 return out_temp_set def build_up(self, diff: float, out_temp_set: float): if diff > 0: 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: 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.running_status or device.equip_switch_set: device.out_temp_set = out_temp_set def run(self): t = arrow.utcnow() 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, interval = self.initialization() self.build_up(device_num_diff, out_temp_set) else: device_num_diff = self.device_num_adjustment() self.build_up(device_num_diff, np.NAN) else: out_temp_set = self.out_temp_set_adjustment() self.build_up(0, out_temp_set)