Homeassistant Tracking Waste Collection Dates

I want to have the next waste disposal dates in Homeassistant. There is an all-solved-solution (hacs_waste_collection_schedule) that also supports the city I live in. But this is a lot of code to copy around when not using Home Assistant Community Store, which I try to not use. The following solution has a lot less code and when it breaks it is easy for me to fix.

In February I wrote a bit of code to parse the ical for waste collection dates from the city of Stuttgart and documented it in a blog post. Let us modify this script to generate the json we need for a template webhook upload in homeassistant.

But first the homeassistant configuration that is needed:

  - trigger:
      - platform: webhook
        webhook_id: !secret waste-disposal
          - POST
        local_only: true
    unique_id: "waste"
      - name: "waste: residual"
        state: "{{ trigger.json.waste_residual }}"
        device_class: date
        unique_id: "waste_residual"
      - name: "waste: paper"
        state: "{{ trigger.json.waste_paper }}"
        device_class: date
        unique_id: "waste_paper"
      - name: "waste: yellow bag"
        state: "{{ trigger.json.waste_yellow_bag }}"
        device_class: date
        unique_id: "waste_yellow_bag"

The script to upload will run on the same host as my homeassistant instance, so local_only can be true.

Now the script that downloads the calendar and uploads to home assistant:

import datetime
import json
import sys
from pathlib import Path
import requests

def parse(stream):
    ds = {}
    for line in stream.split("\n"):
        if line.strip() == "BEGIN:VEVENT" and ds:
            yield ds
            ds = {}
        if line.startswith("DTSTART"):
            ds["date"] = str(
                    line.split(":")[-1].split("T")[0], "%Y%m%d"
        if line.startswith("SUMMARY"):
            ds["summary"] = line.split(":")[-1].strip()

if __name__ == "__main__":
    # copy webhook secret into file .secret
    secret = (Path(__file__).parent / ".secret").open().read().strip()

    # url is set in the first commandline argument
    r = requests.get(url=sys.argv[1])
    key_map = {
        "Restmüll 02-wöchentl.": "waste_residual",
        "Altpapier 03-wöchentl.": "waste_paper",
        "Gelber Sack 03-wöchentl.": "waste_yellow_bag",
    dataset = {}
    for item in parse(r.text):
        _k = item.get("summary")
        if _k in key_map:
            dataset[key_map[_k]] = item.get("date")
            del key_map[_k]

    r = requests.post(f"http://localhost:8123/api/webhook/{secret}", json=dataset)
    assert r.status_code == 200

The parsing part is exactly the same as in the old blog post. The modifications to the February version are:

  • using requests -- because it is already installed

  • the addition of a key_map to rename the keys to the ones expected by the sensor template

  • code to push to homeassistant

The result in homeassistant looks like this:

screenshot of waste collection dates in homeassistant dashboard

I use a cron that runs every day at 4am to push the current collection dates to homeassistant.

Homeassistant Statusbar

I want to show in my statusbar the current CO2 value and the change of CO2 in the last minute. My statusbar is waybar with Sway as windowmanager.

So first add a new sensor that shows the change of the CO2 sensor in the last minute to the homeassistant configuration.yaml. This is a Derivative sensor with a time window of one minute.

  - platform: derivative
    source: sensor.scd30_co2
    name: CO2 per last minute
    round: 1
    unit_time: min
    time_window: "00:01:00"

This looks like this in homeassistant:

screenshot of derivative sensor in homeassistant

Next we create another sensor that formats the string we want for the statusbar:

  - platform: template
        value_template: "CO2: {{ states('sensor.scd30_co2') }} [{{'%+3.2f'|format(states('sensor.co2_per_last_minute')|float)}}]"

The value_template is in jinja syntax and the value in the bracket is forced to have a sign in front (+ or -) using a Python-style format string. The result looks for example like this: CO2: 723.3 [+7.00].

Now we need to get the value from the homeassistant API to show them in the statusbar. Prerequisite is a "Long-lived access token" which we can generate on the profile view of homeassistant.

The curl and jq line to get the "statusbar_co2" sensor:

curl \
  -s -H "Authorization: Bearer YOUR_LONG_LIVED_TOKEN" \
  -H "Content-Type: application/json" http://localhost:8123/api/states | \
  jq -r '.[] | select(.entity_id == "sensor.statusbar_co2") | .state'

The jq command filters the list of entities to the "statusbar_co2" entity and returns the state value. The -r removes the quotes around the result of jq. I put the curl command into co2_status.sh and call it from waybar like this:

"custom/co2": {
   "exec": "/home/user/bin/co2_status.sh",
   "interval": 30

Homeassistant Sensor Push

I have a lot of sensors connected to the GPIO pins of Raspberry PIs. They sent their data to my Flask based http API. To make the switch to homeassistant with as little changes as possible I decided to first try template sensors with webhooks.

These sensors are defined in the configuration.yaml for example for a SCD30 CO2 sensor:

  - trigger:
      - platform: webhook
        webhook_id: !secret rpi-05-co2
          - POST
        # my local network and VPN need this
        local_only: false
    unique_id: "scd30"
      - name: "SCD30 Temperature"
        state: "{{ trigger.json.temperature }}"
        unit_of_measurement: "°C"
        device_class: temperature
        unique_id: "scd30_temperature"
      - name: "SCD30 Humidity"
        state: "{{ trigger.json.humidity }}"
        unit_of_measurement: "%"
        device_class: humidity
        unique_id: "scd30_humidity"
      - name: "SCD30 CO2"
        state: "{{ trigger.json.co2 }}"
        unit_of_measurement: ppm
        device_class: carbon_dioxide
        unique_id: "scd30_co2"

The "device_class" automatically sets the correct icons and the "name" is the one shown in the entity list. The "unique_id" on toplevel is used to prefix all sensors and the "unique_id" per sensor allows custom definitions in the frontend, i.e. set an area.

The code to push sensor values from the Raspberry PI (with SCD30 connected) to homeassistant:

import requests
import time
from pathlib import Path
from scd30_i2c import SCD30

def push(data, secret):
    r = requests.post(f"{secret}", json={
        "co2": round(data[0], 2),
        "temperature": round(data[1], 1),
        "humidity": int(data[2]),
    assert r.status_code == 200

# webhook secret is read from file ".secret"
secret = (Path(__file__).parent / ".secret").open().read().strip()

scd30 = SCD30()
while True:
    if scd30.get_data_ready():
        m = scd30.read_measurement()
        if m is not None:
            push(m, secret)

The sensor is connected via i2c. The Python library to get the values from the sensor is scd30-i2c. One special thing for the sensor is that the i2c timing has to be changed (see "I²C clock stretching" in the scd30-i2c readme).

I will change my other sensors (DHT22, BME280, BH1750 and ADS1015 (i.e. for photo resistors)) in the next weeks.