Source code for aioswitcher.schedules

"""Switcher water heater unofficial API and bridge, Schedules.

.. codeauthor:: Tomer Figenblat <>


# fmt: off
from asyncio import AbstractEventLoop, Future, ensure_future
from binascii import unhexlify
from datetime import datetime
from typing import List

                     SUNDAY, WAITING_TEXT, WEEKDAY_TUP)
from .tools import get_days_list_from_bytes, get_time_from_bytes

# fmt: on

[docs]class SwitcherV2Schedule: """Represnation of the SwitcherV2 schedule slot. Args: loop: the event loop to perform schedule operation in. idx: the index of the schedule slot (0-7). schedule_details: the string raw schedule data details. Todo: * Replace ``init_future`` attribute with ``get_init_future`` method. """ def __init__( self, loop: AbstractEventLoop, idx: int, schedule_details: List[bytes] ) -> None: """Initialize the schedule.""" self._loop = loop self._enabled = False self._recurring = False self._schedule_id = str(int(schedule_details[idx][0:2], 16)) self._days = [] # type: List[str] self._schedule_data = b"" self._start_time = WAITING_TEXT self._end_time = WAITING_TEXT self._duration = WAITING_TEXT self._init_future = loop.create_future() ensure_future(self.initialize(idx, schedule_details), loop=loop)
[docs] async def initialize( self, idx: int, schedule_details: List[bytes] ) -> None: """Finish the initialization of the schedule.""" try: if int(schedule_details[idx][2:4], 16) == 1: self._enabled = True if not schedule_details[idx][4:6] == "00": self._recurring = True if schedule_details[idx][4:6] == "fe": self._days.append(ALL_DAYS) else: self._days = await get_days_list_from_bytes( self._loop, bytearray(unhexlify((schedule_details[idx][4:6])))[0], ) self._start_time = await get_time_from_bytes( self._loop, schedule_details[idx][8:16] ) self._end_time = await get_time_from_bytes( self._loop, schedule_details[idx][16:24] ) self._duration = str( datetime.strptime(self._end_time, "%H:%M") - datetime.strptime(self._start_time, "%H:%M") ) time_id = schedule_details[idx][0:2] on_off = schedule_details[idx][2:4] week = schedule_details[idx][4:6] timestate = schedule_details[idx][6:8] start_time = schedule_details[idx][8:16] end_time = schedule_details[idx][16:24] self._schedule_data = ( time_id + on_off + week + timestate + start_time + end_time ) self.init_future.set_result(self) except RuntimeError as exc: self.init_future.set_exception(exc) return None
@property def schedule_id(self) -> str: """str: Return the schedule id.""" return self._schedule_id @property def enabled(self) -> bool: """bool: Return true if enabled, setter included.""" return self._enabled @enabled.setter def enabled(self, value: bool) -> None: """Setter to set the device as enabled or disabled.""" if isinstance(value, bool): self._enabled = value else: raise TypeError("enabled is of type bool") @property def recurring(self) -> bool: """bool: Return true if recurring.""" return self._recurring @property def days(self) -> List[str]: """list(str): Return the weekdays of the schedule.""" return self._days @property def start_time(self) -> str: """str: Return the start time of the schedule.""" return self._start_time @property def end_time(self) -> str: """str: Return the end time of the schedule.""" return self._end_time @property def duration(self) -> str: """str: Return the duration of the schedule.""" return self._duration @property def schedule_data(self) -> bytes: """bytes: Return the schedule data, setter included.""" return self._schedule_data @schedule_data.setter def schedule_data(self, data: bytes) -> None: """Setter to set the schedule data for managing the schedule.""" if isinstance(data, bytes): self._schedule_data = data else: raise TypeError("schedeule_data is of type bytes") @property def init_future(self) -> Future: """asyncio.Future: Return the future of the initialization.""" return self._init_future
[docs] def as_dict(self): """Return as dict. Returns: A dictionary represntation of the object properties. Used to make the object json serializable. """ return self.__dict__
def _calc_next_run_for_schedule(schedule_details: SwitcherV2Schedule) -> str: """Calculate the next runtime of the schedule. Args: schedule_details: ``SwitcherV2Schedule`` representing the schedule slot. Returns: A pretty string describing the next due run. e.g. "Due tommorow at 17:00". Note: This is a private function containing blocking code. Please consider using ``calc_next_run_for_schedule`` (without the `_`), to schedule as a task on the event loop. """ if schedule_details.recurring: today_datetime = start_time = datetime.strptime(schedule_details.start_time, "%H:%M") current_time = datetime.strptime( ("0" + str(today_datetime.hour))[-2:] + ":" + ("0" + str(today_datetime.minute))[-2:], "%H:%M", ) current_weekday = today_datetime.weekday() found_day = -1 if schedule_details.days == [ALL_DAYS]: if current_time < start_time: return SCHEDULE_DUE_TODAY_FORMAT.format( schedule_details.start_time ) return SCHEDULE_DUE_TOMMOROW_FORMAT.format( schedule_details.start_time ) for day in schedule_details.days: set_weekday = WEEKDAY_TUP.index(day) if set_weekday == current_weekday and current_time < start_time: return SCHEDULE_DUE_TODAY_FORMAT.format( schedule_details.start_time ) if found_day == -1 or found_day > set_weekday: found_day = set_weekday if found_day - 1 == current_weekday or ( found_day == WEEKDAY_TUP.index(MONDAY) and current_weekday == WEEKDAY_TUP.index(SUNDAY) ): return SCHEDULE_DUE_TOMMOROW_FORMAT.format( schedule_details.start_time ) return SCHEDULE_DUE_ANOTHER_DAY_FORMAT.format( WEEKDAY_TUP[found_day], schedule_details.start_time ) return SCHEDULE_DUE_TODAY_FORMAT.format(schedule_details.start_time) async def calc_next_run_for_schedule( loop: AbstractEventLoop, schedule_details: SwitcherV2Schedule ) -> str: """Asynchronous wrapper for _calc_next_run_for_schedule. Use as async wrapper for calling _calc_next_run_for_schedule, calculating the next runtime of the schedule. Args: loop: the event loop to execute the function in. schedule_details: ``SwitcherV2Schedule`` representing the schedule slot. Returns: A pretty string describing the next due run. e.g. "Due tommorow at 17:00". """ return await loop.run_in_executor( None, _calc_next_run_for_schedule, schedule_details )