...
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 |
---|
# 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") |
...
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") |
...