switch.py 2.8 KB

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