initial commit

CalendarSync.py

@@ -0,0 +1,75 @@
+import caldav
+import datetime
+import logging
+import re
+import vobject
+
+class CalendarSync(object):
+    log = logging.getLogger("CalendarSync")
+
+    startTime = "06:00"
+
+    def __init__(self):
+        self._calendar = None
+        self._client = None
+
+    def connect(self, url, user, passwd):
+        self._client = caldav.DAVClient(url=url, username=user, password=passwd)
+
+    @property
+    def is_connected(self) -> bool:
+        return self._client is not None
+
+    def getExistingMuellEvents(self) -> list[caldav.Event]:
+        events = self._calendar.events()
+        ret = []
+        for e in events:
+            try:
+                uid = e.vobject_instance.vevent.uid.value
+                m = re.match(r'(.+)@MyMuell', uid)
+                if m is not None:
+                    ret.append(e)
+            except AttributeError:
+                continue
+        return ret
+
+    def createEvent(self, uid, summary, date):
+        date_start = datetime.datetime.combine(date, datetime.time.fromisoformat(CalendarSync.startTime))
+        date_end = date_start + datetime.timedelta(minutes=5)
+
+        cal = vobject.iCalendar()
+        ev = cal.add('vevent')
+        ev.add("summary").value = summary
+        ev.add("dtstart").value = date_start
+        ev.add("dtend").value = date_end
+        ev.add("uid").value = str(uid) + "@MyMuell"
+        ev.add("valarm").add("trigger").value = datetime.timedelta(hours=-12)
+        ev.add("valarm").add("trigger").value = datetime.timedelta(hours=0)
+
+        self._calendar.save_event(cal.serialize())
+
+    def syncEvents(self, e):
+        self.createEvent(
+            e["id"],
+            e["title"],
+            datetime.datetime.strptime(e["day"], "%Y-%m-%d"))
+
+    def getCalendars(self):
+        principal = self._client.principal()
+        cals = principal.calendars()
+        ret = []
+        for c in cals:
+            ret.append(c.name)
+        return ret
+
+    def createCalendar(self, cal):
+        principal = self._client.principal()
+        cals = principal.calendars()
+
+        for c in cals:
+            if c.name == cal:
+                self._calendar = c
+
+        if self._calendar is None:
+            CalendarSync.log.info("creating new calendar \"{}\"".format(cal))
+            self._calendar = principal.make_calendar(name=cal)

GuiWorker.py

@@ -0,0 +1,35 @@
+from PyQt5.QtCore import QThread, pyqtSignal
+from PyQt5.QtWidgets import QProgressBar, QPushButton, QStatusBar
+
+
+class GuiWorker(QThread):
+    finished = pyqtSignal(bool, str)
+    rangeChanged = pyqtSignal(int, int)
+    progressChanged = pyqtSignal(int)
+    stateChanged = pyqtSignal(str)
+
+    _button = None
+
+    def __init__(self, runnable, parent=None):
+        QThread.__init__(self, parent)
+        self._callable = runnable
+
+    def connectProgressBar(self, progress_bar: QProgressBar):
+        self.progressChanged.connect(progress_bar.setValue)
+        self.rangeChanged.connect(progress_bar.setRange)
+        progress_bar.setTextVisible(True)
+
+
+    def connectButton(self, button: QPushButton):
+        self._button = button
+        self._button.clicked.connect(self.start)
+
+    def connectStatusBar(self, statusbar: QStatusBar):
+        self.stateChanged.connect(lambda val: statusbar.showMessage(val))
+
+    def run(self):
+        if self._button:
+            self._button.setEnabled(False)
+        ret = self._callable(self)
+        self._button.setEnabled(True)
+        self.finished.emit(ret[0], ret[1])

LocalDataStorage.py

@@ -0,0 +1,75 @@
+from appdirs import *
+import json
+import os
+import re
+import copy
+from cryptography.fernet import Fernet
+
+class LocalDataStorage(object):
+    def __init__(self):
+        self.appname = "MyMuellDav"
+        self.appauthor = "Av3m"
+        self.__fernet = Fernet(b'kWUFurHmtMWX6nOMhpFR45DpuNVPckSQ9t95_ADG2dA=')
+
+        if not os.path.exists(self.user_data_dir):
+            os.makedirs(self.user_data_dir)
+
+    DefaultSettings = {
+        'url': '',
+        'user': '',
+        'password': '',
+        'calendar': '',
+        'mymuellcity': '',
+    }
+
+    @property
+    def user_data_dir(self):
+        return user_data_dir(self.appname, self.appauthor)
+
+    @property
+    def file_settings(self):
+        return os.path.join(self.user_data_dir, "settings.json")
+    @property
+    def file_city_data(self):
+        return os.path.join(self.user_data_dir, "city_data.json")
+
+    @property
+    def settings(self):
+        if os.path.exists(self.file_settings):
+            with open(self.file_settings, "r") as f:
+                j = json.load(f)
+                j["password"] = str(self.__fernet.decrypt(bytes(j["password"], encoding="utf-8")), encoding="utf-8")
+                return j
+        else:
+            return LocalDataStorage.DefaultSettings
+
+    @settings.setter
+    def settings(self, val):
+        if val is None and os.path.exists(self.file_settings):
+            os.remove(self.file_settings)
+            return
+
+        with open(self.file_settings, "w+") as f:
+            v = copy.copy(val)
+            v["password"] = str(self.__fernet.encrypt(bytes(v["password"], encoding="utf-8")), encoding="utf-8")
+            json.dump(v, f)
+
+        os.chmod(self.file_settings, 0o0600)
+
+    @property
+    def city_data(self):
+        if os.path.exists(self.file_city_data):
+            with open(self.file_city_data, "r") as f:
+                return json.load(f)
+        else:
+            return None
+
+    @city_data.setter
+    def city_data(self, val):
+        if val is None and os.path.exists(self.file_city_data):
+            os.remove(self.file_city_data)
+            return
+
+        with open(self.file_city_data, "w+") as f:
+            json.dump(val, f)
+

MyMuell2CalDavGui.py

@@ -0,0 +1,316 @@
+from PyQt5.QtWidgets import \
+    QApplication, \
+    QWidget, \
+    QListWidget, \
+    QListWidgetItem, \
+    QVBoxLayout, \
+    QHBoxLayout, \
+    QGridLayout, \
+    QLineEdit, \
+    QGroupBox, \
+    QLabel, \
+    QPushButton, \
+    QMessageBox, \
+    QComboBox, \
+    QProgressBar, \
+    QStatusBar, \
+    QSizePolicy, \
+    QMainWindow
+
+
+from PyQt5.QtCore import QModelIndex, Qt
+import MyMuellDataModel
+import sys
+import CalendarSync
+
+from GuiWorker import GuiWorker
+
+
+class MyMuell2CalDavGui(QMainWindow):
+    def __init__(self, parent=None):
+        super().__init__(parent)
+
+        self._dataModel = MyMuellDataModel.MyMuellDataModel()
+        self._davClient = CalendarSync.CalendarSync()
+
+        self._selectedCity = None
+
+        self._cities = []
+
+        self._citiesWidget = QListWidget()
+        self._filterText = QLineEdit()
+
+        self._url = QLineEdit()
+        self._user = QLineEdit()
+        self._password = QLineEdit()
+        self._calendarNames = QComboBox()
+        self._connectButton = QPushButton("connect")
+        self._syncButton = QPushButton("sync events")
+        self._deleteButton = QPushButton("delete existing events")
+        self._errorMessage = QMessageBox()
+        self._progressBar = QProgressBar()
+        self._statusBar = QStatusBar()
+
+        self._settings = self._dataModel.storage.settings
+
+        self._url.setText(self._settings["url"])
+        self._password.setText(self._settings["password"])
+        self._user.setText(self._settings["user"])
+
+        self._workerConnect = GuiWorker(self.runnable_connect_caldav)
+        self._workerSync = GuiWorker(self.runnable_sync_events)
+        self._workerDelete = GuiWorker(self.runnable_delete_events)
+
+
+        self.initUI()
+
+        self.__fillCities()
+
+    def entrySelected(self, i: QListWidgetItem):
+        city = self._dataModel.get_city_by_id(i.data(QListWidgetItem.UserType))
+
+        self._selectedCity = (city["id"], city["area_id"])
+
+    def saveSettings(self, val):
+        self._settings["url"] = self._url.text()
+        self._settings["user"] = self._user.text()
+        self._settings["password"] = self._password.text()
+
+        if self._calendarNames.currentText() != '':
+            self._settings["calendar"] = self._calendarNames.currentText()
+
+        if len(self._citiesWidget.selectedItems()) > 0:
+            self._settings["mymuellcity"] = self._citiesWidget.selectedItems()[0].text()
+
+        self._dataModel.storage.settings = self._settings
+
+    def runnable_sync_events(self, worker: GuiWorker) -> tuple[bool, str]:
+        if self._selectedCity is None:
+            return False, "please select a city"
+
+        if self._calendarNames.currentText() == '':
+            return False, "please select a calendar"
+
+        worker.stateChanged.emit("create calendar {} if not existent".format(self._calendarNames.currentText()))
+
+        self._davClient.createCalendar(self._calendarNames.currentText())
+
+        worker.stateChanged.emit("get events from MyMüll.de (city id {}, aread id {}".format(*self._selectedCity))
+        events = self._dataModel.get_events(*self._selectedCity)
+        worker.rangeChanged.emit(0, len(events))
+        worker.progressChanged.emit(0)
+
+        for i in range(0, len(events)):
+            worker.stateChanged.emit("creating event {} {}".format(events[i]["title"], events[i]["day"]))
+            self._davClient.syncEvents(events[i])
+            worker.progressChanged.emit(i+1)
+
+        worker.stateChanged.emit("syncing events finished")
+        return True, "syncing events finished"
+
+    def runnable_delete_events(self, worker: GuiWorker) -> tuple[bool, str]:
+        if self._calendarNames.currentText() == '':
+            return False, "please select a calendar"
+
+        worker.stateChanged.emit("deleting existing events from calendar")
+
+        self._davClient.createCalendar(self._calendarNames.currentText())
+        events = self._davClient.getExistingMuellEvents()
+        if len(events) > 0:
+            worker.rangeChanged.emit(0, len(events))
+            worker.progressChanged.emit(0)
+        else:
+            worker.rangeChanged.emit(0, 1)
+            worker.progressChanged.emit(1)
+
+        for i in range(0, len(events)):
+            worker.stateChanged.emit("deleting event {}".format(events[i].vobject_instance.vevent.uid.value))
+            events[i].delete()
+            worker.progressChanged.emit(i+1)
+
+        worker.stateChanged.emit("deletion finished.")
+        return True, "deleting events finished"
+
+    def runnable_connect_caldav(self, worker: GuiWorker) -> tuple[bool, str]:
+
+
+        try:
+            worker.stateChanged.emit("connecting to {}".format(self._url.text()))
+            self._davClient.connect(self._url.text(), self._user.text(), self._password.text())
+            self._calendarNames.blockSignals(True)
+            for i in self._davClient.getCalendars():
+                self._calendarNames.addItem(i)
+
+            self._calendarNames.blockSignals(False)
+            self._calendarNames.setEnabled(True)
+
+            if self._settings["calendar"] != '':
+                self._calendarNames.setCurrentText(self._settings["calendar"])
+
+
+
+            worker.stateChanged.emit("connected.")
+
+
+
+
+        except Exception as e:
+            worker.stateChanged.emit("connection failed.")
+            return False, str(e)
+
+        return True, "connect successful"
+
+    def initUI(self):
+
+        tlWidget = QWidget()
+        layout = QGridLayout()
+        tlWidget.setLayout(layout)
+
+        self.setCentralWidget(tlWidget)
+
+
+
+        groupBoxMyMuell = QGroupBox("MyMüll.de Cities")
+        layoutGroupBoxMyMuell = QGridLayout()
+        groupBoxMyMuell.setLayout(layoutGroupBoxMyMuell)
+
+        groupBoxCalDav = QGroupBox("CalDAV Settings")
+        layoutGroupBoxCalDav = QGridLayout()
+        groupBoxCalDav.setLayout(layoutGroupBoxCalDav)
+
+        groupBoxProgress = QGroupBox("Progress")
+        layoutGroupBoxProgress = QVBoxLayout()
+        groupBoxProgress.setLayout(layoutGroupBoxProgress)
+
+        layoutGroupBoxMyMuell.addWidget(self._citiesWidget, 0, 0, 4, 6)
+        layoutGroupBoxMyMuell.addWidget(QLabel("Filter Cities"), 4, 0, 1, 1)
+        layoutGroupBoxMyMuell.addWidget(self._filterText, 4, 1, 1, 5)
+
+        layout.addWidget(groupBoxMyMuell)
+        layout.addWidget(groupBoxCalDav)
+        layout.addWidget(groupBoxProgress)
+
+        self.setStatusBar(self._statusBar)
+
+        layoutGroupBoxCalDav.addWidget(QLabel("url"), 0, 0, 1, 1)
+        layoutGroupBoxCalDav.addWidget(self._url, 0, 1, 1, 5)
+        layoutGroupBoxCalDav.addWidget(QLabel("username"), 1, 0, 1, 1)
+        layoutGroupBoxCalDav.addWidget(self._user, 1, 1, 1, 5)
+        layoutGroupBoxCalDav.addWidget(QLabel("password"), 2, 0, 1, 1)
+        layoutGroupBoxCalDav.addWidget(self._password, 2, 1, 1, 5)
+        layoutGroupBoxCalDav.addWidget(QLabel("calendar"), 3, 0, 1, 1)
+        layoutGroupBoxCalDav.addWidget(self._calendarNames, 3, 1, 1, 5)
+
+        buttonLayout = QHBoxLayout()
+        buttonLayout.addWidget(self._connectButton)
+        buttonLayout.addWidget(self._syncButton)
+        buttonLayout.addWidget(self._deleteButton)
+
+        layoutGroupBoxCalDav.addLayout(buttonLayout, 4, 0, 6, 6)
+
+        layoutGroupBoxProgress.addWidget(self._progressBar)
+        layoutGroupBoxCalDav.setSpacing(0)
+        layoutGroupBoxCalDav.setContentsMargins(0, 0, 0, 0)
+
+        self._password.setEchoMode(QLineEdit.Password)
+
+        self._calendarNames.setEditable(True)
+
+        self._filterText.show()
+        self._citiesWidget.show()
+
+        self._syncButton.setEnabled(False)
+        self._calendarNames.setEnabled(False)
+
+        self._citiesWidget.currentItemChanged.connect(lambda cur, prev: self.entrySelected(cur))
+        self._filterText.textChanged.connect(self.__fillCities)
+
+        self._url.textChanged.connect(self.saveSettings)
+        self._user.textChanged.connect(self.saveSettings)
+        self._password.textChanged.connect(self.saveSettings)
+
+        self._citiesWidget.itemSelectionChanged.connect(lambda: self.saveSettings(0))
+
+        self._workerSync.connectProgressBar(self._progressBar)
+        self._workerSync.connectButton(self._syncButton)
+
+        self._workerConnect.connectButton(self._connectButton)
+        self._workerDelete.connectButton(self._deleteButton)
+
+        self._workerSync.connectStatusBar(self._statusBar)
+        self._workerConnect.connectStatusBar(self._statusBar)
+        self._workerDelete.connectStatusBar(self._statusBar)
+        self._workerDelete.connectProgressBar(self._progressBar)
+
+        self._workerConnect.finished.connect(self.slot_process_finished)
+        self._workerSync.finished.connect(self.slot_process_finished)
+        self._workerDelete.finished.connect(self.slot_process_finished)
+
+        self._calendarNames.currentTextChanged.connect(self.slot_calendar_selected)
+
+        self._progressBar.setSizePolicy(QSizePolicy(QSizePolicy.Minimum, QSizePolicy.Fixed))
+        groupBoxProgress.setSizePolicy(QSizePolicy(QSizePolicy.Minimum, QSizePolicy.Fixed))
+        groupBoxCalDav.setSizePolicy(QSizePolicy(QSizePolicy.Minimum, QSizePolicy.Fixed))
+
+        self._url.setContentsMargins(0, 10, 0, 10)
+        self._password.setContentsMargins(0, 10, 0, 10)
+        self._user.setContentsMargins(0, 10, 0, 10)
+        self._calendarNames.setContentsMargins(0, 10, 0, 10)
+
+
+        #self.setGeometry(300, 300, 1000, 800)
+        self.setMinimumWidth(800)
+        self.setWindowTitle("MyMuell DAV GUI")
+
+
+
+    def slot_process_finished(self, result: bool, msg: str):
+        if result:
+            self._errorMessage.information(self, "info", msg)
+            self._syncButton.setEnabled(True)
+        else:
+            self._errorMessage.critical(self, "error", msg)
+            self._syncButton.setEnabled(False)
+
+    def slot_calendar_selected(self):
+
+        if self._calendarNames.currentText() != '' and self._davClient.is_connected:
+            self.saveSettings(0)
+            self._syncButton.setEnabled(True)
+        else:
+            self._syncButton.setEnabled(False)
+
+    def __fillCities(self, pattern=".+"):
+        self._citiesWidget.blockSignals(True)
+        self._citiesWidget.clear()
+
+        self._cities = self._dataModel.match_city(pattern)
+
+        for i in self._cities:
+            c = self._dataModel.get_city_by_index(i)
+            item = QListWidgetItem()
+
+            item.setData(QListWidgetItem.UserType, c["id"])
+            item.setText(c["name"])
+
+            self._citiesWidget.addItem(item)
+
+        self._citiesWidget.blockSignals(False)
+
+        if self._settings["mymuellcity"] != '':
+            items = self._citiesWidget.findItems(self._settings["mymuellcity"], Qt.MatchExactly)
+            if len(items) > 0:
+                self._citiesWidget.setCurrentItem(items[0])
+
+
+
+
+def main():
+    app = QApplication(sys.argv)
+    w = MyMuell2CalDavGui()
+    w.show()
+    sys.exit(app.exec_())
+
+
+if __name__ == "__main__":
+    main()

MyMuellDataModel.py

@@ -0,0 +1,78 @@
+import http.client
+import json
+import LocalDataStorage
+from appdirs import *
+import logging
+import re
+
+class MyMuellDataModel(object):
+    MyMuellHost = 'mymuell.jumomind.com'
+    log = logging.getLogger("MyMuellDataModel")
+
+    def __init__(self):
+        self.client = http.client.HTTPSConnection(MyMuellDataModel.MyMuellHost)
+
+        self.storage = LocalDataStorage.LocalDataStorage()
+        self.cities = self.__get_cities()
+
+    def get_cities_by_request(self):
+        self.client.request('GET', '/mmapp/loxone/lox.php?r=cities')
+        response = self.client.getresponse()
+        return json.loads(str(response.read(), encoding='utf-8'))
+
+    def get_events(self, city_id, area_id):
+        self.client.request('GET', '/mmapp/loxone/lox.php?r=dates/0&city_id={city_id}&area_id={area_id}'.format(city_id=city_id, area_id=area_id))
+        response = self.client.getresponse()
+        ret = str(response.read(), encoding='utf-8')
+        return json.loads(ret)
+
+    def __get_cities(self):
+        cities = self.storage.city_data
+        if cities is not None:
+            MyMuellDataModel.log.debug("using stored values")
+            return cities
+        else:
+            cities = self.get_cities_by_request()
+            self.storage.city_data = cities
+
+        return cities
+
+    def get_city_names(self, indices):
+        ret = []
+        for idx in indices:
+            e = self.get_city_by_index(idx)
+            if e is not None:
+                ret.append(e["name"])
+        return ret
+
+    def match_city(self, pattern):
+        ret = []
+
+        n = 0
+        for i in self.cities:
+
+            m = re.search(pattern, i["name"], re.IGNORECASE)
+            if m is not None:
+                ret.append(n)
+            n = n + 1
+
+        return ret
+
+    def get_city_by_index(self, idx):
+        if len(self.cities) < idx:
+            return None
+
+        return self.cities[idx]
+
+    def get_city_by_id(self, id):
+        for i in self.cities:
+            if i["id"] == id:
+                return i
+
+        return None
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    model = MyMuellDataModel()
+    matches = model.match_city("eich")

README.md

@@ -0,0 +1,65 @@
+# MyMuell 2 CalDAV
+
+Simple tool to retrive dates for garbage disposal from _MyMüll.de_ and sync it with a CalDAV Server, e.g. Nextcloud
+
+
+## Motivation
+_MyMüll.de_ is a web service / app provider which is used by some bavarian municipalities to digitally announce upcoming dates for garbage disposal.
+Although this is basically a good idea, _MyMüll.de_ does not provide any convenience functionality to import these dates in a standard calendar to use your favourite calendar app.
+
+Instead, users are forced to install the buggy and heavy battery draining _MyMüll.de_ smartphone app. On some android devices (e.g. Huawei), even the notifications of the app do not work reliable by default
+(only after changing some settings related to app start policies)
+
+So this tool aims to get rid of the app by parsing data from _MyMüll.de_ web service and synchronize the events with a conventional CalDAV server.
+
+
+## implementation status
+- all required functionality implemented for basic usage
+- functional GUI written in PyQt5
+- persist all settings for later use
+- works on Ubuntu 20.04 and Windows 10
+- works reliable in combination with Nextcloud 20 with official Calendar plugin
+
+
+## ToDos
+- more GUI settings
+  - adjustable notification triggers for upcoming events
+  - adjustable start time and event duration
+    
+- test/implement other CalDAV providers
+  - iCloud
+  - Google
+    
+- export the CalDAV events to a local file for manual import to a calendar
+
+
+#usage
+- get url of your CalDAV principal.
+  For Nextcloud, this `http://your.nextcloud.host/nextcloud/remote.php/dav/calendars`
+  
+- Start GUI 
+
+
+#Disclaimer
+This tool was developed by an annoyed _MyMüll.de_ app user as a free contribute 
+to the open source community and is licenced under the **GPLv3** Licence.
+
+This software does not stand in any relation to the official _MyMüll.de_ app or the company that provides/developes this service.
+Although this software was developed by an experienced software developer to the best of his knowledge and belief 
+and was basically tested on different platforms, the author can not guarantee for the proper functionality of the software.
+
+So the user uses this software at his own risk and he is completely responsible for any damage, security issues, data loss or any additional costs, 
+that might occur when using this software.
+
+
+
+
+
+
+
+
+
+
+
+
+

requirements.txt

@@ -0,0 +1,5 @@
+pyqt5
+caldav
+vobject
+appdirs
+cryptography