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, speed: str) -> bool: if speed == 'off': switch_set = False else: switch_set = True 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: self.device.temperature_set = self.target + 1 return self.target def get_speed_set(self) -> str: if self.realtime > self.device.return_air_temp: diff = self.realtime - self.target else: diff = self.device.return_air_temp - (self.target + 1) if diff < 0: speed = 'off' elif diff < 0.5: speed = 'LL' elif diff < 1.0: speed = 'L' elif diff < 1.5: speed = 'M' elif diff < 2.0: speed = 'H' elif diff > 2.0: speed = 'HH' else: speed = 'off' self.device.speed_set = speed return speed async def run(self): speed = self.get_speed_set() self.get_switch_set(speed) 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