Home-Assistant integration of 'Abfall.IO' waste collection dates



I wanted the garbage collection dates on my HA dashboard. Unfortunately, the city of Landshut doesn't offer an API to access them directly. Instead you can download a garbage collection calendar in PDF format from the garbage disposal portal of the city of Landshut. After some analysis I found out that a third party service is used here: "Abfall.IO/AbfallPlus"

Waste Collection Schedule as PDF

After viewing the information on the homepage of the provider (unfortunately not too many), I managed to get the data in the form of a CSV file.

I can now parse this data and display it as a sensor in Home-Assistant.

Next Waste Collection Dates in Home-Assistant Lovelace

And as mentioned in the article originally, I want to see them in my HA Dashboard of course:

Next Waste Collection Dates in Home-Assistant Lovelace

Configuration

/home-assistant/customcomponents/sensor/abfalllandshut.py

import logging
import requests
import csv
from datetime import datetime
from datetime import timedelta
import voluptuous as vol
from pprint import pprint

from homeassistant.components.sensor import PLATFORM_SCHEMA
import homeassistant.helpers.config_validation as cv
from homeassistant.const import (CONF_RESOURCES)
from homeassistant.util import Throttle
from homeassistant.helpers.entity import Entity

_LOGGER = logging.getLogger(__name__)

MIN_TIME_BETWEEN_UPDATES = timedelta(days=1)

SENSOR_PREFIX = 'Waste '

SENSOR_TYPES = {
    'restmuell': ['Restmüll', '', 'mdi:recycle'],
    'gelbersack': ['Gelber Sack', '', 'mdi:recycle'],
    'papiertonne': ['Papier', '', 'mdi:recycle'],
}

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_RESOURCES, default=[]):
        vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]),
})


def setup_platform(hass, config, add_entities, discovery_info=None):
    _LOGGER.debug("Setup Abfall API retriever")

    try:
        data = AbfallData()
    except requests.exceptions.HTTPError as error:
        _LOGGER.error(error)
        return False

    entities = []

    for resource in config[CONF_RESOURCES]:
        sensor_type = resource.lower()

        if sensor_type not in SENSOR_TYPES:
            SENSOR_TYPES[sensor_type] = [
                sensor_type.title(), '', 'mdi:flash']

        entities.append(AbfallSensor(data, sensor_type))

    add_entities(entities)


class AbfallData(object):

    def __init__(self):
        self.data = None

    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        _LOGGER.debug("Updating Abfall dates using remote API")
        try:
            payload = {
                "f_id_kommune": "2655",
                "f_id_bezirk": "2655",
                "f_id_strasse": "763",
                "f_id_abfalltyp_0": "31",
                "f_id_abfalltyp_1": "17",
                "f_id_abfalltyp_2": "19",
                "f_id_abfalltyp_3": "218",
                "f_abfallarten_index_max": "4",
                "f_abfallarten": "31,17,19,218,31,17,19,218",
                "f_zeitraum": "20190101-20301231"
            }

            j = requests.post(
                "http://api.abfall.io/?key=bd0c2d0177a0849a905cded5cb734a6f&modus=d6c5855a62cf32a4dadbc2831f0f295f&waction=export_csv", data=payload, timeout=10)

            apiRequest = j.text.split('\n')
            reader = csv.reader(apiRequest, delimiter=";")
            rowCounter = 0
            columns = None
            gelberSack = []
            restMuell = []
            papierTonne = []

            for row in reader:
                if rowCounter == 0:
                    columns = {k:row.index(k) for k in row}
                
                else:
                    if (row[columns["Gelber Sack"]] != ""):
                        gelberSack.append(datetime.strptime(row[columns["Gelber Sack"]], "%d.%m.%Y"))

                    if (row[columns["Restabfall"]] != ""):
                        restMuell.append(datetime.strptime(row[columns["Restabfall"]], "%d.%m.%Y"))

                    if (row[columns["Papiertonne"]] != ""):
                        papierTonne.append(datetime.strptime(row[columns["Papiertonne"]], "%d.%m.%Y"))

                rowCounter = rowCounter + 1

            gelberSack.sort(key=lambda date: date)
            restMuell.sort(key=lambda date: date)
            papierTonne.sort(key=lambda date: date)

            nextDates = {}

            for nextDate in gelberSack:
                if nextDate > datetime.now():
                    nextDates["gelberSack"] = nextDate
                    break

            for nextDate in restMuell:
                if nextDate > datetime.now():
                    nextDates["restMuell"] = nextDate
                    break

            for nextDate in papierTonne:
                if nextDate > datetime.now():
                    nextDates["papierTonne"] = nextDate
                    break

            self.data = nextDates

        except requests.exceptions.RequestException as exc:
            _LOGGER.error("Error occurred while fetching data: %r", exc)
            self.data = None
            return False


class AbfallSensor(Entity):

    def __init__(self, data, sensor_type):
        self.data = data
        self.type = sensor_type
        self._name = SENSOR_PREFIX + SENSOR_TYPES[self.type][0]
        self._unit = SENSOR_TYPES[self.type][1]
        self._icon = SENSOR_TYPES[self.type][2]
        self._state = None
        self._attributes = {}

    @property
    def name(self):
        return self._name

    @property
    def icon(self):
        return self._icon

    @property
    def state(self):
        return self._state

    @property
    def unit_of_measurement(self):
        return self._unit

    @property
    def device_state_attributes(self):
        """Return attributes for the sensor."""
        return self._attributes

    def update(self):
        self.data.update()
        abfallData = self.data.data

        try:
            if self.type == 'gelbersack':
                self._state = abfallData.get("gelberSack")
            elif self.type == 'restmuell':
                self._state = abfallData.get("restMuell")
            elif self.type == 'papiertonne':
                self._state = abfallData.get("papierTonne")

            if self._state is not None:
                weekdays = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]
                self._attributes['days'] = (self._state.date() - datetime.now().date()).days
                if self._attributes['days'] == 0:
                    printtext = "heute"
                elif self._attributes['days'] == 1:
                    printtext = "morgen"
                else:
                    printtext = 'in {} Tagen'.format(self._attributes['days'])
                self._attributes['display_text'] = self._state.strftime(
                    '{}, %d.%m.%Y ({})').format(weekdays[self._state.weekday()], printtext)

        except ValueError:
            self._state = None

/home-assistant/configuration.yaml

- platform: abfall_landshut
  resources:
    - gelbersack
    - restmuell
    - papiertonne

- platform: template
  sensors:
    gelbersack_text:
      value_template: "{{states.sensor.waste_gelber_sack.attributes.display_text}}"
      friendly_name: 'Gelber Sack'
      icon_template: mdi:sack
    papiertonne_text:
      value_template: "{{states.sensor.waste_papier.attributes.display_text}}"
      friendly_name: 'Papier'
      icon_template: mdi:delete-empty
    restmuell_text:
      value_template: "{{states.sensor.waste_restmull.attributes.display_text}}"
      friendly_name: 'Restmüll'
      icon_template: mdi:trash-can-outline

/home-assistant/automations/notifications.yaml

- alias: 'Notify: Garbage (Gelber Sack)'
  trigger:
  - at: '18:30:00'
    platform: time
  condition:
  - condition: template
    value_template: '{{ states.sensor.waste_gelber_sack.attributes.days <= 1 }}'
  action:
    service: notify.notify
    data_template:
      title: Gelber Sack
      message: '*Gelber Sack* rausbringen. Abholung *morgen*, am: {{states.sensor.waste_gelber_sack.attributes.display_text}}'


- alias: 'Notify: Garbage (Papiertonne)'
  trigger:
  - at: '18:30:00'
    platform: time
  condition:
  - condition: template
    value_template: '{{ states.sensor.waste_papier.attributes.days <= 1 }}'
  action:
    service: notify.notify
    data_template:
      title: Papiertonne
      message: '*Papiertonne* rausbringen. Abholung *morgen*, am: {{states.sensor.waste_papier.attributes.display_text}}'


- alias: 'Notify: Garbage (Restmüll)'
  trigger:
  - at: '18:30:00'
    platform: time
  condition:
  - condition: template
    value_template: '{{ states.sensor.waste_restmull.attributes.days <= 1 }}'
  action:
    service: notify.notify
    data_template:
      title: Restmüll
      message: '*Restmüll* rausbringen. Abholung *morgen*, am: {{states.sensor.waste_restmull.attributes.display_text}}'

Links

I posted an link to this article in the community of Home-Assistant. I got a lot of good feedback there and also some improvements for the scripts above. I added these improvements already on this page. Thats the link to the post (with the comments):


Comments