1
0
Fork 0
rki-impfparser/plot.py

885 lines
28 KiB
Python
Raw Normal View History

#!/usr/bin/python
# vim: set fileencoding=utf-8 :
# Copyright (C) Benedikt Bastin, 2021
# SPDX-License-Identifier: EUPL-1.2
2021-01-14 23:16:40 +00:00
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
import datetime
import re
import requests as req
import locale
import os.path
import shutil
2021-02-17 10:55:30 +00:00
import math
from functools import reduce
from matplotlib.dates import date2num, DateFormatter, WeekdayLocator
import matplotlib.dates as mdates
import matplotlib.ticker as mtick
from isoweek import Week
locale.setlocale(locale.LC_ALL, 'de_DE.UTF-8')
site_folder = 'site/'
data_folder = 'data/'
2021-01-14 23:16:40 +00:00
einwohner_deutschland = 83190556
herd_immunity = 0.7
2021-01-14 23:16:40 +00:00
today = datetime.date.today()
print_today = today.isoformat()
filename_now = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
2021-01-14 23:16:40 +00:00
force_renew_plots = False
force_renew_dashboard = False
# https://www.tagesschau.de/ausland/europa/ursula-von-der-leyen-zu-corona-impfstoffen-101.html
target_date_for_herd_immunity = datetime.date(2021, 9, 22)
days_until_target = (target_date_for_herd_immunity - today).days - 21
2021-01-14 23:16:40 +00:00
# DIN A4 Plots
plt.rcParams["figure.figsize"] = [11.69, 8.27]
# Download
data_filename = '{}/{}_Impfquotenmonitoring.xlsx'.format(data_folder, filename_now)
r = req.get('https://www.rki.de/DE/Content/InfAZ/N/Neuartiges_Coronavirus/Daten/Impfquotenmonitoring.xlsx?__blob=publicationFile')
with open(data_filename, 'wb') as outfile:
outfile.write(r.content)
#data_filename = 'data/20210118151908_Impfquotenmonitoring.xlsx'
rki_file = pd.read_excel(data_filename, sheet_name=None, engine='openpyxl')
raw_data = rki_file['Impfungen_proTag']
2021-01-14 23:16:40 +00:00
impfungen = raw_data[:-1].dropna(subset=['Datum'])#.fillna(0)
2021-01-14 23:16:40 +00:00
2021-04-09 13:26:30 +00:00
impfungen.drop(impfungen.tail(2).index,inplace=True) # remove Gesamt row
2021-01-14 23:16:40 +00:00
dates = impfungen['Datum']
start_of_reporting_date = dates.iloc[0].date()
2021-01-14 23:16:40 +00:00
def calculate_vaccination_data(data):
total = int(np.sum(data))
total_percentage = float(total) / einwohner_deutschland * 100
to_be_vaccinated = einwohner_deutschland - total
last_date = dates.iloc[-1].date()
start_of_vaccination_index = (data != 0).argmax(axis=0)
start_of_vaccination_date = dates[start_of_vaccination_index].date()
days_since_start_of_vaccination = (last_date - start_of_vaccination_date).days
days_since_start_of_reporting = (last_date - start_of_reporting_date).days
2021-02-17 10:55:30 +00:00
valid_data = data[start_of_vaccination_index:]
cumulative = np.concatenate(([math.nan] * (days_since_start_of_reporting - days_since_start_of_vaccination), np.cumsum(valid_data)))
mean_all_time = np.mean(valid_data)
mean_seven_days = np.mean(data[-7:])
vaccinations_by_week_map = map(lambda x: (Week.withdate(x[0]), x[1]), zip(dates, data))
vaccinations_by_week = {}
for w, v in vaccinations_by_week_map:
if w in vaccinations_by_week:
vaccinations_by_week[w] = vaccinations_by_week[w] + v
else:
vaccinations_by_week[w] = v
def extrapolate(rate, to_be_vaccinated):
days_extrapolated = int(np.ceil(to_be_vaccinated / rate))
extrapolated_dates = np.array([dates[0] + datetime.timedelta(days=i) for i in range(days_extrapolated)])
2021-01-14 23:16:40 +00:00
date_done = extrapolated_dates[-1]
date_herd_immunity = extrapolated_dates[int(np.ceil(days_extrapolated * herd_immunity))]
extrapolated_vaccinations = total + rate * range(-days_since_start_of_reporting, days_extrapolated - days_since_start_of_reporting)
return {
'rate': rate,
'rate_int': int(np.round(rate)),
'days_extrapolated': days_extrapolated,
'dates': extrapolated_dates,
'date_done': date_done,
'date_done_str': date_done.strftime('%d. %B %Y'),
'date_herd_immunity': date_herd_immunity,
'date_herd_immunity_str': date_herd_immunity.strftime('%d. %B %Y'),
'extrapolated_vaccinations': extrapolated_vaccinations
}
extrapolation_mean_all_time = extrapolate(mean_all_time, to_be_vaccinated)
extrapolation_last_rate = extrapolate(data.iloc[-1], to_be_vaccinated)
extrapolation_mean_seven_days = extrapolate(mean_seven_days, to_be_vaccinated)
mean_vaccination_rates_daily = np.round(cumulative / range(1, len(cumulative) + 1))
2021-02-17 10:55:30 +00:00
vaccination_rates_daily_rolling_average = data.rolling(7).mean()
vaccinations_missing_until_target = einwohner_deutschland * 0.7 - total
vaccination_rate_needed_for_target = vaccinations_missing_until_target / days_until_target
vaccination_rate_needed_for_target_percentage = mean_all_time / vaccination_rate_needed_for_target * 100
return {
'daily': data,
'cumulative': cumulative,
'total': total,
'total_percentage': total_percentage,
'to_be_vaccinated': to_be_vaccinated,
'last_date': last_date,
'last_date_str': last_date.strftime('%d. %B %Y'),
'days_since_start': days_since_start_of_vaccination + 1, # Shift from zero to one-based-index
'start_of_vaccination_date': start_of_vaccination_date,
'start_of_vaccination_date_str': start_of_vaccination_date.strftime('%d. %B %Y'),
'vaccinations_by_week': vaccinations_by_week,
'extrapolation_mean_all_time': extrapolation_mean_all_time,
'extrapolation_last_rate': extrapolation_last_rate,
'extrapolation_mean_seven_days': extrapolation_mean_seven_days,
'mean_vaccination_rates_daily': mean_vaccination_rates_daily,
2021-02-17 10:55:30 +00:00
'vaccination_rates_daily_rolling_average': vaccination_rates_daily_rolling_average,
'vaccinations_missing_until_target': int(np.floor(vaccinations_missing_until_target)),
'vaccination_rate_needed_for_target': int(np.floor(vaccination_rate_needed_for_target)),
'vaccination_rate_needed_for_target_percentage': vaccination_rate_needed_for_target_percentage
}
data_first_vaccination = calculate_vaccination_data(impfungen['Erstimpfung'])
data_second_vaccination = calculate_vaccination_data(impfungen['Zweitimpfung'])
2021-01-14 23:16:40 +00:00
# Stand aus Daten auslesen
#stand = dates.iloc[-1]
#print_stand = stand.isoformat()
# Stand aus offiziellen Angaben auslesen
stand = rki_file['Erläuterung'].iloc[1][0]
2021-01-14 23:16:40 +00:00
2021-02-21 14:50:11 +00:00
stand_regex = re.compile('^Datenstand: (\d\d.\d\d.\d\d\d\d, \d?\d:\d\d) Uhr$')
2021-01-14 23:16:40 +00:00
m = stand_regex.match(stand)
stand_date = datetime.datetime.strptime(m.groups()[0], '%d.%m.%Y, %H:%M')
print_stand = stand_date.isoformat()
filename_stand = stand_date.strftime("%Y%m%d%H%M%S")
'''
# Infos der einzelnen Länder
details_sheet_name = (set(rki_file.keys()) - {'Erläuterung', 'Impfungen_proTag'}).pop()
details_sheet = rki_file[details_sheet_name]
regionalcodes = details_sheet['RS'].iloc[0:17]
land_names = details_sheet['Bundesland'].iloc[0:17]
total_vaccinations_by_land = details_sheet['Impfungen kumulativ'].iloc[0:17]
vaccination_per_mille_by_land = details_sheet['Impfungen pro 1.000 Einwohner'].iloc[0:17]
vaccination_reason_age_by_land = details_sheet['Indikation nach Alter*'].iloc[0:17]
vaccination_reason_job_by_land = details_sheet['Berufliche Indikation*'].iloc[0:17]
vaccination_reason_medical_by_land = details_sheet['Medizinische Indikation*'].iloc[0:17]
vaccination_reason_oldhome_by_land = details_sheet['Pflegeheim-bewohnerIn*'].iloc[0:17]
details_per_land = {}
details_per_land_formatted = {}
# Regionalcodes der Länder zu Abkürzung und Name (Plus gesamt)
laendernamen = [
('SH', 'Schleswig-Holstein'),
('HH', 'Hamburg'),
('NI', 'Niedersachsen'),
('HB', 'Bremen'),
('NW', 'Nordrhein-Westfalen'),
('HE', 'Hessen'),
('RP', 'Rheinland-Pfalz'),
('BW', 'Baden-Württemberg'),
('BY', 'Bayern'),
('SL', 'Saarland'),
('BE', 'Berlin'),
('BB', 'Brandenburg'),
('MV', 'Mecklenburg-Vorpommern'),
('SN', 'Sachsen'),
('ST', 'Sachsen-Anhalt'),
('TH', 'Thüringen'),
('𝚺', 'Gesamt')
]
def row_to_details(i):
regionalcode = regionalcodes[i] if i != 16 else 16
print(laendernamen[regionalcode])
shortname, name = laendernamen[regionalcode]
return {
'name': name,
'shortname': shortname,
'total_vaccinations': int(total_vaccinations_by_land[i]),
'total_vaccinations_percentage': vaccination_per_mille_by_land[i] / 10,
'vaccination_reason_age': int(vaccination_reason_age_by_land[i]),
'vaccination_reason_age_percentage': np.round(vaccination_reason_age_by_land[i] / total_vaccinations_by_land[i] * 100),
'vaccination_reason_job': int(vaccination_reason_job_by_land[i]),
'vaccination_reason_job_percentage': np.round(vaccination_reason_job_by_land[i] / total_vaccinations_by_land[i] * 100),
'vaccination_reason_medical': int(vaccination_reason_medical_by_land[i]),
'vaccination_reason_medical_percentage': np.round(vaccination_reason_medical_by_land[i] / total_vaccinations_by_land[i] * 100),
'vaccination_reason_oldhome': int(vaccination_reason_oldhome_by_land[i]),
'vaccination_reason_oldhome_percentage': np.round(vaccination_reason_oldhome_by_land[i] / total_vaccinations_by_land[i] * 100),
}
def row_to_details_formatted(i):
regionalcode = regionalcodes[i] if i != 16 else 16
print(laendernamen[regionalcode])
shortname, name = laendernamen[regionalcode]
return {
'name': name,
'shortname': shortname,
'total_vaccinations': '{:n}'.format(int(total_vaccinations_by_land[i])).replace('.', ''),
'total_vaccinations_percentage': '{:.3n}'.format(np.round(vaccination_per_mille_by_land[i] / 10, 2)),
'vaccination_reason_age': '{:n}'.format(int(vaccination_reason_age_by_land[i])).replace('.', ''),
'vaccination_reason_age_percentage': '{:n}'.format(np.round(vaccination_reason_age_by_land[i] / total_vaccinations_by_land[i] * 100)),
'vaccination_reason_job': '{:n}'.format(int(vaccination_reason_job_by_land[i])).replace('.', ''),
'vaccination_reason_job_percentage': '{:n}'.format(np.round(vaccination_reason_job_by_land[i] / total_vaccinations_by_land[i] * 100)),
'vaccination_reason_medical': '{:n}'.format(int(vaccination_reason_medical_by_land[i])).replace('.', ''),
'vaccination_reason_medical_percentage': '{:n}'.format(np.round(vaccination_reason_medical_by_land[i] / total_vaccinations_by_land[i] * 100)),
'vaccination_reason_oldhome': '{:n}'.format(int(vaccination_reason_oldhome_by_land[i])).replace('.', ''),
'vaccination_reason_oldhome_percentage': '{:n}'.format(np.round(vaccination_reason_oldhome_by_land[i] / total_vaccinations_by_land[i] * 100))
}
for i in range(len(land_names) - 1):
details_per_land[land_names[i]] = row_to_details(i)
details_per_land_formatted[land_names[i]] = row_to_details_formatted(i)
details_total = row_to_details(16)
details_total_formatted = row_to_details_formatted(16)
'''
archive_folder = site_folder + 'archive/' + filename_stand
if os.path.isdir(archive_folder):
print('Archive folder {} already exists'.format(archive_folder))
else:
os.mkdir(archive_folder)
def check_recreate_plot(plot_name):
archive_plot_filename = '{}/{}'.format(archive_folder, plot_name)
if os.path.isfile(archive_plot_filename + '.pdf') and not force_renew_plots:
print('Plot {} already exists'.format(plot_name))
return False
return True
def save_plot(plot_name):
folders = [archive_folder, site_folder]
file_formats = ['pdf', 'png', 'svg']
file_template = '{folder}/{plot_name}.{format}'
for folder in folders:
for format in file_formats:
plt.savefig(file_template.format(folder=folder, plot_name=plot_name, format=format))
print('Created plot {} as {}'.format(plot_name, file_formats))
def labeled_timeperiod(ax, start, end, text, color='lightgrey', y_height = 100):
centre = start + (end - start) / 2
ax.axvspan(start, end, color=color, alpha=0.5)
ax.text(centre, ax.get_ylim()[1] * (y_height / 100), text, bbox={
'boxstyle': 'square',
'fc': color,
'ec': 'black'
}, ha='center')
def add_annotations(ax):
labeled_timeperiod(ax, datetime.date(2021, 3, 15), datetime.date(2021, 3, 19), 'AZ-Stopp', 'silver')
labeled_timeperiod(ax, datetime.date(2021, 3, 29), today, 'AZ-Stopp u. 60', 'lightgrey', 96)
labeled_timeperiod(ax, datetime.date(2021, 4, 6), today, 'Arztpraxen impfen', 'lightgreen')
def plot_vaccination_bar_graph_total_time():
plot_name = 'vaccination_bar_graph_total_time'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Tägliche Impfrate (Erst- und Zweitimpfung übereinander)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.grid()
ax.bar(dates, data_first_vaccination['daily'], label='Tägliche Erstimpfungen', color='blue')
ax.bar(dates, data_second_vaccination['daily'], label='Tägliche Zweitimpfungen', color='lightblue', bottom=data_first_vaccination['daily'])
ax.set_ylim([0, np.max(data_first_vaccination['daily']) + np.max(data_second_vaccination['daily'])])
ax.legend(loc='upper left')
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Tägliche Impfungen')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_vaccination_bar_graph_total_time()
def plot_vaccination_bar_graph_total_time_by_week():
plot_name = 'vaccination_bar_graph_total_time_by_week'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Wöchentliche Impfrate (Erst- und Zweitimpfung übereinander)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.grid()
w = [w.day(3) for w in data_first_vaccination['vaccinations_by_week'].keys()]
f = list(data_first_vaccination['vaccinations_by_week'].values())
s = list(data_second_vaccination['vaccinations_by_week'].values())
bar1 = ax.bar(w, f, label='Wöchentliche Erstimpfungen', color='blue', width=6.8)
bar2 = ax.bar(w, s, label='Wöchentliche Zweitimpfungen', color='lightblue', width=6.8, bottom=f)
i = 0
nan_to_zero = lambda x: 0 if math.isnan(x) else x
for r1, r2 in zip(bar1, bar2):
x = r1.get_x() + r1.get_width() / 2.0
rh1 = nan_to_zero(r1.get_height())
rh2 = nan_to_zero(r2.get_height())
h1 = math.floor(rh1 / 1000)
h2 = math.floor(rh2 / 1000)
hg = math.floor((rh1 + rh2) / 1000)
if h1 > 30:
plt.text(x, h1 * 500, f'{h1:5n} k'.replace('.', ''), ha='center', va='center', color='white')
if h2 > 30:
plt.text(x, h1 * 1000 + h2 * 500, f'{h2:5n} k'.replace('.', ''), ha='center', va='center', color='black')
plt.text(x, hg * 1000, f'{hg:5n} k'.replace('.', ''), ha='center', va='bottom')
if i == 12:
# Woche der AstraZeneca-Aussetzung
plt.annotate('AZ-Stopp', (x, hg * 1000 + 50000),
xytext=(x, ax.get_ylim()[1]),
arrowprops={
'arrowstyle': '->'
},
bbox={
'boxstyle': 'square',
'fc': 'white',
'ec': 'black'
},
ha='center')
if i == len(bar1) - 1:
plt.annotate('Diese Woche', (x, hg * 1000 + 50000),
xytext=(x, ax.get_ylim()[1]),
arrowprops={
'arrowstyle': '->',
'relpos': (0, 0)
},
bbox={
'boxstyle': 'square',
'fc': 'white',
'ec': 'black'
},
ha='left')
i = i + 1
ax.legend(loc='upper left')
ax.get_xaxis().set_major_formatter(DateFormatter('%Y-w%W'))
ax.get_xaxis().set_major_locator(WeekdayLocator(3, 2))
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Wöchentliche Impfungen')
save_plot(plot_name)
plt.close()
plot_vaccination_bar_graph_total_time_by_week()
def plot_vaccination_bar_graph_total_time_two_bars():
plot_name = 'vaccination_bar_graph_total_time_two_bars'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Tägliche Impfrate (Erst- und Zweitimpfung nebeneinander)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.grid()
date_numbers = date2num(dates)
ax.bar(date_numbers - 0.2, data_first_vaccination['daily'], width=0.4, label='Tägliche Erstimpfungen', color='blue')
ax.bar(date_numbers + 0.2, data_second_vaccination['daily'], width=0.4, label='Tägliche Zweitimpfungen', color='lightblue')
ax.set_ylim([0, np.max(data_first_vaccination['daily']) + np.max(data_second_vaccination['daily'])])
ax.legend(loc='upper left')
ax.xaxis_date()
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Tägliche Impfungen')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_vaccination_bar_graph_total_time_two_bars()
def plot_vaccination_bar_graph_compare_both_vaccinations():
plot_name = 'vaccination_bar_graph_compare_both_vaccinations'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Tägliche Impfrate (Erst- und Zweitimpfung um 21 Tage versetzt)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.grid()
date_numbers_first = date2num(dates + datetime.timedelta(days=21))
date_numbers_second = date2num(dates)
ax.bar(date_numbers_first - 0.2, data_first_vaccination['daily'], width=0.4, label='Tägliche Erstimpfungen', color='blue')
ax.bar(date_numbers_second + 0.2, data_second_vaccination['daily'], width=0.4, label='Tägliche Zweitimpfungen', color='lightblue')
ax.set_ylim([0, np.max([np.max(data_first_vaccination['daily']), np.max(data_second_vaccination['daily'])])])
ax.legend(loc='upper left')
ax.xaxis_date()
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Tägliche Impfungen')
save_plot(plot_name)
plt.close()
plot_vaccination_bar_graph_compare_both_vaccinations()
def plot_cumulative_two_vaccinations():
plot_name = 'cumulative_two_vaccinations'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Kumulative Impfrate (Erst- und Zweitimpfung)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.grid()
first_vaccinations_cumulative = data_first_vaccination['cumulative']
second_vaccinations_cumulative = data_second_vaccination['cumulative']
ax.fill_between(dates, first_vaccinations_cumulative, label='Erstimpfungen', color='blue')
ax.fill_between(dates, second_vaccinations_cumulative, label='Zweitimpfungen', color='lightblue')
2021-02-17 10:55:30 +00:00
ax.set_ylim([0, first_vaccinations_cumulative[-1]])
ax.legend(loc='upper left')
ax.xaxis_date()
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Kumulative Impfungen')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_cumulative_two_vaccinations()
def plot_cumulative_two_vaccinations_percentage():
plot_name = 'cumulative_two_vaccinations_percentage'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Kumulative Impfrate (Erst- und Zweitimpfung) in Prozent der Bevökerung Deutschlands ({} Einwohner)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
'{:n}'.format(einwohner_deutschland).replace('.', ''),
print_stand, print_today
)
)
ax.grid()
first_vaccinations_cumulative = data_first_vaccination['cumulative'] / einwohner_deutschland
second_vaccinations_cumulative = data_second_vaccination['cumulative'] / einwohner_deutschland
ax.fill_between(dates, first_vaccinations_cumulative, label='Erstimpfungen', color='blue')
ax.fill_between(dates, second_vaccinations_cumulative, label='Zweitimpfungen', color='lightblue')
ax.set_ylim([0, 1])
ax.legend(loc='upper left')
ax.xaxis_date()
ax.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))
ax.set_xlabel('Datum')
ax.set_ylabel('Kumulative Impfungen')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_cumulative_two_vaccinations_percentage()
def plot_people_between_first_and_second():
plot_name = 'people_between_first_and_second'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Personen zwischen Erst- und Zweitimpfung\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.grid()
first_vaccinations_cumulative = data_first_vaccination['cumulative']
2021-02-17 10:55:30 +00:00
second_vaccinations_cumulative = np.nan_to_num(data_second_vaccination['cumulative'], nan=0)
people_between = first_vaccinations_cumulative - second_vaccinations_cumulative
ax.plot(dates, people_between, label='Personen zwischen Erst- und Zweitimpfung', color='darkblue')
ax.bar(dates, data_first_vaccination['daily'], color='blue', label='Erstimpfungen')
ax.bar(dates, -data_second_vaccination['daily'], color='lightblue', label='Zweitimpfungen')
ax.legend(loc='upper left')
ax.xaxis_date()
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Personen zwischen Erst- und Zweitimpfung')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_people_between_first_and_second()
2021-02-17 10:55:30 +00:00
def plot_vaccination_rate():
plot_name = 'vaccination_rate'
if not check_recreate_plot(plot_name):
return
2021-02-17 10:55:30 +00:00
fig, ax = plt.subplots(1)
plt.title(
'Tägliche Impfrate sowie durchschnittliche Impfrate\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
ax.plot(dates, data_first_vaccination['daily'], label='Tägliche Erstimpfrate', color='blue', linewidth=0.5)
ax.plot(dates, data_second_vaccination['daily'], label='Tägliche Zweitimpfrate', color='lightblue', linewidth=0.5)
ax.plot(dates, data_first_vaccination['vaccination_rates_daily_rolling_average'], color='blue', linewidth=2, label='Erstimpfrate über sieben Tage')
ax.plot(dates, data_second_vaccination['vaccination_rates_daily_rolling_average'], color='lightblue', linewidth=2, label='Zweitimpfrate über sieben Tage')
ax.plot(dates, data_first_vaccination['mean_vaccination_rates_daily'], color='violet', label='Durchschnittliche Erstimpfrate\nbis zu diesem Tag (inkl.)')
ax.plot(dates, data_second_vaccination['mean_vaccination_rates_daily'], color='magenta', label='Durchschnittliche Zweitimpfrate\nbis zu diesem Tag (inkl.)')
ax.grid(True)
ax.legend(loc='upper left')
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Impfrate [Impfungen/Tag]')
add_annotations(ax)
save_plot(plot_name)
2021-02-17 10:55:30 +00:00
plt.close()
plot_vaccination_rate()
def plot_vaccination_done_days():
plot_name = 'vaccination_done_days'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Lin. Extrapolation der Erstimpfungen bis 70 % der Bevölkerung anhand der durchschn. Impfrate (Anzahl Tage, Gesamt und 7 Tage)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
d = data_first_vaccination
days_remaining_daily = np.ceil((einwohner_deutschland * 0.7 - d['cumulative']) / (d['mean_vaccination_rates_daily']))
days_remaining_rolling = np.ceil((einwohner_deutschland * 0.7 - d['cumulative']) / (d['vaccination_rates_daily_rolling_average']))
ax.set_ylim(0, 2500)
ax.plot(dates, days_remaining_daily, label='Durchschnitt Gesamt', linewidth=0.5)
ax.plot(dates, days_remaining_rolling, label='Durchschnitt 7 Tage', linewidth=2)
ax.grid(True)
ax.legend(loc='upper right')
ax.get_yaxis().get_major_formatter().set_scientific(False)
ax.set_xlabel('Datum')
ax.set_ylabel('Tage, bis 70 % erreicht sind')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_vaccination_done_days()
def plot_vaccination_done_dates():
plot_name = 'vaccination_done_dates'
if not check_recreate_plot(plot_name):
return
fig, ax = plt.subplots(1)
plt.title(
'Lin. Extrapolation der Erstimpfungen bis 70 % der Bevölkerung anhand der durchschn. Impfrate (Datum, Gesamt und 7 Tage)\n'
'Datenquelle: RKI, Stand: {}. Erstellung: {}, Ersteller: Benedikt Bastin, Lizenz: CC BY-SA 4.0\n'.format(
print_stand, print_today
)
)
d = data_first_vaccination
#print(d['cumulative'])
#print(np.sum(d['daily']))
#print((einwohner_deutschland - d['cumulative'])[:-1] - d['to_be_vaccinated'])
days_remaining_daily = np.ceil((einwohner_deutschland * 0.7 - d['cumulative']) / (d['mean_vaccination_rates_daily']))
days_remaining_rolling = np.ceil((einwohner_deutschland * 0.7 - d['cumulative']) / (d['vaccination_rates_daily_rolling_average']))
dates_daily = [today + datetime.timedelta(days) for days in days_remaining_daily]
dates_rolling = [today + datetime.timedelta(days) for days in days_remaining_rolling.dropna()]
#print(dates_rolling)
ax.set_ylim(today, today + datetime.timedelta(int(np.max(days_remaining_rolling) * 1.05)))
ax.plot(dates, dates_daily, label='Durchschnitt Gesamt', linewidth=0.5)
ax.plot(dates[6:], dates_rolling, label='Durchschnitt 7 Tage', linewidth=2)
ax.grid(True)
ax.legend(loc='upper right')
ax.set_xlabel('Datum')
ax.set_ylabel('Datum, an dem 70 % erreicht sind')
add_annotations(ax)
save_plot(plot_name)
plt.close()
plot_vaccination_done_dates()
def render_dashboard():
dashboard_filename = 'site/index.xhtml'
dashboard_archive_filename = 'site/archive/{}/index.xhtml'.format(filename_stand)
stylesheet_filename = 'site/rki-dashboard.css'
stylesheet_archive_filename = 'site/archive/{}/rki-dashboard.css'.format(filename_stand)
if os.path.isfile(dashboard_archive_filename) and not force_renew_dashboard:
print('Dashboard {} already exists'.format(dashboard_archive_filename))
return
from jinja2 import Template, Environment, FileSystemLoader, select_autoescape
env = Environment(
loader=FileSystemLoader('./'),
autoescape=select_autoescape(['html', 'xml', 'xhtml'])
)
german_text_date_format = '%d. %B %Y'
df = german_text_date_format
german_text_datetime_format = '%d. %B %Y, %H:%M:%S Uhr'
dtf = german_text_datetime_format
latest_dashboard_filename = site_folder + 'index.xhtml'
archive_dashboard_filename = archive_folder
template = env.get_template('dashboard_template.xhtml')
template.stream(
stand = stand_date.strftime(dtf),
filename_stand = filename_stand,
einwohner_deutschland = '{:n}'.format(einwohner_deutschland).replace('.', ''),
herd_immunity = '{:n}'.format(int(herd_immunity * 100)),
target_date_for_herd_immunity = target_date_for_herd_immunity,
target_date_for_herd_immunity_str = target_date_for_herd_immunity.strftime('%d. %B %Y'),
days_until_target = days_until_target,
data_first_vaccination = data_first_vaccination,
data_second_vaccination = data_second_vaccination,
#details_per_land = dict(sorted(details_per_land_formatted.items(), key=lambda item: item[0])),
#details_total = details_total_formatted
figures = [
{
'index': 1,
'filename': 'vaccination_bar_graph_total_time',
'caption': 'Tägliche Impfrate (Erst- und Zweitimpfung übereinander)'
},{
'index': 2,
'filename': 'vaccination_bar_graph_total_time_by_week',
'caption': 'Wöchentliche Impfrate (Erst- und Zweitimpfung übereinander)'
},{
'index': 3,
'filename': 'vaccination_bar_graph_total_time_two_bars',
'caption': 'Tägliche Impfrate (Erst- und Zweitimpfung nebeneinander)'
},{
'index': 4,
'filename': 'vaccination_bar_graph_compare_both_vaccinations',
'caption': 'Tägliche Impfrate (Erst- und Zweitimpfung nebeneinander)'
},{
'index': 5,
'filename': 'cumulative_two_vaccinations',
'caption': 'Kumulative Impfrate (Erst- und Zweitimpfung)'
},{
'index': 6,
'filename': 'cumulative_two_vaccinations_percentage',
'caption': 'Kumulative Impfrate (Erst- und Zweitimpfung) in Prozent der Bevölkerung Deutschlands'
},{
'index': 7,
'filename': 'people_between_first_and_second',
'caption': 'Anzahl der Personen zwischen Erst- und Zweitimpfung, also Personen, die die erste Impfung erhalten haben, die zweite aber noch nicht'
},{
'index': 8,
'filename': 'vaccination_rate',
'caption': 'Tägliche Impfrate sowie durchschnittliche Impfrate'
},{
'index': 9,
'filename': 'vaccination_done_days',
'caption': 'Lineare Extrapolation bis 70 % der Bevölkerung anhand der Erstimpfungen der durchschnittlichen Impfrate (Anzahl Tage, Gesamt und 7 Tage)'
},{
'index': 10,
'filename': 'vaccination_done_dates',
'caption': 'Lineare Extrapolation bis 70 % der Bevölkerung anhand der Erstimpfungen der durchschnittlichen Impfrate (Datum, Gesamt und 7 Tage)'
}
]
).dump('site/index.xhtml')
shutil.copyfile(dashboard_filename, dashboard_archive_filename)
shutil.copyfile(stylesheet_filename, stylesheet_archive_filename)
print('Created dashboard')
render_dashboard()