# -*- coding: utf-8 -*- 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