Explorar el Código

add basic switch logic

chenhaiyang hace 4 años
padre
commit
99bcfe770e
Se han modificado 1 ficheros con 72 adiciones y 0 borrados
  1. 72 0
      app/controllers/equipment/switch.py

+ 72 - 0
app/controllers/equipment/switch.py

@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+
+from typing import Dict, Tuple
+
+import arrow
+from httpx import AsyncClient
+
+from app.schemas.equipment import BaseEquipment
+from app.services.platform import DataPlatformService, InfoCode
+from app.services.transfer import Duoduo
+from app.utils.date import get_time_str, TIME_FMT
+
+
+class Switch:
+
+    def __init__(self, equipment: BaseEquipment):
+        super(Switch, self).__init__()
+        self._equip = equipment
+        self._now_time = arrow.get(get_time_str(), TIME_FMT).time().strftime('%H%M%S')
+
+    async def build_next_action(self, day_type: str) -> str:
+        if self._equip.in_cloud_status:
+            if day_type == 'WeekDay':
+                if self._equip.on_time <= self._now_time <= self._equip.off_time:
+                    switch_flag = True
+                else:
+                    switch_flag = False
+
+                if switch_flag and not self._equip.running_status:
+                    action = 'on'
+                elif not switch_flag and self._equip.running_status:
+                    action = 'off'
+                else:
+                    action = 'hold'
+            else:
+                action = 'hold'
+        else:
+            action = 'hold'
+
+        return action
+
+
+async def fetch_data(project_id: str, equipment_id: str) -> Tuple[Dict, Dict]:
+    async with AsyncClient() as client:
+        platform = DataPlatformService(client, project_id)
+        duo_duo = Duoduo(client, project_id)
+
+        day_type = await duo_duo.get_day_type()
+        running_status = await platform.get_realtime_running_status(equipment_id)
+        cloud_status = await platform.get_cloud_status(equipment_id)
+        on_time, off_time = await platform.get_schedule(equipment_id)
+
+    equip_params = {
+        'running_status': running_status,
+        'in_cloud_status': True if cloud_status == 1.0 else False,
+        'on_time': on_time,
+        'off_time': off_time
+    }
+
+    return equip_params, day_type
+
+
+async def send_switch_command(project_id: str, equipment_id: str, action: str) -> None:
+    async with AsyncClient() as client:
+        platform = DataPlatformService(client, project_id)
+
+        if action == 'on':
+            await platform.set_code_value(equipment_id, InfoCode.equip_switch_set, 1.0)
+        elif action == 'off':
+            await platform.set_code_value(equipment_id, InfoCode.equip_switch_set, 0.0)
+        else:
+            pass