import datetime
import pandas as pd
from cowidev.testing import CountryTestBase
[docs]class Mexico(CountryTestBase):
location = "Mexico"
units = "people tested"
source_label = "Health Secretary"
source_url = "https://datos.covid-19.conacyt.mx/#DownZCSV"
source_url_ref = source_url
[docs] def url_melt(self, url: str, name: str) -> pd.DataFrame:
df_melt = pd.read_csv(url).melt(id_vars=["cve_ent", "poblacion", "nombre"], var_name="Date", value_name=name)
df_melt = df_melt[df_melt["nombre"] == "Nacional"]
df_melt = df_melt.drop(["cve_ent", "poblacion", "nombre"], axis=1)
df_melt["Date"] = pd.to_datetime(df_melt["Date"], format="%d-%m-%Y")
return df_melt
[docs] def read(self) -> pd.DataFrame:
yesterday = (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y%m%d")
base_url = "https://datos.covid-19.conacyt.mx/Downloads/Files/Casos_Diarios_Estado_Nacional_"
return (
pd.merge(
self.url_melt(f"{base_url}Confirmados_{yesterday}.csv", "positive"),
self.url_melt(f"{base_url}Negativos_{yesterday}.csv", "negative"),
on=["Date"],
how="right",
)
.fillna(0)
.sort_values("Date")
)
[docs] def pipe_daily_change_in_cum_total(self, df: pd.DataFrame) -> pd.DataFrame:
df["Daily change in cumulative total"] = df["positive"] + df["negative"]
df["Daily change in cumulative total"] = pd.to_numeric(
df["Daily change in cumulative total"], downcast="integer"
)
df = df[df["Daily change in cumulative total"] != 0]
return df
[docs] def pipe_positive_rate(self, df: pd.DataFrame) -> pd.DataFrame:
return df.assign(
**{
"Positive rate": (
(df.positive.rolling(7).sum() / df["Daily change in cumulative total"].rolling(7).sum()).round(3)
)
}
)
# df["Positive rate"] = (
# df["positive"].rolling(7).sum() / df["Daily change in cumulative total"].rolling(7).sum()
# ).round(3)
# df.loc[:, "Positive rate"] = df["Positive rate"].fillna(0)
# return df
[docs] def pipe_filter_rows(self, df: pd.DataFrame) -> pd.DataFrame:
return df[["Date", "Daily change in cumulative total", "Positive rate"]]
[docs] def pipeline(self, df: pd.DataFrame) -> pd.DataFrame:
return (
df.pipe(self.pipe_daily_change_in_cum_total)
.pipe(self.pipe_positive_rate)
.pipe(self.pipe_filter_rows)
.pipe(self.pipe_metadata)
)
[docs] def export(self):
df = self.read().pipe(self.pipeline)
self.export_datafile(df)
[docs]def main():
Mexico().export()