waldhart_availability.py (7747B)
1 #!/usr/bin/env python 2 3 import logging 4 import secrets 5 import os 6 import requests 7 import caldav 8 from datetime import datetime 9 from enum import Enum 10 11 class Duration(Enum): 12 NV = 0 # not available, not skiing 13 AM = 1 # skiing at forenoon (08:00 - 12:00) 14 PM = 2 # skiing at afternoon (12:00 - 16:00) 15 ALLDAY = 3 # skiing all day 16 CALL = 4 # on-call, skiing when required 17 18 class WaldhartAvailability: 19 """ 20 Read availability data from Waldhart software 21 """ 22 23 def __init__(self, **kwargs): 24 """ 25 Init CalDAV client 26 """ 27 28 # setup logger 29 self.logger = logging.getLogger(__name__) 30 LOGLEVEL = os.environ.get('LOGLEVEL', 'INFO').upper() 31 logging.basicConfig(level=LOGLEVEL, format="%(asctime)s %(message)s") 32 33 # read input keyword arguments 34 if "waldhart_url" in kwargs: 35 self.waldhart_url = kwargs["waldhart_url"] 36 elif "WALDHART_URL" in os.environ: 37 self.waldhart_url = os.environ["WALDHART_URL"] 38 else: 39 raise NameError("WALDHART_URL undefined") 40 41 if "caldav_calendar_name" in kwargs: 42 self.caldav_calendar_name = kwargs["caldav_calendar_name"] 43 elif "CALDAV_CALENDAR_NAME" in os.environ: 44 self.caldav_calendar_name = os.environ["CALDAV_CALENDAR_NAME"] 45 else: 46 raise NameError("WALDHART_CALENDAR_NAME undefined") 47 48 if "waldhart_username" in kwargs: 49 self.waldhart_username = kwargs["waldhart_username"] 50 elif "WALDHART_USERNAME" in os.environ: 51 self.waldhart_username = os.environ["WALDHART_USERNAME"] 52 else: 53 raise NameError("WALDHART_USERNAME undefined") 54 55 if "waldhart_password" in kwargs: 56 self.waldhart_password = kwargs["waldhart_password"] 57 elif "WALDHART_PASSWORD" in os.environ: 58 self.waldhart_password = os.environ["WALDHART_PASSWORD"] 59 else: 60 raise NameError("WALDHART_PASSWORD undefined") 61 62 if "caldav_url" in kwargs: 63 self.caldav_url = kwargs["caldav_url"] 64 elif "CALDAV_URL" in os.environ: 65 self.caldav_url = os.environ["CALDAV_URL"] 66 else: 67 raise NameError("CALDAV_URL undefined") 68 69 if "caldav_username" in kwargs: 70 self.caldav_username = kwargs["caldav_username"] 71 elif "CALDAV_USERNAME" in os.environ: 72 self.caldav_username = os.environ["CALDAV_USERNAME"] 73 else: 74 raise NameError("CALDAV_USERNAME undefined") 75 76 if "caldav_password" in kwargs: 77 self.caldav_password = kwargs["caldav_password"] 78 elif "CALDAV_PASSWORD" in os.environ: 79 self.caldav_password = os.environ["CALDAV_PASSWORD"] 80 else: 81 raise NameError("CALDAV_PASSWORD undefined") 82 83 if "season" in kwargs: 84 self.season = int(kwargs["season"]) 85 elif "SEASON" in os.environ: 86 self.season = int(os.environ["SEASON"]) 87 else: 88 raise NameError("SEASON undefined") 89 90 # DAVClient 91 # * https://caldav.readthedocs.io/en/latest/caldav/davclient.html 92 # * https://github.com/python-caldav/caldav/blob/master/examples/basic_usage_examples.py 93 self.dav_client = caldav.DAVClient(url=self.caldav_url, 94 username=self.caldav_username, 95 password=self.caldav_password) 96 97 self.logger.info(f"Connected to CalDAV API {self.caldav_url}") 98 99 def login(self): 100 """ 101 Login to Waldhart session. 102 103 Returns true when login was successful. 104 """ 105 106 # set random session id 107 self.session = requests.Session() 108 self.session.cookies.set("session_id", secrets.token_hex(nbytes=20)) 109 110 payload = { 111 "username": f"{self.waldhart_username}", 112 "password": f"{self.waldhart_password}" 113 } 114 response = self.session.post(f"{self.waldhart_url}/login", 115 data=payload) 116 117 return response.status_code == 200 118 119 def read_availability(self, year: int = 2024, month: int = 12): 120 """ 121 Read availability data from Waldhart software. 122 123 Required arguments are year and month. 124 """ 125 126 self.logger.info(f"Reading availability for year '{year}' and month '{month}'") 127 128 resp = self.session.get(f"{self.waldhart_url}/myavailability/myavailability_month/get_availability_json?year={year}&month={month}") 129 self.logger.debug(f"Received json response {resp.content}") 130 return resp.json() 131 132 def update_availability(self, year: int = 2024, month: int = 12, day: int = 1, duration: Duration = Duration.NV): 133 principal = self.dav_client.principal() 134 calendars = principal.calendars() 135 self.logger.debug(f"Calendars: {calendars}") 136 calendar = [c for c in calendars if c.name == self.caldav_calendar_name] 137 138 if len(calendar) <= 0: 139 raise Exception(f"Calendar with name {self.caldav_calendar_name} does not exist") 140 141 self.logger.info(f"Selected calendar: {calendar[0]}") 142 143 events = calendar[0].search( 144 start=datetime(year, month, day, 6), 145 end=datetime(year, month, day, 18), 146 event=True, 147 expand=False, # ignore recurrences 148 ) 149 150 self.logger.debug(f"Found events: {events}") 151 152 # remove events 153 for e in events: 154 e.delete() 155 156 # don't create a blocker when not available for skiing or on-call 157 if duration in [Duration.NV, Duration.CALL]: 158 self.logger.debug(f"Not available (NV) or on-call, not creating any blocker") 159 return 160 161 # create new event from Waldhart availability when skiing 162 start_time = 8 if duration == Duration.AM or duration == Duration.ALLDAY else 12 163 end_time = 16 if duration == Duration.PM or duration == Duration.ALLDAY else 12 164 e = calendar[0].save_event( 165 dtstart=datetime(year, month, day, start_time), 166 dtend=datetime(year, month, day, end_time), 167 summary="Skiing", 168 ) 169 170 self.logger.debug(f"Created event: {e}") 171 172 def duration_type(self, entry: str) -> Duration: 173 """ 174 Define duration type of a Waldhart availability entry. 175 """ 176 177 self.logger.debug(f"Availability entry: {entry}") 178 179 if entry["not_available"]: 180 return Duration.NV 181 elif entry["on_call"]: 182 return Duration.CALL 183 elif entry["pm"] and not entry["am"]: 184 return Duration.PM 185 elif entry["am"] and not entry["pm"]: 186 return Duration.AM 187 elif entry["pm"] and entry["am"]: 188 return Duration.ALLDAY 189 else: 190 return None 191 192 193 def enumerate_and_update(availability, year, month): 194 for i, key in enumerate(availability): 195 duration = waldhart.duration_type(availability[key]) 196 if duration == None: 197 raise Exception("Cannot determine availability") 198 waldhart.update_availability(year, month, int(key), duration) 199 200 if __name__== "__main__": 201 waldhart = WaldhartAvailability() 202 waldhart.login() 203 204 # read availability in current season 205 dec = waldhart.read_availability(waldhart.season, 12) 206 jan = waldhart.read_availability(waldhart.season + 1, 1) 207 feb = waldhart.read_availability(waldhart.season + 1, 2) 208 mar = waldhart.read_availability(waldhart.season + 1, 3) 209 210 # update availability December - March 211 enumerate_and_update(dec, waldhart.season, 12) 212 enumerate_and_update(jan, waldhart.season + 1, 1) 213 enumerate_and_update(feb, waldhart.season + 1, 2) 214 enumerate_and_update(mar, waldhart.season + 1, 3)