Corona ist in aller Munde, daher kann es nicht schaden, sich seine eigene Visualisierung zu basteln, um das Virus im Blick zu halten. Die folgende Visu ist so gemacht, dass sie auf dem 7" Display meines RaspberryPi ansehbar und bedienbar ist. Wer einen größeren Bildschirm hat, kann natürlich noch mehr Informationen anzeigen. Das Kochrezept ist aber immer das selbe.
Voraussetzungen
Damit die App funktioniert solltet Ihr Python, Kivy und die Matplotlib installiert haben.
Die App / Das Widget
Das folgende Programm kann entweder standalone als eigenständige App betrieben werden oder in eine andere App als Widget eingebunden werden. Am Ende des Eintrags findet Ihr den Link zu Github, wo Ihr euch den vollständig kommentierten Quellcode runterladen könnt.
Daten besorgen
Die Daten zur Visualisierung finden wir ganz einfach bei bspw. dem European Center for Disease Control. Kurz ECDC. Unter anderem kann man sich hier die täglichen Infektionszahlen in allen Ländern der Welt herunterladen. Ebenso steht einem der Zugang zu den aktuellen Fortschritten der Impfungen zur Verfügung. Es gibt sogar verschiedene Formate zur Auswahl. Ich entscheide mich hier für JSON. Solltet Ihr eine eigene App mit diesen Daten veröffentliche, so beachtet bitte die Copyright Hinweise des ECDC.
Das kv File
Wie in Kivy üblich, erstelle ich das Layout im kv File.
Zunächst definieren wir ein Layout, das die Darstellung aller weiteren Elemente übernimmt. Später im py File werdet Ihr sehen, dass es sich hierbei um ein Boxlayout handelt. Das sorgt dafür, dass meine Darstellung immer das gesamte Fenster einnimmt. Das Boxlayout stellt nur ein einziges Element, nämlich das CoronaWidget dar.
<CoronaTestLayout>:
orientation: 'vertical'
CoronaWidget:
id: wdgt1Das eigentliche Widget besteht im wesentlichen aus zwei, sogenannten Spinnern. Das sind DropDown Felder. Der erste Spinner zeigt die Kontinente an und der zweite die Länder des jeweiligen Kontinents. Das Updaten der Inhalte der Spinner läuft hier automatisch, denn die Eigenschaft values ist mit den jeweiligen Variablen in der zugehörigen Klasse im py File gebunden.
Gleiches mache ich mit den beiden Labels, die das Datum des letzten verfügbaren Eintrags, sowie die dazugehörige Zahl an Neuinfektionen anzeigen.
Der Plot wird ebenso mit den täglichen und kummulierten Infektionszahlen verbunden. Die Darstellung des Plots kann man über die Vorgabe der Linienfarben und Beschriftungen selbst noch etwas anpassen. Die verfügbaren Farben sind in der Doku von Matplotlib nachzulesen.
<CoronaWidget>:
Spinner:
id: spn1
values: root.continents
size_hint: .25, .15
pos_hint: {'top':1, 'left':1}
on_text:
root.activeContinent = self.text
root.countries = root.continentsAndCountries[self.text]
Spinner:
id: spn2
values: root.countries
size_hint: .25, .15
pos_hint: {'top':1, 'x':.25}
on_text:
root.update_active_country_infection(self.text)
root.update_active_country_vaccination(self.text)
Label:
id: lbl1
text: 'Date'
font_size: 15
size_hint: .25, .05
pos_hint: {'top':1, 'x':.5}
Label:
id: lbl2
text: root.newInfectionsDate
font_size: 25
size_hint: .25, .1
pos_hint: {'top':.95, 'x':.5}
Label:
id: lbl3
text: 'Infections'
font_size: 15
size_hint: .25, .05
pos_hint: {'top':1, 'x':.75}
Label:
id: lbl4
text: root.newInfections
font_size: 25
size_hint: .25, .1
pos_hint: {'top':.95, 'x':.75}
ScrollView:
id: scroll1
size_hint: 1, .85
pos_hint: {'top': .85, 'left':1}
GridLayout:
id: grid1
size_hint: 1, None
height: self.minimum_height
cols: 1
row_force_default: True
row_default_height: root.height * 0.85
TwoPlotsSharedXWidget:
id: plt1
sourceX: root.datesOfCases
sourceY: [root.casesWeekly, root.cumulativeCases]
units : ['','']
titles : ['Cases / week','Cumulative Cases']
ax_colors : ['r','b']
formatter : '%d-%b'
TwoPlotsSharedXWidget:
id: plt2
size_hint:1,1
#pos_hint: {'top': .85, 'left':1}
sourceX: root.datesOfVaccination
sourceY: [root.vaccinationWeekly,[]]
units : ['']
titles : ['Vaccinations / week']
ax_colors : ['r']
formatter : '%d-%b'Das Python File
Zunächst binden wir die notwendingen Module ein.
# encoding: utf-8
from kivy.app import App
from kivy.uix.relativelayout import RelativeLayout
from kivy.uix.boxlayout import BoxLayout
from kivy.properties import DictProperty, ObjectProperty, StringProperty, ListProperty
from kivy.uix.spinner import Spinner
from kivy.garden.matplotlib.backend_kivyagg import FigureCanvasKivyAgg
import matplotlib.pyplot as plt
import matplotlib.dates as dt
from datetime import datetime
from time import mktime, strptime, strftime
import os
from kivy.network.urlrequest import UrlRequest
from TwoPlotsSharedXWidgetApp import TwoPlotsSharedXWidget as TwoPlotsSharedXWidget
import pandas as pd
from kivy.resources import resource_add_path
import os
Anschließend bauen wir das CoronaWidget auf.
Zunächsteinmal definieren wir alle Eigenschaften, die über Ihren Namen and die jeweiligen Variablen im kv File gebunden werden. Wo es Sinn macht, vergebe ich hier schon Standardwerte, damit die Anzeigen nicht so leer sind.
Die Funktionen download_data und get_http_to_json beschaffen sich die, auf der Homepage des ECDC vorhandenen Daten und speichern diese in ein JSON File mit dem Namen covid.json.
Die Funktion update_dataset wird aufgerufen, wenn ein neuer Datensatz heruntergeladen wurde. Alle Informationen werden dann aus der covid.json Datei in die Variable dataset gespeichert.
Wenn ein neues Land ausgewählt wird, holt sich die Funktion update_active_country alle Daten und befüllt die Variablen für die täglichen Fälle, kummulierten Fälle und die aktuellen Fälle.
Die Funktion update_continent_spinner wird benötigt, um die Spinner initial mit allen Kontinenten und Ländern zu befüllen. Als Standardwert wähle ich hier Deutschland aus.
class CoronaWidget(RelativeLayout):
'''Shows infections for the current date and plots for daily and cumulative infections.
Attributes:
The attributes (excluding dataset[]) are bound by name to propertis in the kv file. Updating them will automatically update the displayed data in the visualisation
continentsAndCountries (DictProperty):
contains continents and correspondant countries
{'europe': ['germany', 'france', ...], 'asia': ['china', 'singapore', ...]}
continents (ListProperty):
list holding all continents. Retrieved from continentsAndCountries.
countries (ListProperty):
list holding all countries of the selected continent.
Retrieved from continentsAndCountries
activeCountry (StringProperty, str):
currently active country
activeContinent (StringProperty, str):
currently active continent
newInfections (StringProperty, str):
Infections which were reported for the latest date within activeCountry
newInfectionsDate (StringProperty, str):
Date at which the newInfections were reported
weeklyCases (ListProperty):
List holding all infections per week in the order from latest to oldest
cumulativeCases (ListProperty):
List holding all added up infections in the order from latest to oldest
datesOfCases (ListProperty):
List holding all dates for the dailyCases and cumulativeCases
notification (StringProperty, str):
Error string. Shows exceptions, like no data available.
Initially set to --.
dataset (list):
list holding all the downloaded data
'''
continentsAndCountries = DictProperty({})
continents = ListProperty([])
countries = ListProperty([])
activeCountry = StringProperty('Germany')
activeContinent = StringProperty('Europe')
newInfections = StringProperty('--')
newInfectionsDate = StringProperty('--')
casesWeekly = ListProperty([])
cumulativeCases = ListProperty([])
datesOfCases = ListProperty([])
notification = StringProperty('')
dataset = []
dataset_vaccination = []
vaccinationWeekly = ListProperty([])
datesOfVaccination = ListProperty([])
vaccines = {
'AZ' : 'AstraZeneca',
'JANSS' : 'Janssen',
'MOD' : 'Moderna',
'COM' : 'Pfizer / BioNTech',
'SIN' : 'Sinovac',
'SPU' : 'Sputnik V',
'CN' : 'CNBG',
'UNK' : 'Unknown'
}
def __init__(self, **kwargs):
super(CoronaWidget, self).__init__(**kwargs)
def download_data_infection(self, *args, **kwargs):
'''download the current data from the ECDC. Infections and deaths by week from
https://www.ecdc.europa.eu/en/publications-data/data-national-14-day-notification-rate-covid-19
'''
url = 'https://opendata.ecdc.europa.eu/covid19/nationalcasedeath/json/'
UrlRequest(url=url, on_success=self.update_dataset_infection, on_error=self.download_error, on_progress=self.progress,
chunk_size=40960)
def download_data_vaccination(self, *args, **kwargs):
'''download the current data from the ECDC. Vaccination data from
https://www.ecdc.europa.eu/en/publications-data/data-covid-19-vaccination-eu-eea
'''
url = 'https://opendata.ecdc.europa.eu/covid19/vaccine_tracker/json/'
UrlRequest(url=url, on_success=self.update_dataset_vaccination, on_error=self.download_error, on_progress=self.progress,
chunk_size=40960)
def update_dataset_infection(self, request, result):
'''write result of request into variable self.dataset'''
self.dataset = result # data was json. Therefore directly decoded by the UrlRequest
self.update_continent_spinner()
self.update_active_country_infection(country=self.activeCountry)
self.notification = ''
def update_dataset_vaccination(self, request, result):
self.dataset_vaccination = result['records']
self.update_active_country_vaccination(country=self.activeCountry)
def download_error(self, request, error):
'''notify on error'''
self.notification = 'data could not be downloaded'
def progress(self, request, current_size, total_size):
'''show progress to the user'''
#modify. Two notifications necessary since two downloads in prallel
self.notification = ('Downloading data: {} bytes of {} bytes'.format(current_size, total_size))
def update_active_country_infection(self, country, *args, **kwargs):
'''update all infection data for a new selected country'''
self.activeCountry = country
# update plot data for infections
self.casesWeekly.clear()
self.cumulativeCases.clear()
self.datesOfCases.clear()
for element in self.dataset:
if element['country'] == country and element['indicator'] == 'cases':
self.casesWeekly.append(element['weekly_count'])
self.cumulativeCases.append(element['cumulative_count'])
self.datesOfCases.append(
datetime.fromtimestamp(mktime(strptime(element['year_week'] + '-1', '%Y-%W-%w'))))
try:
self.ids['plt1'].update_plot()
except:
print('Infetions could not be updated. Check data')
# update last week labels
indexLastDate = self.datesOfCases.index(max(self.datesOfCases))
self.newInfections = str(self.casesWeekly[indexLastDate])
self.newInfectionsDate = self.datesOfCases[indexLastDate].strftime('%Y-%W')
def update_active_country_vaccination(self, country, *args, **kwargs):
'''update all vaccination data for a new selected country'''
if not self.dataset_vaccination:
return
country_code = self.get_country_code(country_name=country, code='Alpha-2 code')
#data sorted by week and all available vaccines
self.vaccinationWeekly.clear()
self.datesOfVaccination.clear()
dates = []
doses = []
vaccines = []
dates.clear()
doses.clear()
vaccines.clear()
for element in self.dataset_vaccination:
if element['ReportingCountry'] == country_code and element['TargetGroup'] == 'ALL':
dates.append(datetime.fromtimestamp(mktime(strptime(element['YearWeekISO'].replace('W','') + '-1', '%Y-%W-%w'))))
doses.append(element['FirstDose'])
vaccines.append(self.vaccines[element['Vaccine']])
#re-sort data by week and sum of all vaccines by week
dates_unique = []
doses_sum = []
dates_unique.clear()
doses_sum.clear()
for date, dose in zip(dates, doses):
if not date in dates_unique:
dates_unique.append(date)
doses_sum.append(dose)
else:
doses_sum[-1] = doses_sum[-1] + dose
self.datesOfVaccination = dates_unique
self.vaccinationWeekly = doses_sum
try:
self.ids['plt2'].update_plot()
except Exception as e:
print('Vaccination could not be updated')
print(e)
def update_continent_spinner(self, *args, **kwargs):
'''Update the spinners with all available continents and countries.
Use preset values from self.activeContinent and self.activeCountry'''
continentsAndCountries = {}
for element in self.dataset:
try:
if not element.get('continent') in continentsAndCountries.keys():
continentsAndCountries[element['continent']] = []
if not element.get('country') in continentsAndCountries[element['continent']]:
continentsAndCountries[element['continent']].append(element['country'])
except:
self.notification = ('Unknown error in : {}').format(element)
self.continentsAndCountries = continentsAndCountries
self.continents = continentsAndCountries.keys()
if self.activeContinent in self.continents:
self.ids['spn1'].text = self.activeContinent
if self.activeCountry in self.countries:
self.ids['spn2'].text = self.activeCountry
def get_country_code(self, country_name, code):
xlsx = pd.ExcelFile('CountryCodes.xlsx')
df = pd.read_excel(xlsx, sheet_name = 'Tabelle1', index_col = 0, usecols = 'A:D', engine = 'openpyxl')
two_digit_code = df.loc[country_name, code]
return two_digit_code
class CoronaTestLayout(BoxLayout):
def __init__(self, **kwargs):
super(CoronaTestLayout, self).__init__(**kwargs)
self.ids['wdgt1'].download_data_infection()
self.ids['wdgt1'].download_data_vaccination()
class CoronaWidgetApp(App):
def build(self):
return CoronaTestLayout()
if __name__ == '__main__':
#add the parent directories of your referenced kv files or they will not be found
resource_add_path(r'C:\Users\49172\PycharmProjects')
CoronaWidgetApp().run()Die Klasse TwoPlotsSharedXWidget ist ein separates Widget, das hier eingebunden und genutzt wird. Eine detaillierte Beschreibung gibt der Artikel: Matplotlib Widget in Kivy mit zwei Plots erstellen
Zu guter Letzt benötigen wir noch die Klasse für das initiale Layout. Sobald das instanziert wird, laden wir die Daten runter, aktualisieren das dataset und befüllen die Spinner mit den vorhandenen Werten. Durch das Binden der Eigenschaften, auch innerhalb des Plots, werden so auch die Plots aktualisiert.
class CoronaTestLayout(BoxLayout):
def __init__(self, **kwargs):
super(CoronaTestLayout, self).__init__(**kwargs)
self.ids['wdgt1'].download_data_infection()
self.ids['wdgt1'].download_data_vaccination()
class CoronaWidgetApp(App):
def build(self):
return CoronaTestLayout()
if __name__ == '__main__':
#add the parent directories of your referenced kv files or they will not be found
resource_add_path(r'C:\Users\49172\PycharmProjects')
CoronaWidgetApp().run()
Quellcode
Der komplette Code ist auf Github zum Download verfügbar: https://github.com/HaikoKrais/CoronaWidget