donnees_meteo/docs/03 - Premiers graphiques/scripts/plot_calendar_overview.py

# docs/03 - Premiers graphiques/scripts/plot_calendar_overview.py
from __future__ import annotations

from pathlib import Path
import sys

import calendar
import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[3]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from meteo.dataset import load_raw_csv
from meteo.analysis import compute_daily_rainfall_totals
from meteo.plots import export_plot_dataset, plot_calendar_heatmap, plot_weekday_profiles
from meteo.variables import VARIABLES_BY_KEY


DOC_DIR = Path(__file__).resolve().parent.parent
CSV_PATH = PROJECT_ROOT / "data" / "weather_minutely.csv"
OUTPUT_DIR = DOC_DIR / "figures" / "calendar"

WEEKDAY_VARIABLE_KEYS = ["temperature", "humidity", "wind_speed", "illuminance"]


def _format_calendar_matrix(series: pd.Series, year: int, agg_label: str) -> pd.DataFrame:
    """
    Transforme une série quotidienne en matrice mois x jours (1-31).
    """
    start = pd.Timestamp(year=year, month=1, day=1, tz=series.index.tz)
    end = pd.Timestamp(year=year, month=12, day=31, tz=series.index.tz)
    filtered = series.loc[(series.index >= start) & (series.index <= end)]

    matrix = pd.DataFrame(
        np.nan,
        index=[calendar.month_name[m][:3] for m in range(1, 13)],
        columns=list(range(1, 32)),
    )

    for timestamp, value in filtered.items():
        month = timestamp.month
        day = timestamp.day
        matrix.at[calendar.month_name[month][:3], day] = value

    matrix.index.name = f"{agg_label} ({year})"
    return matrix


def compute_daily_mean(df: pd.DataFrame, column: str) -> pd.Series:
    return df[column].resample("1D").mean()


def main() -> None:
    if not CSV_PATH.exists():
        print(f"⚠ Fichier introuvable : {CSV_PATH}")
        return

    df = load_raw_csv(CSV_PATH)
    if df.empty:
        print("⚠ Dataset vide.")
        return

    if not isinstance(df.index, pd.DatetimeIndex):
        print("⚠ Le dataset doit avoir un index temporel.")
        return

    print(f"Dataset minuté chargé : {CSV_PATH}")
    print(f"  Lignes   : {len(df)}")
    print(f"  Colonnes : {list(df.columns)}")
    print()

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    latest_year = df.index.year.max()
    print(f"Année retenue pour le calendrier : {latest_year}")

    daily_totals = compute_daily_rainfall_totals(df=df)
    daily_rain = daily_totals["daily_total"]
    rain_matrix = _format_calendar_matrix(daily_rain, latest_year, "Pluie (mm)")
    rain_matrix.attrs["cmap"] = "Blues"
    rain_matrix.attrs["colorbar_label"] = "mm"
    rain_path = OUTPUT_DIR / f"calendar_rain_{latest_year}.png"
    plot_calendar_heatmap(
        matrix=rain_matrix,
        output_path=rain_path,
        title=f"Pluie quotidienne - {latest_year}",
        cmap="Blues",
        colorbar_label="mm",
    )
    print(f"✔ Heatmap pluie {latest_year} : {rain_path}")

    daily_temp = compute_daily_mean(df, "temperature")
    temp_matrix = _format_calendar_matrix(daily_temp, latest_year, "Température (°C)")
    temp_matrix.attrs["cmap"] = "coolwarm"
    temp_matrix.attrs["colorbar_label"] = "°C"
    temp_path = OUTPUT_DIR / f"calendar_temperature_{latest_year}.png"
    plot_calendar_heatmap(
        matrix=temp_matrix,
        output_path=temp_path,
        title=f"Température moyenne quotidienne - {latest_year}",
        cmap="coolwarm",
        colorbar_label="°C",
    )
    print(f"✔ Heatmap température {latest_year} : {temp_path}")

    if "pressure" in df.columns:
        daily_pressure = compute_daily_mean(df, "pressure")
        pressure_matrix = _format_calendar_matrix(daily_pressure, latest_year, "Pression (hPa)")
        pressure_matrix.attrs["cmap"] = "Greens"
        pressure_matrix.attrs["colorbar_label"] = "hPa"
        pressure_path = OUTPUT_DIR / f"calendar_pressure_{latest_year}.png"
        plot_calendar_heatmap(
            matrix=pressure_matrix,
            output_path=pressure_path,
            title=f"Pression moyenne quotidienne - {latest_year}",
            cmap="Greens",
            colorbar_label="hPa",
        )
        print(f"✔ Heatmap pression {latest_year} : {pressure_path}")

    if "illuminance" in df.columns:
        daily_lux = compute_daily_mean(df, "illuminance")
        lux_matrix = _format_calendar_matrix(daily_lux, latest_year, "Illuminance (lux)")
        lux_matrix.attrs["cmap"] = "YlOrBr"
        lux_matrix.attrs["colorbar_label"] = "lux"
        lux_path = OUTPUT_DIR / f"calendar_illuminance_{latest_year}.png"
        plot_calendar_heatmap(
            matrix=lux_matrix,
            output_path=lux_path,
            title=f"Illuminance moyenne quotidienne - {latest_year}",
            cmap="YlOrBr",
            colorbar_label="lux",
        )
        print(f"✔ Heatmap illuminance {latest_year} : {lux_path}")

    if "wind_speed" in df.columns:
        daily_wind = compute_daily_mean(df, "wind_speed")
        wind_matrix = _format_calendar_matrix(daily_wind, latest_year, "Vent (km/h)")
        wind_matrix.attrs["cmap"] = "Purples"
        wind_matrix.attrs["colorbar_label"] = "km/h"
        wind_path = OUTPUT_DIR / f"calendar_wind_{latest_year}.png"
        plot_calendar_heatmap(
            matrix=wind_matrix,
            output_path=wind_path,
            title=f"Vitesse moyenne du vent - {latest_year}",
            cmap="Purples",
            colorbar_label="km/h",
        )
        print(f"✔ Heatmap vent {latest_year} : {wind_path}")

    hourly = df[WEEKDAY_VARIABLE_KEYS].resample("1h").mean()
    weekday_stats = hourly.groupby(hourly.index.dayofweek).mean()
    weekday_path = OUTPUT_DIR / "weekday_profiles.png"
    variables = [VARIABLES_BY_KEY[key] for key in WEEKDAY_VARIABLE_KEYS]
    plot_weekday_profiles(
        weekday_df=weekday_stats,
        variables=variables,
        output_path=weekday_path,
        title="Profils moyens par jour de semaine",
    )
    print(f"✔ Profils hebdomadaires : {weekday_path}")

    print("✔ Graphiques calendrier générés.")


if __name__ == "__main__":
    main()