|
@@ -0,0 +1,92 @@
|
|
|
+from typing import Dict, Optional
|
|
|
+
|
|
|
+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, speed: Optional[str] = 'off') -> bool:
|
|
|
+ if self.target and self.target:
|
|
|
+ if speed == 'off':
|
|
|
+ switch_set = False
|
|
|
+ else:
|
|
|
+ switch_set = True
|
|
|
+ else:
|
|
|
+ switch_set = False
|
|
|
+
|
|
|
+ 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 = '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 = 'unknown'
|
|
|
+
|
|
|
+ self.device.speed_set = speed
|
|
|
+
|
|
|
+ return speed
|
|
|
+
|
|
|
+ async def run(self):
|
|
|
+ if self.target and self.realtime:
|
|
|
+ speed = self.get_speed_set()
|
|
|
+ self.get_switch_set(speed)
|
|
|
+ self.get_mode_set()
|
|
|
+ self.get_temperature_set()
|
|
|
+ else:
|
|
|
+ self.get_switch_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)
|
|
|
+ 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
|