import logging import datetime import homeassistant.util.dt as dt_util from homeassistant.const import ( WEEKDAYS, STATE_ON, STATE_OFF, ) from homeassistant.core import ( HomeAssistant, callback, ) from homeassistant.helpers.event import ( async_track_point_in_time, async_track_state_change_event, ) from homeassistant.helpers.dispatcher import ( async_dispatcher_connect, async_dispatcher_send, ) from . import const from .store import async_get_registry _LOGGER = logging.getLogger(__name__) ATTR_NEXT_RISING = "next_rising" ATTR_NEXT_SETTING = "next_setting" ATTR_WORKDAYS = "workdays" def has_sun(time_str: str): return const.OffsetTimePattern.match(time_str) def is_same_day(dateA: datetime.datetime, dateB: datetime.datetime): return dateA.date() == dateB.date() def days_until_date(date_string: str, ts: datetime.datetime): date = dt_util.parse_date(date_string) diff = date - ts.date() return diff.days def find_closest_from_now(date_arr: list): now = dt_util.as_local(dt_util.utcnow()) minimum = None for item in date_arr: if item is not None: if minimum is None: minimum = item elif item > now: if item < minimum or minimum < now: minimum = item else: if item < minimum and minimum < now: minimum = item return minimum class TimerHandler: def __init__(self, hass: HomeAssistant, id: str): """init""" self.hass = hass self.id = id self._weekdays = [] self._start_date = None self._end_date = None self._timeslots = [] self._timer = None self._next_trigger = None self._next_slot = None self._sun_tracker = None self._workday_tracker = None self._watched_times = [] self.slot_queue = [] self.timestamps = [] self.current_slot = None self.hass.loop.create_task(self.async_reload_data()) @callback async def async_item_updated(id: str): if id == self.id: await self.async_reload_data() self._update_listener = async_dispatcher_connect( self.hass, const.EVENT_ITEM_UPDATED, async_item_updated ) async def async_reload_data(self): """load schedule data into timer class object and start timer""" store = await async_get_registry(self.hass) data = store.async_get_schedule(self.id) self._weekdays = data[const.ATTR_WEEKDAYS] self._start_date = data[const.ATTR_START_DATE] self._end_date = data[const.ATTR_END_DATE] self._timeslots = [ dict((k, slot[k]) for k in [const.ATTR_START, const.ATTR_STOP] if k in slot) for slot in data[const.ATTR_TIMESLOTS] ] await self.async_start_timer() async def async_unload(self): """unload a timer class object""" await self.async_stop_timer() self._update_listener() self._next_trigger = None async def async_start_timer(self): [current_slot, timestamp_end] = self.current_timeslot() [next_slot, timestamp_next] = self.next_timeslot() self._watched_times = [] if timestamp_next is not None: self._watched_times.append(self._timeslots[next_slot][const.ATTR_START]) if timestamp_end is not None: self._watched_times.append(self._timeslots[current_slot][const.ATTR_STOP]) # the next trigger time is next slot or end of current slot (whichever comes first) timestamp = find_closest_from_now([timestamp_end, timestamp_next]) self._timer_is_endpoint = ( timestamp != timestamp_next and timestamp == timestamp_end ) if timestamp == timestamp_next and timestamp is not None: self._next_slot = next_slot else: self._next_slot = None self.current_slot = current_slot self._next_trigger = timestamp await self.async_start_sun_tracker() now = dt_util.as_local(dt_util.utcnow()) if timestamp is not None: if self._timer: self._timer() if (timestamp - now).total_seconds() < 0: self._timer = None _LOGGER.debug( "Timer of {} is not set because it is in the past ({})".format( self.id, timestamp ) ) else: self._timer = async_track_point_in_time( self.hass, self.async_timer_finished, timestamp ) _LOGGER.debug("Timer of {} set for {}".format(self.id, timestamp)) await self.async_start_workday_tracker() async_dispatcher_send(self.hass, const.EVENT_TIMER_UPDATED, self.id) async def async_stop_timer(self): """stop the timer""" if self._timer: self._timer() self._timer = None await self.async_stop_sun_tracker() await self.async_stop_workday_tracker() async def async_start_sun_tracker(self): """check for changes in the sun sensor""" if ( self._next_trigger is not None and any(has_sun(x) for x in self._watched_times) ) or ( self._next_trigger is None and all(has_sun(x[const.ATTR_START]) for x in self._timeslots) ): # install sun tracker for updating timer when sun changes # initially the time calculation may fail due to the sun entity being unavailable if self._sun_tracker is not None: # the tracker is already running return @callback async def async_sun_updated(_event): """the sun entity was updated""" # sun entity changed if self._next_trigger is None: # sun entity has initialized await self.async_start_timer() return ts = find_closest_from_now( self.calculate_timestamp(x) for x in self._watched_times ) if not ts or not self._next_trigger: # sun entity became unavailable (or other corner case) await self.async_start_timer() return # we are re-scheduling an existing timer delta = (ts - self._next_trigger).total_seconds() if abs(delta) >= 60 and abs(delta) < 2000: # only reschedule if the difference is at least a minute # only reschedule if this doesnt cause the timer to shift to another day (+/- 24 hrs delta) # only reschedule if this doesnt cause the timer to shift to another hour (due to DST change) await self.async_start_timer() self._sun_tracker = async_track_state_change_event( self.hass, const.SUN_ENTITY, async_sun_updated ) else: # clear existing tracker await self.async_stop_sun_tracker() async def async_stop_sun_tracker(self): """stop checking for changes in the sun sensor""" if self._sun_tracker: self._sun_tracker() self._sun_tracker = None async def async_start_workday_tracker(self): """check for changes in the workday sensor""" if ( const.DAY_TYPE_WORKDAY in self._weekdays or const.DAY_TYPE_WEEKEND in self._weekdays ): # install tracker for updating timer when workday sensor changes if self._workday_tracker is not None: # the tracker is already running return @callback async def async_workday_updated(): """the workday sensor was updated""" [current_slot, timestamp_end] = self.current_timeslot() [next_slot, timestamp_next] = self.next_timeslot() ts_next = find_closest_from_now([timestamp_end, timestamp_next]) # workday entity changed if not ts_next or not self._next_trigger: # timer was not yet set await self.async_start_timer() else: # we are re-scheduling an existing timer delta = (ts_next - self._next_trigger).total_seconds() if abs(delta) >= 60: # only reschedule if the difference is at least a minute await self.async_start_timer() self._workday_tracker = async_dispatcher_connect( self.hass, const.EVENT_WORKDAY_SENSOR_UPDATED, async_workday_updated ) else: # clear existing tracker await self.async_stop_workday_tracker() async def async_stop_workday_tracker(self): """stop checking for changes in the workday sensor""" if self._workday_tracker: self._workday_tracker() self._workday_tracker = None async def async_timer_finished(self, _time): """the timer is finished""" if not self._timer_is_endpoint: # timer marks the start of a new timeslot self.current_slot = self._next_slot _LOGGER.debug( "Timer {} has reached slot {}".format(self.id, self.current_slot) ) async_dispatcher_send(self.hass, const.EVENT_TIMER_FINISHED, self.id) # don't automatically reset, wait for external reset after 1 minute # await self.async_start_timer() await self.async_stop_timer() else: # timer marks the end of a timeslot _LOGGER.debug( "Timer {} has reached end of timeslot, resetting..".format(self.id) ) await self.async_start_timer() def day_in_weekdays(self, ts: datetime.datetime) -> bool: """check if the day of a datetime object is in the allowed list of days""" day = WEEKDAYS[ts.weekday()] workday_sensor = self.hass.states.get(const.WORKDAY_ENTITY) if ( workday_sensor and workday_sensor.state in [STATE_ON, STATE_OFF] and is_same_day(ts, dt_util.as_local(dt_util.utcnow())) ): # state of workday sensor is used for evaluating workday vs weekend if const.DAY_TYPE_WORKDAY in self._weekdays: return workday_sensor.state == STATE_ON elif const.DAY_TYPE_WEEKEND in self._weekdays: return workday_sensor.state == STATE_OFF if workday_sensor and ATTR_WORKDAYS in workday_sensor.attributes: # workday sensor defines a list of workdays workday_list = workday_sensor.attributes[ATTR_WORKDAYS] weekend_list = [e for e in WEEKDAYS if e not in workday_list] else: # assume workdays are mon-fri workday_list = WEEKDAYS[0:5] weekend_list = WEEKDAYS[5:7] if const.DAY_TYPE_DAILY in self._weekdays or not len(self._weekdays): return True elif const.DAY_TYPE_WORKDAY in self._weekdays and day in workday_list: return True elif const.DAY_TYPE_WEEKEND in self._weekdays and day in weekend_list: return True return day in self._weekdays def calculate_timestamp( self, time_str, now: datetime.datetime = None, iteration: int = 0, reverse_direction: bool = False, ) -> datetime.datetime: """calculate the next occurence of a time string""" if time_str is None: return None if now is None: now = dt_util.as_local(dt_util.utcnow()) res = has_sun(time_str) if not res: # fixed time time = dt_util.parse_time(time_str) ts = dt_util.find_next_time_expression_time( now, [time.second], [time.minute], [time.hour] ) else: # relative to sunrise/sunset sun = self.hass.states.get(const.SUN_ENTITY) if not sun: return None ts = None if ( res.group(1) == const.SUN_EVENT_SUNRISE and ATTR_NEXT_RISING in sun.attributes ): ts = dt_util.parse_datetime(sun.attributes[ATTR_NEXT_RISING]) elif ( res.group(1) == const.SUN_EVENT_SUNSET and ATTR_NEXT_SETTING in sun.attributes ): ts = dt_util.parse_datetime(sun.attributes[ATTR_NEXT_SETTING]) if not ts: return None ts = dt_util.as_local(ts) ts = ts.replace(second=0) time_sun = datetime.timedelta( hours=ts.hour, minutes=ts.minute, seconds=ts.second ) offset = dt_util.parse_time(res.group(3)) offset = datetime.timedelta( hours=offset.hour, minutes=offset.minute, seconds=offset.second ) if res.group(2) == "-": if (time_sun - offset).total_seconds() >= 0: ts = ts - offset else: # prevent offset to shift the time past the extends of the day ts = ts.replace(hour=0, minute=0, second=0) else: if (time_sun + offset).total_seconds() <= 86340: ts = ts + offset else: # prevent offset to shift the time past the extends of the day ts = ts.replace(hour=23, minute=59, second=0) ts = dt_util.find_next_time_expression_time( now, [ts.second], [ts.minute], [ts.hour] ) time_delta = datetime.timedelta(seconds=1) if self.day_in_weekdays(ts) and ( (ts - now).total_seconds() > 0 or iteration > 0 ): if self._start_date and days_until_date(self._start_date, ts) > 0: # start date is in the future, jump to start date end_of_day = ts.replace(hour=0, minute=0, second=0, microsecond=0) + datetime.timedelta(days=1) days_delta = days_until_date(self._start_date, end_of_day) if days_delta: time_delta = datetime.timedelta(days=days_delta) elif self._end_date and days_until_date(self._end_date, ts) < 0: # end date is in the past, jump to end date time_delta = datetime.timedelta( days=days_until_date(self._end_date, ts) ) reverse_direction = True else: # date restrictions are met return ts elif reverse_direction: time_delta = datetime.timedelta(days=-1) # calculate next timestamp next_day = dt_util.find_next_time_expression_time( now + time_delta, [0], [0], [0] ) if iteration > 15: _LOGGER.warning( "failed to calculate next timeslot for schedule {}".format(self.id) ) return None return self.calculate_timestamp( time_str, next_day, iteration + 1, reverse_direction ) def next_timeslot(self): """calculate the closest timeslot from now""" now = dt_util.as_local(dt_util.utcnow()) # calculate next start of all timeslots timestamps = [ self.calculate_timestamp(slot[const.ATTR_START], now) for slot in self._timeslots ] # calculate timeslot that will start soonest (or closest in the past) remaining = [ abs((ts - now).total_seconds()) if ts is not None else now.timestamp() for ts in timestamps ] slot_order = sorted(range(len(remaining)), key=lambda k: remaining[k]) # filter out timeslots that cannot be computed for i in range(len(timestamps)): if timestamps[i] is None: slot_order.remove(i) timestamps = [e for e in timestamps if e is not None] self.slot_queue = slot_order self.timestamps = timestamps next_slot = slot_order[0] if len(slot_order) > 0 else None return (next_slot, timestamps[next_slot] if next_slot is not None else None) def current_timeslot(self, now: datetime.datetime = None): """calculate the end of the timeslot that is overlapping now""" if now is None: now = dt_util.as_local(dt_util.utcnow()) def unwrap_end_of_day(time_str: str): if time_str == "00:00:00": return "23:59:59" else: return time_str # calculate next stop of all timeslots timestamps = [] for slot in self._timeslots: if slot[const.ATTR_STOP] is not None: timestamps.append( self.calculate_timestamp( unwrap_end_of_day(slot[const.ATTR_STOP]), now ) ) else: ts = self.calculate_timestamp(slot[const.ATTR_START], now) if ts is None: timestamps.append(None) else: ts = ts + datetime.timedelta(minutes=1) timestamps.append( self.calculate_timestamp(ts.strftime("%H:%M:%S"), now) ) # calculate timeslot that will end soonest remaining = [ (ts - now).total_seconds() if ts is not None else now.timestamp() for ts in timestamps ] (next_slot_end, val) = sorted( enumerate(remaining), key=lambda i: (i[1] < 0, abs(i[1])) )[0] stop = timestamps[next_slot_end] if stop is not None: # calculate last start of timeslot that will end soonest if (stop - now).total_seconds() < 0: # end of timeslot is in the past return (None, None) start = self.calculate_timestamp( self._timeslots[next_slot_end][const.ATTR_START], stop - datetime.timedelta(days=1), ) if start is not None: elapsed = (now - start).total_seconds() if elapsed > 0: # timeslot is currently overlapping return ( next_slot_end, stop if self._timeslots[next_slot_end][const.ATTR_STOP] is not None else None, ) return (None, None)