Reading Afvalwijzer in Jython

Inspired by this topic and wanting to train my python/jython skills, I tried a different approach to reading the garbage pickup dates. In this HA forum topic I found that mijnafvalwijzer.nl can be queried to get the same date in a more practical way. That was my starting point.

The nice thing about the new rules engine/jython is that we can define functions to break up the functionality in small testable components. I use cpython in vscode to develop and test the functions, because I can use a debugger there (not directly in openhab as far as I know). When the basic functionality is working, I then transfer it to openhab and hook it up with items.

In this way, I came to the following script to get the garbage collection data:

'''
This script contains a rule that consults afvalkalenders 
for garbage pickup times twice a day and update Items 
with the dates. The script can create the Items needed, if they do not exist.

Please configure this script by filling in your address data in afvalkalender_configuration below. 
Please select the base URI for your garbage collector provider from the list below

MijnAfvalWijzer: https://json.mijnafvalwijzer.nl
Cyclus NV: https://afvalkalender.cyclusnv.nl
HVC: https://apps.hvcgroep.nl
Dar: https://afvalkalender.dar.nl
Afvalvrij / Circulus-Berkel: https://afvalkalender.circulus-berkel.nl
Meerlanden: https://afvalkalender.meerlanden.nl
Cure: https://afvalkalender.cure-afvalbeheer.nl
Avalex: https://www.avalex.nl
RMN: https://inzamelschema.rmn.nl
Venray: https://afvalkalender.venray.nl
Den Haag: https://huisvuilkalender.denhaag.nl
Berkelland: https://afvalkalender.gemeenteberkelland.nl
Alphen aan den Rijn: https://afvalkalender.alphenaandenrijn.nl
Waalre: https://afvalkalender.waalre.nl
ZRD: https://afvalkalender.zrd.nl
Spaarnelanden: https://afvalwijzer.spaarnelanden.nl
Montfoort: https://afvalkalender.montfoort.nl
GAD: https://inzamelkalender.gad.nl
Cranendonck: https://afvalkalender.cranendonck.nl

List taken from: https://www.gadget-freakz.com/domoticz-dzvents-getgarbagedates-script/
'''
import json
from datetime import date, datetime, timedelta

from core.rules import rule
from core.triggers import when
from core.items import add_item
from core.actions import HTTP
from core.utils import postUpdate as post_update
from personal.etxean_utils import push_notification

afvalkalender_configuration = {
    # 'afvalkalender_url': 'https://json.mijnafvalwijzer.nl,
    'huisnummer': "5",
    'toevoeging': '',
    'postcode': "1234AB",
    'create_missing_items': False,
}

MIJNAFVALWIJZER_APIKEY = "<specify-key-here>"

@rule("TrashPickupdates", description="Haal de gegevens over ophalen van afval op uit de afvalwijzer", tags=["afvalkalender"])
@when("Time cron 0 30 7 * * ?")
@when("Time cron 0 30 19 * * ?")
def send_trash_pickup_notifications(event):
    
    # Get data from mijnafvalwijzer
    postcode = afvalkalender_configuration['postcode']
    huisnummer = afvalkalender_configuration['huisnummer']
    toevoeging = afvalkalender_configuration['toevoeging']

    send_trash_pickup_notifications.log.debug('Getting pickup dates for {}-{}{}'.format(postcode, huisnummer, toevoeging))
    if 'afvalkalender_uri' in afvalkalender_configuration and afvalkalender_configuration['afvalkalender_uri'] and afvalkalender_configuration['afvalkalender_uri'] != 'https://json.mijnafvalwijzer.nl':
        pickupdates = get_pickupdates_afvalkalender(postcode, huisnummer, afvalkalender_configuration['afvalkalender_uri'])
    else:
        pickupdates = get_pickupdates_mijnafvalwijzer(postcode, huisnummer, toevoeging)

    process_pickupdates(pickupdates, 'papier', 'oud papier', 'AfvalKalender_papier')
    process_pickupdates(pickupdates, 'gft', 'groente-, fruit- en tuinafval', 'AfvalKalender_gft')
    process_pickupdates(pickupdates, 'restafval', 'restafval', 'AfvalKalender_restafval')
    process_pickupdates(pickupdates, 'pmd', 'plastic- en metaalafval en drinkkartons', 'AfvalKalender_pmd')
    process_pickupdates(pickupdates, 'kca', 'klein chemisch afval', 'AfvalKalender_kca')

the_log = send_trash_pickup_notifications.log

def process_pickupdates(dates, type, description, item, now=None):
    log = the_log
    if not dates or len(dates) == 0:
        return
    pickup_dates = [x['date'] for x in dates if x['type'] == type]
    if len(pickup_dates) == 0:
        return
    if now is None:
        now = datetime.now()
    log.debug('Processing pickup dates for {}: {} and item {} on {}'.format(type, description, item, now))
    if ir.getItems(item) == []:
        if afvalkalender_configuration['create_missing_items']:
            add_item(item, item_type="DateTime", label="{} [%%1$td-%1$tm]".format(description), category="Calendar", groups=[], tags=[])
        else:
            return
    today = str(now.date())
    tomorrow = str(now.date() + timedelta(days=1))
    next_date = next((d for d in pickup_dates if d >= today), pickup_dates[-1])
    next_date_list = next_date.split("-")
    pickup_date = DateTimeType().zonedDateTime.withYear(int(next_date_list[0])).withMonth(int(next_date_list[1])).withDayOfMonth(int(next_date_list[2])).withHour(0).withMinute(0).withSecond(0).withNano(0)
    post_update(item, str(pickup_date))
    current_time = now.time()
    if today == next_date and current_time.hour <= 12:
        message = "Vandaag wordt het {} opgehaald.".format(description)
        log.info(message)
        push_notification(message)
    if tomorrow == next_date and current_time.hour > 12:
        message = "Morgen wordt het {} opgehaald.".format(description)
        log.info(message)
        push_notification(message)


def get_json_response(uri):
    log = the_log
    the_page = HTTP.sendHttpGetRequest(uri)
    if not the_page:
        log.warn("Could not get response from {}".format(uri))
        return None
    try:
        return json.loads(the_page.decode('utf-8'))
    except Exception as ex:
        log.warn('Could not interpret json response from {}: {}'.format(uri, ex))
        return None


def get_pickupdates_mijnafvalwijzer(postcode, huisnummer, toevoeging=''):
    '''Gets pickup dates from different garbage collectors from mijnafvalwijzer.'''
    log = the_log
    today = date.today().strftime("%d-%m-%Y")
    uri = 'https://api.mijnafvalwijzer.nl/webservices/appsinput/?method=postcodecheck&street=&postcode={0}&huisnummer={1}&toevoeging={2}&apikey={3}&app_name=afvalwijzer&platform=phone&mobiletype=android&afvaldata={4}&version=58&langs=nl'.format(postcode, huisnummer, toevoeging, MIJNAFVALWIJZER_APIKEY,today)
    log.info('Afvalkalender uri: {}'.format(uri))
    pickupdates = {}
    try:
        response = get_json_response(uri)
        pickupdates = response['ophaaldagen']
        if pickupdates[u'response'] != u'OK':
            log.warn('Invalid response while getting garbage pickup dates from {}: {}'.format(uri, pickupdates['error']))
            return None
        return pickupdates['data']
    except Exception as ex:
        log.warn('Could not interpret garbage pickup dates from {}: {}'.format(uri, ex))
        return None


def get_pickupdates_afvalkalender(postcode, huisnummer, afvalkalender_uri):
    '''Gets pickup dates from different garbage collectors.'''
    log = the_log
    uri_address='{0}/rest/adressen/{1}-{2}'.format(afvalkalender_uri, postcode, huisnummer)
    response_address = get_json_response(uri_address)
    if not response_address:
        log.warn('Could not get garbage pickup information from {}'.format(uri_address))
        return None
    uri_pickupdates='{0}/rest/adressen/{1}/afvalstromen'.format(afvalkalender_uri, response_address[0]['bagId'])
    response_pickupdates = get_json_response(uri_pickupdates)
    if not response_pickupdates:
        log.warn('Could not get garbage pickup dates from {}'.format(uri_pickupdates))
        return None
    pickupdates = []
    for pickupdate in response_pickupdates:
        newdate = {}
        if pickupdate['icon'] and pickupdate['ophaaldatum']:
            newdate['type'] = pickupdate['icon']
            newdate['date'] = pickupdate['ophaaldatum']
            if newdate['type'] == 'rest':
                newdate['type'] = 'restafval'
            pickupdates.append(newdate)
 
    return pickupdates
2 Likes

Nice! After running this through pylint :wink:, please consider contributing this to the community section of the helper library repo. I have a commit for what I’d built for the other topic, but I’ll hold off on pushing it.

Hmmm… maybe we should add both…

After some more Googling, I came up with a solution for the communities that do not use mijnafvalwijzer.nl. Just exchange the original get_pickupdates with the code below. Not thoroughly tested though.

postcode='5581BG'
huisnummer='17'

'''
Gets pickup dates from different garbage collectors. Please select the base URI for 
your garbage collector from the list below

Cyclus NV: https://afvalkalender.cyclusnv.nl
HVC: https://apps.hvcgroep.nl
Dar: https://afvalkalender.dar.nl
Afvalvrij / Circulus-Berkel: https://afvalkalender.circulus-berkel.nl
Meerlanden: https://afvalkalender.meerlanden.nl
Cure: https://afvalkalender.cure-afvalbeheer.nl
Avalex: https://www.avalex.nl
RMN: https://inzamelschema.rmn.nl
Venray: https://afvalkalender.venray.nl
Den Haag: https://huisvuilkalender.denhaag.nl
Berkelland: https://afvalkalender.gemeenteberkelland.nl
Alphen aan den Rijn: https://afvalkalender.alphenaandenrijn.nl
Waalre: https://afvalkalender.waalre.nl
ZRD: https://afvalkalender.zrd.nl
Spaarnelanden: https://afvalwijzer.spaarnelanden.nl
Montfoort: https://afvalkalender.montfoort.nl
GAD: https://inzamelkalender.gad.nl
Cranendonck: https://afvalkalender.cranendonck.nl

List taken from: https://www.gadget-freakz.com/domoticz-dzvents-getgarbagedates-script/
'''
def get_pickupdates_alternative(postcode, huisnummer, base_URI = 'https://afvalkalender.waalre.nl'):
    log = send_trash_pickup_notifications.log
    URI1='{0}/rest/adressen/{1}-{2}'.format(base_URI, postcode, huisnummer)
    the_page = HTTP.sendHttpGetRequest(URI1)
    if not the_page:
        log.warn("Could not get response from {}".format(URI1))
        return
    response1 = json.loads(the_page.decode('utf-8'))
    URI2='{0}/rest/adressen/{1}/afvalstromen'.format(base_URI, response1[0]['bagId'])
    the_page = HTTP.sendHttpGetRequest(URI2)
    if not the_page:
        log.warn("Could not get response from {}".format(URI2))
        return
    response2 = json.loads(the_page.decode('utf-8'))
    pickupdates = []
    for pd in response2:
        newdate = {}
        if pd['icon'] and pd['ophaaldatum']:
            newdate['type'] = pd['icon']
            newdate['date'] = pd['ophaaldatum']
            if newdate['type'] == 'rest':
                newdate['type'] = 'restafval'
            pickupdates.append(newdate)
    
    return pickupdates
1 Like

Nice :slight_smile:

Sorry, it DOES seem to work!

removed

The OP has been updated. I’ve refactored my code:

  • added the querying of afvalkalender as an alternative to mijnafvalwijzer;
  • improved formatting after using pylint.
2 Likes

Really would like to use something like this. But i can;t get it to work. If i read Please verify the url:' + jsonUrl + 'manually in your browser · Issue #72 · xirixiz/homeassistant-afvalwijzer · GitHub the json.mijnafvalwijzer.nl no longer works. they moved to an api with apikey.

As the topic at the HA forum suggests, this API key can be obtained with some “inspection” of the Android app. I managed to get it that way and supplied in the script. See the OP for the revised python script. You will have to find and enter the APi key yourself though.