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()