123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687 |
- # -*- 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
|