basic.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. from typing import Dict
  2. import numpy as np
  3. from loguru import logger
  4. from app.controllers.equipment.controller import EquipmentController
  5. from app.models.domain.devices import ACATVIInstructionsRequest
  6. from app.schemas.equipment import VRF
  7. from app.utils.math import round_half_up
  8. class VRFController(EquipmentController):
  9. def __init__(self, device: VRF, target: float, realtime: float):
  10. super(VRFController, self).__init__()
  11. self.device = device
  12. self.target = target
  13. self.realtime = realtime
  14. def get_switch_set(self) -> str:
  15. if np.isnan(self.target):
  16. switch_set = 'off'
  17. else:
  18. switch_set = 'on'
  19. self.device.equip_switch_set = switch_set
  20. return switch_set
  21. def get_mode_set(self) -> str:
  22. if self.target > self.realtime:
  23. mode_set = 'cooling'
  24. else:
  25. mode_set = 'cooling'
  26. self.device.mode_set = mode_set
  27. return mode_set
  28. def get_temperature_set(self) -> float:
  29. if self.realtime >= 24.0:
  30. diff = self.device.return_air_temp - self.realtime
  31. if diff > 0:
  32. new_temperature_set = self.target + diff
  33. elif diff <= 0:
  34. new_temperature_set = self.target
  35. else:
  36. if np.isnan(self.device.current_temperature_set):
  37. new_temperature_set = self.realtime
  38. else:
  39. new_temperature_set = self.device.current_temperature_set
  40. elif self.realtime <= 22.0:
  41. new_temperature_set = 28.0
  42. else:
  43. if np.isnan(self.device.current_temperature_set):
  44. new_temperature_set = 25.0
  45. else:
  46. new_temperature_set = self.device.current_temperature_set
  47. new_temperature_set = round_half_up(new_temperature_set)
  48. self.device.temperature_set = new_temperature_set
  49. return self.device.temperature_set
  50. def get_speed_set(self) -> str:
  51. if self.realtime >= 24.0:
  52. speed = 'M'
  53. elif self.realtime <= 22.0:
  54. speed = 'LL'
  55. elif self.device.return_air_temp - self.target >= 3.0:
  56. speed = 'HH'
  57. elif self.device.return_air_temp - self.target >= 1.0:
  58. speed = 'M'
  59. else:
  60. if self.device.speed == 'missing':
  61. speed = 'M'
  62. else:
  63. speed = self.device.speed
  64. self.device.speed_set = speed
  65. return speed
  66. async def run(self):
  67. need_send = False
  68. if self.get_switch_set() == 'on':
  69. self.get_speed_set()
  70. self.get_mode_set()
  71. self.get_temperature_set()
  72. if self.device.temperature_set != self.device.current_temperature_set:
  73. need_send = True
  74. if self.device.speed_set != self.device.speed:
  75. need_send = True
  76. if self.device.equip_switch_set == 'on' and not self.device.running_status:
  77. need_send = True
  78. if self.device.equip_switch_set == 'off' and self.device.running_status:
  79. need_send = True
  80. return need_send
  81. def get_results(self):
  82. return self.device
  83. @logger.catch()
  84. async def build_acatvi_instructions(params: ACATVIInstructionsRequest) -> Dict:
  85. target = np.NAN if params.space_temperature_target == -1.0 else params.space_temperature_target
  86. realtime = np.NAN if params.space_realtime_temperature == -1.0 else params.space_realtime_temperature
  87. return_air_temp = np.NAN if params.return_air_temperature == -1.0 else params.return_air_temperature
  88. if params.running_status == 'missing':
  89. return {'need_send': False}
  90. elif params.running_status == 'on':
  91. running_status = True
  92. else:
  93. running_status = False
  94. vrf = VRF(
  95. return_air_temp=return_air_temp,
  96. current_temperature_set=np.NAN if params.current_temperature_set == -1.0 else params.current_temperature_set,
  97. speed=params.current_speed,
  98. running_status=running_status
  99. )
  100. controller = VRFController(vrf, target, realtime)
  101. need_send = await controller.run()
  102. regulated_vrf = controller.get_results()
  103. instructions = {
  104. 'switch_set': regulated_vrf.equip_switch_set,
  105. 'need_send': need_send
  106. }
  107. if regulated_vrf.equip_switch_set == 'on':
  108. instructions.update({'speed_set': regulated_vrf.speed_set})
  109. instructions.update({'temperature_set': regulated_vrf.temperature_set})
  110. instructions.update({'mode_set': regulated_vrf.mode_set})
  111. return instructions