switch.py 2.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879
  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, is_workday: bool) -> str:
  15. if self._equip.in_cloud_status:
  16. if is_workday:
  17. if self._equip.on_time <= self._equip.off_time:
  18. if self._equip.on_time <= self._now_time <= self._equip.off_time:
  19. switch_flag = True
  20. else:
  21. switch_flag = False
  22. else:
  23. if self._equip.off_time <= self._now_time <= self._equip.on_time:
  24. switch_flag = False
  25. else:
  26. switch_flag = True
  27. if switch_flag and not self._equip.running_status:
  28. action = 'on'
  29. elif not switch_flag and self._equip.running_status:
  30. action = 'off'
  31. else:
  32. action = 'hold'
  33. else:
  34. action = 'hold'
  35. else:
  36. action = 'hold'
  37. return action
  38. async def fetch_data(project_id: str, equipment_id: str) -> Tuple[Dict, Dict]:
  39. async with AsyncClient() as client:
  40. platform = DataPlatformService(client, project_id)
  41. duo_duo = Duoduo(client, project_id)
  42. day_type = await duo_duo.get_day_type()
  43. running_status = await platform.get_realtime_running_status(equipment_id)
  44. cloud_status = await platform.get_cloud_status(equipment_id)
  45. on_time, off_time = await platform.get_schedule(equipment_id)
  46. equip_params = {
  47. 'id': equipment_id,
  48. 'running_status': running_status,
  49. 'in_cloud_status': True if cloud_status == 1.0 else False,
  50. 'on_time': on_time,
  51. 'off_time': off_time
  52. }
  53. return equip_params, day_type
  54. async def send_switch_command(project_id: str, equipment_id: str, action: str) -> None:
  55. async with AsyncClient() as client:
  56. platform = DataPlatformService(client, project_id)
  57. if action == 'on':
  58. await platform.set_code_value(equipment_id, InfoCode.equip_switch_set, 1.0)
  59. elif action == 'off':
  60. await platform.set_code_value(equipment_id, InfoCode.equip_switch_set, 0.0)
  61. else:
  62. pass