Source code for cowidev.vax.batch.united_kingdom

import locale

import pandas as pd
from uk_covid19 import Cov19API

from cowidev.vax.utils.utils import make_monotonic
from cowidev.vax.utils.base import CountryVaxBase


[docs]class UnitedKingdom(CountryVaxBase): location = "United Kingdom" source_url = "https://coronavirus.data.gov.uk/details/vaccinations"
[docs] def read(self): dfs = [ self._read_metrics("areaType=overview"), self._read_metrics("areaType=nation"), ] df = pd.concat(dfs).reset_index(drop=True) return df
[docs] def _read_metrics(self, filters): metrics = { "date": "date", "location": "areaName", "areaCode": "areaCode", "people_vaccinated": "cumPeopleVaccinatedFirstDoseByPublishDate", "people_fully_vaccinated": "cumPeopleVaccinatedSecondDoseByPublishDate", "total_vaccinations": "cumVaccinesGivenByPublishDate", "total_boosters": "cumPeopleVaccinatedThirdInjectionByPublishDate", "vaccinations_age": "vaccinationsAgeDemographics", } api = Cov19API( filters=[filters], structure=metrics, ) df = api.get_dataframe() return df
[docs] def pipe_source_url(self, df: pd.DataFrame) -> pd.DataFrame: return df.assign(source_url=self.source_url)
[docs] def pipe_vaccine(self, df: pd.DataFrame) -> pd.DataFrame: def _enrich_vaccine(date: str) -> str: if date < "2021-01-04": return "Pfizer/BioNTech" elif "2021-04-07" > date >= "2021-01-04": return "Oxford/AstraZeneca, Pfizer/BioNTech" elif date >= "2021-04-07": # https://www.reuters.com/article/us-health-coronavirus-britain-moderna-idUSKBN2BU0KG return "Moderna, Oxford/AstraZeneca, Pfizer/BioNTech" return df.assign(vaccine=df.date.apply(_enrich_vaccine))
[docs] def pipe_select_output_cols(self, df: pd.DataFrame) -> pd.DataFrame: return df[ [ "location", "date", "vaccine", "source_url", "total_vaccinations", "people_vaccinated", "people_fully_vaccinated", "total_boosters", ] ]
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame: return ( df.pipe(self.pipe_source_url) .pipe(self.pipe_vaccine) .pipe(self.pipe_select_output_cols) .sort_values(by=["location", "date"]) .dropna(subset=["total_vaccinations"]) )
[docs] def _filter_location(self, df: pd.DataFrame, location: str) -> pd.DataFrame: return df[df.location == location].assign(location=location)
[docs] def export(self): df_base = self.read().pipe(self.pipeline) for location in set(df_base.location): df = df_base.pipe(self._filter_location, location).pipe(self.make_monotonic, max_removed_rows=45) self.export_datafile(df, filename=location)
[docs]def main(): UnitedKingdom().export()