123456789101112131415161718192021222324252627282930313233343536373839404142434445 |
- # -*- coding: utf-8 -*-
- from loguru import logger
- from app.controllers.equipment.switch import Switch, fetch_data, send_switch_command
- from app.models.domain.devices import ACATFUSwitchSetRequest
- from app.schemas.equipment import PAU
- class PAUSwitch(Switch):
- def __init__(self, equipment: PAU):
- super(PAUSwitch, self).__init__(equipment)
- @logger.catch()
- async def pau_switch_control(project_id: str, equipment_id: str) -> None:
- equip_params, day_type = await fetch_data(project_id, equipment_id)
- action = await PAUSwitch(PAU(**equip_params)).build_next_action(day_type.get('day_type'))
- logger.debug(f'PAU-{equipment_id}: {action}')
- await send_switch_command(project_id, equipment_id, action)
- @logger.catch()
- async def get_switch_action(project_id: str, device_id: str) -> str:
- device_params, day_type = await fetch_data(project_id, device_id)
- is_workday = True if day_type.get('day_type') == 'WeekDay' else False
- action = await PAUSwitch(PAU(**device_params)).build_next_action(is_workday)
- return action
- @logger.catch()
- async def build_acatfu_switch_set(params: ACATFUSwitchSetRequest) -> str:
- pau = PAU(
- running_status=params.running_status,
- in_cloud_status=params.in_cloud_status,
- on_time=params.on_time,
- off_time=params.off_time
- )
- action = await PAUSwitch(pau).build_next_action(params.is_workday)
- if not params.is_workday:
- action = 'off'
- return action
|