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.schemas.equipment import VRF 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): if self.device.running_status == 'on': switch_set = 'off' else: switch_set = 'hold' else: if self.device.running_status == 'off': switch_set = 'on' else: switch_set = 'hold' 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: self.device.temperature_set = self.target + diff else: self.device.temperature_set = self.target elif self.realtime < 22.0: self.device.temperature_set = self.device.return_air_temp + 1 else: self.device.temperature_set = self.device.current_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: speed = self.device.speed self.device.speed_set = speed return speed async def run(self): self.get_speed_set() self.get_switch_set() self.get_mode_set() self.get_temperature_set() def get_results(self): return self.device @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 vrf = VRF(return_air_temp=return_air_temp) controller = VRFController(vrf, target, realtime) await controller.run() regulated_vrf = controller.get_results() instructions = {'switch_set': 'on' if regulated_vrf.equip_switch_set else 'off'} if regulated_vrf.equip_switch_set: instructions.update({'speed_set': regulated_vrf.speed_set}) instructions.update({'temperature_set': regulated_vrf.temperature_set}) instructions.update({'mode_set': regulated_vrf.mode_set}) return instructions