switch.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. # -*- coding: utf-8 -*-
  2. from loguru import logger
  3. from app.models.domain.devices import ACVTSFSwitchSetRequest
  4. from app.controllers.equipment.switch import Switch, fetch_data, send_switch_command
  5. from app.schemas.equipment import VentilationFan
  6. class VentilationFanSwitch(Switch):
  7. def __init__(self, equipment: VentilationFan):
  8. super(VentilationFanSwitch, self).__init__(equipment)
  9. @logger.catch()
  10. async def ventilation_fan_switch_control(project_id: str, equipment_id: str) -> None:
  11. equip_params, day_type = await fetch_data(project_id, equipment_id)
  12. is_workday = True if day_type.get("day_type") == "WeekDay" else False
  13. action = await VentilationFanSwitch(
  14. VentilationFan(**equip_params)
  15. ).build_next_action(is_workday)
  16. if not is_workday:
  17. action = "off"
  18. logger.debug(f"VTSF-{equipment_id}: {action}")
  19. await send_switch_command(project_id, equipment_id, action)
  20. @logger.catch()
  21. async def build_acvtsf_switch_set(params: ACVTSFSwitchSetRequest) -> str:
  22. vrf = VentilationFan(
  23. running_status=params.running_status,
  24. in_cloud_status=params.in_cloud_status,
  25. on_time=params.on_time,
  26. off_time=params.off_time,
  27. )
  28. action = await VentilationFanSwitch(vrf).build_next_action(params.is_workday)
  29. if not params.is_workday:
  30. action = "off"
  31. return action