12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485 |
- from enum import Enum
- 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 SwitchSet(str, Enum):
- on = "on"
- off = "off"
- hold = "hold"
- 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, is_workday: bool) -> SwitchSet:
- if self._equip.in_cloud_status:
- if is_workday:
- if self._equip.on_time <= self._equip.off_time:
- if self._equip.on_time <= self._now_time <= self._equip.off_time:
- switch_flag = True
- else:
- switch_flag = False
- else:
- if self._equip.off_time <= self._now_time <= self._equip.on_time:
- switch_flag = False
- else:
- switch_flag = True
- if switch_flag and not self._equip.running_status:
- action = SwitchSet.on
- elif not switch_flag and self._equip.running_status:
- action = SwitchSet.off
- else:
- action = SwitchSet.hold
- else:
- action = SwitchSet.hold
- else:
- action = SwitchSet.off
- 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 = {
- "id": equipment_id,
- "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
|