Source code for cowidev.testing.incremental.azerbaijan
from bs4 import BeautifulSoup, element
import pandas as pd
from cowidev.utils import get_soup, clean_count
from cowidev.utils.clean.dates import localdate
from cowidev.testing import CountryTestBase
[docs]class Azerbaijan(CountryTestBase):
location: str = "Azerbaijan"
units: str = "tests performed"
source_label: str = "Cabinet of Ministers of Azerbaijan"
source_url: str = "https://koronavirusinfo.az/az/page/statistika/azerbaycanda-cari-veziyyet"
source_url_ref: str = "https://koronavirusinfo.az/az/page/statistika/azerbaycanda-cari-veziyyet"
[docs] def read(self) -> pd.DataFrame:
"""Read data from source"""
soup = get_soup(self.source_url, verify=False)
df = self._parse_data(soup)
return df
[docs] def _parse_data(self, soup: BeautifulSoup) -> pd.DataFrame:
"""Parse data from soup"""
# Get the element
elem = soup.find(text="Müayinə aparılıb").parent
if not elem:
raise ValueError("Element not found, please update the script")
# Get the metrics
count = self._parse_metrics(elem)
df = pd.DataFrame(
{
"Date": [localdate("Asia/Baku")],
"Cumulative total": [count],
}
)
return df
[docs] def _parse_metrics(self, elem: element.Tag) -> int:
"""Parse metrics from element"""
count = clean_count(elem.find_previous_sibling("strong").text)
return count
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame:
"""Pipeline for data processing"""
return (
df.pipe(self.pipe_metadata)
.pipe(self.pipe_merge_current)
.drop_duplicates(subset=["Cumulative total"], keep="first")
)
[docs] def export(self):
"""Export data to csv"""
df = self.read().pipe(self.pipeline)
self.export_datafile(df)