File size: 3,288 Bytes
10f22e0 5ccd9a9 10f22e0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
## Imports ##
import requests
import pandas as pd
from datetime import datetime
from datetime import timezone
from pages.Location import *
class ApiCall:
def getData(self):
## Api Variablen für NINA
api_variablen = {
"hochwasser": "/lhp/mapData",
"polizei": "/police/mapData",
"wetter": "/dwd/mapData",
"katwarn": "/katwarn/mapData",
"mowas": "/mowas/mapData"
}
# Dataframe Collumns definieren
columns = ['ID', 'Urgency','Area', 'Titel', 'Event', 'Datum']
# NINA Api URLs
ninaBaseUrl = "https://warnung.bund.de/api31"
ninaWarningsUrl = "https://nina.api.proxy.bund.dev/api31/warnings/"
# Einzelne NINA Warnings abfragen
def get_api_warning(meldung):
response = requests.get(ninaBaseUrl+meldung+".json")
return response.json()
# Details zu NINA Warning abfragen: Return DataFrame mit allen Ergebnissen
def get_api_details(warning):
n = 0
response = get_api_warning(warning)
df = pd.DataFrame(columns=columns)
# GetDetails for warning
for responses in response:
id = responses["id"]
warningDetails = requests.get(ninaWarningsUrl+id+".json").json()
meldungsText = warningDetails["info"][0]["headline"]+ ": "+warningDetails["info"][0]["description"]
warnung = warningDetails["info"][0]["event"]
start = warningDetails["sent"]
urgency = warningDetails["info"][0]["urgency"]
area = warningDetails["info"][0]["area"][0]["areaDesc"]
d = datetime.fromisoformat(start).astimezone(timezone.utc)
d.strftime('%Y-%m-%d %H:%M:')
time = d.replace(tzinfo=None)
df.loc[n] = [id, urgency, area, meldungsText, warnung, time]
n = n+1
return df
## Dataframe to CSV
def df_to_csv(data, filename):
data.to_csv(filename+'.csv')
# Dataframe definieren
data = pd.DataFrame(columns=columns)
## API abfragen für alle NINA Variablen/Warnings ##
for x in api_variablen:
data_api = get_api_details(api_variablen[x])
data = pd.concat([data, data_api])
# DataFrame df2 erstellen
data['Area'] = data['Area'].str.split(', ')
df2 = data.explode('Area')
# Neue Zeile mit Plz erstellen
def berechnung_func(row):
loca = row['Area']
ort = loca.replace("Gemeinde", "")
ort = loca.replace("Stadt", "")
ort = loca.replace("Mitgliedsgemeinde in Verwaltungsgemeinschaft", "")
###### TODO #####
# Orte klein schreiben und filtern nach und ... #
###### TODO #####
location = Location(ort.strip()).getPostalCode()
plz = location.iloc[0]['name']
return plz
df2['Plz'] = df2.apply(berechnung_func, axis=1)
# API Daten in CSV speichern
df_to_csv(df2, "Data")
# Return
return df2
# Testing
# l1 = ApiCall().getData()
|