switch.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  1. # -*- coding: utf-8 -*-
  2. from typing import Tuple
  3. from httpx import AsyncClient
  4. from loguru import logger
  5. from app.controllers.equipment.switch import Switch, fetch_data, send_switch_command
  6. from app.models.domain.devices import ACATAHSwitchSetRequest
  7. from app.schemas.equipment import AHU
  8. from app.services.platform import DataPlatformService
  9. class AHUSwitch(Switch):
  10. def __init__(self, equipment: AHU):
  11. super(AHUSwitch, self).__init__(equipment)
  12. def break_time_action(self, begin: str, end: str, is_workday: bool) -> str:
  13. if self._equip.in_cloud_status:
  14. if is_workday:
  15. if begin and end:
  16. switch_flag = True
  17. if begin <= end:
  18. if begin <= self._now_time <= end:
  19. switch_flag = False
  20. else:
  21. if not end <= self._now_time <= begin:
  22. switch_flag = False
  23. if not switch_flag:
  24. action = 'off'
  25. else:
  26. action = 'hold'
  27. else:
  28. action = 'hold'
  29. else:
  30. action = 'hold'
  31. else:
  32. action = 'hold'
  33. return action
  34. async def fetch_break_time(project_id: str, device_id: str) -> Tuple[str, str]:
  35. async with AsyncClient() as client:
  36. platform = DataPlatformService(client, project_id)
  37. begin = await platform.get_static_info('ctm-BeginBreakTime', device_id)
  38. end = await platform.get_static_info('ctm-EndBreakTime', device_id)
  39. return begin, end
  40. @logger.catch()
  41. async def ahu_switch_control(project_id: str, equipment_id: str) -> None:
  42. equip_params, day_type = await fetch_data(project_id, equipment_id)
  43. is_workday = True if day_type.get('day_type') == 'WeekDay' else False
  44. begin, end = await fetch_break_time(project_id, equipment_id)
  45. switch_controller = AHUSwitch(AHU(**equip_params))
  46. action = await switch_controller.build_next_action(is_workday)
  47. break_action = switch_controller.break_time_action(begin, end, is_workday)
  48. if break_action == 'off':
  49. action = 'off'
  50. logger.debug(f'AHU-{equipment_id}: {action}')
  51. await send_switch_command(project_id, equipment_id, action)
  52. @logger.catch()
  53. async def build_acatah_switch_set(params: ACATAHSwitchSetRequest) -> str:
  54. ahu = AHU(
  55. running_status=params.running_status,
  56. in_cloud_status=params.in_cloud_status,
  57. on_time=params.on_time,
  58. off_time=params.off_time
  59. )
  60. ahu_switch_controller = AHUSwitch(ahu)
  61. action = await ahu_switch_controller.build_next_action(params.is_workday)
  62. break_action = ahu_switch_controller.break_time_action(
  63. params.break_start_time,
  64. params.break_end_time,
  65. params.is_workday
  66. )
  67. if break_action == 'off':
  68. action = 'off'
  69. return action