from typing import Dict, Optional import numpy as np from loguru import logger from app.controllers.equipment.controller import EquipmentController 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) -> bool: if np.isnan(self.target): 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 return self.target def get_speed_set(self) -> str: diff = self.realtime - self.target if diff <= 0: speed = 'LL' elif diff < 0.8: speed = 'L' elif diff < 1.5: speed = 'M' elif diff < 2.0: speed = 'H' elif diff >= 2.0: speed = 'HH' else: speed = 'LL' self.device.speed_set = speed return speed async def run(self): if self.get_switch_set(): self.get_speed_set() self.get_mode_set() self.get_temperature_set() def get_results(self): return self.device @logger.catch() async def get_vrf_instructions(device_id: str, target: float, realtime: float) -> Dict: vrf = VRF(id=device_id) target = np.NAN if target == -1.0 else target realtime = np.NAN if realtime == -1.0 else realtime controller = VRFController(vrf, target, realtime) await controller.run() regulated_vrf = controller.get_results() output = {'switch_set': 'on' if regulated_vrf.equip_switch_set else 'off'} if regulated_vrf.equip_switch_set: output.update({'speed_set': regulated_vrf.speed_set}) output.update({'temperature_set': regulated_vrf.temperature_set}) output.update({'mode_set': regulated_vrf.mode_set}) return output