Source code for cowidev.oxcgrt.grapher

from datetime import datetime

import pandas as pd
from cowidev.grapher.db.base import GrapherBaseUpdater
from cowidev.utils.utils import time_str_grapher, get_filename
from cowidev.utils.clean.dates import DATE_FORMAT

ZERO_DAY = "2020-01-01"
zero_day = datetime.strptime(ZERO_DAY, DATE_FORMAT)

URL_VACCINE = "https://raw.githubusercontent.com/OxCGRT/covid-policy-tracker/master/data/OxCGRT_vaccines_full.csv"


[docs]def run_grapheriser(input_path: str, input_path_country_std: str, output_path: str): cgrt = pd.read_csv( input_path, low_memory=False, usecols=[ "CountryName", "Date", "RegionCode", "C1M_School closing", "C2M_Workplace closing", "C3M_Cancel public events", "C4M_Restrictions on gatherings", "C5M_Close public transport", "C6M_Stay at home requirements", "C7M_Restrictions on internal movement", "C8EV_International travel controls", "E1_Income support", "E2_Debt/contract relief", "E3_Fiscal measures", "E4_International support", "H1_Public information campaigns", "H2_Testing policy", "H3_Contact tracing", "H4_Emergency investment in healthcare", "H5_Investment in vaccines", "H6M_Facial Coverings", "H7_Vaccination policy", "StringencyIndex_Average", "ContainmentHealthIndex_Average", "V2A_Vaccine Availability (summary)", "V2B_Vaccine age eligibility/availability age floor (general population summary)", "V2C_Vaccine age eligibility/availability age floor (at risk summary)", ], ) country_mapping = pd.read_csv(input_path_country_std) cgrt = cgrt[cgrt.RegionCode.isnull()].drop(columns="RegionCode") vax = pd.read_csv( URL_VACCINE, low_memory=False, usecols=["CountryName", "Date", "V2_Vaccine Availability (summary)", "V2_Pregnant people"], ) cgrt = pd.merge(cgrt, vax, how="outer", on=["CountryName", "Date"], validate="one_to_one") cgrt.loc[:, "Date"] = pd.to_datetime(cgrt["Date"], format="%Y%m%d").map(lambda date: (date - zero_day).days) cgrt = country_mapping.merge(cgrt, on="CountryName", how="right") missing_from_mapping = cgrt[cgrt["Country"].isna()]["CountryName"].unique() if len(missing_from_mapping) > 0: raise Exception(f"Missing countries in mapping: {missing_from_mapping}") cgrt = cgrt.drop(columns=["CountryName"]) rename_dict = { "Date": "Year", "C1M_School closing": "school_closures", "C2M_Workplace closing": "workplace_closures", "C3M_Cancel public events": "cancel_public_events", "C5M_Close public transport": "close_public_transport", "H1_Public information campaigns": "public_information_campaigns", "C7M_Restrictions on internal movement": "restrictions_internal_movements", "C8EV_International travel controls": "international_travel_controls", "E3_Fiscal measures": "fiscal_measures", "H4_Emergency investment in healthcare": "emergency_investment_healthcare", "H5_Investment in vaccines": "investment_vaccines", "H3_Contact tracing": "contact_tracing", "H6M_Facial Coverings": "facial_coverings", "StringencyIndex_Average": "stringency_index", "ContainmentHealthIndex_Average": "containment_index", "C4M_Restrictions on gatherings": "restriction_gatherings", "C6M_Stay at home requirements": "stay_home_requirements", "E1_Income support": "income_support", "E2_Debt/contract relief": "debt_relief", "E4_International support": "international_support", "H7_Vaccination policy": "vaccination_policy", "H2_Testing policy": "testing_policy", "V2_Vaccine Availability (summary)": "vaccine_eligibility", "V2A_Vaccine Availability (summary)": "vaccine_availability", "V2B_Vaccine age eligibility/availability age floor (general population summary)": "vaccine_eligibility_age", "V2C_Vaccine age eligibility/availability age floor (at risk summary)": "vaccine_eligibility_age_at_risk", "V2_Pregnant people": "pregnant_vaccine_eligibility", } cgrt = cgrt.rename(columns=rename_dict).sort_values(["Country", "Year"]) cgrt.to_csv(output_path, index=False)
[docs]def run_db_updater(input_path: str): dataset_name = get_filename(input_path) GrapherBaseUpdater( dataset_name=dataset_name, source_name=( "Oxford COVID-19 Government Response Tracker, Blavatnik School of Government, University of Oxford" f" – Last updated {time_str_grapher()}" ), zero_day=ZERO_DAY, slack_notifications=False, ).run()