Source code for cowidev.testing.batch.trinidad_and_tobago

import pandas as pd

from cowidev.utils.web import request_json
from cowidev.utils.clean import clean_date_series
from cowidev.testing.utils import make_monotonic
from cowidev.testing import CountryTestBase


[docs]class TrinidadAndTobago(CountryTestBase): location: str = "Trinidad and Tobago" units: str = "people tested" source_label: str = "Ministry of Health" source_url: str = "https://services3.arcgis.com/x3I4DqUw3b3MfTwQ/arcgis/rest/services/service_7a519502598f492a9094fd0ad503cf80/FeatureServer/0/query?f=json&resultOffset=0&resultRecordCount=32000&where=report_dt%20IS%20NOT%20NULL%20AND%20report_dt%20BETWEEN%20timestamp%20%272020-12-20%2021%3A00%3A00%27%20AND%20CURRENT_TIMESTAMP&orderByFields=report_dt%20asc&outFields=unique_public_private_test,CreationDate,positive_tests&resultType=standard&returnGeometry=false&spatialRel=esriSpatialRelIntersects" source_url_ref: str = "https://www.covid19.gov.tt/" rename_columns: dict = { "attributes.unique_public_private_test": "Cumulative total", "attributes.CreationDate": "Date", "attributes.positive_tests": "positive", }
[docs] def read(self) -> pd.DataFrame: """Reads data from source.""" data = request_json(self.source_url) df = pd.json_normalize(data, record_path=["features"]).dropna(subset=["attributes.unique_public_private_test"]) return df.drop_duplicates(subset="attributes.unique_public_private_test")
[docs] def pipe_date(self, df: pd.DataFrame) -> pd.DataFrame: """Cleans date column""" return df.assign(Date=clean_date_series(df["Date"], unit="ms"))
[docs] def pipe_pr(self, df: pd.DataFrame) -> pd.DataFrame: """Calculate Positive Rate""" df["Positive rate"] = ( df.positive.diff().rolling(7).sum().div(df["Cumulative total"].diff().rolling(7).sum()).round(3) ).fillna(0) df = df[df["Date"] != "2021-03-27"] return df.drop_duplicates(subset="Date")
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame: """pipeline for data""" return ( df.pipe(self.pipe_rename_columns) .pipe(self.pipe_date) .pipe(self.pipe_pr) .pipe(self.pipe_metadata) .pipe(make_monotonic) .sort_values("Date") )
[docs] def export(self): """Exports data to csv""" df = self.read().pipe(self.pipeline) self.export_datafile(df)
[docs]def main(): TrinidadAndTobago().export()