switch.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172
  1. # -*- coding: utf-8 -*-
  2. from typing import Dict, Tuple
  3. import arrow
  4. from httpx import AsyncClient
  5. from app.schemas.equipment import BaseEquipment
  6. from app.services.platform import DataPlatformService, InfoCode
  7. from app.services.transfer import Duoduo
  8. from app.utils.date import get_time_str, TIME_FMT
  9. class Switch:
  10. def __init__(self, equipment: BaseEquipment):
  11. super(Switch, self).__init__()
  12. self._equip = equipment
  13. self._now_time = arrow.get(get_time_str(), TIME_FMT).time().strftime('%H%M%S')
  14. async def build_next_action(self, day_type: str) -> str:
  15. if self._equip.in_cloud_status:
  16. if day_type == 'WeekDay':
  17. if self._equip.on_time <= self._now_time <= self._equip.off_time:
  18. switch_flag = True
  19. else:
  20. switch_flag = False
  21. if switch_flag and not self._equip.running_status:
  22. action = 'on'
  23. elif not switch_flag and self._equip.running_status:
  24. action = 'off'
  25. else:
  26. action = 'hold'
  27. else:
  28. action = 'hold'
  29. else:
  30. action = 'hold'
  31. return action
  32. async def fetch_data(project_id: str, equipment_id: str) -> Tuple[Dict, Dict]:
  33. async with AsyncClient() as client:
  34. platform = DataPlatformService(client, project_id)
  35. duo_duo = Duoduo(client, project_id)
  36. day_type = await duo_duo.get_day_type()
  37. running_status = await platform.get_realtime_running_status(equipment_id)
  38. cloud_status = await platform.get_cloud_status(equipment_id)
  39. on_time, off_time = await platform.get_schedule(equipment_id)
  40. equip_params = {
  41. 'running_status': running_status,
  42. 'in_cloud_status': True if cloud_status == 1.0 else False,
  43. 'on_time': on_time,
  44. 'off_time': off_time
  45. }
  46. return equip_params, day_type
  47. async def send_switch_command(project_id: str, equipment_id: str, action: str) -> None:
  48. async with AsyncClient() as client:
  49. platform = DataPlatformService(client, project_id)
  50. if action == 'on':
  51. await platform.set_code_value(equipment_id, InfoCode.equip_switch_set, 1.0)
  52. elif action == 'off':
  53. await platform.set_code_value(equipment_id, InfoCode.equip_switch_set, 0.0)
  54. else:
  55. pass