123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213 |
- from typing import Dict
- import numpy as np
- from loguru import logger
- from app.controllers.equipment.controller import EquipmentController
- from app.models.domain.devices import ACATVIInstructionsRequest
- from app.models.domain.feedback import FeedbackValue
- from app.schemas.equipment import VRF
- from app.utils.math import round_half_up
- class VRFController(EquipmentController):
- def __init__(self, device: VRF, target: float, realtime: float):
- super(VRFController, self).__init__()
- self.device = device
- self.target = target
- self.realtime = realtime
- def get_switch_set(self) -> str:
- if np.isnan(self.target):
- switch_set = 'off'
- else:
- switch_set = 'on'
- self.device.equip_switch_set = switch_set
- return switch_set
- def get_mode_set(self) -> str:
- if self.target > self.realtime:
- mode_set = 'cooling'
- else:
- mode_set = 'cooling'
- self.device.mode_set = mode_set
- return mode_set
- def get_temperature_set(self) -> float:
- if self.realtime >= 24.0:
- diff = self.device.return_air_temp - self.realtime
- if diff > 0:
- new_temperature_set = self.target + diff
- elif diff <= 0:
- new_temperature_set = self.target
- else:
- if np.isnan(self.device.current_temperature_set):
- new_temperature_set = self.realtime
- else:
- new_temperature_set = self.device.current_temperature_set
- elif self.realtime <= 22.0:
- new_temperature_set = 28.0
- else:
- if np.isnan(self.device.current_temperature_set):
- new_temperature_set = 25.0
- else:
- new_temperature_set = self.device.current_temperature_set
- new_temperature_set = round_half_up(new_temperature_set)
- self.device.temperature_set = new_temperature_set
- return self.device.temperature_set
- def get_speed_set(self) -> str:
- if self.realtime >= 24.0:
- speed = 'M'
- elif self.realtime <= 22.0:
- speed = 'LL'
- elif self.device.return_air_temp - self.target >= 3.0:
- speed = 'HH'
- elif self.device.return_air_temp - self.target >= 1.0:
- speed = 'M'
- else:
- if self.device.speed == 'missing':
- speed = 'M'
- else:
- speed = self.device.speed
- self.device.speed_set = speed
- return speed
- async def run(self):
- if self.get_switch_set() == 'on':
- self.get_speed_set()
- self.get_mode_set()
- self.get_temperature_set()
- if self.device.temperature_set == self.device.current_temperature_set:
- self.device.temperature_set = 0.0
- if self.device.speed_set == self.device.speed:
- self.device.speed_set = 'hold'
- self.device.equip_switch_set = 'hold'
- if not self.device.running_status:
- self.device.equip_switch_set = 'on'
- self.get_speed_set()
- else:
- if not self.device.running_status:
- self.device.equip_switch_set = 'hold'
- def get_results(self):
- return self.device
- class VRFController1(VRFController):
- def __init__(self, device: VRF, target: float, realtime: float, feedback: FeedbackValue):
- super(VRFController1, self).__init__(device, target, realtime)
- self.feedback = feedback
- def get_temperature_set(self) -> float:
- new_temperature_set = self.device.current_temperature_set
- # Default temperature set.
- if not self.device.running_status:
- new_temperature_set = self.target
- # lower limit.
- if self.realtime <= 22.0:
- new_temperature_set = 28.0
- # recovery
- # if self.device.temperature_set == 28.0 and self.realtime > 25.0:
- # new_temperature_set = self.target
- # feedback
- if self.feedback == FeedbackValue.a_little_cold or self.feedback == FeedbackValue.so_cold:
- if self.feedback == FeedbackValue.so_cold and self.device.speed == 'LL':
- new_temperature_set = self.device.current_temperature_set + 2.0
- else:
- new_temperature_set = self.device.current_temperature_set + 1.0
- if self.device.return_air_temp <= self.device.current_temperature_set:
- if self.feedback == FeedbackValue.so_cold:
- new_temperature_set = self.device.current_temperature_set + 2.0
- elif self.feedback == FeedbackValue.a_little_hot or self.feedback == FeedbackValue.so_hot:
- if self.feedback == FeedbackValue.so_hot and self.device.speed == 'HH':
- new_temperature_set = self.device.current_temperature_set - 2.0
- else:
- new_temperature_set = self.device.current_temperature_set - 1.0
- new_temperature_set = round_half_up(new_temperature_set)
- self.device.temperature_set = new_temperature_set
- return new_temperature_set
- def get_speed_set(self) -> str:
- new_speed = self.device.speed
- # Default speed set:
- if not self.device.running_status:
- new_speed = 'M'
- # Lower limit.
- if self.realtime <= 22.0:
- new_speed = 'LL'
- # Feedback.
- if self.feedback == FeedbackValue.so_cold:
- if self.device.return_air_temp > self.device.current_temperature_set:
- if self.device.speed == 'HH':
- new_speed = 'M'
- elif self.device.speed == 'M':
- new_speed = 'LL'
- elif self.feedback == FeedbackValue.so_hot:
- if self.device.speed == 'LL':
- new_speed = 'M'
- elif self.device.speed == 'M':
- new_speed = 'HH'
- elif self.feedback == FeedbackValue.noisy_or_blowy:
- if self.device.speed == 'HH':
- new_speed = 'M'
- elif self.device.speed == 'M':
- new_speed = 'LL'
- else:
- new_speed = 'LL'
- self.device.speed_set = new_speed
- return new_speed
- @logger.catch()
- async def build_acatvi_instructions(params: ACATVIInstructionsRequest) -> Dict:
- target = np.NAN if params.space_temperature_target == -1.0 else params.space_temperature_target
- realtime = np.NAN if params.space_realtime_temperature == -1.0 else params.space_realtime_temperature
- return_air_temp = np.NAN if params.return_air_temperature == -1.0 else params.return_air_temperature
- current_temperature_set = np.NAN if params.current_temperature_set == -1.0 else round_half_up(
- params.current_temperature_set
- )
- vrf = VRF(
- return_air_temp=return_air_temp,
- current_temperature_set=current_temperature_set,
- speed=params.current_speed,
- running_status=params.running_status
- )
- controller = VRFController1(vrf, target, realtime, params.feedback)
- await controller.run()
- regulated_vrf = controller.get_results()
- instructions = {
- 'switch_set': regulated_vrf.equip_switch_set,
- 'speed_set': regulated_vrf.speed_set,
- 'temperature_set': regulated_vrf.temperature_set,
- 'mode_set': regulated_vrf.mode_set
- }
- return instructions
|