Versions Compared

Version Old Version 7 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_
        elif message == "INVALID_PASSWORD":
            print(f"Wachtwoord is onjuist!") 
        else:
            print(f"Bad request, controleer API verzoek!")
    elif response_authenticate.status_code == 401:
        print(f"Gebruikersnaam en/of wachtwoord zijn onjuist!")
    elif response_authenticate.status_code == 405:
        print(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

...

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):def check_status_API(username, password):
    API_available = False
    
    # 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_cert)

        #controleer reactie Rainguru API
        API_available = False
        if response_get_info.status_code == 200:
            # 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 
latitude = 51.99772 
longitude = 4.381829

# Kwantiel: in deze versie van Rainguru kan alleen het 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 = {
        "provider": f"{firebase_provider}",
        "latitude": 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 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 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 is een json bestand 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.

...