diff --git a/module3/exo3/COV19.IPYNB b/module3/exo3/COV19.IPYNB new file mode 100644 index 0000000000000000000000000000000000000000..5df2ded2eba93706f705a0877549b867ef4dc925 --- /dev/null +++ b/module3/exo3/COV19.IPYNB @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Titre +----- +Courbes cumulées COVID-19 (style SCMP) pour une liste de pays. + +Objectif pédagogique +-------------------- +Reproduire des graphes à la manière du South China Morning Post (SCMP) montrant, +pour une sélection de pays, l'évolution du nombre **cumulé** de cas confirmés +depuis le début de la pandémie, avec deux échelles (linéaire et logarithmique), +et des **étiquettes décalées** à droite pour garantir la lisibilité. + +Reproductibilité (à lire avant d'exécuter) +------------------------------------------ +- Code 100% autonome : télécharge les données à l'exécution. +- Aucune étape manuelle. +- Tous les traitements sont explicités et automatisés. +- Les graphiques sont sauvés en PNG dans le dossier courant. + +Données & Sources (liens pérennes) +---------------------------------- +- Confirmed cases (JHU CSSE time series): + https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/csse_covid_19_data/csse_covid_19_time_series/time_series_covid19_confirmed_global.csv + +Environnement logiciel / bibliothèques +-------------------------------------- +- Python >= 3.9 +- pandas, numpy, matplotlib, urllib (stdlib) + Installation: `pip install pandas numpy matplotlib` + +Critères de sélection des pays (conformément à l'énoncé) +-------------------------------------------------------- +- Belgium +- China **sans** Hong-Kong (somme de toutes les provinces de "China" hors HK) +- Hong Kong (China, Hong-Kong) : traité séparément +- France **métropolitaine** uniquement (ligne "France" avec Province/State manquant) +- Germany, Iran, Italy, Japan, Korea, South +- Netherlands (sans colonies : par construction des fichiers JHU) +- Portugal, Spain, United Kingdom (sans colonies), US + +Points méthodologiques +---------------------- +- Les données JHU sont **cumulatives** par date (jour civil). +- On agrège par date au niveau pays (somme des provinces si nécessaire). +- Pour France métropolitaine: on conserve la ligne "France" où Province/State est NaN. +- Pour China excl. HK: on agrège toutes les provinces de "China" sauf celles + dont Province/State contient "Hong". +- Les étiquettes de fin sont décalées à droite, avec un *dodge* vertical minimal + pour éviter les chevauchements; des traits connectent la fin des courbes aux étiquettes. + +Sorties générées +---------------- +- scmp_linear.png (échelle linéaire) +- scmp_log.png (échelle logarithmique) +- panel_cumulative_cases.csv (tableau final pays x dates) + +Auteurs & attribution +--------------------- +- Script préparé pour un rendu Mooc (évaluation par les pairs). +- Merci de citer JHU CSSE pour les données. + +""" + +from __future__ import annotations + +import io +from urllib.request import urlopen +from typing import Dict, Iterable + +import numpy as np +import pandas as pd +import matplotlib.pyplot as plt + + +# ========================= +# 1) Chargement des données +# ========================= + +URL_CONFIRMED = ( + "https://raw.githubusercontent.com/CSSEGISandData/COVID-19/master/" + "csse_covid_19_data/csse_covid_19_time_series/" + "time_series_covid19_confirmed_global.csv" +) + + +def load_jhu_timeseries_csv(url: str) -> pd.DataFrame: + """ + Télécharge et charge en DataFrame la série temporelle JHU (time series, wide). + Retourne le tableau "wide" tel que fourni par JHU (colonnes dates à partir de la 5e). + """ + raw = urlopen(url).read() + df = pd.read_csv(io.BytesIO(raw)) + # Hygiène : s'assurer que les 4 premières colonnes sont celles attendues + expected = ["Province/State", "Country/Region", "Lat", "Long"] + assert list(df.columns[:4]) == expected, "Format JHU inattendu." + return df + + +def to_long(df_wide: pd.DataFrame) -> pd.DataFrame: + """ + Convertit le format wide (colonnes de dates) en format long (date, cum_cases). + """ + date_cols = df_wide.columns[4:] + long = df_wide.melt( + id_vars=["Province/State", "Country/Region", "Lat", "Long"], + value_vars=date_cols, + var_name="date", + value_name="cum_cases", + ) + long["date"] = pd.to_datetime(long["date"], format="%m/%d/%y") + # S'assure que cum_cases est numérique + long["cum_cases"] = pd.to_numeric(long["cum_cases"], errors="coerce").fillna(0) + return long + + +# ======================================== +# 2) Construction des séries par "pays" +# ======================================== + +def get_country_series(long_df: pd.DataFrame, row_mask: pd.Series, label: str) -> pd.Series: + """ + Agrège les lignes sélectionnées (mask) au niveau de la date et renvoie une Series + (index=dates, valeurs=cumul cas), triée par date et nommée `label`. + """ + sub = long_df.loc[row_mask, ["date", "cum_cases"]] + s = sub.groupby("date", as_index=True)["cum_cases"].sum().sort_index() + s.name = label + return s + + +def build_panel(long_df: pd.DataFrame) -> pd.DataFrame: + """ + Construit le DataFrame final 'panel' (colonnes = pays, index = dates) + pour la liste de pays exigée par l'énoncé. + """ + # Détection de Hong-Kong + is_hk = ( + long_df["Province/State"].fillna("").str.contains("Hong", case=False) + | long_df["Country/Region"].str.contains("Hong", case=False, na=False) + ) + + series = [] + + # Sélections + series.append(get_country_series(long_df, long_df["Country/Region"].eq("Belgium"), "Belgium")) + series.append(get_country_series(long_df, long_df["Country/Region"].eq("China") & (~is_hk), + "China (excl. Hong Kong)")) + series.append(get_country_series(long_df, is_hk, "Hong Kong")) + + # France métropolitaine : Province/State manquant (NaN) => approx. métropole + france_metro_mask = long_df["Country/Region"].eq("France") & long_df["Province/State"].isna() + series.append(get_country_series(long_df, france_metro_mask, "France (metropolitan)")) + + # Autres pays (sans colonies implicites dans JHU : UK/NL sont déjà propres) + for country in [ + "Germany", "Iran", "Italy", "Japan", "Korea, South", + "Netherlands", "Portugal", "Spain", "United Kingdom", "US" + ]: + series.append(get_country_series(long_df, long_df["Country/Region"].eq(country), country)) + + panel = pd.concat(series, axis=1) + + # Tri des dates (par sécurité) et forward-fill des manques (rare) + panel = panel.sort_index().ffill() + + return panel + + +# ================================== +# 3) Tracé "style SCMP" (annoté) +# ================================== + +def style_scmp(ax: plt.Axes, title: str, log: bool = False) -> None: + """ + Applique un style épuré (type SCMP) à un axe matplotlib. + - Titre aligné à gauche + - Grille légère + - Axes droit/haut masqués + - Libellés adaptés selon l'échelle + """ + ax.set_title(title, loc="left", fontsize=16, weight="bold") + ax.grid(True, which="major", linewidth=0.6, alpha=0.3) + ax.spines["top"].set_visible(False) + ax.spines["right"].set_visible(False) + ax.set_xlabel("") + ylabel = "Cumulative confirmed cases" + (" (log scale)" if log else "") + ax.set_ylabel(ylabel) + + +def add_end_labels( + ax: plt.Axes, + data: pd.DataFrame, + *, + right_pad_days: int = 28, + min_sep_frac: float = 0.035, + log: bool = False, + text_kwargs: Dict = None, + line_kwargs: Dict = None, +) -> None: + """ + Place des étiquettes à droite, avec un *dodge* vertical minimal pour éviter les chevauchements. + - right_pad_days : décalage horizontal des labels (en jours). + - min_sep_frac : séparation verticale minimale entre étiquettes (en fraction de la hauteur). + - log : s'aligne avec l'échelle actuelle (lin vs log) pour la séparation verticale. + - text_kwargs : style du texte (fontsize, etc.). + - line_kwargs : style des traits de liaison. + """ + text_kwargs = {"fontsize": 9} | (text_kwargs or {}) + line_kwargs = {"linewidth": 1.0, "alpha": 0.6} | (line_kwargs or {}) + + x_last = data.index.max() + x_lab = x_last + pd.Timedelta(days=right_pad_days) + + # Dernière valeur par série + last_vals = {col: data[col].dropna().iloc[-1] for col in data.columns} + + # Empilage du bas vers le haut (ordre croissant des valeurs) + items = sorted(last_vals.items(), key=lambda kv: kv[1]) + + y0, y1 = ax.get_ylim() + to_ax = (np.log10 if log else (lambda y: y)) + from_ax = (lambda z: 10 ** z) if log else (lambda z: z) + span = to_ax(y1) - to_ax(y0) + min_sep = min_sep_frac * span + + ys_ax = [to_ax(v) for _, v in items] + + # Dodge vertical: impose un espacement minimal successif + for i in range(1, len(ys_ax)): + if ys_ax[i] - ys_ax[i - 1] < min_sep: + ys_ax[i] = ys_ax[i - 1] + min_sep + + # Tracer traits + placer étiquettes + for (name, y_end), y_ax in zip(items, ys_ax): + y_target = from_ax(y_ax) + ax.plot([x_last, x_lab], [y_end, y_target], **line_kwargs) + ax.text(x_lab, y_target, f" {name}", va="center", ha="left", **text_kwargs) + + # Laisser de la marge à droite pour les étiquettes + ax.set_xlim(data.index.min(), x_lab + pd.Timedelta(days=14)) + + +def plot_panel(panel: pd.DataFrame) -> None: + """ + Produit les deux graphiques (linéaire et log), applique le style et les labels, + et enregistre les PNG dans le dossier courant. + """ + # ---------- Linéaire ---------- + fig, ax = plt.subplots(figsize=(13, 7)) + panel.plot(ax=ax, lw=2, legend=False) + style_scmp(ax, "Cumulative COVID-19 confirmed cases (selected countries) — linear scale", log=False) + add_end_labels(ax, panel, right_pad_days=28, min_sep_frac=0.03, log=False) + plt.tight_layout() + plt.savefig("scmp_linear.png", dpi=180) + + # ---------- Logarithmique ---------- + fig2, ax2 = plt.subplots(figsize=(13, 7)) + panel.plot(ax=ax2, lw=2, legend=False) + ax2.set_yscale("log") + style_scmp(ax2, "Cumulative COVID-19 confirmed cases (selected countries) — log scale", log=True) + add_end_labels(ax2, panel, right_pad_days=28, min_sep_frac=0.04, log=True) + plt.tight_layout() + plt.savefig("scmp_log.png", dpi=180) + + plt.show() + + +# ========================== +# 4) Point d'entrée (main) +# ========================== + +def main() -> None: + # 1) Charger la table JHU + df_wide = load_jhu_timeseries_csv(URL_CONFIRMED) + + # 2) Passer en format long (date, cum_cases) + long_df = to_long(df_wide) + + # 3) Construire le panel pays x dates selon les règles de sélection + panel = build_panel(long_df) + + # 4) Exporter le tableau pour faciliter la relecture/rejeu + panel.to_csv("panel_cumulative_cases.csv", index_label="date") + + # 5) Tracer et enregistrer les figures + plot_panel(panel) + + # Message final (console) + print("✅ Fini. Fichiers exportés : scmp_linear.png, scmp_log.png, panel_cumulative_cases.csv") + + +if __name__ == "__main__": + main()