switch.py 1.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546
  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(
  13. day_type.get("day_type")
  14. )
  15. logger.debug(f"PAU-{equipment_id}: {action}")
  16. await send_switch_command(project_id, equipment_id, action)
  17. @logger.catch()
  18. async def get_switch_action(project_id: str, device_id: str) -> str:
  19. device_params, day_type = await fetch_data(project_id, device_id)
  20. is_workday = True if day_type.get("day_type") == "WeekDay" else False
  21. action = await PAUSwitch(PAU(**device_params)).build_next_action(is_workday)
  22. return action
  23. @logger.catch()
  24. async def build_acatfu_switch_set(params: ACATFUSwitchSetRequest) -> str:
  25. pau = PAU(
  26. running_status=params.running_status,
  27. in_cloud_status=params.in_cloud_status,
  28. on_time=params.on_time,
  29. off_time=params.off_time,
  30. )
  31. action = await PAUSwitch(pau).build_next_action(params.is_workday)
  32. if not params.is_workday:
  33. action = "off"
  34. return action