Un script qu'un autre chargé peut exécuter — Python pour les données de programme — Analyse de données

Le test de la transmission

Une analyse est terminée quand quelqu’un d’autre peut l’exécuter sur l’export du trimestre prochain sans modifier le code. Non pas « sans trop le modifier » — sans le modifier.

Ce critère élimine les trois habitudes que tout notebook accumule : un chemin écrit pour une machine, une valeur changée à la main entre deux exécutions, et une étape qui ne fonctionne que si l’on sait déjà quelle cellule lancer d’abord.

Des arguments, non des modifications

# scripts/indicator_table.py
import argparse
from pathlib import Path

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Calcule les taux de MAG et MAS par commune depuis un registre."
    )
    parser.add_argument(
        "--input", type=Path, required=True,
        help="Chemin du CSV du registre de depistage.",
    )
    parser.add_argument(
        "--output", type=Path, required=True,
        help="Repertoire ou ecrire les tableaux. Cree s'il manque.",
    )
    parser.add_argument(
        "--gam-threshold", type=int, default=125,
        help="Seuil de PB en mm pour la malnutrition aigue globale (defaut : 125).",
    )
    parser.add_argument(
        "--min-assessment-rate", type=float, default=0.8,
        help="Les communes sous ce taux d'evaluation sont signalees, non retirees.",
    )
    return parser.parse_args()

uv run python scripts/indicator_table.py \
    --input data/raw/muac-screening-artibonite-2024.v1.csv \
    --output outputs/tables

type=Path convertit la chaîne pour vous, et required=True fait qu’un argument oublié produit un message clair plutôt qu’un KeyError deux cents lignes plus loin.

Mettez les seuils dans les arguments et les valeurs par défaut dans le code. Un seuil qu’on ne change qu’en éditant une ligne finira par être changé et non remis. Un seuil qui est un argument avec un défaut documenté est visible dans la commande qui a produit la sortie — ce que vous voulez quand on vous demandera comment le chiffre du trimestre dernier a été calculé.

--help est généré depuis les mêmes déclarations, et c’est la documentation que les gens lisent réellement :

uv run python scripts/indicator_table.py --help

Échouer bruyamment, tôt

Un script qui produit un chiffre faux en silence est pire qu’un script qui plante. Vérifiez ce que vous supposez, à l’endroit où vous le supposez.

def read_register(path: Path) -> pd.DataFrame:
    if not path.exists():
        raise SystemExit(f"Fichier d'entree introuvable : {path}")

    muac = pd.read_csv(
        path,
        dtype={"child_id": "string", "commune": "string"},
        na_values={"muac_mm": ["-99"]},
    )

    expected = {"child_id", "commune", "screening_date", "muac_mm", "oedema"}
    missing = expected - set(muac.columns)
    if missing:
        raise SystemExit(f"Colonnes absentes de l'entree : {sorted(missing)}")

    if muac["child_id"].isna().any():
        raise SystemExit("Des lignes n'ont pas de child_id ; arret.")

    return muac

SystemExit avec un message sort en statut non nul et affiche une ligne lisible, plutôt qu’une trace d’appels que le lecteur doit interpréter. Employez-le pour « cette entrée n’est pas celle qu’on m’avait promise » ; employez assert pour « ceci ne peut pas arriver à moins que le code soit faux ».

Transformez en erreur l’avertissement pandas qui signale une défaillance silencieuse :

import warnings
import pandas as pd

warnings.simplefilter("error", pd.errors.ChainedAssignmentError)

L’affectation chaînée ne modifie jamais le tableau, et par défaut elle se contente d’avertir. Dans un script dont quelqu’un rapportera la sortie, s’arrêter est la bonne réponse.

Journaliser, non imprimer

print part sur la sortie standard au milieu des résultats, n’a pas de niveau de gravité et ne peut pas être atténué.

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)-8s %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)

log.info("Lu %d lignes depuis %s", len(muac), args.input)
log.warning("%d communes sous le plancher de taux d'evaluation", len(flagged))
log.info("Ecrit %s", output_path)

09:14:02 INFO     Lu 4218 lignes depuis data/raw/muac-...v1.csv
09:14:02 WARNING  2 communes sous le plancher de taux d'evaluation
09:14:02 INFO     Ecrit outputs/tables/gam_by_commune.csv

Trois choses que print n’apporte pas : un horodatage, un niveau de gravité filtrable, et la possibilité d’ajouter --verbose sans toucher à chaque appel.

Utilisez la forme %s plutôt qu’une f-string — le formatage est entièrement évité quand le niveau est désactivé.

Journalisez les nombres qui permettraient de reconstituer l’exécution : lignes lues, lignes exclues et pourquoi, dénominateur, chemin de sortie. Un journal qui dit « terminé » n’apprend rien à personne.

La forme du script

# scripts/indicator_table.py
"""Calcule les taux de MAG et MAS par commune depuis un registre de PB."""

import argparse
import logging
from pathlib import Path

import pandas as pd

log = logging.getLogger(__name__)

GAM_DEFAULT, SAM_DEFAULT = 125, 115


def read_register(path: Path) -> pd.DataFrame:
    ...


def indicator_table(muac: pd.DataFrame, gam_mm: int, sam_mm: int) -> pd.DataFrame:
    measured = muac["muac_mm"].notna() | muac["oedema"].notna()
    gam = measured & ((muac["muac_mm"] < gam_mm) | (muac["oedema"] == True))
    sam = measured & ((muac["muac_mm"] < sam_mm) | (muac["oedema"] == True))

    table = (
        muac.assign(measured=measured, gam=gam, sam=sam)
        .groupby("commune", dropna=False, observed=True)
        .agg(
            screened=("child_id", "size"),
            assessed=("measured", "sum"),
            gam_cases=("gam", "sum"),
            sam_cases=("sam", "sum"),
        )
    )
    table["assessment_rate"] = (table["assessed"] / table["screened"]).round(3)
    table["gam_rate"] = (table["gam_cases"] / table["assessed"]).round(3)
    table["sam_rate"] = (table["sam_cases"] / table["assessed"]).round(3)
    return table.reset_index()


def main() -> None:
    args = parse_args()
    logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(message)s")

    muac = read_register(args.input)
    log.info("Lu %d lignes", len(muac))

    table = indicator_table(muac, args.gam_threshold, SAM_DEFAULT)

    thin = table.loc[table["assessment_rate"] < args.min_assessment_rate, "commune"]
    if len(thin):
        log.warning("Taux d'evaluation sous le plancher : %s", ", ".join(thin))

    args.output.mkdir(parents=True, exist_ok=True)
    destination = args.output / "gam_by_commune.csv"
    table.to_csv(destination, index=False)
    log.info("Ecrit %s (%d communes)", destination, len(table))


if __name__ == "__main__":
    main()

Quatre aspects de cette structure sont délibérés.

Les fonctions prennent des arguments et renvoient des valeurs. indicator_table ne lit aucun fichier, ignore où va la sortie et ne touche à aucune variable globale. C’est ce qui la rend testable et ce qui permet au notebook de l’importer.

main() est le seul endroit qui fait des entrées-sorties. Lecture, écriture et journalisation y vivent ; tout le reste est du calcul.

Le garde if __name__ == "__main__" fait qu’importer ce module depuis un notebook n’exécute rien. Sans lui, from indicator_table import indicator_table exécute tout le script.

Les données minces sont signalées, non supprimées. Une commune sous le plancher de taux d’évaluation reste dans le tableau avec un avertissement à côté. La supprimer changerait le dénominateur du district et rien dans la sortie ne le dirait.

Tester la partie qui calcule

Une fois le calcul devenu une fonction prenant un tableau, le tester tient en trois lignes :

# tests/test_indicator_table.py
import pandas as pd
from indicator_table import indicator_table


def test_denominator_excludes_unmeasured_children():
    muac = pd.DataFrame({
        "child_id": ["a", "b", "c"],
        "commune": ["X", "X", "X"],
        "muac_mm": [110, 130, None],
        "oedema": [False, False, None],
    })

    table = indicator_table(muac, gam_mm=125, sam_mm=115)

    assert table.loc[0, "screened"] == 3
    assert table.loc[0, "assessed"] == 2      # l'enfant non mesure n'est pas au denominateur
    assert table.loc[0, "gam_rate"] == 0.5

Ce test encode la décision de la leçon 6 — quels enfants entrent au dénominateur — sous une forme qui échoue si quelqu’un la change. Il mérite d’être écrit pour tout indicateur dont la définition a fait débat.

Rendre l’exécution reproductible

Consignez ce qui a produit la sortie, à côté de la sortie :

import json
import subprocess
from datetime import datetime, timezone


def run_metadata(args) -> dict:
    return {
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "input": str(args.input),
        "gam_threshold": args.gam_threshold,
        "pandas": pd.__version__,
        "commit": subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            capture_output=True, text=True,
        ).stdout.strip() or "pas un depot git",
    }


(args.output / "run.json").write_text(json.dumps(run_metadata(args), indent=2))

Six mois plus tard, « quel seuil a produit ce tableau » a une réponse qui ne dépend de la mémoire de personne.

Puis le contrôle de la leçon 2, qui est le véritable test :

rm -rf outputs/
uv run python scripts/indicator_table.py --input data/raw/muac.csv --output outputs/tables
git status

Si cela ne restaure pas les sorties à l’identique, quelque chose n’est pas reproductible.

Transmettre

Le README de la leçon 2 a besoin d’une ligne de plus désormais — la commande avec ses vrais arguments :

## Executer

    uv sync
    uv run python scripts/indicator_table.py \
        --input data/raw/muac-screening-artibonite-2024.v1.csv \
        --output outputs/tables

Les seuils valent par defaut PB < 125 mm (MAG) et < 115 mm (MAS).
Passez --gam-threshold pour changer ; la valeur employee est consignee
dans outputs/tables/run.json.

Deux commandes, et les seuils visibles sans ouvrir le code. Toute la transmission est là.

Où ce cours s’arrête

Vous savez construire un environnement réinstallable hors ligne, lire n’importe quel export sans le corrompre, transformer ses codes en valeurs, agréger vers un numérateur et un dénominateur issus de la même opération, maîtriser l’arithmétique des dates, et remettre le résultat à quelqu’un d’autre sous forme de script plutôt que de service rendu.

Ce que ce cours n’a délibérément pas enseigné, c’est quoi calculer — quel indicateur, quel dénominateur, quel seuil, et comment le défendre. C’est l’objet de Fondamentaux de l’analyse de données, et les cours sectoriels du module Analyses sectorielles de la feuille de route le poussent plus loin, jusqu’aux classifications que votre cluster vous impose.

Si votre équipe travaille en R plutôt qu’en Python, R pour les données de programme couvre le même terrain dans ce langage.