Source code for cowidev.testing.batch.united_states
import pandas as pd
from cowidev.testing import CountryTestBase
from cowidev.utils import clean_date_series
[docs]class UnitedStates(CountryTestBase):
location = "United States"
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", "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"]
.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.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"].rolling(7).mean())
).round(3)
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