Source code for cowidev.testing.batch.netherlands
import pandas as pd
from cowidev.utils.web import request_json
from cowidev.testing import CountryTestBase
[docs]class Netherlands(CountryTestBase):
location = "Netherlands"
units = "tests performed"
source_label = "Dutch National Institute for Public Health and the Environment"
source_url = "https://data.rivm.nl/covid-19/COVID-19_uitgevoerde_testen.json"
source_url_ref = "https://data.rivm.nl/covid-19/"
notes = pd.NA
rename_columns = {"Date_of_statistics": "Date", "Tested_with_result": "Daily change in cumulative total"}
[docs] def read(self) -> pd.DataFrame:
data = request_json(self.source_url)
df = pd.DataFrame.from_records(
data, columns=["Date_of_statistics", "Tested_with_result", "Security_region_name", "Tested_positive"]
)
df = df.groupby("Date_of_statistics").sum().reset_index()
return df
[docs] def pipe_positive_rate(self, df: pd.DataFrame):
df["Positive rate"] = (
(df["Tested_positive"].rolling(7).mean()).div((df["Tested_with_result"].rolling(7).mean())).round(3)
)
return df
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame:
return df.pipe(self.pipe_positive_rate).pipe(self.pipe_rename_columns).pipe(self.pipe_metadata)