import json
from cowidev.vax.utils.base import CountryVaxBase
import requests
import pandas as pd
from cowidev.vax.utils.utils import make_monotonic
[docs]class Lithuania(CountryVaxBase):
location: str = "Lithuania"
source_url_ref: str = "https://experience.arcgis.com/experience/cab84dcfe0464c2a8050a78f817924ca/page/page_3/"
vaccine_mapping = {
"AstraZeneca": "Oxford/AstraZeneca",
"Johnson & Johnson": "Johnson&Johnson",
"Moderna": "Moderna",
"Pfizer-BioNTech": "Pfizer/BioNTech",
"Pfizer-BioNTech BA.4-5": "Pfizer/BioNTech",
"Pfizer-BioNTech BA.1": "Pfizer/BioNTech",
"Novavax": "Novavax",
}
source_url_coverage: str = "https://services3.arcgis.com/MF53hRPmwfLccHCj/arcgis/rest/services/covid_vaccinations_chart_new/FeatureServer/0/query"
query_params_coverage: dict = {
"f": "json",
"where": "municipality_code='00' AND vaccination_state<>'01dalinai'",
"returnGeometry": False,
"spatialRel": "esriSpatialRelIntersects",
"outFields": "date,vaccination_state,all_cum",
"resultOffset": 0,
"resultRecordCount": 32000,
"resultType": "standard",
}
source_url_doses: str = "https://services3.arcgis.com/MF53hRPmwfLccHCj/arcgis/rest/services/covid_vaccinations_by_drug_name_new/FeatureServer/0/query"
query_params_doses: dict = {
"f": "json",
"where": "municipality_code='00'",
"returnGeometry": False,
"spatialRel": "esriSpatialRelIntersects",
"outFields": "date,vaccines_used_cum,vaccine_name",
"resultOffset": 0,
"resultRecordCount": 32000,
"resultType": "standard",
}
[docs] def read(self, url, params):
res = requests.get(url, params=params)
if res.ok:
data = [elem["attributes"] for elem in json.loads(res.content)["features"]]
df = pd.DataFrame.from_records(data)
return df
raise ValueError("Source not valid/available!")
[docs] def pipe_parse_dates(self, df: pd.DataFrame) -> pd.DataFrame:
df["date"] = pd.to_datetime(df["date"], unit="ms").dt.date.astype(str)
return df
[docs] def pipe_clean_doses(self, df: pd.DataFrame) -> pd.DataFrame:
known_vaccines = set(self.vaccine_mapping) | {"visos"}
vax_wrong = set(df.vaccine_name).difference(known_vaccines)
if vax_wrong:
raise ValueError(f"Some unknown vaccines were found {vax_wrong}")
self.vaccine_start_dates = (
df[(df.vaccines_used_cum > 0) & (df.vaccine_name != "visos")]
.replace(self.vaccine_mapping)
.groupby("vaccine_name", as_index=False)
.min()
.drop(columns="vaccines_used_cum")
)
return (
df[(df.vaccines_used_cum > 0) & (df.vaccine_name == "visos")]
.drop(columns="vaccine_name")
.rename(columns={"vaccines_used_cum": "total_vaccinations"})
)
[docs] def pipe_clean_coverage(self, df: pd.DataFrame) -> pd.DataFrame:
assert set(df.vaccination_state) == {"00visos", "03pakartotinai", "02pilnai"}
df = (
df.pivot(index="date", columns="vaccination_state", values="all_cum")
.reset_index()
.rename(
columns={
"00visos": "people_vaccinated",
"02pilnai": "people_fully_vaccinated",
"03pakartotinai": "total_boosters",
}
)
)
# 02pilnai actually includes only people fully vaccinated *without* boosters
# People who get boosters are transferred from 02pilnai to 03pakartotinai
df["people_fully_vaccinated"] = df.people_fully_vaccinated + df.total_boosters
return df[df.people_vaccinated > 0]
[docs] def _find_vaccines(self, date):
vaccines = self.vaccine_start_dates.loc[self.vaccine_start_dates.date <= date, "vaccine_name"].values
return ", ".join(sorted(vaccines))
[docs] def pipe_add_vaccines(self, df: pd.DataFrame) -> pd.DataFrame:
df["vaccine"] = df.date.apply(self._find_vaccines)
return df
[docs] def export(self):
coverage = (
self.read(self.source_url_coverage, self.query_params_coverage)
.pipe(self.pipe_parse_dates)
.pipe(self.pipe_clean_coverage)
)
doses = (
self.read(self.source_url_doses, self.query_params_doses)
.pipe(self.pipe_parse_dates)
.pipe(self.pipe_clean_doses)
)
df = (
pd.merge(coverage, doses, how="inner", on="date")
.pipe(self.pipe_add_vaccines)
.pipe(self.pipe_metadata)
.pipe(make_monotonic, max_removed_rows=20)
)
self.export_datafile(df)
[docs]def main():
Lithuania().export()