switch.py 1.8 KB

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