commit f5a57a7cced2052fe66f2f8d578ef61f325bc85e
parent c1e1da9f9ff2982c33c3fccb44f4820bb3af03c3
Author: Andreas Gruhler <andreas.gruhler@adfinis.com>
Date: Mon, 2 Dec 2024 00:15:58 +0100
feat: read availability and update calendar
Diffstat:
7 files changed, 505 insertions(+), 0 deletions(-)
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,6 @@
+.env
+.coverage
+cookies.txt
+**.swp
+.venv
+__pycache__
diff --git a/.pytest.ini b/.pytest.ini
@@ -0,0 +1,4 @@
+[pytest]
+# https://docs.python.org/3/library/logging.html#levels
+log_cli = true
+log_cli_level = 20
diff --git a/Dockerfile b/Dockerfile
@@ -0,0 +1,9 @@
+FROM docker.io/python:3.12
+
+COPY waldhart_availability.py /app/
+COPY requirements.txt /app/
+
+WORKDIR /app/
+RUN pip install -r requirements.txt
+
+CMD ["./waldhart_availability.py"]
diff --git a/env b/env
@@ -0,0 +1,10 @@
+export WALDHART_URL=
+export WALDHART_USERNAME=
+export WALDHART_PASSWORD=
+export CALDAV_URL=
+export CALDAV_CALENDAR_NAME=
+export CALDAV_USERNAME=
+export CALDAV_PASSWORD=
+export SEASON=2024 # December - March next year
+#export PYTHON_CALDAV_DEBUGMODE=DEBUG
+#export LOGLEVEL=DEBUG
diff --git a/requirements.txt b/requirements.txt
@@ -0,0 +1,5 @@
+pytest
+coverage
+requests
+requests_mock
+caldav
diff --git a/waldhart_availability.py b/waldhart_availability.py
@@ -0,0 +1,214 @@
+#!/usr/bin/env python
+
+import logging
+import secrets
+import os
+import requests
+import caldav
+from datetime import datetime
+from enum import Enum
+
+class Duration(Enum):
+ NV = 0 # not available, not skiing
+ AM = 1 # skiing at forenoon (08:00 - 12:00)
+ PM = 2 # skiing at afternoon (12:00 - 16:00)
+ ALLDAY = 3 # skiing all day
+ CALL = 4 # on-call, skiing when required
+
+class WaldhartAvailability:
+ """
+ Read availability data from Waldhart software
+ """
+
+ def __init__(self, **kwargs):
+ """
+ Init CalDAV client
+ """
+
+ # setup logger
+ self.logger = logging.getLogger(__name__)
+ LOGLEVEL = os.environ.get('LOGLEVEL', 'INFO').upper()
+ logging.basicConfig(level=LOGLEVEL, format="%(asctime)s %(message)s")
+
+ # read input keyword arguments
+ if "waldhart_url" in kwargs:
+ self.waldhart_url = kwargs["waldhart_url"]
+ elif "WALDHART_URL" in os.environ:
+ self.waldhart_url = os.environ["WALDHART_URL"]
+ else:
+ raise NameError("WALDHART_URL undefined")
+
+ if "caldav_calendar_name" in kwargs:
+ self.caldav_calendar_name = kwargs["caldav_calendar_name"]
+ elif "CALDAV_CALENDAR_NAME" in os.environ:
+ self.caldav_calendar_name = os.environ["CALDAV_CALENDAR_NAME"]
+ else:
+ raise NameError("WALDHART_CALENDAR_NAME undefined")
+
+ if "waldhart_username" in kwargs:
+ self.waldhart_username = kwargs["waldhart_username"]
+ elif "WALDHART_USERNAME" in os.environ:
+ self.waldhart_username = os.environ["WALDHART_USERNAME"]
+ else:
+ raise NameError("WALDHART_USERNAME undefined")
+
+ if "waldhart_password" in kwargs:
+ self.waldhart_password = kwargs["waldhart_password"]
+ elif "WALDHART_PASSWORD" in os.environ:
+ self.waldhart_password = os.environ["WALDHART_PASSWORD"]
+ else:
+ raise NameError("WALDHART_PASSWORD undefined")
+
+ if "caldav_url" in kwargs:
+ self.caldav_url = kwargs["caldav_url"]
+ elif "CALDAV_URL" in os.environ:
+ self.caldav_url = os.environ["CALDAV_URL"]
+ else:
+ raise NameError("CALDAV_URL undefined")
+
+ if "caldav_username" in kwargs:
+ self.caldav_username = kwargs["caldav_username"]
+ elif "CALDAV_USERNAME" in os.environ:
+ self.caldav_username = os.environ["CALDAV_USERNAME"]
+ else:
+ raise NameError("CALDAV_USERNAME undefined")
+
+ if "caldav_password" in kwargs:
+ self.caldav_password = kwargs["caldav_password"]
+ elif "CALDAV_PASSWORD" in os.environ:
+ self.caldav_password = os.environ["CALDAV_PASSWORD"]
+ else:
+ raise NameError("CALDAV_PASSWORD undefined")
+
+ if "season" in kwargs:
+ self.season = int(kwargs["season"])
+ elif "SEASON" in os.environ:
+ self.season = int(os.environ["SEASON"])
+ else:
+ raise NameError("SEASON undefined")
+
+ # DAVClient
+ # * https://caldav.readthedocs.io/en/latest/caldav/davclient.html
+ # * https://github.com/python-caldav/caldav/blob/master/examples/basic_usage_examples.py
+ self.dav_client = caldav.DAVClient(url=self.caldav_url,
+ username=self.caldav_username,
+ password=self.caldav_password)
+
+ self.logger.info(f"Connected to CalDAV API {self.caldav_url}")
+
+ def login(self):
+ """
+ Login to Waldhart session.
+
+ Returns true when login was successful.
+ """
+
+ # set random session id
+ self.session = requests.Session()
+ self.session.cookies.set("session_id", secrets.token_hex(nbytes=20))
+
+ payload = {
+ "username": f"{self.waldhart_username}",
+ "password": f"{self.waldhart_password}"
+ }
+ response = self.session.post(f"{self.waldhart_url}/login",
+ data=payload)
+
+ return response.status_code == 200
+
+ def read_availability(self, year: int = 2024, month: int = 12):
+ """
+ Read availability data from Waldhart software.
+
+ Required arguments are year and month.
+ """
+
+ self.logger.info(f"Reading availability for year '{year}' and month '{month}'")
+
+ resp = self.session.get(f"{self.waldhart_url}/myavailability/myavailability_month/get_availability_json?year={year}&month={month}")
+ self.logger.debug(f"Received json response {resp.content}")
+ return resp.json()
+
+ def update_availability(self, year: int = 2024, month: int = 12, day: int = 1, duration: Duration = Duration.NV):
+ principal = self.dav_client.principal()
+ calendars = principal.calendars()
+ self.logger.debug(f"Calendars: {calendars}")
+ calendar = [c for c in calendars if c.name == self.caldav_calendar_name]
+
+ if len(calendar) <= 0:
+ raise Exception(f"Calendar with name {self.caldav_calendar_name} does not exist")
+
+ self.logger.info(f"Selected calendar: {calendar[0]}")
+
+ events = calendar[0].search(
+ start=datetime(year, month, day, 6),
+ end=datetime(year, month, day, 18),
+ event=True,
+ expand=False, # ignore recurrences
+ )
+
+ self.logger.debug(f"Found events: {events}")
+
+ # remove events
+ for e in events:
+ e.delete()
+
+ # don't create a blocker when not available for skiing or on-call
+ if duration in [Duration.NV, Duration.CALL]:
+ self.logger.debug(f"Not available (NV) or on-call, not creating any blocker")
+ return
+
+ # create new event from Waldhart availability when skiing
+ start_time = 8 if duration == Duration.AM or duration == Duration.ALLDAY else 12
+ end_time = 16 if duration == Duration.PM or duration == Duration.ALLDAY else 12
+ e = calendar[0].save_event(
+ dtstart=datetime(year, month, day, start_time),
+ dtend=datetime(year, month, day, end_time),
+ summary="Skiing",
+ )
+
+ self.logger.debug(f"Created event: {e}")
+
+ def duration_type(self, entry: str) -> Duration:
+ """
+ Define duration type of a Waldhart availability entry.
+ """
+
+ self.logger.debug(f"Availability entry: {entry}")
+
+ if entry["not_available"]:
+ return Duration.NV
+ elif entry["on_call"]:
+ return Duration.CALL
+ elif entry["pm"] and not entry["am"]:
+ return Duration.PM
+ elif entry["am"] and not entry["pm"]:
+ return Duration.AM
+ elif entry["pm"] and entry["am"]:
+ return Duration.ALLDAY
+ else:
+ return None
+
+
+def enumerate_and_update(availability, year, month):
+ for i, key in enumerate(availability):
+ duration = waldhart.duration_type(availability[key])
+ if duration == None:
+ raise Exception("Cannot determine availability")
+ waldhart.update_availability(year, month, int(key), duration)
+
+if __name__== "__main__":
+ waldhart = WaldhartAvailability()
+ waldhart.login()
+
+ # read availability in current season
+ dec = waldhart.read_availability(waldhart.season, 12)
+ jan = waldhart.read_availability(waldhart.season + 1, 1)
+ feb = waldhart.read_availability(waldhart.season + 1, 2)
+ mar = waldhart.read_availability(waldhart.season + 1, 3)
+
+ # update availability December - March
+ enumerate_and_update(dec, waldhart.season, 12)
+ enumerate_and_update(jan, waldhart.season + 1, 1)
+ enumerate_and_update(feb, waldhart.season + 1, 2)
+ enumerate_and_update(mar, waldhart.season + 1, 3)
diff --git a/waldhart_availability_test.py b/waldhart_availability_test.py
@@ -0,0 +1,257 @@
+import logging
+import pytest
+import requests
+import requests_mock
+from datetime import datetime
+
+from waldhart_availability import Duration
+from waldhart_availability import WaldhartAvailability
+
+logger = logging.getLogger(__name__)
+
+class TestWaldhartAvailability:
+ """
+ Test WaldhartAvailability functionality.
+ """
+
+ @requests_mock.Mocker(kw="mock")
+ def test_login(self, **kwargs):
+ kwargs['mock'].post("http://waldhart/login", text="OK")
+ kwargs['mock'].get("http://caldav", text="dav")
+
+ waldhart = WaldhartAvailability(waldhart_url="http://waldhart",
+ waldhart_username="test",
+ waldhart_password="123",
+ caldav_url="http://caldav",
+ caldav_username="testc",
+ caldav_password="1234")
+ r = waldhart.login()
+ assert r
+
+ @requests_mock.Mocker(kw="mock")
+ def test_read_availability(self, **kwargs):
+ kwargs['mock'].post("http://waldhart/login", text="OK")
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_3)
+ kwargs['mock'].get("http://caldav", text="dav")
+
+ waldhart = WaldhartAvailability(waldhart_url="http://waldhart",
+ waldhart_username="test",
+ waldhart_password="123",
+ caldav_url="http://caldav",
+ caldav_username="testc",
+ caldav_password="1234")
+ waldhart.login()
+
+ day = waldhart.read_availability(2025, 1)["3"]
+ logger.info(day)
+
+ @requests_mock.Mocker(kw="mock")
+ def test_duration(self, **kwargs):
+ kwargs['mock'].post("http://waldhart/login", text="OK")
+ kwargs['mock'].get("http://caldav", text="dav")
+
+ waldhart = WaldhartAvailability(waldhart_url="http://waldhart",
+ waldhart_username="test",
+ waldhart_password="123",
+ caldav_url="http://caldav",
+ caldav_username="testc",
+ caldav_password="1234")
+ waldhart.login()
+
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_3)
+ day = waldhart.read_availability(2025, 1)["3"]
+ duration = waldhart.duration_type(day)
+ assert duration == Duration.ALLDAY
+
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_6)
+ day = waldhart.read_availability(2025, 1)["6"]
+ duration = waldhart.duration_type(day)
+ assert duration == Duration.AM
+
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_20)
+ day = waldhart.read_availability(2025, 1)["20"]
+ duration = waldhart.duration_type(day)
+ assert duration == Duration.PM
+
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_26)
+ day = waldhart.read_availability(2025, 1)["26"]
+ duration = waldhart.duration_type(day)
+ assert duration == Duration.NV
+
+ @requests_mock.Mocker(kw="mock")
+ def test_update_availability(self, **kwargs):
+ kwargs['mock'].post("http://waldhart/login", text="OK")
+ kwargs['mock'].get("mock://caldav", text="dav")
+
+ # TODO: mock PROPFIND requests
+ #session = requests.Session()
+ #adapter = requests_mock.Adapter()
+ #adapter.register_uri("PROPFIND", "mock://caldav", text="dav")
+ #session.mount("mock://", adapter)
+
+ waldhart = WaldhartAvailability(waldhart_url="http://waldhart",
+ waldhart_username="test",
+ waldhart_password="123",
+ caldav_url="mock://caldav",
+ caldav_username="testc",
+ caldav_password="1234")
+ waldhart.login()
+
+ # expect ALLDAY
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_3)
+ day = waldhart.read_availability(2025, 1)["3"]
+ duration = waldhart.duration_type(day)
+ waldhart.update_availability(2025, 1, 3, duration)
+
+ # expect forenoon (AM)
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_6)
+ day = waldhart.read_availability(2025, 1)["6"]
+ duration = waldhart.duration_type(day)
+ waldhart.update_availability(2025, 1, 6, duration)
+
+ # expect afternoon (PM)
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_20)
+ day = waldhart.read_availability(2025, 1)["20"]
+ duration = waldhart.duration_type(day)
+ waldhart.update_availability(2025, 1, 20, duration)
+
+ # expect not available (NV)
+ kwargs['mock'].get("http://waldhart/myavailability/myavailability_month/get_availability_json?year=2025&month=1", text=JAN_26)
+ day = waldhart.read_availability(2025, 1)["26"]
+ duration = waldhart.duration_type(day)
+ waldhart.update_availability(2025, 1, 26, duration)
+
+
+# Waldhart json response mock
+
+JAN_3 = """
+{
+ "3": {
+ "has_course_pm": false,
+ "has_course_md": false,
+ "pm": true,
+ "on_call": false,
+ "ready_3": null,
+ "ready_1": null,
+ "to_2": null,
+ "has_course_am": false,
+ "md": true,
+ "group_type": null,
+ "to_1": null,
+ "vacation": false,
+ "from_2": null,
+ "has_course_all_day": false,
+ "am": true,
+ "ready_2": null,
+ "from_3": null,
+ "ready_state": 1,
+ "from_1": null,
+ "ill": false,
+ "courses": null,
+ "show_hatching": true,
+ "not_available": false,
+ "group_type_2": null,
+ "to_3": null,
+ "has_course": false
+ }
+}
+"""
+
+JAN_6 = """
+{
+ "6": {
+ "has_course_pm": false,
+ "has_course_md": false,
+ "pm": false,
+ "on_call": false,
+ "ready_3": 0,
+ "ready_1": 1,
+ "to_2": 840,
+ "has_course_am": false,
+ "md": false,
+ "group_type": null,
+ "to_1": 720,
+ "vacation": false,
+ "from_2": 720,
+ "has_course_all_day": false,
+ "am": true,
+ "ready_2": 0,
+ "from_3": 720,
+ "ready_state": 5,
+ "from_1": 480,
+ "ill": false,
+ "courses": null,
+ "show_hatching": true,
+ "not_available": false,
+ "group_type_2": null,
+ "to_3": 960,
+ "has_course": false
+ }
+}
+"""
+
+JAN_20 = """
+{
+ "20": {
+ "has_course_pm": false,
+ "has_course_md": false,
+ "pm": true,
+ "on_call": false,
+ "ready_3": 1,
+ "ready_1": 0,
+ "to_2": 840,
+ "has_course_am": false,
+ "md": false,
+ "group_type": null,
+ "to_1": 720,
+ "vacation": false,
+ "from_2": 720,
+ "has_course_all_day": false,
+ "am": false,
+ "ready_2": 0,
+ "from_3": 720,
+ "ready_state": 5,
+ "from_1": 480,
+ "ill": false,
+ "courses": null,
+ "show_hatching": true,
+ "not_available": false,
+ "group_type_2": null,
+ "to_3": 960,
+ "has_course": false
+ }
+}
+"""
+
+JAN_26 = """
+{
+ "26": {
+ "has_course_pm": false,
+ "has_course_md": false,
+ "pm": false,
+ "on_call": false,
+ "ready_3": null,
+ "ready_1": null,
+ "to_2": null,
+ "has_course_am": false,
+ "md": false,
+ "group_type": null,
+ "to_1": null,
+ "vacation": false,
+ "from_2": null,
+ "has_course_all_day": false,
+ "am": false,
+ "ready_2": null,
+ "from_3": null,
+ "ready_state": 0,
+ "from_1": null,
+ "ill": false,
+ "courses": null,
+ "show_hatching": true,
+ "not_available": true,
+ "group_type_2": null,
+ "to_3": null,
+ "has_course": false
+ }
+}
+"""
