Source code for cowidev.testing.batch.canada

import pandas as pd
import requests
from io import StringIO

from cowidev.testing import CountryTestBase




[docs]
class Canada(CountryTestBase):
    location = "Canada"
    source_url = "https://health-infobase.canada.ca/src/data/covidLive/covid19-download.csv"
    source_url_ref = "https://health-infobase.canada.ca/src/data/covidLive/covid19-download.csv"
    rename_columns = {
        "numtests": "Cumulative total",
        "prname": "Country",
        "date": "Date",
    }
    source_label = "Government of Canada"



[docs]
    def read(self):
        requests.packages.urllib3.disable_warnings()
        requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS += ":HIGH:!DH:!aNULL"
        try:
            requests.packages.urllib3.contrib.pyopenssl.util.ssl_.DEFAULT_CIPHERS += ":HIGH:!DH:!aNULL"
        except AttributeError:
            # no pyopenssl support used / needed / available
            pass

        req = requests.get(self.source_url, verify=False)
        data = StringIO(req.text)

        df = pd.read_csv(data, usecols=["prname", "date", "numtested", "numtests"])
        df = df[df.prname == "Canada"]
        return df





[docs]
    def pipeline_metric(self, df: pd.DataFrame, units: str, metric_field: str) -> pd.DataFrame:
        if metric_field == "numtested":
            metric_drop = "numtests"
        elif metric_field == "numtests":
            metric_drop = "numtested"
        else:
            raise ValueError("Invalid metric")
        df = (
            df.copy()
            .assign(Units=units)
            .drop(columns=[metric_drop])
            .pipe(self.pipe_rename_columns)
            .sort_values("Date")
            .drop_duplicates(subset=["Cumulative total"], keep="last")
        )
        df = df[~df["Cumulative total"].isna()]
        return df





[docs]
    def pipeline(self, df: pd.DataFrame, units: str, metric_field: str) -> pd.DataFrame:
        return df.pipe(self.pipeline_metric, units, metric_field).pipe(self.pipe_metadata)





[docs]
    def export(self):
        df = self.read()  # .pipe(self.pipeline_base)
        # People
        # df_ = df.pipe(self.pipeline, "people tested", "numtested")
        # self.export_datafile(df_, filename=f"{self.location} - people tested")
        # Tests
        df_ = df.pipe(self.pipeline, "tests performed", "numtests")
        self.export_datafile(df_)








[docs]
def main():
    Canada().export()