import appdaemon.plugins.hass.hassapi as hass from ad_toolbox.smartobject import SmartObject import ad_toolbox.smartcondition as SmartCondition import datetime from threading import Lock from requests import get import requests import icalendar from enum import Enum class EventType(Enum): Vacation = 1 Flex = 2 Busy = 3 Tentative = 4 class CalendarEvent: def __init__(self,start_time,stop_time,event_type): self.start_time = start_time self.stop_time = stop_time self.type = event_type def __str__(self): return f"[{self.type}]Start at {self.start_time},Finish at {self.stop_time}" def __eq__(self, obj): return isinstance(obj, CalendarEvent) and obj.type == self.type and obj.start_time == self.start_time and obj.stop_time == self.stop_time def __ne__(self, obj): return not self == obj class AlarmClock(SmartObject): def on_initialize_smart_object(self): self.switch_alarmclock_main = self.args['switch_alarmclock_main'] self.switch_alarmclock_secondary = self.args['switch_alarmclock_secondary'] self.inputdate_alarmclock_main = self.args['inputdate_alarmclock_main'] self.inputdate_alarmclock_secondary = self.args['inputdate_alarmclock_secondary'] self.input_text_wakeup_time = self.args['input_text_wakeup_time'] #self.sensor_wakeup_datetime = self.args['sensor_wakeup_datetime'] self.sleeping_condition = SmartCondition.Evaluator(self, self.args['sleeping_condition'], condition_name="sleeping_condition", constants=self.constants, templates_library=self.templates_library) self.lock = Lock() self.wakeup_time_str = "" self.handle_program_next_wakueup_cb = None self.handle_prewakueup_cb = None self.handle_wakueup_cb = None self.handle_switch_reveil_principal = self.listen_state(self.on_alarm_clock_set, self.switch_alarmclock_main) self.handle_switch_reveil_secondaire = self.listen_state(self.on_alarm_clock_set, self.switch_alarmclock_secondary) self.listen_state(self.on_alarm_clock_set, self.inputdate_alarmclock_main) self.listen_state(self.on_alarm_clock_set, self.inputdate_alarmclock_secondary) self.run_hourly(self.on_refresh_cal,datetime.time(0, 0, 0)) self.sensor_wakeup_datetime_entity = self.create_entity("sensor.wakeup_datetime", icon="mdi:alarm") #sensor_wakeup_datetime_entity = self.get_entity(self.sensor_wakeup_datetime) #if not sensor_wakeup_datetime_entity.exists(): # sensor_wakeup_datetime_entity.add() self.calendar = list() self.refresh_cal() self.update_wakeup_time() def refresh_cal(self): new_cal = list() #if self.load_cal_from_google_calendar(new_cal): if self.load_cal_from_ics(new_cal): if not self.are_cal_identical(self.calendar,new_cal): self.calendar = new_cal self.log("Calendar has been updated!") self.print_cal(self.calendar) return True return False def are_cal_identical(self,calendar1,calendar2): if len(calendar1) != len(calendar2): return False for i, event in enumerate(calendar1): if calendar2[i] != event: return False return True def print_cal(self,calendar): for event in calendar: self.log(f"{event}") def load_cal_from_ics(self,calendar): calendar.clear() try: self.log(f"Retrieving calengar from outlook.office365.com") r = get("https://outlook.office365.com/owa/calendar/0c64359c70eb41cd9713fe92c470d3df@ubisoft.com/5edb3fb5150f4203b66d9cf8db3184be11626033572842691749/calendar.ics", timeout=10)#, headers=headers, verify=False) #self.log(f"Processing Calendar") today = datetime.datetime.combine(datetime.date.today(), datetime.time(0,0, 0)) start_load_date = today - datetime.timedelta(days=1) end_load_date = today + datetime.timedelta(days=2) #self.log("[" + str(start_load_date) + "]-[" + str(end_load_date) + "]") calendar_ubi = icalendar.Calendar.from_ical(r.text) for component in calendar_ubi.walk(): if component.name == "VEVENT": event_summary = component.get('summary') if event_summary == "Away": event_type = EventType.Vacation elif event_summary == "Busy": event_type = EventType.Busy elif event_summary == "Tentative": event_type = EventType.Tentative elif event_summary == "Working elsewhere" or event_summary == "Free": event_type = EventType.Flex else: event_type = None start_datetime = component.get('dtstart').dt end_datetime = component.get('dtend').dt #regular event have a datetime and full day events only a date #let's make both event a time zone unaware datetime if isinstance(start_datetime,datetime.datetime): start_datetime = start_datetime.replace(tzinfo=None) else: start_datetime = datetime.datetime.combine(start_datetime,datetime.time(0,0,0)) if isinstance(end_datetime,datetime.datetime): end_datetime = end_datetime.replace(tzinfo=None) else: end_datetime = datetime.datetime.combine(end_datetime,datetime.time(0,0,0)) outside_of_loading_period = False if start_datetime < start_load_date and end_datetime < start_load_date: outside_of_loading_period = True if start_datetime > end_load_date and end_datetime > end_load_date: outside_of_loading_period = True if not outside_of_loading_period: if event_type: calendar.append(CalendarEvent(start_datetime,end_datetime,event_type)) else: self.log(f"[Unknow event : {event_summary}]Start at {start_datetime},Finish at {end_datetime}",level = "WARNING") except ValueError: self.log_error("Couldn't load calendar") return False except (requests.exceptions.ConnectionError,requests.exceptions.ReadTimeout): self.log("Couldn't load calendar, request Timeout",level = "WARNING") return False #self.log(f"Calendar loaded") return True def on_alarm_clock_set(self, entity, attribute, old, new, kwargs): self.update_wakeup_time() def on_refresh_cal(self, kwargs): if self.refresh_cal(): self.update_wakeup_time() def on_program_next_wakeup(self, kwargs): self.handle_program_next_wakueup_cb = None no_switch_changed = True #le reveil principal se reactive chaque matin if self.get_state(self.switch_alarmclock_main) == 'off': self.log("Turning on main alarm clock") no_switch_changed = False self.turn_on(self.switch_alarmclock_main) #le reveil secondaire ne sonne qu'une fois, il est donc desactivé chaque matin if self.get_state(self.switch_alarmclock_secondary) == 'on': self.log("Turning off secondary alarm clock") no_switch_changed = False self.turn_off(self.switch_alarmclock_secondary) if no_switch_changed: self.update_wakeup_time() def on_pre_wakeup(self, kwargs): self.handle_prewakueup_cb = None for key, value in kwargs.items(): if key == 'wakeup_in' : wakeup_in = value if wakeup_in > 0: self.handle_prewakueup_cb = self.run_in(self.on_pre_wakeup, 5 * 60,wakeup_in = wakeup_in - 5) wakeup_message = "On se reveil dans " + str(wakeup_in) + " minutes" self.log(wakeup_message) if self.sleeping_condition.evaluate() == SmartCondition.Result.Succeeded: self.fire_event("WAKEUP_IN", wakeup_in = wakeup_in) #self.call_service("notify/ios_iphone_de_pierre", title = "Reveil", message = wakeup_message) def on_wakeup(self, kwargs): self.handle_wakueup_cb = None wakeup_message = "WAKE UP!!!! It's " + self.wakeup_time_str self.log(wakeup_message) if self.sleeping_condition.evaluate() == SmartCondition.Result.Succeeded: self.fire_event("WAKEUP") #self.call_service("notify/ios_iphone_de_pierre", title = "Reveil", message = wakeup_message) def is_wakeup_time_on_a_working_day(self, wakeup_datetime): if self.is_during_vacation(wakeup_datetime): return False if wakeup_datetime.weekday() >= 5: #saturday is 5, sunday is 6 return False return True def is_during_vacation(self, date_time): for event in self.calendar: if event.type == EventType.Vacation and event.start_time <= date_time and date_time <= event.stop_time: return True return False def is_wakeup_time_on_a_flex_day(self, date_time): return False #we'll see about that later for event in self.calendar: if event.type == EventType.Flex and event.start_time <= date_time and date_time <= event.stop_time: return True return False def get_next_busy(self,date_time): for event in self.calendar: if event.type == EventType.Busy and event.start_time >= date_time: return event return None def offset_wakeup_time_for_meeting(self,wakeup_datetime): date_time = wakeup_datetime - datetime.timedelta(hours=wakeup_datetime.hour,minutes=wakeup_datetime.minute) busy_event = self.get_next_busy(date_time) if busy_event and busy_event.start_time.date() == date_time.date() and busy_event.start_time.hour > 6 and busy_event.start_time <= wakeup_datetime: wake_up_datetime = busy_event.start_time - datetime.timedelta(minutes=30) new_wake_up_time = wake_up_datetime.strftime("%H:%M") self.log(f"A meeting have been found at {busy_event.start_time}, we'll wake up at {new_wake_up_time}") return new_wake_up_time new_wake_up_time = wakeup_datetime.strftime("%H:%M") #self.log(f"No meeting have been found before {wake_up_time}, we'll wake up at {wake_up_time}") return new_wake_up_time def calculate_wakeup_time(self): new_wakeup_time = "--:--" if self.get_state(self.switch_alarmclock_secondary) == 'on': new_wakeup_time = self.get_state(self.inputdate_alarmclock_secondary)[0:5] elif self.get_state(self.switch_alarmclock_main) == 'on': max_flex_wakeup_time = "10:00" #todo: make that data driven max_flex_wakeup_datetime = self.generate_wakeup_datetime_from_string(max_flex_wakeup_time) new_wakeup_time = self.get_state(self.inputdate_alarmclock_main)[0:5] new_wakeup_datetime = self.generate_wakeup_datetime_from_string(new_wakeup_time) if not self.is_wakeup_time_on_a_working_day(new_wakeup_datetime): self.log(f"Primary Wake up time ({new_wakeup_datetime}) is not on a working day") new_wakeup_time = "--:--" elif self.is_wakeup_time_on_a_flex_day(max_flex_wakeup_datetime): self.log(f"Flex Wake up time ({max_flex_wakeup_datetime}) is on a flex office day") new_wakeup_time = self.offset_wakeup_time_for_meeting(max_flex_wakeup_datetime) else: new_wakeup_time = self.offset_wakeup_time_for_meeting(new_wakeup_datetime) return new_wakeup_time def generate_wakeup_datetime_from_string(self,wakeup_string): wakeup_time = datetime.time(int(wakeup_string[0:2]), int(wakeup_string[3:5]), 0) wakeup_datetime = datetime.datetime.combine(datetime.date.today(),wakeup_time) if (datetime.datetime.combine(datetime.date.today(), wakeup_time) - datetime.datetime.now()).total_seconds() <= 0: wakeup_datetime = wakeup_datetime + datetime.timedelta(days=1) return wakeup_datetime def register_callbacks(self, wakeup_time_str): if self.wakeup_time_str != wakeup_time_str or not self.handle_program_next_wakueup_cb: with self.lock: #first, cancel all previous callbacks if self.handle_program_next_wakueup_cb: self.log("Canceling previous program next wakeup cb") self.cancel_timer(self.handle_program_next_wakueup_cb) self.handle_program_next_wakueup_cb = None if self.handle_prewakueup_cb: self.log("Canceling previous pre wakeup cb") self.cancel_timer(self.handle_prewakueup_cb) self.handle_prewakueup_cb = None if self.handle_wakueup_cb: self.log("Canceling previous wakeup cb") self.cancel_timer(self.handle_wakueup_cb) self.handle_wakueup_cb = None if wakeup_time_str == '--:--': #if their is no alarm clock set up, we just need to program a callback to update the wakeup time tomorow # 12:00 should do it program_next_wakeup_time = datetime.datetime.combine(datetime.date.today(), datetime.time(12,0, 0)) #if we toggle off an alalrm clock before 4, we assume it's for the 'next' day (meaning tbe same day between 0 and 4) #if we toggle off an alalrm clock after 4, it's suppose to stay so untill next day if datetime.datetime.now().hour >= 4: program_next_wakeup_time = program_next_wakeup_time + datetime.timedelta(days=1) self.log("program next wakeup cb registered for " + str(program_next_wakeup_time)) self.handle_program_next_wakueup_cb = self.run_at(self.on_program_next_wakeup, program_next_wakeup_time) self.sensor_wakeup_datetime_entity.set_state(state = 'unavailable') else: wakeup_datetime = self.generate_wakeup_datetime_from_string(wakeup_time_str) self.sensor_wakeup_datetime_entity.set_state(state = wakeup_datetime.strftime("%H:%M")) #the program next wakeup CB need to be slightly after the wakeup CB to avoid the risk of wakeup CB cleaning it's handle AFTER program next wakup set up the next one program_next_wakeup_datetime = wakeup_datetime + datetime.timedelta(seconds=1) prewakeup_offset = min(60,(wakeup_datetime - datetime.datetime.now()).total_seconds() / 60) prewakeup_offset = int(prewakeup_offset / 10) * 10 prewakeup_datetime = wakeup_datetime - datetime.timedelta(minutes=prewakeup_offset) self.handle_prewakueup_cb = self.run_at(self.on_pre_wakeup, prewakeup_datetime,wakeup_in = prewakeup_offset) self.log("prewakup cb registered for " + str(prewakeup_datetime)) self.handle_wakueup_cb = self.run_at(self.on_wakeup, wakeup_datetime) self.log("wakeup cb registered for " + str(wakeup_datetime)) self.handle_program_next_wakueup_cb = self.run_at(self.on_program_next_wakeup, program_next_wakeup_datetime) self.log("program next wakeup cb registered for " + str(program_next_wakeup_datetime)) self.wakeup_time_str = wakeup_time_str self.call_service("input_text/set_value", entity_id = self.input_text_wakeup_time, value = self.wakeup_time_str) def update_wakeup_time(self): self.register_callbacks(self.calculate_wakeup_time())