Read_EGN_Abfallkalender_Programmatically_With_Python.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Read-EGN-Abfallkalender-Programmatically-With-Python.ipynb",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0Iq-Qol5cypY",
        "colab_type": "text"
      },
      "source": [
        "# Read EGN Abfallkalender programmatically with Python\n",
        "\n",
        "This code enables you to read the EGN Abfallkalender (https://www.egn-abfallkalender.de/kalender) programmatically with Python. \n",
        "\n",
        "The EGN Abfallkalender offers appointments for the following German cities:\n",
        "* Brüggen\n",
        "* Dormagen\n",
        "* Gefrath\n",
        "* Grevenbroich\n",
        "* Nettetal\n",
        "* Niederkrüchten\n",
        "* Schwalmtal\n",
        "* Tönisvorst\n",
        "* Willich\n",
        "\n",
        "You can use it to integrate the EGN appointments e.g. within your own Smart Home solution.\n",
        "\n",
        "Note: \n",
        "* This is only a small excerpt from my entire solution for the EGN calendar integration. I focus only on the general mechanism to request the data from the calendar for subsequent use\n",
        "* My experience is also that the online calendar is sometimes not in sync with the paper-based calendar, in most cases the paper-based calendar \"wins\". I assume that this is only the result of a wrong (maybe also handmade?!) maintenance of the digital version...\n",
        "\n",
        "## Installation\n",
        "1. Download this Jupyter notebook or copy the code e.g. within a new [Google Colab notebook](https://colab.research.google.com/)\n",
        "2. Replace the given address parts in line 19 - 22 with your personal address\n",
        "3. Run the code in your environment, and enjoy the output!"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "NJ6AlDWsQs-Y",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 258
        },
        "outputId": "e2afc876-b680-485f-df37-f3e26b57ad88"
      },
      "source": [
        "\"\"\"\n",
        "    Author: Anja Kuchenbecker\n",
        "    License: MIT\n",
        "    Email: anja@anjalytics.com\n",
        "\"\"\"\n",
        "\n",
        "# External Imports\n",
        "import requests\n",
        "from bs4 import BeautifulSoup\n",
        "import datetime\n",
        "# Internal Imports\n",
        "#NA\n",
        "\n",
        "class EgnCalendar:\n",
        "  def __init__(self):\n",
        "    # Target URL\n",
        "    self.EGN_URL = \"https://www.egn-abfallkalender.de/kalender\"\n",
        "\n",
        "    # Replace the following lines with your specific data\n",
        "    self.EGN_CITY = \"Brüggen\"\n",
        "    self.EGN_DISTRICT = \"Brüggen\"\n",
        "    self.EGN_STREET = \"Hochstrasse\"\n",
        "    self.EGN_STREET_NUMBER = \"68\"\n",
        "\n",
        "    # Determine current datetime and extract relevant datetime information\n",
        "    now = datetime.datetime.now()\n",
        "    self.current_year = now.year\n",
        "    self.current_month = now.month\n",
        "    self.current_day = now.day\n",
        "    self.current_cweek = datetime.date(int(self.current_year), int(self.current_month), int(self.current_day)).isocalendar()[1]\n",
        "\n",
        "    # Dictionary for en_de weekday mapping\n",
        "    self.trans_dayname_en_de = {\n",
        "      \"Monday\": \"Montag\",\n",
        "      \"Tuesday\": \"Dienstag\",\n",
        "      \"Wednesday\": \"Mittwoch\",\n",
        "      \"Thursday\": \"Donnerstag\",\n",
        "      \"Friday\": \"Freitag\",\n",
        "      \"Saturday\": \"Samstag\",\n",
        "      \"Sunday\": \"Sonntag\"\n",
        "    }\n",
        "\n",
        "    # Dictionary for EGN abbreviation color mapping\n",
        "    self.trans_egnabbreviation_color = {\n",
        "      \"PPK\": \"Blaue Tonne\",  # Blaue-Tonne\n",
        "      \"HML\": \"Graue Tonne\",  # Graue-Tonne\n",
        "      \"LVP\": \"Gelbe Tonne\",  # Gelbe-Tonne\n",
        "      \"BIO\": \"Bio Tonne\",  # Bio-Tonne\n",
        "      \"WBM\": \"Bio Tonne\"  # Bio-Tonne\n",
        "    }\n",
        "\n",
        "  # Determines and returns dayname from date information\n",
        "  def date_to_dayname(self, input_date):\n",
        "    day_name = datetime.datetime.strptime(input_date, '%Y-%m-%d')\n",
        "    return day_name.strftime(\"%A\")\n",
        "\n",
        "  # Request the https://www.egn-abfallkalender.de/kalender page\n",
        "  def request_calendar(self):\n",
        "    try:\n",
        "      ######  Perform first request ######\n",
        "      # in order to retrieve xsrf-token, egn-session and csrf-token for second request\n",
        "\n",
        "      # Step 1: Define request headers\n",
        "      request_headers = {\n",
        "        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:78.0) Gecko/20100101 Firefox/78.0',\n",
        "        'Accept': '*/*'\n",
        "      }\n",
        "\n",
        "      # Step 2: Perform request\n",
        "      response = requests.get(self.EGN_URL,\n",
        "                              headers=request_headers,\n",
        "                              verify=True)\n",
        "\n",
        "      # Step 3: Process response\n",
        "      # Get xsrf-token and egn-session from first request (coming from response)\n",
        "      xsrf_token = response.cookies.get(\"XSRF-TOKEN\")\n",
        "      egn_abfallkalender_session = response.cookies.get(\"egn_abfallkalender_session\")\n",
        "      # Get csrf-token (coming from meta-tag in reponse body)\n",
        "      soup = BeautifulSoup(response.text, features=\"html.parser\")\n",
        "      metatag_csrf_token = soup.find_all('meta', attrs={'name': 'csrf-token'})\n",
        "      csrf_token = metatag_csrf_token[0].get(\"content\")\n",
        "\n",
        "      ######  Perform second request ######\n",
        "      # in order to retrieve calendar data (in json format) we're interested in\n",
        "\n",
        "      # Step 1: Define request cookies, headers and data\n",
        "      request_cookies = {\n",
        "        'XSRF-TOKEN': xsrf_token,\n",
        "        'egn_abfallkalender_session': egn_abfallkalender_session\n",
        "      }\n",
        "\n",
        "      request_headers = {\n",
        "        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:78.0) Gecko/20100101 Firefox/78.0',\n",
        "        'Accept': '*/*',\n",
        "        'X-CSRF-TOKEN': csrf_token,\n",
        "      }\n",
        "\n",
        "      request_data = {\n",
        "        'city': self.EGN_CITY,\n",
        "        'district': self.EGN_DISTRICT,\n",
        "        'street': self.EGN_STREET,\n",
        "        'street_number': self.EGN_STREET_NUMBER\n",
        "      }\n",
        "\n",
        "      # Step 2: Perform second request to retrieve calendar data\n",
        "      response = requests.post(self.EGN_URL,\n",
        "                               headers=request_headers,\n",
        "                               cookies=request_cookies,\n",
        "                               data=request_data)\n",
        "\n",
        "      # Step 3: Process reponse\n",
        "      # Convert reponse to json format\n",
        "      # This is the original output coming from egn calendar request\n",
        "      waste_discharge = response.json()\n",
        "\n",
        "      # Create dictionary that holds our filtered calendar\n",
        "      egn_appointments = dict()\n",
        "\n",
        "      # Iterate through each element\n",
        "      for month_iter in range(1, 13):\n",
        "        current_item = waste_discharge[\"waste_discharge\"][str(self.current_year)][str(month_iter)]\n",
        "        # Consider only current month\n",
        "        if self.current_month == month_iter:\n",
        "          # Handle current month items\n",
        "          for key, value in current_item.items():\n",
        "            # Determine calender week of current item\n",
        "            item_cweek = datetime.date(int(self.current_year), int(self.current_month), int(key)).isocalendar()[1]\n",
        "            # Consider only current and next week calendar week items\n",
        "            if (self.current_cweek == item_cweek) or (self.current_cweek+1 == item_cweek):\n",
        "              date_key = str(self.current_year) + \"-\" + str(self.current_month).rjust(2, '0') + \"-\" + str(key).rjust(2, '0')\n",
        "              egn_appointments.update({date_key: value})\n",
        "\n",
        "      # ----------------> # Return (OK)\n",
        "      return waste_discharge, egn_appointments\n",
        "\n",
        "    except Exception as e:\n",
        "      # ----------------> # Return (NOK)\n",
        "      return f'{e}'\n",
        "\n",
        "\n",
        "# Create EGN Calendar instance and request the calendar\n",
        "egn_calendar = EgnCalendar()\n",
        "egn_calendar_original, egn_calendar_filtered = egn_calendar.request_calendar()\n",
        "\n",
        "# Print original output coming from egn calendar request\n",
        "print(\"Original EGN Calendar Output:\\n\", egn_calendar_original)\n",
        "\n",
        "# Print filtered output considers only current and next week calender items\n",
        "print(\"\\n\\n\")\n",
        "print(\"Filtered output, this and next week calendar items:\\n\", egn_calendar_filtered)\n",
        "\n",
        "# Print formatted text based on filtered calendar\n",
        "print(\"\\n\\n\")\n",
        "print(\"Müllabfuhrtermine diese und nächste Woche:\")\n",
        "for key, value in egn_calendar_filtered.items():\n",
        "  key_date = datetime.datetime.strptime(key, '%Y-%m-%d')\n",
        "  current_weekday_name = egn_calendar.date_to_dayname(key)\n",
        "  for i in value:\n",
        "    # Skip hml2-value\n",
        "    if i == \"hml2\":\n",
        "      continue\n",
        "    print(f\"---> Am {egn_calendar.trans_dayname_en_de[current_weekday_name]} ({key}) wird die {egn_calendar.trans_egnabbreviation_color[str(value[0])]} abgeholt!\")"
      ],
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Original EGN Calendar Output:\n",
            " {'waste_discharge': {'2020': {'1': {'2': ['BIO'], '9': ['LVP'], '14': ['HML'], '15': ['BIO'], '16': ['PPK'], '23': ['LVP'], '28': ['HML'], '29': ['BIO']}, '2': {'6': ['LVP'], '11': ['HML'], '12': ['BIO'], '13': ['PPK'], '20': ['LVP'], '26': ['HML'], '27': ['BIO']}, '3': {'5': ['LVP'], '10': ['HML'], '11': ['BIO'], '12': ['PPK'], '19': ['LVP'], '24': ['HML'], '25': ['BIO']}, '4': {'2': ['LVP'], '6': ['HML'], '7': ['BIO'], '9': ['PPK'], '17': ['LVP'], '21': ['HML'], '22': ['BIO'], '29': ['LVP']}, '5': {'5': ['HML'], '6': ['BIO'], '7': ['PPK'], '14': ['LVP'], '19': ['HML'], '20': ['BIO'], '28': ['LVP']}, '6': {'3': ['HML', 'BIO'], '5': ['PPK'], '12': ['LVP'], '16': ['HML'], '17': ['BIO'], '25': ['LVP'], '30': ['HML']}, '7': {'1': ['BIO'], '2': ['PPK'], '9': ['LVP'], '14': ['HML'], '15': ['BIO'], '23': ['LVP'], '28': ['HML'], '29': ['BIO'], '30': ['PPK']}, '8': {'6': ['LVP'], '11': ['HML'], '12': ['BIO'], '20': ['LVP'], '25': ['HML'], '26': ['BIO'], '27': ['PPK']}, '9': {'3': ['LVP'], '8': ['HML'], '9': ['BIO'], '17': ['LVP'], '22': ['HML'], '23': ['BIO'], '24': ['PPK']}, '10': {'1': ['LVP'], '6': ['HML'], '7': ['BIO'], '15': ['LVP'], '20': ['HML'], '21': ['BIO'], '22': ['PPK'], '29': ['LVP']}, '11': {'3': ['HML'], '4': ['BIO'], '12': ['LVP'], '17': ['HML'], '18': ['BIO'], '19': ['PPK'], '26': ['LVP']}, '12': {'1': ['HML'], '2': ['BIO'], '10': ['LVP'], '15': ['HML'], '16': ['BIO'], '17': ['PPK'], '23': ['LVP'], '29': ['HML'], '30': ['BIO']}}}, 'holidays': {'01-01-2020': 'Neujahr', '05-01-2020': 'Tag der Arbeit', '10-03-2020': 'Tag der Deutschen Einheit', '12-25-2020': '1. Weihnachtstag', '12-26-2020': '2. Weihnachtstag', '11-01-2020': 'Allerheiligen', '04-10-2020': 'Karfreitag', '04-12-2020': 'Ostersonntag', '04-13-2020': 'Ostermontag', '05-21-2020': 'Christi Himmelfahrt', '06-01-2020': 'Pfingstmontag', '06-11-2020': 'Fronleichnam'}, 'trash_type_colors': {'ppk': 'blau', 'hml': 'grau', 'hml2': None, 'lvp': 'gelb', 'bio': 'braun', 'wbm': 'braun'}}\n",
            "\n",
            "\n",
            "\n",
            "Filtered output, this and next week calendar items:\n",
            " {'2020-06-25': ['LVP'], '2020-06-30': ['HML']}\n",
            "\n",
            "\n",
            "\n",
            "Müllabfuhrtermine diese und nächste Woche:\n",
            "---> Am Donnerstag (2020-06-25) wird die Gelbe Tonne abgeholt!\n",
            "---> Am Dienstag (2020-06-30) wird die Graue Tonne abgeholt!\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JWDhqKW8dWw7",
        "colab_type": "text"
      },
      "source": [
        "# Summary\n",
        "* The first output represents the original EGN Calendar output\n",
        "* The second output represents a filtered output, considering only this and next week calendar items\n",
        "* The third output gives you a formatted text version, based on the filtered output\n",
        "\n",
        "\n",
        "# What's next?\n",
        "You can integrate this logic within different web service (e.g. RESTful APIs with flask) as I did within my self-made Smart Home solution. \n",
        "\n",
        "E.g. a web service \n",
        "* that can be called repeatedly (e.g. with support of a Cron job) to request and save the calendar data in filtered form \n",
        "* that can be called repeatedly to check if there are upcoming appointments the next day and if yes sends an e-mail reminder\n",
        "* that returns ready-to-use html content that you can directly integrate within you web frontend (e.g. AJAX request against this web service)"
      ]
    }
  ]
}