# -*- coding: utf-8 -*- from typing import Tuple from httpx import AsyncClient from loguru import logger from app.controllers.equipment.switch import Switch, fetch_data, send_switch_command from app.schemas.equipment import AHU from app.services.platform import DataPlatformService class AHUSwitch(Switch): def __init__(self, equipment: AHU): super(AHUSwitch, self).__init__(equipment) def break_time_action(self, begin: str, end: str, day_type: str) -> str: if self._equip.in_cloud_status: if day_type == 'WeekDay': if begin and end: switch_flag = True if begin <= end: if begin <= self._now_time <= end: switch_flag = False else: if not end <= self._now_time <= begin: switch_flag = False if not switch_flag and self._equip.running_status: action = 'off' else: action = 'hold' else: action = 'hold' else: action = 'hold' else: action = 'hold' return action async def fetch_break_time(project_id: str, device_id: str) -> Tuple[str, str]: async with AsyncClient() as client: platform = DataPlatformService(client, project_id) begin = await platform.get_static_info('ctm-BeginBreakTime', device_id) end = await platform.get_static_info('ctm-EndBreakTime', device_id) return begin, end @logger.catch() async def ahu_switch_control(project_id: str, equipment_id: str) -> None: equip_params, day_type = await fetch_data(project_id, equipment_id) begin, end = await fetch_break_time(project_id, equipment_id) switch_controller = AHUSwitch(AHU(**equip_params)) action = await switch_controller.build_next_action(day_type.get('day_type')) break_action = switch_controller.break_time_action(begin, end, day_type.get('day_type')) if break_action == 'off': action = 'off' logger.debug(f'AHU-{equipment_id}: {action}') await send_switch_command(project_id, equipment_id, action)