Source code for cowidev.testing.batch.hungary
import pandas as pd
from cowidev.testing import CountryTestBase
from cowidev.utils.gdrive import GSheetApi
from cowidev.utils.clean import clean_date_series
[docs]class Hungary(CountryTestBase):
location = "Hungary"
units = "tests performed"
source_label = "Government of Hungary"
source_url_ref = "https://atlo.team/koronamonitor/"
notes = "Made available by Atlo.team"
rename_columns = {
"Dátum": "Date",
"Új mintavételek száma": "Daily change in cumulative total",
}
[docs] def read(self) -> pd.DataFrame:
api = GSheetApi()
sheet = api.get_worksheet("1e4VEZL1xvsALoOIq9V2SQuICeQrT5MtWfBm32ad7i8Q", "koronahun")
df = sheet.to_frame()
return df
[docs] def pipe_filter(self, df):
return df[(df["Daily change in cumulative total"] != 0) & ~df["Daily change in cumulative total"].isna()]
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame:
return df.pipe(self.pipe_rename_columns).pipe(self.pipe_filter).pipe(self.pipe_date).pipe(self.pipe_metadata)
[docs] def export(self):
df = self.read().pipe(self.pipeline)
self.export_datafile(df, float_format="%.5f")