# -*- 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._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 = '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 = { '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