56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Structures et helpers communs pour les analyses météorologiques."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
__all__ = ['MONTH_ORDER', 'DiurnalCycleStats', 'BinnedStatistics']
|
|
|
|
MONTH_ORDER = list(range(1, 13))
|
|
|
|
|
|
@dataclass
|
|
class DiurnalCycleStats:
|
|
"""Conteneur pour les statistiques agrégées par heure (moyenne, médiane et quantiles optionnels)."""
|
|
|
|
mean: pd.DataFrame
|
|
median: pd.DataFrame
|
|
quantile_low: pd.DataFrame | None
|
|
quantile_high: pd.DataFrame | None
|
|
quantile_low_level: float | None = None
|
|
quantile_high_level: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class BinnedStatistics:
|
|
"""Structure englobant les résultats calculés sur des intervalles (bins) réguliers ou personnalisés."""
|
|
|
|
centers: np.ndarray
|
|
intervals: pd.IntervalIndex
|
|
counts: pd.Series
|
|
mean: pd.DataFrame
|
|
median: pd.DataFrame
|
|
quantile_low: pd.DataFrame | None
|
|
quantile_high: pd.DataFrame | None
|
|
quantile_low_level: float | None = None
|
|
quantile_high_level: float | None = None
|
|
|
|
def _ensure_datetime_index(df: pd.DataFrame) -> pd.DatetimeIndex:
|
|
"""Valide la présence d'un index temporel et le retourne pour uniformiser les traitements."""
|
|
|
|
if not isinstance(df.index, pd.DatetimeIndex):
|
|
raise TypeError("Cette fonction nécessite un DataFrame indexé par le temps.")
|
|
return df.index
|
|
|
|
|
|
def _infer_time_step(index: pd.DatetimeIndex) -> pd.Timedelta:
|
|
"""Estime la résolution temporelle représentative (médiane) d'un index daté."""
|
|
|
|
diffs = index.to_series().diff().dropna()
|
|
if diffs.empty:
|
|
return pd.Timedelta(minutes=1)
|
|
return diffs.median()
|