Premières analyses de corrélation

meteo/analysis.py

@@ -0,0 +1,117 @@
+# meteo/analysis.py
+from __future__ import annotations
+
+from typing import Literal
+
+import numpy as np
+import pandas as pd
+
+from .variables import Variable
+
+
+def compute_correlation_matrix(
+    df: pd.DataFrame,
+    *,
+    method: Literal["pearson", "spearman"] = "pearson",
+) -> pd.DataFrame:
+    """
+    Calcule la matrice de corrélation entre toutes les colonnes numériques
+    du DataFrame.
+
+    Attention :
+    - La direction du vent est traitée ici comme une variable scalaire 0–360°,
+      ce qui n'est pas idéal pour une analyse circulaire. On affinera plus tard
+      si besoin (représentation en sin/cos).
+    """
+    numeric_df = df.select_dtypes(include=["number"])
+    corr = numeric_df.corr(method=method)
+    return corr
+
+
+def compute_correlation_matrix_for_variables(
+    df: pd.DataFrame,
+    variables: Sequence[Variable],
+    *,
+    method: Literal["pearson", "spearman"] = "pearson",
+) -> pd.DataFrame:
+    """
+    Calcule la matrice de corrélation pour un sous-ensemble de variables,
+    dans un ordre bien défini.
+
+    Paramètres
+    ----------
+    df :
+        DataFrame contenant les colonnes à analyser.
+    variables :
+        Séquence de Variable décrivant les colonnes à prendre en compte.
+    method :
+        Méthode de corrélation pandas (pearson, spearman, ...).
+
+    Retour
+    ------
+    DataFrame :
+        Matrice de corrélation, index et colonnes dans le même ordre que
+        `variables`, avec les colonnes pandas correspondant aux noms de colonnes
+        du DataFrame (ex: "temperature", "humidity", ...).
+    """
+    columns = [v.column for v in variables]
+    missing = [c for c in columns if c not in df.columns]
+    if missing:
+        raise KeyError(f"Colonnes manquantes dans le DataFrame : {missing!r}")
+
+    numeric_df = df[columns].astype(float)
+    corr = numeric_df.corr(method=method)
+
+    # On s'assure de l'ordre
+    corr = corr.loc[columns, columns]
+    return corr
+
+
+def compute_lagged_correlation(
+    df: pd.DataFrame,
+    var_x: Variable,
+    var_y: Variable,
+    *,
+    max_lag_minutes: int = 360,
+    step_minutes: int = 10,
+    method: Literal["pearson", "spearman"] = "pearson",
+) -> pd.DataFrame:
+    """
+    Calcule la corrélation entre deux variables pour une série de décalages
+    temporels (lags).
+
+    Convention :
+    - lag > 0 : X "précède" Y de `lag` minutes.
+      On corrèle X(t) avec Y(t + lag).
+    - lag < 0 : Y "précède" X de |lag| minutes.
+      On corrèle X(t) avec Y(t + lag), lag étant négatif.
+
+    Implémentation :
+    - On utilise un DataFrame avec les deux colonnes,
+      puis on applique un `shift` sur Y.
+    """
+    if var_x.column not in df.columns or var_y.column not in df.columns:
+        raise KeyError("Les colonnes demandées ne sont pas présentes dans le DataFrame.")
+
+    series_x = df[var_x.column]
+    series_y = df[var_y.column]
+
+    lags = range(-max_lag_minutes, max_lag_minutes + 1, step_minutes)
+    results: list[tuple[int, float]] = []
+
+    for lag in lags:
+        # Y décalé de -lag : pour lag positif, on corrèle X(t) à Y(t + lag)
+        shifted_y = series_y.shift(-lag)
+        pair = pd.concat([series_x, shifted_y], axis=1).dropna()
+
+        if pair.empty:
+            corr = np.nan
+        else:
+            corr = pair.iloc[:, 0].corr(pair.iloc[:, 1], method=method)
+
+        results.append((lag, corr))
+
+    lag_df = pd.DataFrame(results, columns=["lag_minutes", "correlation"])
+    lag_df = lag_df.set_index("lag_minutes")
+
+    return lag_df

meteo/plots.py

@@ -0,0 +1,149 @@
+# meteo/plots.py
+from __future__ import annotations
+
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+
+from .variables import Variable
+
+
+def plot_scatter_pair(
+    df: pd.DataFrame,
+    var_x: Variable,
+    var_y: Variable,
+    output_path: str | Path,
+    *,
+    sample_step: int = 10,
+) -> Path:
+    """
+    Trace un nuage de points (scatter) pour une paire de variables.
+
+    - On sous-échantillonne les données avec `sample_step` (par exemple,
+      1 point sur 10) pour éviter un graphique illisible.
+    """
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # On ne garde que les colonnes pertinentes et les lignes complètes
+    df_pair = df[[var_x.column, var_y.column]].dropna()
+
+    if sample_step > 1:
+        df_pair = df_pair.iloc[::sample_step, :]
+
+    plt.figure()
+    plt.scatter(df_pair[var_x.column], df_pair[var_y.column], s=5, alpha=0.5)
+    plt.xlabel(f"{var_x.label} ({var_x.unit})")
+    plt.ylabel(f"{var_y.label} ({var_y.unit})")
+    plt.title(f"{var_y.label} en fonction de {var_x.label}")
+    plt.tight_layout()
+    plt.savefig(output_path, dpi=150)
+    plt.close()
+
+    return output_path.resolve()
+
+
+def plot_lagged_correlation(
+    lag_df: pd.DataFrame,
+    var_x: Variable,
+    var_y: Variable,
+    output_path: str | Path,
+) -> Path:
+    """
+    Trace la corrélation en fonction du lag (en minutes) entre deux variables.
+    """
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    plt.figure()
+    plt.plot(lag_df.index, lag_df["correlation"])
+    plt.axvline(0, linestyle="--")  # lag = 0
+    plt.xlabel("Décalage (minutes)\n(lag > 0 : X précède Y)")
+    plt.ylabel("Corrélation")
+    plt.title(f"Corrélation décalée : {var_x.label} → {var_y.label}")
+    plt.grid(True)
+    plt.tight_layout()
+    plt.savefig(output_path, dpi=150)
+    plt.close()
+
+    return output_path.resolve()
+
+
+
+def plot_correlation_heatmap(
+    corr: pd.DataFrame,
+    variables: Sequence[Variable],
+    output_path: str | Path,
+    *,
+    annotate: bool = True,
+) -> Path:
+    """
+    Trace une heatmap de la matrice de corrélation.
+
+    Paramètres
+    ----------
+    corr :
+        Matrice de corrélation (index et colonnes doivent correspondre
+        aux noms de colonnes des variables).
+    variables :
+        Liste de Variable, dans l'ordre où elles doivent apparaître.
+    output_path :
+        Chemin du fichier image à écrire.
+    annotate :
+        Si True, affiche la valeur numérique dans chaque case.
+    """
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    columns = [v.column for v in variables]
+    labels = [v.label for v in variables]
+
+    # On aligne la matrice sur l'ordre désiré
+    corr = corr.loc[columns, columns]
+
+    data = corr.to_numpy()
+
+    fig, ax = plt.subplots()
+    im = ax.imshow(data, vmin=-1.0, vmax=1.0)
+
+    # Ticks et labels
+    ax.set_xticks(np.arange(len(labels)))
+    ax.set_yticks(np.arange(len(labels)))
+    ax.set_xticklabels(labels, rotation=45, ha="right")
+    ax.set_yticklabels(labels)
+
+    # Axe en haut/bas selon préférence (ici on laisse en bas)
+    ax.set_title("Matrice de corrélation (coef. de Pearson)")
+
+    # Barre de couleur
+    cbar = plt.colorbar(im, ax=ax)
+    cbar.set_label("Corrélation")
+
+    # Annotation des cases
+    if annotate:
+        n = data.shape[0]
+        for i in range(n):
+            for j in range(n):
+                if i == j:
+                    text = "—"
+                else:
+                    val = data[i, j]
+                    if np.isnan(val):
+                        text = ""
+                    else:
+                        text = f"{val:.2f}"
+                ax.text(
+                    j,
+                    i,
+                    text,
+                    ha="center",
+                    va="center",
+                )
+
+    plt.tight_layout()
+    plt.savefig(output_path, dpi=150)
+    plt.close(fig)
+
+    return output_path.resolve()

meteo/variables.py

@@ -0,0 +1,82 @@
+# meteo/variables.py
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List
+
+
+@dataclass(frozen=True)
+class Variable:
+    """
+    Décrit une variable météorologique exploitable dans les analyses.
+
+    - key : identifiant interne (stable, sans espace)
+    - column : nom de la colonne dans le DataFrame
+    - label : libellé humain pour les graphiques
+    - unit : unité d'affichage (texte libre)
+    """
+
+    key: str
+    column: str
+    label: str
+    unit: str
+
+
+VARIABLES: List[Variable] = [
+    Variable(
+        key="temperature",
+        column="temperature",
+        label="Température",
+        unit="°C",
+    ),
+    Variable(
+        key="humidity",
+        column="humidity",
+        label="Humidité relative",
+        unit="%",
+    ),
+    Variable(
+        key="pressure",
+        column="pressure",
+        label="Pression atmosphérique",
+        unit="hPa",
+    ),
+    Variable(
+        key="rain_rate",
+        column="rain_rate",
+        label="Précipitations",
+        unit="mm/h",
+    ),
+    Variable(
+        key="illuminance",
+        column="illuminance",
+        label="Luminance",
+        unit="lx",
+    ),
+    Variable(
+        key="wind_speed",
+        column="wind_speed",
+        label="Vitesse du vent",
+        unit="km/h",
+    ),
+    Variable(
+        key="wind_direction",
+        column="wind_direction",
+        label="Direction du vent",
+        unit="°",
+    ),
+]
+
+VARIABLES_BY_KEY: Dict[str, Variable] = {v.key: v for v in VARIABLES}
+
+
+def iter_variable_pairs() -> list[tuple[Variable, Variable]]:
+    """
+    Retourne la liste de toutes les paires (i, j) avec i < j, pour
+    éviter les doublons et les diagonales.
+    """
+    pairs: list[tuple[Variable, Variable]] = []
+    for i, vi in enumerate(VARIABLES):
+        for vj in VARIABLES[i + 1 :]:
+            pairs.append((vi, vj))
+    return pairs