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: switch_set = 'off' else: switch_set = 'hold' else: if not self.device.running_status: 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: if np.isnan(self.device.current_temperature_set): self.device.temperature_set = 25.0 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: if self.device.speed == 'missing': 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() need_send = False if self.get_switch_set() != 'hold': need_send = True if self.device.temperature_set != self.device.current_temperature_set: need_send = True if self.device.speed_set != self.device.speed: need_send = True return need_send 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 if params.running_status == 'missing': return {'need_send': False} elif params.running_status == 'on': running_status = True else: running_status = False vrf = VRF( return_air_temp=return_air_temp, current_temperature_set=np.NAN if params.current_temperature_set == -1.0 else params.current_temperature_set, speed=params.current_speed, running_status=running_status ) controller = VRFController(vrf, target, realtime) need_send = await controller.run() regulated_vrf = controller.get_results() instructions = { 'switch_set': regulated_vrf.equip_switch_set, 'need_send': need_send } if regulated_vrf.equip_switch_set != 'off': 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