import pandas as pd
from cowidev.utils.clean import clean_date
from cowidev.utils.web import request_json
from cowidev.vax.utils.incremental import enrich_data, increment
vaccine_mapping = {
"AstraZeneca / БНЭУ 🤝 + КОВАКС 🤝": "Oxford/AstraZeneca",
"Pfizer-BioNTech / КОВАКС 🤝 + ЯПОН 🤝": "Pfizer/BioNTech",
"Pfizer-BioNTech / КОВАКС 🤝": "Pfizer/BioNTech",
"Синофарм / БНХАУ 🤝+ 💵": "Sinopharm/Beijing",
"Спутник V / ОХУ 💵": "Sputnik V",
"Спутник V / ОХУ 🤝 + 💵": "Sputnik V",
}
[docs]def read(source: str) -> pd.Series:
data = request_json(source, parser="html.parser")
return parse_data(data)
[docs]def parse_data(data: dict) -> pd.Series:
date = clean_date(data["updated"], "%Y/%m/%d")
people_vaccinated = data["progress"]
people_fully_vaccinated = data["completed"]
boosters = data["d3"]
return pd.Series(
data={
"date": date,
"people_vaccinated": people_vaccinated,
"people_fully_vaccinated": people_fully_vaccinated,
"total_boosters": boosters,
"vaccine": ", ".join(_get_vaccine_names(data, translate=True)),
}
)
[docs]def _get_vaccine_names(data: dict, translate: bool = False) -> list:
vaccine_names = [v["name"] for v in data["vaccines"]]
_check_vaccine_names(vaccine_names)
if translate:
return sorted([vaccine_mapping[v] for v in vaccine_names])
else:
return sorted(vaccine_names)
[docs]def _check_vaccine_names(vaccine_names: list):
unknown_vaccines = set(vaccine_names).difference(vaccine_mapping.keys())
if unknown_vaccines:
raise ValueError("Found unknown vaccines: {}".format(unknown_vaccines))
[docs]def add_totals(ds: pd.Series) -> pd.Series:
total_vaccinations = ds.people_vaccinated + ds.people_fully_vaccinated + ds.total_boosters
return enrich_data(ds, "total_vaccinations", total_vaccinations)
[docs]def enrich_location(ds: pd.Series) -> pd.Series:
return enrich_data(ds, "location", "Mongolia")
[docs]def enrich_source(ds: pd.Series) -> pd.Series:
return enrich_data(ds, "source_url", "https://ikon.mn/")
[docs]def pipeline(ds: pd.Series) -> pd.Series:
return ds.pipe(add_totals).pipe(enrich_location).pipe(enrich_source)
[docs]def main():
source = "https://ikon.mn/api/json/vaccine"
data = read(source).pipe(pipeline)
increment(
location=str(data["location"]),
total_vaccinations=int(data["total_vaccinations"]),
people_vaccinated=int(data["people_vaccinated"]),
people_fully_vaccinated=int(data["people_fully_vaccinated"]),
total_boosters=int(data["total_boosters"]),
date=str(data["date"]),
source_url=str(data["source_url"]),
vaccine=str(data["vaccine"]),
)