# -*- coding: utf-8 -*- from typing import Tuple from httpx import AsyncClient from loguru import logger from app.controllers.equipment.switch import Switch, SwitchSet 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) -> SwitchSet: 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 = SwitchSet.off else: action = SwitchSet.hold else: action = SwitchSet.hold else: action = SwitchSet.hold else: action = SwitchSet.hold return action @logger.catch() async def build_acatah_switch_set(params: ACATAHSwitchSetRequest) -> SwitchSet: 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 = SwitchSet.off return action