on_ratio.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122
  1. import time
  2. from typing import Dict, Tuple
  3. import arrow
  4. from httpx import AsyncClient
  5. from loguru import logger
  6. from app.utils.math import round_half_up
  7. from app.utils.date import get_time_str, TIME_FMT
  8. class OnRatioController:
  9. def __init__(self, target: float, return_air: float, period_num: int):
  10. super(OnRatioController, self).__init__()
  11. self.target = target
  12. self.return_air = return_air
  13. self.period_num = period_num
  14. def select_mode(self) -> str:
  15. if self.target < self.return_air:
  16. mode = 'off'
  17. else:
  18. if self.target - self.return_air > 1:
  19. mode = 'normal'
  20. else:
  21. mode = 'on_ratio'
  22. return mode
  23. def select_speed(self, last_return_air: float) -> str:
  24. mode = self.select_mode()
  25. if mode == 'off':
  26. speed = 'off'
  27. elif mode == 'normal':
  28. if (self.return_air - last_return_air) > 0.8:
  29. speed = 'medium'
  30. else:
  31. speed = 'high'
  32. else:
  33. speed = 'medium'
  34. return speed
  35. def select_water_valve(self) -> bool:
  36. mode = self.select_mode()
  37. if mode == 'off':
  38. switch = True
  39. elif mode == 'normal':
  40. switch = False
  41. else:
  42. switch = True
  43. return switch
  44. def calculate_on_ratio(self, delta_on: float, delta_off: float) -> float:
  45. if self.period_num == 0:
  46. ratio = 0.5
  47. else:
  48. if delta_on <= 0:
  49. ratio = 1.0
  50. else:
  51. try:
  52. ratio = (0.5 * (self.return_air - self.target) + delta_off) / (delta_off - delta_on)
  53. except ZeroDivisionError:
  54. ratio = 0.0
  55. return ratio
  56. async def send_instructions(switch: bool, speed: str, water_valve: bool) -> None:
  57. pass
  58. async def fetch_params(device_id: str, period_time: float, last_ratio: float) -> Tuple[float, float, float, float]:
  59. pass
  60. @logger.catch()
  61. async def on_ratio_experiment(device_id):
  62. _TARGET = 26.0
  63. _DAILY_ON_TIME = '060000'
  64. _DAILY_OFF_TIME = '220000'
  65. period_time = 10 * 60
  66. period_num = 0
  67. last_on_ratio = 1.0
  68. life_count = 0
  69. while life_count < 1000:
  70. time_str = get_time_str()
  71. if _DAILY_ON_TIME <= arrow.get(time_str, TIME_FMT).time().strftime('%H%M%S') < _DAILY_OFF_TIME:
  72. return_air, last_return_air, delta_on, delta_off = await fetch_params(device_id, period_time, last_on_ratio)
  73. controller = OnRatioController(_TARGET, return_air, period_num)
  74. mode = controller.select_mode()
  75. speed = controller.select_speed(last_return_air)
  76. switch = False if speed == 'off' else True
  77. water_valve = controller.select_water_valve()
  78. if mode == 'on_ratio':
  79. on_ratio = controller.calculate_on_ratio(delta_on, delta_off)
  80. on_range = round_half_up(period_time * on_ratio)
  81. off_range = period_time - on_range
  82. await send_instructions(switch, speed, water_valve)
  83. time.sleep(on_range)
  84. await send_instructions(switch, speed, False)
  85. time.sleep(off_range)
  86. period_num += 1
  87. last_on_ratio = on_ratio
  88. else:
  89. await send_instructions(switch, speed, water_valve)
  90. time.sleep(period_time)
  91. period_num = 0
  92. last_on_ratio = 1.0
  93. life_count += 1
  94. else:
  95. await send_instructions(False, 'off', False)
  96. period_num = 0
  97. last_on_ratio = 0.0