# -*- 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.models.domain.devices import ACATAHSwitchSetRequest 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, is_workday: bool) -> str: if self._equip.in_cloud_status: if is_workday: 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: 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) is_workday = True if day_type.get('day_type') == 'WeekDay' else False begin, end = await fetch_break_time(project_id, equipment_id) switch_controller = AHUSwitch(AHU(**equip_params)) action = await switch_controller.build_next_action(is_workday) break_action = switch_controller.break_time_action(begin, end, is_workday) if break_action == 'off': action = 'off' logger.debug(f'AHU-{equipment_id}: {action}') await send_switch_command(project_id, equipment_id, action) @logger.catch() async def build_acatah_switch_set(params: ACATAHSwitchSetRequest) -> str: ahu = AHU( running_status=params.running_status, in_cloud_status=params.in_cloud_status, on_time=params.on_time, off_time=params.off_time ) ahu_switch_controller = AHUSwitch(ahu) action = await ahu_switch_controller.build_next_action(params.is_workday) break_action = ahu_switch_controller.break_time_action( params.break_start_time, params.break_end_time, params.is_workday ) if break_action == 'off': action = 'off' return action