switch.py 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. # -*- coding: utf-8 -*-
  2. from typing import Tuple
  3. from httpx import AsyncClient
  4. from loguru import logger
  5. from app.controllers.equipment.switch import Switch, SwitchSet
  6. from app.models.domain.devices import ACATAHSwitchSetRequest
  7. from app.schemas.equipment import AHU
  8. from app.services.platform import DataPlatformService
  9. class AHUSwitch(Switch):
  10. def __init__(self, equipment: AHU):
  11. super(AHUSwitch, self).__init__(equipment)
  12. def break_time_action(self, begin: str, end: str, is_workday: bool) -> SwitchSet:
  13. if self._equip.in_cloud_status:
  14. if is_workday:
  15. if begin and end:
  16. switch_flag = True
  17. if begin <= end:
  18. if begin <= self._now_time <= end:
  19. switch_flag = False
  20. else:
  21. if not end <= self._now_time <= begin:
  22. switch_flag = False
  23. if not switch_flag:
  24. action = SwitchSet.off
  25. else:
  26. action = SwitchSet.hold
  27. else:
  28. action = SwitchSet.hold
  29. else:
  30. action = SwitchSet.hold
  31. else:
  32. action = SwitchSet.hold
  33. return action
  34. @logger.catch()
  35. async def build_acatah_switch_set(params: ACATAHSwitchSetRequest) -> SwitchSet:
  36. ahu = AHU(
  37. running_status=params.running_status,
  38. in_cloud_status=params.in_cloud_status,
  39. on_time=params.on_time,
  40. off_time=params.off_time,
  41. )
  42. ahu_switch_controller = AHUSwitch(ahu)
  43. action = await ahu_switch_controller.build_next_action(params.is_workday)
  44. break_action = ahu_switch_controller.break_time_action(
  45. params.break_start_time, params.break_end_time, params.is_workday
  46. )
  47. if break_action == "off":
  48. action = SwitchSet.off
  49. return action