Versions Compared

Version Old Version 5 New Version Current
Changes made by

Bart Thonus

Bart Thonus

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Code Block
# Dit is een voorbeeld notebook voor het gebruik van de Rainguru API.
# Zie ook onze Wiki pagina (https://hkvconfluence.atlassian.net/wiki/spaces/Rainguru/overview?homepageId=2083520655) voor uitleg over Rainguru en nowcasting en het gebruik van de API (inclusief het voorbeeld uit dit notebook)

# Benodigde packages
import requests
import json
import os
import datetime

# Server (API)
server = "https://api.rainguru.nl"
# Rainguru HTTPS certificaat
rainguru_cert=r"Rainguru.pem"
# Firebase authentication provider
firebase_provider="mfa"

# Folder om resultaatbestanden in op te slaan
result_folder = os.path.join(os.getcwd(), "Results")

...

Code Block
def authenticate(username, password):
    """
    Authenticate user
    
    Parameters:
    ----------
        username: str
            Name of the user
        password: str
            Password of the user
    """ 
    
    # Firebase instellingen voor authenticatie
    firebase_url = "https://identitytoolkit.googleapis.com/v1"
    firebase_key = "AIzaSyCjGibOfg9LKxoFhKAIzaSyBhun_JSiE1_903H6M4vrh3UkAbQz48VXRFq0eKrErI4UT3ES0"
    
    # Authentiseer gebruiker 
    response_authenticate = requests.post(f"{firebase_url}/accounts:signInWithPassword?key={firebase_key}", data={'email': username, 'password': password, 'returnSecureToken': 'true'})
    
    # Controle reactie firebase
    my_headers = {}
    if response_authenticate.status_code == 200:
        # Succesvol: sla header op
        id_token = response_authenticate.json()['idToken']
        my_headers = {'Authorization': f"Bearer {id_token}"}      
    
    # Niet succescol print foutmeldingen
    elif response_authenticate.status_code == 400:
        message print(f"Bad request, controleer API verzoek!")= json.loads(response_authenticate.text)["error"]["errors"][0]["message"]
        elif response_authenticate.status_codeif message == 401:"EMAIL_NOT_FOUND":
            print(f"Gebruikersnaam en/of wachtwoord zijnis onjuist!")       elif response_authenticate.status_code 
        elif message == 405"INVALID_PASSWORD":
            print(f"APIWachtwoord is niet beschikbaaronjuist!") 
     elif response_authenticate.status_code == 500else:
            print(f"ErBad isrequest, eencontroleer interneAPI fout opgetredenverzoek!")
    elif  elseresponse_authenticate.status_code == 401:
        print(f"ErGebruikersnaam isen/of eenwachtwoord ongedefinieerdezijn fout opgetredenonjuist!")
         elif response_authenticate.status_code == 405:
         return my_headers

Daarna wordt een POST request naar de Google API gedefinieerd, waarbij de key en de dictionary met de login cedentials meegestuurd wordt. Als de HTML statuscode van de response gelijk is aan 200 (de Google API geeft succesvol een resultaat), dan bevat de response een zogenaamde id_token. Deze wordt tot slot omgezet in een Bearer token (een gecodeerde token) die daarna als header meegeven kan worden aan toekomstige verzoeken aan de Rainguru API.

Wanneer het verzoek niet sucesvol afgehandeld kan worden, schrijven we de foutmelding weg naar het scherm:

  • Status_code=400 → het API verzoek is onjuist, controleer de url en parameters

  • Status_code=401 → de gebruikersnaam en/of wachtwoord zijn onjuist

  • Status_code=405 → de API is niet beschikbaar

  • Status_code=500 → er is een fout opgetreden

De funtie retourneert tot slot de header in de variabele 'my_headers'.

Stap 3: Functie om de status van de Rainguru API te controleren

Voordat de Rainguru API gebruikt wordt, is het verstandig om eerst de status van de API te controleren. We hebben een aparte Python functie aangemaakt (def) om de status van de API op te vragen. Deze kan aangeroepen worden waarbij een gebruikersnaam en wachtwoord als parameters meegegeven moeten worden. Dit is nodig, omdat de functie als eerste de authenticatie functie aanroept (zie Stap 2).

Code Block
def check_status_API(username, password):

    # Inloggen en authentiseren
    my_headers = authenticate(username, password) 

    # Alleen doorgaan als my_headers beschikbaar is (authenticatie geslaagd)
    if not my_headers == {}: 
        
        # Check status Rainguru API: https://api.rainguru.nl/
        response_get_info = requests.get(f"{server}/", headers=my_headers, verify=rainguru_certprint(f"API is niet beschikbaar!")
    elif response_authenticate.status_code == 500:
        print(f"Er is een interne fout opgetreden!")
    else:
        print(f"Er is een ongedefinieerde fout opgetreden!")
              
    return my_headers

Daarna wordt een POST request naar de Google API gedefinieerd, waarbij de key en de dictionary met de login cedentials meegestuurd wordt. Als de HTML statuscode van de response gelijk is aan 200 (de Google API geeft succesvol een resultaat), dan bevat de response een zogenaamde id_token. Deze wordt tot slot omgezet in een Bearer token (een gecodeerde token) die daarna als header meegeven kan worden aan toekomstige verzoeken aan de Rainguru API.

Wanneer het verzoek niet sucesvol afgehandeld kan worden, schrijven we de foutmelding weg naar het scherm:

  • Status_code=400 → het API verzoek is onjuist, controleer de url en parameters

  • Status_code=401 → de gebruikersnaam en/of wachtwoord zijn onjuist

  • Status_code=405 → de API is niet beschikbaar

  • Status_code=500 → er is een fout opgetreden

De funtie retourneert tot slot de header in de variabele 'my_headers'.

Stap 3: Functie om de status van de Rainguru API te controleren

Voordat de Rainguru API gebruikt wordt, is het verstandig om eerst de status van de API te controleren. We hebben een aparte Python functie aangemaakt (def) om de status van de API op te vragen. Deze kan aangeroepen worden waarbij een gebruikersnaam en wachtwoord als parameters meegegeven moeten worden. Dit is nodig, omdat de functie als eerste de authenticatie functie aanroept (zie Stap 2).

Code Block
def check_status_API(username, password):
    API_available = False
    
    # Inloggen en authentiseren
    my_headers = authenticate(username, password) 

    # Alleen doorgaan #controleerals reactiemy_headers Raingurubeschikbaar APIis (authenticatie geslaagd)
    if not APImy_availableheaders == {}: False
        if
response_get_info.status_code == 200:
            # Succesvol:Check print json met statusstatus Rainguru API
   : https://api.rainguru.nl/
        print(response_get_info = requests.json())get(f"{server}/", headers=my_headers, verify=rainguru_cert)

        #controleer reactie Rainguru API_available
= True       if response_get_info.status_code == 200:
       # Niet succesvol, print foutmeldingen # Succesvol: print json met status API
            print(response_get_info.json())
            API_available = True
        
        # Niet succesvol, print foutmeldingen
        elif response_get_info.status_code == 400:
            print(f"Bad request, controleer API verzoek!")
        elif response_get_info.status_code == 401:
            print(f"Gebruikersnaam en/of wachtwoord zijn onjuist!")
        elif response_get_info.status_code == 405:
            print(f"API is niet beschikbaar!")
        elif response_get_info.status_code == 500:
            print(f"Er is een interne fout opgetreden!")
        else:
            print(f"Er is een ongedefinieerde fout opgetreden!")        
    
    return API_available, my_headers

...

Code Block
# Vraag Rainguru data op, op basis van locatie (latitude, longitude)

# Inlog gegevens gebruiker
username = "gebruiker@rainguru.nl"
password = "*********"

# Locatie: kies een locatie op basis van een latutide en longitude
# Voorbeeld locatie HKV Delft 
#latitudelatitude = 51.99772 
#longitudelongitude = 4.381829

# Kwantiel: in deze versie van Rainguru kan alleen het 5-%50% kwantiel gebruikt worden. Mogelijk wordt dat in de toekomst uitgebreid.
quantile = "0.5"

# Uitvoer formaat: kies uit 'json' of 'csv'
format = "csv"

# Controleer of Rainguru API beschikbaar is
API_available, my_headers = check_status_API(username, password) 

if API_available == True: 
    
    # Voeg parameters toe aan parameters object t.b.v. API verzoek
    my_parameters = {
        "latitudeprovider": f"{latitudefirebase_provider}",
        "longitudelatitude": f"{latitude}",
        "longitude": f"{longitude}",
        "quantile": f"{quantile}" 
    }

    # Verzoek Rainguru API op basis van locatie (latutide, longitude): https://api.rainguru.nl/data.csv of https://api.rainguru.nl/data.json
    response_get_data_by_location = requests.get(f"{server}/data.{format}", params=my_parameters, headers=my_headers, verify=rainguru_cert)
    
    # Controleer reactie Rainguru API
    if response_get_data_by_location.status_code == 200:    
        
        # Succesvol: druk resultaat API af
        if format == "csv":
            print(response_get_data_by_location.text)
            # Resultaat opslaan in CSV bestand
            result_data = bytes(response_get_data_by_location.content)
            if not os.path.exists(result_folder):
                os.makedirs(result_folder)
            open(os.path.join(result_folder,datetime.datetime.now().strftime("%Y%m%d%H%M%S") + "_Rainguru.csv"), 'wb').write(response_get_data_by_location.content)
        elif format == "json":
            print(response_get_data_by_location.json())
            # Resultaat opslaan in JSON bestand
            result_data = bytes(response_get_data_by_location.content)
            if not os.path.exists(result_folder):
                os.makedirs(result_folder)
            open(os.path.join(result_folder,datetime.datetime.now().strftime("%Y%m%d%H%M%S") + "_Rainguru.json"), 'wb').write(response_get_data_by_location.content)
        else: 
            print(f"Er is een onverwachte fout opgetreden!")
    
    # Niet succesvol print foutmeldingen
    elif response_get_data_by_location.status_code == 400:
        print(f"Bad request, controleer API verzoek!")
    elif response_get_data_by_location.status_code == 401:
        print(f"Gebruikersnaam en/of wachtwoord zijn onjuist!")
    elif response_get_data_by_location.status_code == 405:
        print(f"API is niet beschikbaar!")
    elif response_get_data_by_location.status_code == 500:
        print(f"Er is een interne fout opgetreden!")
    else:
        print(f"Er is een ongedefinieerde fout opgetreden!")    
else:
    # De API is niet beschikbaar; neem contact op met de helpdesk
    print(f"Rainguru API is niet beschikbaar! Neem contact op met helpdesk-rainguru@hkv.nl")

...

Code Block
# Vraag Rainguru data op, op basis van adres (straatnaam + huisnummer, plaatsnaam en/of postcode)

# Inlog gegevens gebruiker
username = "gebruiker@rainguru.nl"
password = "*********"

# locatie: geef een adres op (straatnaam + huisnummer, plaatsnaam en/of postcode (bepalen met PDOK)
# Voorbeeld locatie HKV Lelystad
adres = "8232JN, Botter 11"

# Kwantiel: in deze versie van Rainguru kan alleen het 5-%50% kwantiel gebruikt worden. Mogelijk wordt dat in de toekomst uitgebreid.
quantile = "0.5"

# Uitvoer formaat: kies uit 'json' of 'csv'
format = "json"

# Controleer of Rainguru API beschikbaar is
API_available, my_headers = check_status_API(username, password) 

if API_available == True: 
    
    # Voeg parameters toe aan parameters object t.b.v. API verzoek
    my_parameters = {
        "provider": f"{firebase_provider}",
        "address": f"{adres}" ,
        "quantile": f"{quantile}" 
    }
    
    # Verzoek Rainguru API op basis van adres: https://api.rainguru.nl/data.csv of https://api.rainguru.nl/data.json
    response_get_data_by_address = requests.get(f"{server}/data.{format}", params=my_parameters, headers=my_headers, verify=rainguru_cert)

    if response_get_data_by_address.status_code == 200:    
        # Succesvol: druk resultaat API af
        if format == "csv":
            print(response_get_data_by_address.text)
            result_data = bytes(response_get_data_by_address.content)
            if not os.path.exists(result_folder):
                os.makedirs(result_folder)
            open(os.path.join(result_folder,datetime.datetime.now().strftime("%Y%m%d%H%M%S") + "_Rainguru.csv"), 'wb').write(response_get_data_by_address.content)
        elif format == "json":
            print(response_get_data_by_address.json())
            # Resultaat opslaan in JSON bestand
            result_data = bytes(response_get_data_by_address.content)
            if not os.path.exists(result_folder):
                os.makedirs(result_folder)
            open(os.path.join(result_folder,datetime.datetime.now().strftime("%Y%m%d%H%M%S") + "_Rainguru.json"), 'wb').write(response_get_data_by_address.content)
        else: 
            print(f"Er is een onverwachte fout opgetreden!")
    
    # Niet succesvol print foutmeldingen
    elif response_get_data_by_address.status_code == 400:
        print(f"Bad request, controleer API verzoek!")
    elif response_get_data_by_address.status_code == 401:
        print(f"Gebruikersnaam en/of wachtwoord zijn onjuist!")
    elif response_get_data_by_address.status_code == 405:
        print(f"API is niet beschikbaar!")
    elif response_get_data_by_address.status_code == 500:
        print(f"Er is een interne fout opgetreden!")
    else:
        print(f"Er is een ongedefinieerde fout opgetreden!")    
else:
    # De API is niet beschikbaar; neem contact op met de helpdesk
    print(f"Rainguru API is niet beschikbaar! Neem contact op met helpdesk-rainguru@hkv.nl")

...

Het (fictieve) resultaat (in csvjson) ziet er als volgt uit:

Code Block
{
	'format': 'json',
	'latitude': 52.50669969,
	'longitude': 5.46887425,
	'adres': 'Botter 11  8232JN Lelystad',
	'quantile': [0.5],
	'data': [{
		'date': '2023-07-14T13:40:00',
		'quantile': 0.5,
		'value': 0.00039187,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T13:45:00',
		'quantile': 0.5,
		'value': 0.00036777,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T13:50:00',
		'quantile': 0.5,
		'value': 0.00036918,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T13:55:00',
		'quantile': 0.5,
		'value': 0.00037081,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:00:00',
		'quantile': 0.5,
		'value': 0.00036859,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:05:00',
		'quantile': 0.5,
		'value': 0.00036742,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:10:00',
		'quantile': 0.5,
		'value': 0.00036766,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:15:00',
		'quantile': 0.5,
		'value': 0.00036844,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:20:00',
		'quantile': 0.5,
		'value': 0.00036827,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:25:00',
		'quantile': 0.5,
		'value': 0.00036851,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:30:00',
		'quantile': 0.5,
		'value': 0.00036876,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:35:00',
		'quantile': 0.5,
		'value': 0.00036863,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:40:00',
		'quantile': 0.5,
		'value': 0.0003687,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:45:00',
		'quantile': 0.5,
		'value': 0.00036877,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:50:00',
		'quantile': 0.5,
		'value': 0.00036874,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T14:55:00',
		'quantile': 0.5,
		'value': 0.00036859,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T15:00:00',
		'quantile': 0.5,
		'value': 0.00036823,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T15:05:00',
		'quantile': 0.5,
		'value': 0.00036793,
		'units': 'mm/hr'
	}, {
		'date': '2023-07-14T15:10:00',
		'quantile': 0.5,
		'value': 0.00036751,
		'units': 'mm/hr'
	}]
}

...

Code Block
# Download Rainguru raster-bestanden voor heel Nederland

# Inlog gegevens gebruiker
username = "gebruiker@rainguru.nl"
password = "*********"

# Kwantiel: in deze versie van Rainguru kan alleen het 5-%50% kwantiel gebruikt worden. Mogelijk wordt dat in de toekomst uitgebreid.
quantile = "0.5"

# Uitvoer formaat: kies altijd "zip" voor het downloaden van raster-bestanden
format = "zip"

# Controleer of Rainguru API beschikbaar is
API_available, my_headers = check_status_API(username, password) 

if API_available == True: 
    # Voeg parameters toe aan API verzoek
    my_parameters = {
        "provider": f"{firebase_provider}",
        "quantile": f"{quantile}" 
    }
     
    # Verzoek Rainguru API op basis van adres
    response_get_raster = requests.get(f"{server}/raster.{format}", params=my_parameters, headers=my_headers, verify=rainguru_cert)

    if response_get_raster.status_code == 200:    
        # Succesvol: druk resultaat API af
        if format == "zip":
            print("Bestanden zijn succesvol gedownload; zie result_folder!")
            result_data = bytes(response_get_raster.content)
            if not os.path.exists(result_folder):
                os.makedirs(result_folder)
            open(os.path.join(result_folder,datetime.datetime.now().strftime("%Y%m%d%H%M%S") + "_Rainguru.zip"), 'wb').write(response_get_raster.content)    
        else: 
            print(f"Er is een onverwachte fout opgetreden!")
    
    # Niet succesvol print foutmeldingen
    elif response_get_raster.status_code == 400:
        print(f"Bad request, controleer API verzoek!")
    elif response_get_raster.status_code == 401:
        print(f"Gebruikersnaam en/of wachtwoord zijn onjuist!")
    elif response_get_raster.status_code == 405:
        print(f"API is niet beschikbaar!")
    elif response_get_raster.status_code == 500:
        print(f"Er is een interne fout opgetreden!")
    else:
        print(f"Er is een ongedefinieerde fout opgetreden!")    
else:
    # De API is niet beschikbaar; neem contact op met de helpdesk
    print(f"Rainguru API is niet beschikbaar! Neem contact op met helpdesk-rainguru@hkv.nl")

...

Het resultaat ziet er als volgt uit:

...

En in GIS ziet dat er als volg uit:

...

Stap 8: (optioneel)

...

Als een validatie-taak is beeindigd (al dan niet succesvol), kan de metadata van de validatie-taak gedownload worden ter controle van het validatie resultaat. Allereerst wordt een lokale folder op een computersysteem gedefinieerd waar naartoe een bestand met metadata kan worden gedownload. Met een POST request naar de API op het endpoint task/result/metadata wordt een bestand met metadata van de validatie-taak opgevraagd. Naast de id van de validatie-taak (in de URL) worden ook de Bearer token (opgeslagen in my_headers) en het (https) beveiligingscertificaat (verify=hydamo_cert) meegegeven in het request. Als de HTML statuscode van de response gelijk is aan 200 (het post request geeft een succesvol resultaat), kan het bestand met de metadata worden opgeslagen in de lokale folder. Hiertoe wordt het resultaat van het request eerst opgeslaten in de variabele ‘result_data'. Vervolgens wordt gecontroleerd of de map waar het bestand opgeslagen moet worden bestaat en zo niet, dan wordt deze aangemaakt. Tot slot wordt het resultaat van het request opgeslagen in het bestand 'taak-id_metadata.json’.

...

Converteren van adres naar coordinaten met PDOK locatieserver

We hebben een stap 8 toegevoegd aan het notebook, waarmee een adres met behulp van de PDOK locatieserver omgezet kan worden naar coordinaten (zie voor een uitgebreide beschrijving https://www.pdok.nl/pdok-locatieserver ).

Code Block
# Ter illustratie een voorbeeld van de geolocation functie van PDOK die gebruikt wordt om adres gegevens om te zetten coordinaten

# locatie: geef een adres op (straatnaam + huisnummer, plaatsnaam en/of postcode)
adres = "8232JN, Botter"

# PDOK url geolocation functie (zie ook https://www.pdok.nl/pdok-locatieserver)
PDOK_url = "https://api.pdok.nl/bzk/locatieserver/search/v3_1/free"

# Voeg parameters toe aan API verzoek
my_parameters = {
    "q": f"{adres}",
    "rows": 1,
    "fl": "centroide_rd,centroide_ll,straatnaam,woonplaatsnaam,postcode",
}

# Verzoek aan PDOK geolocation API
resonse_PDOK = requests.get(url=PDOK_url, params=my_parameters)

if resonse_PDOK.status_code == 200:    
    # Succesvol: druk resultaat API af
    print(resonse_PDOK.json())

# Niet succesvol print foutmeldingen
elif resonse_PDOK.status_code == 400:
    print(f"Bad request, controleer API verzoek!")
elif resonse_PDOK.status_code == 401:
    print(f"Gebruikersnaam en/of wachtwoord zijn onjuist!")
elif resonse_PDOK.status_code == 200405:
    result_data = bytes(response_get_metadata.content)
    if not os.path.exists(result_folder)print(f"API is niet beschikbaar!")
elif resonse_PDOK.status_code == 500:
    print(f"Er is een  os.makedirs(result_folder)
interne fout opgetreden!")
else:
    open(os.path.join(result_folder,f"{task_id}_metadata.json"), 'wb').write(response_get_metadata.content)print(f"Er is een ongedefinieerde fout opgetreden!")

Het resultaat is een json bestand met de volgende inhoud (zie ook Resultaten):de coordinaten van de locatie.

Notebook

Bovenstaande voorbeelden zijn opgenomen in een Jupyter notebook. Dit notebook kunt u hier downloaden. Let op: u moet wel de paden en login gegevens zelf aanpassen, anders werkt het notebook niet.

View file
nameVoorbeeldrainguru_Wikiwiki.ipynb