Source code for cowidev.testing.batch.northern_mariana_islands

import pandas as pd

from cowidev.testing import CountryTestBase
from cowidev.utils import clean_date_series


[docs]class NorthernMarianaIslands(CountryTestBase): location = "Northern Mariana Islands" source_url = "https://healthdata.gov/api/views/j8mb-icvb/rows.csv" source_url_ref = "https://healthdata.gov/dataset/COVID-19-Diagnostic-Laboratory-Testing-PCR-Testing/j8mb-icvb" source_label = "Department of Health & Human Services" units = "tests performed" rename_columns = {"date": "Date"}
[docs] def read(self): df = pd.read_csv( self.source_url, usecols=["date", "state_name", "new_results_reported", "overall_outcome"], parse_dates=["date"], ) return df
[docs] def pipe_metrics(self, df: pd.DataFrame) -> pd.DataFrame: # daily change in positive tests pos = ( df[(df["overall_outcome"] == "Positive") & (df["state_name"] == "Northern Mariana Islands")] .groupby("Date", as_index=False) .agg(**{"Daily change in positive total": ("new_results_reported", sum)}) ).sort_values("Date") # daily change in total tests df = ( df[df["state_name"] == "Northern Mariana Islands"] .groupby(["Date"], as_index=False) .agg(**{"Daily change in cumulative total": ("new_results_reported", sum)}) ).sort_values("Date") # generate positive rate df["Positive rate"] = ( ( (pos["Daily change in positive total"].rolling(7).mean()) / (df["Daily change in cumulative total"].shift(-4).rolling(7).mean()) ) .round(3) .shift(4) ) df = df[df["Daily change in cumulative total"] != 0] return df
[docs] def pipe_date(self, df: pd.DataFrame) -> pd.DataFrame: return df.assign(Date=clean_date_series(df.Date, "%Y-%m-%d"))
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame: df = df.pipe(self.pipe_rename_columns).pipe(self.pipe_metrics).pipe(self.pipe_date).pipe(self.pipe_metadata) return df
[docs] def export(self): df = self.read().pipe(self.pipeline) self.export_datafile(df)
[docs]def main(): NorthernMarianaIslands().export()