You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

201 lines
7.9 KiB
Python

"""
Quellen:
Johns Hopkins University:
https://gisanddata.maps.arcgis.com/apps/opsdashboard/index.html#/bda7594740fd40299423467b48e9ecf6
https://services1.arcgis.com/0MSEUqKaxRlEPj5g/arcgis/rest/services/ncov_cases/FeatureServer/1/query?f=json&where=Confirmed%20%3E%200&returnGeometry=false&spatialRel=esriSpatialRelIntersects&outFields=*&orderByFields=Confirmed%20desc%2CCountry_Region%20asc%2CProvince_State%20asc&resultOffset=0&resultRecordCount=250&cacheHint=true
WHO:
https://experience.arcgis.com/experience/685d0ace521648f8a5beeeee1b9125cd
https://services.arcgis.com/5T5nSi527N4F7luB/arcgis/rest/services/Cases_by_country_Plg_V3/FeatureServer/0/query?f=json&where=1%3D1&returnGeometry=false&spatialRel=esriSpatialRelIntersects&outFields=*&orderByFields=cum_conf%20desc&resultOffset=0&resultRecordCount=125&cacheHint=true
Deutschland RKI:
https://experience.arcgis.com/experience/478220a4c454480e823b17327b2bf1d4
https://services7.arcgis.com/mOBPykOjAyBO2ZKk/arcgis/rest/services/Coronaf%C3%A4lle_in_den_Bundesl%C3%A4ndern/FeatureServer/0/query?f=json&where=1%3D1&returnGeometry=false&spatialRel=esriSpatialRelIntersects&outFields=*&orderByFields=faelle_100000_EW%20desc&resultOffset=0&resultRecordCount=50&cacheHint=true
Landkreis RKI
https://experience.arcgis.com/experience/478220a4c454480e823b17327b2bf1d4/page/page_1/
https://services7.arcgis.com/mOBPykOjAyBO2ZKk/arcgis/rest/services/RKI_Landkreisdaten/FeatureServer/0/query?f=json&where=1%3D1&returnGeometry=false&spatialRel=esriSpatialRelIntersects&outFields=*&orderByFields=cases%20desc&resultOffset=0&resultRecordCount=1000&cacheHint=true
"""
import json
import requests
import db_model_postgrest as db
import toml
import os
import datetime
QUELLEN = {
"jhu": "https://services1.arcgis.com/0MSEUqKaxRlEPj5g/arcgis/rest/services/ncov_cases/FeatureServer/1/query?f"
"=json&where=Confirmed%20%3E%200&returnGeometry=false&spatialRel=esriSpatialRelIntersects&outFields"
"=*&orderByFields=Confirmed%20desc%2CCountry_Region%20asc%2CProvince_State%20asc&resultOffset=0"
"&resultRecordCount=250&cacheHint=true",
"rki": "https://services7.arcgis.com/mOBPykOjAyBO2ZKk/arcgis/rest/services/Coronaf%C3%A4lle_in_den_Bundesl%C3"
"%A4ndern/FeatureServer/0/query?f=json&where=1%3D1&returnGeometry=false&spatialRel"
"=esriSpatialRelIntersects&outFields=*&orderByFields=faelle_100000_EW%20desc&resultOffset=0"
"&resultRecordCount=50&cacheHint=true",
"rki_landkreis": "https://services7.arcgis.com/mOBPykOjAyBO2ZKk/arcgis/rest/services/RKI_Landkreisdaten"
"/FeatureServer/0/query?f=json&where=1%3D1&returnGeometry=false&spatialRel"
"=esriSpatialRelIntersects&outFields=*&orderByFields=cases%20desc&resultOffset=0"
"&resultRecordCount=1000&cacheHint=true "
}
INTENSIVREGISTER = {
"intensivregister": "https://www.intensivregister.de/api/public/reporting/laendertabelle"
}
def config_laden():
configfile = os.path.join(SKRIPTPFAD, "config.toml")
with open(configfile) as file:
return toml.loads(file.read())
SKRIPTPFAD = os.path.abspath(os.path.dirname(__file__))
CONFIG = config_laden()
def daten_abrufen(url):
response = requests.get(url)
json_text = json.loads(response.text)
return json_text["features"]
def intensivbetten_daten_abrufen(url):
response = requests.get(url)
json_text = json.loads(response.text)
return json_text
def jhu_daten_erstellen(quelle, daten):
daten_dict = {}
for data in daten:
datum = data["attributes"]
country = datum["Country_Region"].replace(" ", "")
if country in daten_dict:
datensatz = daten_dict[country]
datensatz.ts = max(datensatz.ts, datetime.datetime.fromtimestamp(datum["Last_Update"] / 1000))
datensatz.confirmed = datensatz.confirmed + datum["Confirmed"]
datensatz.deaths = datensatz.deaths + datum["Deaths"]
datensatz.recoverd = datensatz.recoverd + datum["Recovered"]
datensatz.active = datensatz.confirmed - datensatz.deaths - datensatz.recoverd
daten_dict[country] = datensatz
else:
daten_dict[country] = (db.CoronaDaten(
datetime.datetime.fromtimestamp(datum["Last_Update"] / 1000),
country.lower().capitalize(),
quelle,
datum["Confirmed"],
datum["Deaths"],
datum["Recovered"],
datum["Confirmed"] - datum["Deaths"] - datum["Recovered"],
-1
))
return daten_dict
def rki_daten_erstellen(quelle, daten):
daten_dict = {}
for data in daten:
datum = data["attributes"]
daten_dict[datum["LAN_ew_GEN"]] = (db.CoronaDaten(
datetime.datetime.fromtimestamp(datum["Aktualisierung"] / 1000),
datum["LAN_ew_GEN"].lower().capitalize(),
quelle,
datum["Fallzahl"],
datum["Death"],
-1,
-1,
round(datum.get("cases7_bl_per_100k", -1), 2)
))
return daten_dict
def rki_landkreis_daten_erstellen(quelle, daten):
daten_dict = {}
for data in daten:
datum = data["attributes"]
daten_dict[datum["county"]] = (db.CoronaDaten(
datetime.datetime.strptime(datum["last_update"], "%d.%m.%Y, %H:%M Uhr"),
datum["county"].lower().capitalize(),
quelle,
datum["cases"],
datum["deaths"],
-1,
-1,
round(datum.get("cases7_per_100k", -1), 2)
))
return daten_dict
def intensivbetten_daten_erstellen(daten):
daten_dict = {}
for data in daten["data"]:
daten_dict[data['bundesland']] = (db.Intensivregister(
datetime.datetime.strptime(data['creationTimestamp'], "%Y-%m-%dT%H:%M:%SZ"),
data['bundesland'],
data['faelleCovidAktuell'],
data['faelleCovidAktuellBeatmet'],
data['intensivBettenBelegt'],
data['intensivBettenFrei'],
data['intensivBettenNotfall7d']
))
daten_dict["GESAMT"] = (db.Intensivregister(
datetime.datetime.strptime(daten["overallSum"]['creationTimestamp'], "%Y-%m-%dT%H:%M:%SZ"),
"GESAMT",
daten["overallSum"]['faelleCovidAktuell'],
daten["overallSum"]['faelleCovidAktuellBeatmet'],
daten["overallSum"]['intensivBettenBelegt'],
daten["overallSum"]['intensivBettenFrei'],
daten["overallSum"]['intensivBettenNotfall7d']
))
return daten_dict
def daten_erstellen(quelle, daten):
if quelle == "jhu":
daten_db = jhu_daten_erstellen(quelle, daten)
elif quelle == "rki":
daten_db = rki_daten_erstellen(quelle, daten)
elif quelle == "rki_landkreis":
daten_db = rki_landkreis_daten_erstellen(quelle, daten)
else:
raise ValueError("Quelle nicht bekannt")
return daten_db
def daten_trennen(daten):
daten_liste = []
for _, value in daten.items():
value.ts = value.ts.strftime("%Y-%m-%d %H:%M:%S")
daten_liste.append(value)
return daten_liste
def headers_erstellen():
headers = {"Authorization": "Bearer {token}".format(token=CONFIG["token"]),
"Prefer": "resolution=ignore-duplicates"}
return headers
def main():
headers = headers_erstellen()
# Coronazahlen
for quelle, url in QUELLEN.items():
json_daten = daten_abrufen(url)
daten = daten_erstellen(quelle, json_daten)
daten_server = daten_trennen(daten)
db.sende_daten(CONFIG["url"], CONFIG["tablename_daten"], headers, daten_server)
# Intensivbetten
for quelle, url in INTENSIVREGISTER.items():
json_daten = intensivbetten_daten_abrufen(url)
daten = intensivbetten_daten_erstellen(json_daten)
daten_server = daten_trennen(daten)
db.sende_daten((CONFIG["url"]), "intensivregister", headers, daten_server)
if __name__ == "__main__":
main()