switch.py 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445
  1. # -*- coding: utf-8 -*-
  2. from loguru import logger
  3. from app.controllers.equipment.switch import Switch, fetch_data, send_switch_command
  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. @logger.catch()
  10. async def pau_switch_control(project_id: str, equipment_id: str) -> None:
  11. equip_params, day_type = await fetch_data(project_id, equipment_id)
  12. action = await PAUSwitch(PAU(**equip_params)).build_next_action(day_type.get('day_type'))
  13. logger.debug(f'PAU-{equipment_id}: {action}')
  14. await send_switch_command(project_id, equipment_id, action)
  15. @logger.catch()
  16. async def get_switch_action(project_id: str, device_id: str) -> str:
  17. device_params, day_type = await fetch_data(project_id, device_id)
  18. is_workday = True if day_type.get('day_type') == 'WeekDay' else False
  19. action = await PAUSwitch(PAU(**device_params)).build_next_action(is_workday)
  20. return action
  21. @logger.catch()
  22. async def build_acatfu_switch_set(params: ACATFUSwitchSetRequest) -> str:
  23. pau = PAU(
  24. running_status=params.running_status,
  25. in_cloud_status=params.in_cloud_status,
  26. on_time=params.on_time,
  27. off_time=params.off_time
  28. )
  29. action = await PAUSwitch(pau).build_next_action(params.is_workday)
  30. if not params.is_workday:
  31. action = 'off'
  32. return action