Merge pull request #172 from mampfes/icalevents

Icalevents

custom_components/waste_collection_schedule/manifest.json

@@ -2,9 +2,9 @@
   "domain": "waste_collection_schedule",
   "name": "waste_collection_schedule",
   "documentation": "https://github.com/mampfes/hacs_waste_collection_schedule#readme",
-  "requirements": ["icalendar", "recurring_ical_events", "bs4"],
+  "requirements": ["icalendar", "recurring_ical_events", "icalevents", "bs4"],
   "dependencies": [],
   "codeowners": ["@mampfes"],
   "iot_class": "cloud_polling",
-  "version": "1.14.0"
+  "version": "1.16.0"
 }

custom_components/waste_collection_schedule/waste_collection_schedule/service/ICS.py

@@ -2,8 +2,7 @@
 import logging
 import re
 
-import icalendar
-import recurring_ical_events
+from icalevents import icalevents
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -17,14 +16,6 @@ def __init__(self, offset=None, regex=None, split_at=None):
         self._split_at = split_at
 
     def convert(self, ics_data):
-        # parse ics file
-        try:
-            calendar = icalendar.Calendar.from_ical(ics_data)
-        except Exception as err:
-            _LOGGER.error(f"Parsing ics data failed:{str(err)}")
-            _LOGGER.debug(ics_data)
-            return []
-
         # calculate start- and end-date for recurring events
         start_date = datetime.datetime.now().replace(
             hour=0, minute=0, second=0, microsecond=0
@@ -33,32 +24,34 @@ def convert(self, ics_data):
             start_date -= datetime.timedelta(days=self._offset)
         end_date = start_date.replace(year=start_date.year + 1)
 
-        events = recurring_ical_events.of(calendar).between(start_date, end_date)
+        # parse ics data
+        events = icalevents.events(
+            start=start_date, end=end_date, string_content=ics_data.encode()
+        )
 
         entries = []
         for e in events:
-            if e.name == "VEVENT":
-                # calculate date
-                dtstart = None
-                if type(e.get("dtstart").dt) == datetime.date:
-                    dtstart = e.get("dtstart").dt
-                elif type(e.get("dtstart").dt) == datetime.datetime:
-                    dtstart = e.get("dtstart").dt.date()
-                if self._offset is not None:
-                    dtstart += datetime.timedelta(days=self._offset)
-
-                # calculate waste type
-                summary = str(e.get("summary"))
-                if self._regex is not None:
-                    match = self._regex.match(summary)
-                    if match:
-                        summary = match.group(1)
-
-                if self._split_at is not None:
-                    summary = re.split(self._split_at, summary)
-                    for t in summary:
-                        entries.append((dtstart, t.strip().title()))
-                else:
-                    entries.append((dtstart, summary))
+            # calculate date
+            dtstart = None
+            if type(e.start) == datetime.date:
+                dtstart = e.start
+            elif type(e.start) == datetime.datetime:
+                dtstart = e.start.date()
+            if self._offset is not None:
+                dtstart += datetime.timedelta(days=self._offset)
+
+            # calculate waste type
+            summary = str(e.summary)
+            if self._regex is not None:
+                match = self._regex.match(summary)
+                if match:
+                    summary = match.group(1)
+
+            if self._split_at is not None:
+                summary = re.split(self._split_at, summary)
+                for t in summary:
+                    entries.append((dtstart, t.strip().title()))
+            else:
+                entries.append((dtstart, summary))
 
         return entries

custom_components/waste_collection_schedule/waste_collection_schedule/service/ICS_v1.py

@@ -0,0 +1,64 @@
+import datetime
+import logging
+import re
+
+import icalendar
+import recurring_ical_events
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ICS_v1:
+    def __init__(self, offset=None, regex=None, split_at=None):
+        self._offset = offset
+        self._regex = None
+        if regex is not None:
+            self._regex = re.compile(regex)
+        self._split_at = split_at
+
+    def convert(self, ics_data):
+        # parse ics file
+        try:
+            calendar = icalendar.Calendar.from_ical(ics_data)
+        except Exception as err:
+            _LOGGER.error(f"Parsing ics data failed:{str(err)}")
+            _LOGGER.debug(ics_data)
+            return []
+
+        # calculate start- and end-date for recurring events
+        start_date = datetime.datetime.now().replace(
+            hour=0, minute=0, second=0, microsecond=0
+        )
+        if self._offset is not None:
+            start_date -= datetime.timedelta(days=self._offset)
+        end_date = start_date.replace(year=start_date.year + 1)
+
+        events = recurring_ical_events.of(calendar).between(start_date, end_date)
+
+        entries = []
+        for e in events:
+            if e.name == "VEVENT":
+                # calculate date
+                dtstart = None
+                if type(e.get("dtstart").dt) == datetime.date:
+                    dtstart = e.get("dtstart").dt
+                elif type(e.get("dtstart").dt) == datetime.datetime:
+                    dtstart = e.get("dtstart").dt.date()
+                if self._offset is not None:
+                    dtstart += datetime.timedelta(days=self._offset)
+
+                # calculate waste type
+                summary = str(e.get("summary"))
+                if self._regex is not None:
+                    match = self._regex.match(summary)
+                    if match:
+                        summary = match.group(1)
+
+                if self._split_at is not None:
+                    summary = re.split(self._split_at, summary)
+                    for t in summary:
+                        entries.append((dtstart, t.strip().title()))
+                else:
+                    entries.append((dtstart, summary))
+
+        return entries

custom_components/waste_collection_schedule/waste_collection_schedule/source/ics.py

@@ -5,6 +5,7 @@
 import requests
 from waste_collection_schedule import Collection  # type: ignore[attr-defined]
 from waste_collection_schedule.service.ICS import ICS
+from waste_collection_schedule.service.ICS_v1 import ICS_v1
 
 TITLE = "ICS"
 DESCRIPTION = "Source for ICS based schedules."
@@ -33,7 +34,8 @@
         "file": str(Path(__file__).resolve().parents[1].joinpath("test/recurring.ics"))
     },
     "München, Bahnstr. 11": {
-        "url": "https://www.awm-muenchen.de/entsorgen/abfuhrkalender?tx_awmabfuhrkalender_abfuhrkalender%5Bhausnummer%5D=11&tx_awmabfuhrkalender_abfuhrkalender%5Bleerungszyklus%5D%5BB%5D=1%2F2%3BU&tx_awmabfuhrkalender_abfuhrkalender%5Bleerungszyklus%5D%5BP%5D=1%2F2%3BG&tx_awmabfuhrkalender_abfuhrkalender%5Bleerungszyklus%5D%5BR%5D=001%3BU&tx_awmabfuhrkalender_abfuhrkalender%5Bsection%5D=ics&tx_awmabfuhrkalender_abfuhrkalender%5Bsinglestandplatz%5D=false&tx_awmabfuhrkalender_abfuhrkalender%5Bstandplatzwahl%5D=true&tx_awmabfuhrkalender_abfuhrkalender%5Bstellplatz%5D%5Bbio%5D=70024507&tx_awmabfuhrkalender_abfuhrkalender%5Bstellplatz%5D%5Bpapier%5D=70024507&tx_awmabfuhrkalender_abfuhrkalender%5Bstellplatz%5D%5Brestmuell%5D=70024507&tx_awmabfuhrkalender_abfuhrkalender%5Bstrasse%5D=bahnstr.&tx_awmabfuhrkalender_abfuhrkalender%5Byear%5D={%Y}"
+        "url": "https://www.awm-muenchen.de/entsorgen/abfuhrkalender?tx_awmabfuhrkalender_abfuhrkalender%5Bhausnummer%5D=11&tx_awmabfuhrkalender_abfuhrkalender%5Bleerungszyklus%5D%5BB%5D=1%2F2%3BU&tx_awmabfuhrkalender_abfuhrkalender%5Bleerungszyklus%5D%5BP%5D=1%2F2%3BG&tx_awmabfuhrkalender_abfuhrkalender%5Bleerungszyklus%5D%5BR%5D=001%3BU&tx_awmabfuhrkalender_abfuhrkalender%5Bsection%5D=ics&tx_awmabfuhrkalender_abfuhrkalender%5Bsinglestandplatz%5D=false&tx_awmabfuhrkalender_abfuhrkalender%5Bstandplatzwahl%5D=true&tx_awmabfuhrkalender_abfuhrkalender%5Bstellplatz%5D%5Bbio%5D=70024507&tx_awmabfuhrkalender_abfuhrkalender%5Bstellplatz%5D%5Bpapier%5D=70024507&tx_awmabfuhrkalender_abfuhrkalender%5Bstellplatz%5D%5Brestmuell%5D=70024507&tx_awmabfuhrkalender_abfuhrkalender%5Bstrasse%5D=bahnstr.&tx_awmabfuhrkalender_abfuhrkalender%5Byear%5D={%Y}",
+        "version": 1,
     },
     "Buxtehude, Am Berg": {
         "url": "https://abfall.landkreis-stade.de/api_v2/collection_dates/1/ort/10/strasse/90/hausnummern/1/abfallarten/R02-R04-B02-D04-D12-P04-R12-R14-W0-R22-R24-R31/kalender.ics"
@@ -129,12 +131,16 @@ def __init__(
         year_field=None,
         method="GET",
         split_at=None,
+        version=2,
     ):
         self._url = url
         self._file = file
         if bool(self._url is not None) == bool(self._file is not None):
             raise RuntimeError("Specify either url or file")
-        self._ics = ICS(offset=offset, split_at=split_at)
+        if version == 1:
+            self._ics = ICS_v1(offset=offset, split_at=split_at)
+        else:
+            self._ics = ICS(offset=offset, split_at=split_at)
         self._params = params
         self._year_field = year_field  # replace this field in params with current year
         self._method = method  # The method to send the params