Source code for cowidev.testing.batch.switzerland

import pandas as pd

from cowidev.testing import CountryTestBase
from cowidev.utils.web import request_json
from cowidev.utils.web.download import read_csv_from_url




[docs]
class Switzerland(CountryTestBase):
    location: str = "Switzerland"
    units: str = "tests performed"
    source_label: str = "Federal Office of Public Health"
    source_url_ref: str = "https://opendata.swiss/en/dataset/covid-19-schweiz"
    source_url: str = "https://www.covid19.admin.ch/api/data/context"
    rename_columns: str = {
        "datum": "Date",
        "entries": "Daily change in cumulative total",
    }



[docs]
    def read(self):
        """Read data from source"""
        url = self._get_related_url_from_source(self.source_url)
        df = read_csv_from_url(url, usecols=["datum", "entries", "entries_pos", "geoRegion"])
        return df








[docs]
    def pipe_metrics(self, df: pd.DataFrame) -> pd.DataFrame:
        """Process metrics"""
        df = df[df.geoRegion == "CH"]
        df = df.groupby("Date", as_index=False).sum()
        df["Positive rate"] = (
            df.entries_pos.rolling(7).sum().div(df["Daily change in cumulative total"].rolling(7).sum()).round(3)
        )
        df = df[df["Daily change in cumulative total"] > 0].drop(columns=["entries_pos"])
        return df





[docs]
    def pipeline(self, df: pd.DataFrame) -> pd.DataFrame:
        """Pipeline"""
        return df.pipe(self.pipe_rename_columns).pipe(self.pipe_metrics).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():
    Switzerland().export()