Handle VALARM subcomponents (notifications).

icalevents/icalparser.py

@@ -14,6 +14,7 @@
 from icalendar.windows_to_olson import WINDOWS_TO_OLSON
 from icalendar.prop import vDDDLists, vText
 from pytz import timezone
+import copy
 
 
 def now():
@@ -52,6 +53,7 @@ def __init__(self):
         self.categories = None
         self.status = None
         self.url = None
+        self.alarms = []
 
     def time_left(self, time=None):
         """
@@ -145,6 +147,7 @@ def copy_to(self, new_start=None, uid=None):
         ne.categories = self.categories
         ne.status = self.status
         ne.url = self.url
+        ne.alarms = copy.deepcopy(self.alarms)
 
         return ne
 
@@ -166,7 +169,6 @@ def create_event(component, tz=UTC):
     :param tz: timezone for start and end times
     :return: event
     """
-
     event = Event()
 
     event.start = normalize(component.get("dtstart").dt, tz=tz)
@@ -291,6 +293,15 @@ def adjust_timezone(component, dates, tz=None):
     return dates
 
 
+def calculate_alarm_dt(trigger_dt, event_start):
+    if isinstance(trigger_dt, timedelta):
+        if type(event_start) == datetime.date:  # support full day events
+            event_start = datetime(event_start.year, event_start.month, event_start.day)
+        return event_start + trigger_dt
+    elif isinstance(trigger_dt, datetime):
+        return trigger_dt
+
+
 def parse_events(content, start=None, end=None, default_span=timedelta(days=7)):
     """
     Query the events occurring in a given time range.
@@ -367,6 +378,36 @@ def parse_events(content, start=None, end=None, default_span=timedelta(days=7)):
                     exdate = ex.to_ical().decode("UTF-8")
                     exceptions[exdate[0:8]] = exdate
 
+            for subcomponent in component.subcomponents:
+                if subcomponent.name == "VALARM":
+                    trigger = subcomponent.get("TRIGGER")
+                    if trigger is None:
+                        continue
+                    alarm_dt = None
+                    trigger_dt = trigger.dt
+                    alarm_dt = calculate_alarm_dt(trigger_dt, e.start)
+                    action = str(subcomponent.get("ACTION", ""))
+                    attachment = str(subcomponent.get("ATTACH", ""))
+                    description = str(subcomponent.get("DESCRIPTION", ""))
+                    alarm_uid = subcomponent.get("UID")
+                    if alarm_uid is None:
+                        # try to get other X-FOO-UID
+                        for key in subcomponent.keys():
+                            if key.endswith("-UID"):
+                                alarm_uid = subcomponent.get(key)
+                    if alarm_uid is None:
+                        alarm_uid = ""
+                    e.alarms.append(
+                        dict(
+                            description=description,
+                            alarm_dt=alarm_dt,
+                            action=action,
+                            attachment=attachment,
+                            uid=str(alarm_uid),
+                            trigger_dt=trigger_dt,
+                        )
+                    )
+
             # Attempt to work out what timezone is used for the start
             # and end times. If the timezone is defined in the calendar,
             # use it; otherwise, attempt to load the rules from pytz.
@@ -430,12 +471,17 @@ def parse_events(content, start=None, end=None, default_span=timedelta(days=7)):
                         ecopy.start.month,
                         ecopy.start.day,
                     )
+                    for alarm in ecopy.alarms:
+                        alarm["alarm_dt"] = calculate_alarm_dt(
+                            alarm["trigger_dt"], ecopy.start
+                        )
                     if exdate not in exceptions:
                         found.append(ecopy)
             elif e.end >= start and e.start <= end:
                 exdate = "%04d%02d%02d" % (e.start.year, e.start.month, e.start.day)
                 if exdate not in exceptions:
                     found.append(e)
+
     # Filter out all events that are moved as indicated by the recurrence-id prop
     return [
         event

test/test_data/recurring_alarm.ics

@@ -0,0 +1,31 @@
+BEGIN:VCALENDAR
+BEGIN:VTIMEZONE
+TZID:Europe/Berlin
+END:VTIMEZONE
+BEGIN:VEVENT
+DTSTART;VALUE=DATE:20181030
+DTEND;VALUE=DATE:20181031
+DESCRIPTION:All-day event recurring on tuesday each week
+SUMMARY:Recurring All-day Event
+RRULE:FREQ=WEEKLY;BYDAY=TU
+BEGIN:VALARM
+ACTION:AUDIO
+TRIGGER:-PT15H
+UID:4BB6A40E-6845-4541-BD87-0962514D03DC
+ATTACH;VALUE=URI:Basso
+X-APPLE-DEFAULT-ALARM:TRUE
+ACKNOWLEDGED:20170319T131719Z
+END:VALARM
+BEGIN:VALARM
+ACTION:DISPLAY
+DESCRIPTION:Funny Description
+TRIGGER;RELATED=START:-P3D
+X-EVOLUTION-ALARM-UID:def4351cbf2dc54c4019fa7a5b8557ec3b9ee26d
+END:VALARM
+BEGIN:VALARM
+ACTION:DISPLAY
+DESCRIPTION:This is an event reminder
+TRIGGER:-P0DT10H0M0S
+END:VALARM
+END:VEVENT
+END:VCALENDAR

test/test_icalevents.py

@@ -450,3 +450,80 @@ def test_status_and_url(self):
         self.assertEqual(ev3.status, "CANCELLED")
         self.assertEqual(ev4.status, "CANCELLED")
         self.assertEqual(ev5.status, None)
+
+    def test_alarms_absolute(self):
+        """Alarms which are set to a fixed datetime are properly
+        returned."""
+        ical = "test/test_data/basic.ics"
+        start = date(2017, 5, 16)
+        evs = icalevents.events(url=None, file=ical, start=start)
+        self.assertEqual(
+            datetime(1976, 4, 1, 0, 55, 45, tzinfo=UTC), evs[0].alarms[0]["alarm_dt"]
+        )
+
+    def test_alarms_relative(self):
+        """Alarms which are set to a relative datetime are properly
+        returned."""
+        ical = "test/test_data/basic.ics"
+        start = date(2017, 3, 19)
+        evs = icalevents.events(url=None, file=ical, start=start)
+        self.assertEqual(
+            datetime(2017, 3, 19, 9, 0, tzinfo=evs[0].start.tzinfo),
+            evs[0].alarms[0]["alarm_dt"],
+        )
+        self.assertEqual(timedelta(hours=-15), evs[0].alarms[0]["trigger_dt"])
+
+    def test_alarms_recurring(self):
+        """Recurrences get their own alarm each."""
+        ical = "test/test_data/recurring_alarm.ics"
+        start = date(2020, 3, 19)
+        end = start + timedelta(days=20)
+        evs = icalevents.events(url=None, file=ical, start=start, end=end)
+        expected_tz = evs[0].start.tzinfo
+        self.assertEqual(
+            datetime(2020, 3, 23, 9, 0, tzinfo=expected_tz),
+            evs[0].alarms[0]["alarm_dt"],
+        )
+        self.assertEqual(
+            datetime(2020, 3, 30, 9, 0, tzinfo=expected_tz),
+            evs[1].alarms[0]["alarm_dt"],
+        )
+        self.assertEqual(
+            datetime(2020, 4, 6, 9, 0, tzinfo=expected_tz), evs[2].alarms[0]["alarm_dt"]
+        )
+
+    def test_alarms_data__1(self):
+        ical = "test/test_data/recurring_alarm.ics"
+        start = date(2020, 3, 19)
+        evs = icalevents.events(url=None, file=ical, start=start)
+        expected_tz = evs[0].start.tzinfo
+        # apple
+        expected = {
+            "action": "AUDIO",
+            "alarm_dt": datetime(2020, 3, 23, 9, 0, tzinfo=expected_tz),
+            "attachment": "Basso",
+            "description": "",
+            "trigger_dt": timedelta(hours=-15),
+            "uid": "4BB6A40E-6845-4541-BD87-0962514D03DC",
+        }
+        self.assertEqual(expected, evs[0].alarms[0])
+        # evolution
+        expected = {
+            "action": "DISPLAY",
+            "alarm_dt": datetime(2020, 3, 21, 0, 0, tzinfo=expected_tz),
+            "attachment": "",
+            "description": "Funny Description",
+            "trigger_dt": timedelta(days=-3),
+            "uid": "def4351cbf2dc54c4019fa7a5b8557ec3b9ee26d",
+        }
+        self.assertEqual(expected, evs[0].alarms[1])
+        # google
+        expected = {
+            "action": "DISPLAY",
+            "alarm_dt": datetime(2020, 3, 23, 14, 0, tzinfo=expected_tz),
+            "attachment": "",
+            "description": "This is an event reminder",
+            "trigger_dt": timedelta(hours=-10),
+            "uid": "",
+        }
+        self.assertEqual(expected, evs[0].alarms[2])

test/test_icalparser.py

@@ -1,6 +1,6 @@
 import unittest
 import icalevents.icalparser
-from datetime import datetime, date
+from datetime import datetime, date, timedelta
 from dateutil.tz import UTC, gettz
 
 
@@ -18,6 +18,18 @@ def setUp(self):
         self.eventA.summary = "Event A"
         self.eventA.attendee = "[email protected]"
         self.eventA.organizer = "[email protected]"
+        trigger_dt = timedelta(days=-1)
+        alarm_dt = self.eventA.start + trigger_dt
+        self.eventA.alarms = [
+            dict(
+                summary="Reminder for Event A",
+                description="",
+                alarm_dt=alarm_dt,
+                action="",
+                uid="alarm_uid",
+                trigger_dt=trigger_dt,
+            )
+        ]
 
         self.eventB = icalevents.icalparser.Event()
         self.eventB.uid = 1234
@@ -59,6 +71,10 @@ def test_event_copy_to(self):
             self.eventA.end - self.eventA.start,
             "new event has same duration",
         )
+        self.assertEqual(len(eventC.alarms), 1)
+        self.eventA.alarms.append("test")
+        self.assertEqual(len(eventC.alarms), 1, "alarms is a copy")
+
         self.assertEqual(eventC.all_day, False, "new event is no all day event")
         self.assertEqual(eventC.summary, self.eventA.summary, "copy to: summary")
         self.assertEqual(