123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 |
- # -*- 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)
|