"""Tracés simples et réutilisables pour les séries temporelles de base.""" from __future__ import annotations from dataclasses import dataclass from enum import Enum from pathlib import Path from typing import Callable import matplotlib.dates as mdates import matplotlib.pyplot as plt import numpy as np import pandas as pd from meteo.dataset import _circular_mean_deg from meteo.variables import Variable from .base import export_plot_dataset __all__ = [ "PlotStyle", "PlotChoice", "recommended_style", "resample_series_for_plot", "plot_basic_series", "plot_dual_time_series", ] class PlotStyle(str, Enum): LINE = "line" AREA = "area" BAR = "bar" SCATTER = "scatter" @dataclass(frozen=True) class PlotChoice: """Configuration par variable : style et fonction d'agrégation.""" style: PlotStyle agg: Callable[[pd.Series], float] | str = "mean" DEFAULT_CHOICES: dict[str, PlotChoice] = { # Variations continues : lignes ou aires. "temperature": PlotChoice(PlotStyle.LINE, "mean"), "pressure": PlotChoice(PlotStyle.LINE, "mean"), "humidity": PlotChoice(PlotStyle.AREA, "mean"), "illuminance": PlotChoice(PlotStyle.AREA, "mean"), "sun_elevation": PlotChoice(PlotStyle.AREA, "mean"), # Variables dont la perception bénéficie d'autres représentations. "rain_rate": PlotChoice(PlotStyle.BAR, "mean"), "wind_speed": PlotChoice(PlotStyle.LINE, "mean"), "wind_direction": PlotChoice(PlotStyle.SCATTER, _circular_mean_deg), } # Palette douce mais contrastée, associée aux variables. PALETTE = { "temperature": "#d1495b", "pressure": "#5c677d", "humidity": "#2c7bb6", "rain_rate": "#1b9e77", "illuminance": "#f4a259", "wind_speed": "#118ab2", "wind_direction": "#8e6c8a", "sun_elevation": "#f08c42", } DEFAULT_COLOR = "#386cb0" def recommended_style(variable: Variable, override: str | None = None) -> PlotChoice: """Retourne le style/agrégation par défaut, ou une surcharge utilisateur.""" if override: style = PlotStyle(override) agg = DEFAULT_CHOICES.get(variable.key, PlotChoice(style)).agg return PlotChoice(style, agg) return DEFAULT_CHOICES.get(variable.key, PlotChoice(PlotStyle.LINE)) def _nice_frequencies() -> list[tuple[str, pd.Timedelta]]: return [ ("5min", pd.Timedelta(minutes=5)), ("10min", pd.Timedelta(minutes=10)), ("15min", pd.Timedelta(minutes=15)), ("30min", pd.Timedelta(minutes=30)), ("1h", pd.Timedelta(hours=1)), ("3h", pd.Timedelta(hours=3)), ("6h", pd.Timedelta(hours=6)), ("12h", pd.Timedelta(hours=12)), ("1d", pd.Timedelta(days=1)), ("3d", pd.Timedelta(days=3)), ("7d", pd.Timedelta(days=7)), ] def _auto_resample_frequency(index: pd.DatetimeIndex, *, target_points: int = 420) -> str: """Choisit une fréquence qui limite le nombre de points tout en conservant la forme générale.""" if index.empty or len(index) < 2: return "1h" span = index.max() - index.min() if span <= pd.Timedelta(0): return "1h" for label, delta in _nice_frequencies(): if span / delta <= target_points: return label return _nice_frequencies()[-1][0] def _format_time_axis(ax: plt.Axes) -> None: locator = mdates.AutoDateLocator(minticks=4, maxticks=8) formatter = mdates.ConciseDateFormatter(locator, formats=["%Y", "%b", "%d", "%d %H:%M", "%H:%M", "%S"]) ax.xaxis.set_major_locator(locator) ax.xaxis.set_major_formatter(formatter) def _infer_bar_width(index: pd.DatetimeIndex) -> float: """ Calcule une largeur de barre raisonnable (en jours) pour les histogrammes temporels. """ if len(index) < 2: return 0.3 # ~7 heures, pour rendre le point visible même isolé diffs = np.diff(index.asi8) # nanosecondes median_ns = float(np.median(diffs)) if not np.isfinite(median_ns) or median_ns <= 0: return 0.1 return pd.to_timedelta(median_ns, unit="ns") / pd.Timedelta(days=1) * 0.8 def _ensure_datetime_index(series: pd.Series) -> pd.Series: if not isinstance(series.index, pd.DatetimeIndex): raise TypeError("Une série temporelle (DatetimeIndex) est attendue pour le tracé.") return series def _series_color(variable: Variable) -> str: if variable.key in PALETTE: return PALETTE[variable.key] return PALETTE.get(variable.column, DEFAULT_COLOR) def _format_label(var: Variable) -> str: unit_text = f" ({var.unit})" if var.unit else "" return f"{var.label}{unit_text}" def resample_series_for_plot( series: pd.Series, *, variable: Variable, freq: str | None = None, target_points: int = 420, ) -> tuple[pd.Series, str]: """ Prépare une série pour l'affichage : resample et agrégation adaptés à la variable. """ _ensure_datetime_index(series) if freq is None: freq = _auto_resample_frequency(series.index, target_points=target_points) agg_func = DEFAULT_CHOICES.get(variable.key, PlotChoice(PlotStyle.LINE)).agg resampled = series.resample(freq).agg(agg_func).dropna() return resampled, freq def plot_basic_series( series: pd.Series, *, variable: Variable, output_path: str | Path, style: PlotStyle, title: str, ylabel: str, annotate_freq: str | None = None, ) -> Path: """ Trace une série temporelle avec un style simple (ligne, aire, barres, nuage de points). """ _ensure_datetime_index(series) if series.empty: raise ValueError(f"Aucune donnée disponible pour {variable.key} après filtrage.") output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) color = _series_color(variable) x = mdates.date2num(series.index) values = series.to_numpy(dtype=float) fig, ax = plt.subplots(figsize=(11, 4.2)) if style is PlotStyle.LINE: ax.plot_date(x, values, "-", linewidth=1.8, color=color, label=variable.label) elif style is PlotStyle.AREA: ax.fill_between(x, values, step="mid", color=color, alpha=0.2) ax.plot_date(x, values, "-", linewidth=1.6, color=color) elif style is PlotStyle.BAR: width = _infer_bar_width(series.index) ax.bar(x, values, width=width, color=color, edgecolor=color, linewidth=0.5, alpha=0.85) elif style is PlotStyle.SCATTER: ax.scatter(x, values, s=16, color=color, alpha=0.9) else: raise ValueError(f"Style inconnu : {style}") ax.set_title(title) ax.set_ylabel(ylabel) _format_time_axis(ax) ax.grid(True, color="#e0e0e0", linewidth=0.8, alpha=0.7) ax.margins(x=0.02, y=0.05) if annotate_freq: ax.text( 0.99, 0.02, f"Agrégation : {annotate_freq}", transform=ax.transAxes, ha="right", va="bottom", fontsize=9, color="#555555", ) fig.tight_layout() fig.savefig(output_path, dpi=150) plt.close(fig) export_plot_dataset(series.to_frame(name=variable.column), output_path) return output_path.resolve() def _draw_series(ax: plt.Axes, series: pd.Series, *, choice: PlotChoice, color: str, label: str): x = mdates.date2num(series.index) values = series.to_numpy(dtype=float) if choice.style is PlotStyle.LINE: return ax.plot_date(x, values, "-", linewidth=1.8, color=color, label=label) if choice.style is PlotStyle.AREA: ax.fill_between(x, values, step="mid", color=color, alpha=0.15) return ax.plot_date(x, values, "-", linewidth=1.6, color=color, label=label) if choice.style is PlotStyle.BAR: width = _infer_bar_width(series.index) * 0.9 return ax.bar(x, values, width=width, color=color, edgecolor=color, linewidth=0.5, alpha=0.75, label=label) if choice.style is PlotStyle.SCATTER: return ax.scatter(x, values, s=16, color=color, alpha=0.9, label=label) raise ValueError(f"Style inconnu : {choice.style}") def plot_dual_time_series( series_left: pd.Series, variable_left: Variable, choice_left: PlotChoice, series_right: pd.Series, variable_right: Variable, choice_right: PlotChoice, *, output_path: str | Path, title: str, annotate_freq: str | None = None, ) -> Path: """Superpose deux séries temporelles (axes Y séparés) avec styles adaptés.""" _ensure_datetime_index(series_left) _ensure_datetime_index(series_right) if series_left.empty or series_right.empty: raise ValueError("Les séries à tracer ne peuvent pas être vides.") output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) color_left = _series_color(variable_left) color_right = _series_color(variable_right) fig, ax_left = plt.subplots(figsize=(11, 4.6)) ax_right = ax_left.twinx() artists_left = _draw_series( ax_left, series_left, choice=choice_left, color=color_left, label=_format_label(variable_left), ) artists_right = _draw_series( ax_right, series_right, choice=choice_right, color=color_right, label=_format_label(variable_right), ) ax_left.set_ylabel(_format_label(variable_left), color=color_left) ax_right.set_ylabel(_format_label(variable_right), color=color_right) ax_left.tick_params(axis="y", labelcolor=color_left) ax_right.tick_params(axis="y", labelcolor=color_right) _format_time_axis(ax_left) ax_left.grid(True, color="#e0e0e0", linewidth=0.8, alpha=0.7) ax_left.margins(x=0.02, y=0.05) ax_right.margins(x=0.02, y=0.05) ax_left.set_title(title) handles = [] labels = [] for artist in artists_left if isinstance(artists_left, list) else [artists_left]: handles.append(artist) labels.append(artist.get_label()) if isinstance(artists_right, list): handles.extend(artists_right) labels.extend([a.get_label() for a in artists_right]) else: handles.append(artists_right) labels.append(artists_right.get_label()) ax_left.legend(handles, labels, loc="upper left") if annotate_freq: ax_left.text( 0.99, 0.02, f"Agrégation : {annotate_freq}", transform=ax_left.transAxes, ha="right", va="bottom", fontsize=9, color="#555555", ) fig.tight_layout() fig.savefig(output_path, dpi=150) plt.close(fig) export_plot_dataset( pd.concat( {variable_left.column: series_left, variable_right.column: series_right}, axis=1, ), output_path, ) return output_path.resolve()