switch.py 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
  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, fetch_data, send_switch_command
  6. from app.schemas.equipment import AHU
  7. from app.services.platform import DataPlatformService
  8. class AHUSwitch(Switch):
  9. def __init__(self, equipment: AHU):
  10. super(AHUSwitch, self).__init__(equipment)
  11. def break_time_action(self, begin: str, end: str, day_type: str) -> str:
  12. if self._equip.in_cloud_status:
  13. if day_type == 'WeekDay':
  14. if begin and end:
  15. switch_flag = True
  16. if begin <= end:
  17. if begin <= self._now_time <= end:
  18. switch_flag = False
  19. else:
  20. if not end <= self._now_time <= begin:
  21. switch_flag = False
  22. if not switch_flag and self._equip.running_status:
  23. action = 'off'
  24. else:
  25. action = 'hold'
  26. else:
  27. action = 'hold'
  28. else:
  29. action = 'hold'
  30. else:
  31. action = 'hold'
  32. return action
  33. async def fetch_break_time(project_id: str, device_id: str) -> Tuple[str, str]:
  34. async with AsyncClient() as client:
  35. platform = DataPlatformService(client, project_id)
  36. begin = await platform.get_static_info('ctm-BeginBreakTime', device_id)
  37. end = await platform.get_static_info('ctm-EndBreakTime', device_id)
  38. return begin, end
  39. @logger.catch()
  40. async def ahu_switch_control(project_id: str, equipment_id: str) -> None:
  41. equip_params, day_type = await fetch_data(project_id, equipment_id)
  42. begin, end = await fetch_break_time(project_id, equipment_id)
  43. switch_controller = AHUSwitch(AHU(**equip_params))
  44. action = await switch_controller.build_next_action(day_type.get('day_type'))
  45. break_action = switch_controller.break_time_action(begin, end, day_type.get('day_type'))
  46. if break_action == 'off':
  47. action = 'off'
  48. logger.debug(f'AHU-{equipment_id}: {action}')
  49. await send_switch_command(project_id, equipment_id, action)