Source code for cowidev.testing.batch.slovakia

import pandas as pd
from cowidev.testing import CountryTestBase


[docs]class Slovakia(CountryTestBase): location: str = "Slovakia" units: str = "tests performed" source_label: str = "Ministry of Health" notes: str = "Ministry of Health via https://github.com/Institut-Zdravotnych-Analyz" source_url: str = "https://github.com/Institut-Zdravotnych-Analyz/covid19-data/raw/main/DailyStats/OpenData_Slovakia_Covid_DailyStats.csv" source_url_ref: str = "https://github.com/Institut-Zdravotnych-Analyz/covid19-data" rename_columns: dict = {"Datum": "Date"}
[docs] def read(self) -> pd.DataFrame: """Read data from source""" df = pd.read_csv( self.source_url, sep=";", usecols=["Datum", "Dennych.PCR.testov", "AgTests", "Dennych.PCR.prirastkov", "AgPosit"], ) return df
[docs] def pipe_metrics(self, df: pd.DataFrame) -> pd.DataFrame: """Pipes metrics""" df = df.sort_values("Date") return df.assign( **{ "positive": df["Dennych.PCR.prirastkov"].fillna(0) + df["AgPosit"].fillna(0), "Daily change in cumulative total": df["Dennych.PCR.testov"].fillna(0) + df["AgTests"].fillna(0), } )
[docs] def pipe_pr(self, df: pd.DataFrame) -> pd.DataFrame: """Calculate positive rate""" return df.assign( **{ "Positive rate": df.positive.rolling(7) .mean() .div(df["Daily change in cumulative total"].rolling(7).mean()) .round(3) } )
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame: """Pipeline for data""" return df.pipe(self.pipe_rename_columns).pipe(self.pipe_metrics).pipe(self.pipe_pr).pipe(self.pipe_metadata)
[docs] def export(self): """Export data to CSV""" df = self.read().pipe(self.pipeline) self.export_datafile(df)
[docs]def main(): Slovakia().export()