Aller au contenu

Session météo (kadi.weather.session)

WeatherSession est le point d'entrée principal du module kadi.weather. Elle orchestre les composants internes et expose une API simple pour toutes les fonctionnalités météo et agronomiques.

Méthodes

forecast(days)

Récupère les prévisions météo court-terme depuis Open-Meteo.

prevision = session.forecast(days=5)

Paramètres :

Nom Type Défaut Description
days int Depuis config.py Nombre de jours (maximum 16)

Retour : dict

Clé Type Description
location dict {'name', 'lat', 'lon'}
data list[dict] Liste des jours avec temperature_min, temperature_max, precipitation
data_source str Source utilisée ('open-meteo' ou 'cache')
last_updated str Horodatage ISO de la mise à jour

historical(metric, months_back)

Retourne les séries météo historiques depuis CHIRPS et Open-Meteo.

# Seulement les précipitations sur 6 mois
df_pluie = session.historical(metric="precipitation", months_back=6)

# Toutes les variables sur 10 ans
df_complet = session.historical(months_back=120)

Paramètres :

Nom Type Défaut Description
metric str 'all' Filtre : 'temperature', 'precipitation', 'humidity', 'all'
months_back int 120 Nombre de mois d'historique

Retour : pd.DataFrame — Données indexées par date.

onset()

Détecte la date de démarrage de la saison des pluies selon la zone climatique.

onset = session.onset()
print(f"Démarrage : {onset['onset_date']}")
print(f"Méthode   : {onset['method']}")  # 'Sivakumar' ou 'Walter-Anyadike'

Retour : dict avec onset_date, method, confidence, zone.

L'algorithme utilisé dépend de la zone automatiquement détectée : - Nord (> 9.5° N) : Sivakumar — saison unimodale - Sud (< 7.5° N) : Walter-Anyadike — saison bimodale

cessation()

Détermine la date de fin des pluies utiles.

cessation = session.cessation()
print(f"Fin saison : {cessation['cessation_date']}")

Retour : dict avec cessation_date, method, confidence.

growing_degree_days(crop, start_date, end_date)

Calcule l'accumulation de degrés-jours de croissance depuis la date de semis. Les GDD mesurent l'énergie thermique disponible pour le développement de la plante.

gdd = session.growing_degree_days(
    crop="maize",
    start_date="2026-05-15",   # Date de semis
    end_date="2026-09-30",     # Optionnel — jusqu'à aujourd'hui si None
)

print(f"GDD accumulés  : {gdd['gdd_accumulated']:.1f} °C·jour")
print(f"Stade phéno    : {gdd['phenology_stage']}")
print(f"Floraison dans : {gdd['days_to_flowering']} jours")

Paramètres :

Nom Type Défaut Description
crop str requis Code de la culture ('maize', 'rice', etc.)
start_date str requis Date de semis au format 'YYYY-MM-DD'
end_date str None Date de fin (aujourd'hui si None)

Retour : dict avec gdd_accumulated, phenology_stage, crop, start_date, days_to_flowering, days_to_maturity.

rain_probability(days_ahead, min_rainfall_mm)

Calcule la probabilité de pluie en combinant les prévisions Open-Meteo et les fréquences historiques (chaînes de Markov).

prob = session.rain_probability(days_ahead=3, min_rainfall_mm=1.0)

print(f"Demain     : {prob['tomorrow'] * 100:.0f}%")
print(f"Recommandation : {prob['recommendation']}")
print(f"Message        : {prob['message']}")

Paramètres :

Nom Type Défaut Description
days_ahead int 1 Horizon de prévision en jours
min_rainfall_mm float 1.0 Seuil de pluie significative en mm

Retour : dict avec tomorrow (probabilité J+1), message, recommendation.

drought_index(method, window_months)

Calcule un indice de sécheresse sur les données historiques.

drought = session.drought_index(method="spi", window_months=3)

print(f"SPI 3 mois  : {drought['spi_3month']:.2f}")
print(f"Sévérité    : {drought['drought_severity']}")

Méthodes disponibles :

Méthode Description
'spi' Standardized Precipitation Index (Z-score sur les précipitations)
'markov' Probabilité de persistance de la sécheresse (Markov)
'hurst' Exposant de Hurst — mémoire longue de la sécheresse
'combined' Combinaison pondérée des 3 méthodes

Sévérités SPI :

SPI Sévérité
> -0.5 no_drought
-0.5 à -1.0 mild
-1.0 à -1.5 moderate
< -1.5 severe

water_balance(crop, soil_type)

Simule le bilan hydrique quotidien du sol selon la méthode FAO-56, en calculant l'évapotranspiration de référence (ET0) par Hargreaves-Samani.

bilan = session.water_balance(crop="maize", soil_type="ferrugineux")
print(bilan.tail(7)[["precipitation", "ET0", "deficit_eau", "reserve_utile"]])

Paramètres :

Nom Type Défaut Description
crop str 'maize' Culture de référence (influence le Kc)
soil_type str 'ferrugineux' Type de sol béninois

Types de sols supportés : 'ferrugineux', 'vertisol', 'hydromorphe', 'sableux'.

Colonnes du DataFrame retourné :

Colonne Description
precipitation Précipitations observées (mm)
ET0 Évapotranspiration de référence (mm)
ETc Évapotranspiration de la culture (ET0 × Kc)
deficit_eau Déficit hydrique journalier (mm)
reserve_utile Eau disponible dans le sol (mm)
runoff Ruissellement journalier (mm)

et0_hargreaves(tmin, tmax, day_of_year)

Calcule l'évapotranspiration de référence (ET0) pour un jour donné avec la méthode Hargreaves-Samani.

et0 = session.et0_hargreaves(tmin=22.0, tmax=35.0, day_of_year=180)
print(f"ET0 : {et0:.2f} mm/jour")

WeatherSession

Session météorologique : point d'entrée principal pour l'utilisateur. Gère le setup, le cache et expose l'API fonctionnelle complète.

Source code in kadi/weather/session.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class WeatherSession:
    """
    Session météorologique : point d'entrée principal pour l'utilisateur.
    Gère le setup, le cache et expose l'API fonctionnelle complète.
    """

    def __init__(self, latitude: float, longitude: float, name: str = None, cache_dir: str = None):
        """
        Initialise une nouvelle session pour une localisation.

        :param latitude: Latitude (en degrés décimaux).
        :param longitude: Longitude (en degrés décimaux).
        :param name: Nom de la localité (optionnel).
        :param cache_dir: Dossier pour le cache (optionnel).
        """
        self.location = Location(latitude, longitude, name)
        self.cache_dir = cache_dir

        # Initialisation de la gestion des données
        self.weather_data = WeatherData(self.location, cache_dir)

        # Composants métiers (initialisés paresseusement ou lors de _init_all_components)
        self.phenology: Optional[Phenology] = None
        self.hydrology: Optional[Hydrology] = None
        self.risk_indicators: Optional[RiskIndicators] = None

    def _ensure_data(self, require_forecast=False, require_historical=False):
        """
        S'assure que les données nécessaires sont chargées.
        """
        if require_forecast and self.weather_data.forecast_data is None:
            self.weather_data.fetch_forecast()

        if require_historical and self.weather_data.historical_data is None:
            self.weather_data.fetch_historical()

    def _ensure_components(self, component: str):
        """
        S'assure que le composant demandé est initialisé.
        """
        if component == 'phenology' and self.phenology is None:
            self._ensure_data(require_historical=True)
            hist = self.weather_data.historical_data
            self.phenology = Phenology(self.location, hist['precipitation'], hist[['temperature_min', 'temperature_max']])

        elif component == 'hydrology' and self.hydrology is None:
            self._ensure_data(require_historical=True)
            hist = self.weather_data.historical_data
            self.hydrology = Hydrology(self.location, hist['precipitation'], hist[['temperature_min', 'temperature_max']])

        elif component == 'risk' and self.risk_indicators is None:
            self._ensure_data(require_forecast=True, require_historical=True)
            self.risk_indicators = RiskIndicators(self.location, self.weather_data.historical_data['precipitation'], self.weather_data.forecast_data)

    def forecast(self, days: int = None) -> dict:
        """
        Récupère la prévision météorologique court-terme.

        :param days: Nombre de jours de prévision (défaut depuis CONFIG).
        :return: Dictionnaire des prévisions.
        """
        if days is None:
            days = CONFIG["weather"]["forecast_days_default"]

        if days > CONFIG["weather"]["max_forecast_days"]:
            days = CONFIG["weather"]["max_forecast_days"]

        df = self.weather_data.fetch_forecast(days=days)

        # Format de retour selon cahier des charges
        return {
            'location': {'name': self.location.name, 'lat': self.location.latitude, 'lon': self.location.longitude},
            'data': df.reset_index().to_dict(orient='records'),
            'data_source': self.weather_data.data_source,
            'last_updated': pd.Timestamp.now().isoformat()
        }

    def historical(self, metric: str = 'all', months_back: int = 120) -> pd.DataFrame:
        """
        Retourne les séries historiques.

        :param metric: Filtre de colonne ('temperature', 'precipitation', 'humidity', 'all').
        :param months_back: Nombre de mois d'historique.
        :return: DataFrame historique.
        """
        df = self.weather_data.fetch_historical(months_back=months_back)

        if metric != 'all':
            cols = [c for c in df.columns if metric in c]
            if cols:
                return df[cols]

        return df

    def growing_degree_days(self, crop: str, start_date: str, end_date: str = None) -> dict:
        """
        Calcule l'accumulation des degrés-jours.

        :param crop: Nom de la culture.
        :param start_date: Date de semis (YYYY-MM-DD).
        :param end_date: Date de fin (optionnel).
        :return: Résultat du cumul GDD.
        """
        self._ensure_components('phenology')
        return self.phenology.growing_degree_days(crop, start_date, end_date)

    def onset(self) -> dict:
        """
        Détecte la date de démarrage de la saison agricole.

        :return: Résultat de l'onset.
        """
        self._ensure_components('phenology')
        return self.phenology.onset()

    def cessation(self) -> dict:
        """
        Détermine la date de fin des pluies utiles.

        :return: Résultat de cessation.
        """
        self._ensure_components('phenology')
        return self.phenology.cessation()

    def drought_index(self, method: str = 'spi', window_months: int = 3) -> dict:
        """
        Calcule l'indice de sécheresse.

        :param method: 'spi', 'markov', 'hurst', 'combined'.
        :param window_months: Fenêtre temporelle en mois.
        :return: Indicateurs de sécheresse.
        """
        self._ensure_components('risk')
        return self.risk_indicators.drought_index(method, window_months)

    def rain_probability(self, days_ahead: int = 1, min_rainfall_mm: float = 1.0) -> dict:
        """
        Prévoit la probabilité de pluie.

        :param days_ahead: Nombre de jours futurs.
        :param min_rainfall_mm: Seuil minimum.
        :return: Probabilité et recommandations.
        """
        self._ensure_components('risk')
        return self.risk_indicators.rain_probability(days_ahead, min_rainfall_mm)

    def water_balance(self, crop: str = 'maize', soil_type: str = 'ferrugineux') -> pd.DataFrame:
        """
        Simule le bilan hydrique quotidien (FAO-56).

        :param crop: Type de culture.
        :param soil_type: Type de sol.
        :return: DataFrame avec le bilan hydrique.
        """
        self._ensure_components('hydrology')
        self.hydrology.crop = crop
        self.hydrology.soil_type = soil_type
        self.hydrology.soil_params = self.hydrology.get_soil_params(soil_type)
        return self.hydrology.compute_water_balance()

    def et0_hargreaves(self, tmin: float, tmax: float, day_of_year: int) -> float:
        """
        Calcule l'ETo par Hargreaves-Samani.

        :param tmin: Temp. min.
        :param tmax: Temp. max.
        :param day_of_year: Jour de l'année.
        :return: ETo en mm/jour.
        """
        self._ensure_components('hydrology')
        return self.hydrology.et0_hargreaves(tmin, tmax, day_of_year)

    def _init_all_components(self) -> None:
        """
        Initialise toutes les classes composantes.
        """
        self._ensure_components('phenology')
        self._ensure_components('hydrology')
        self._ensure_components('risk')

__init__ 

__init__(latitude: float, longitude: float, name: str = None, cache_dir: str = None)

Initialise une nouvelle session pour une localisation.

:param latitude: Latitude (en degrés décimaux). :param longitude: Longitude (en degrés décimaux). :param name: Nom de la localité (optionnel). :param cache_dir: Dossier pour le cache (optionnel).

Source code in kadi/weather/session.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, latitude: float, longitude: float, name: str = None, cache_dir: str = None):
    """
    Initialise une nouvelle session pour une localisation.

    :param latitude: Latitude (en degrés décimaux).
    :param longitude: Longitude (en degrés décimaux).
    :param name: Nom de la localité (optionnel).
    :param cache_dir: Dossier pour le cache (optionnel).
    """
    self.location = Location(latitude, longitude, name)
    self.cache_dir = cache_dir

    # Initialisation de la gestion des données
    self.weather_data = WeatherData(self.location, cache_dir)

    # Composants métiers (initialisés paresseusement ou lors de _init_all_components)
    self.phenology: Optional[Phenology] = None
    self.hydrology: Optional[Hydrology] = None
    self.risk_indicators: Optional[RiskIndicators] = None

forecast 

forecast(days: int = None) -> dict

Récupère la prévision météorologique court-terme.

:param days: Nombre de jours de prévision (défaut depuis CONFIG). :return: Dictionnaire des prévisions.

Source code in kadi/weather/session.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def forecast(self, days: int = None) -> dict:
    """
    Récupère la prévision météorologique court-terme.

    :param days: Nombre de jours de prévision (défaut depuis CONFIG).
    :return: Dictionnaire des prévisions.
    """
    if days is None:
        days = CONFIG["weather"]["forecast_days_default"]

    if days > CONFIG["weather"]["max_forecast_days"]:
        days = CONFIG["weather"]["max_forecast_days"]

    df = self.weather_data.fetch_forecast(days=days)

    # Format de retour selon cahier des charges
    return {
        'location': {'name': self.location.name, 'lat': self.location.latitude, 'lon': self.location.longitude},
        'data': df.reset_index().to_dict(orient='records'),
        'data_source': self.weather_data.data_source,
        'last_updated': pd.Timestamp.now().isoformat()
    }

historical 

historical(metric: str = 'all', months_back: int = 120) -> pd.DataFrame

Retourne les séries historiques.

:param metric: Filtre de colonne ('temperature', 'precipitation', 'humidity', 'all'). :param months_back: Nombre de mois d'historique. :return: DataFrame historique.

Source code in kadi/weather/session.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def historical(self, metric: str = 'all', months_back: int = 120) -> pd.DataFrame:
    """
    Retourne les séries historiques.

    :param metric: Filtre de colonne ('temperature', 'precipitation', 'humidity', 'all').
    :param months_back: Nombre de mois d'historique.
    :return: DataFrame historique.
    """
    df = self.weather_data.fetch_historical(months_back=months_back)

    if metric != 'all':
        cols = [c for c in df.columns if metric in c]
        if cols:
            return df[cols]

    return df

growing_degree_days 

growing_degree_days(crop: str, start_date: str, end_date: str = None) -> dict

Calcule l'accumulation des degrés-jours.

:param crop: Nom de la culture. :param start_date: Date de semis (YYYY-MM-DD). :param end_date: Date de fin (optionnel). :return: Résultat du cumul GDD.

Source code in kadi/weather/session.py 
113
114
115
116
117
118
119
120
121
122
123
def growing_degree_days(self, crop: str, start_date: str, end_date: str = None) -> dict:
    """
    Calcule l'accumulation des degrés-jours.

    :param crop: Nom de la culture.
    :param start_date: Date de semis (YYYY-MM-DD).
    :param end_date: Date de fin (optionnel).
    :return: Résultat du cumul GDD.
    """
    self._ensure_components('phenology')
    return self.phenology.growing_degree_days(crop, start_date, end_date)

onset 

onset() -> dict

Détecte la date de démarrage de la saison agricole.

:return: Résultat de l'onset.

Source code in kadi/weather/session.py 
125
126
127
128
129
130
131
132
def onset(self) -> dict:
    """
    Détecte la date de démarrage de la saison agricole.

    :return: Résultat de l'onset.
    """
    self._ensure_components('phenology')
    return self.phenology.onset()

cessation 

cessation() -> dict

Détermine la date de fin des pluies utiles.

:return: Résultat de cessation.

Source code in kadi/weather/session.py 
134
135
136
137
138
139
140
141
def cessation(self) -> dict:
    """
    Détermine la date de fin des pluies utiles.

    :return: Résultat de cessation.
    """
    self._ensure_components('phenology')
    return self.phenology.cessation()

drought_index 

drought_index(method: str = 'spi', window_months: int = 3) -> dict

Calcule l'indice de sécheresse.

:param method: 'spi', 'markov', 'hurst', 'combined'. :param window_months: Fenêtre temporelle en mois. :return: Indicateurs de sécheresse.

Source code in kadi/weather/session.py 
143
144
145
146
147
148
149
150
151
152
def drought_index(self, method: str = 'spi', window_months: int = 3) -> dict:
    """
    Calcule l'indice de sécheresse.

    :param method: 'spi', 'markov', 'hurst', 'combined'.
    :param window_months: Fenêtre temporelle en mois.
    :return: Indicateurs de sécheresse.
    """
    self._ensure_components('risk')
    return self.risk_indicators.drought_index(method, window_months)

rain_probability 

rain_probability(days_ahead: int = 1, min_rainfall_mm: float = 1.0) -> dict

Prévoit la probabilité de pluie.

:param days_ahead: Nombre de jours futurs. :param min_rainfall_mm: Seuil minimum. :return: Probabilité et recommandations.

Source code in kadi/weather/session.py 
154
155
156
157
158
159
160
161
162
163
def rain_probability(self, days_ahead: int = 1, min_rainfall_mm: float = 1.0) -> dict:
    """
    Prévoit la probabilité de pluie.

    :param days_ahead: Nombre de jours futurs.
    :param min_rainfall_mm: Seuil minimum.
    :return: Probabilité et recommandations.
    """
    self._ensure_components('risk')
    return self.risk_indicators.rain_probability(days_ahead, min_rainfall_mm)

water_balance 

water_balance(crop: str = 'maize', soil_type: str = 'ferrugineux') -> pd.DataFrame

Simule le bilan hydrique quotidien (FAO-56).

:param crop: Type de culture. :param soil_type: Type de sol. :return: DataFrame avec le bilan hydrique.

Source code in kadi/weather/session.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
def water_balance(self, crop: str = 'maize', soil_type: str = 'ferrugineux') -> pd.DataFrame:
    """
    Simule le bilan hydrique quotidien (FAO-56).

    :param crop: Type de culture.
    :param soil_type: Type de sol.
    :return: DataFrame avec le bilan hydrique.
    """
    self._ensure_components('hydrology')
    self.hydrology.crop = crop
    self.hydrology.soil_type = soil_type
    self.hydrology.soil_params = self.hydrology.get_soil_params(soil_type)
    return self.hydrology.compute_water_balance()

et0_hargreaves 

et0_hargreaves(tmin: float, tmax: float, day_of_year: int) -> float

Calcule l'ETo par Hargreaves-Samani.

:param tmin: Temp. min. :param tmax: Temp. max. :param day_of_year: Jour de l'année. :return: ETo en mm/jour.

Source code in kadi/weather/session.py 
179
180
181
182
183
184
185
186
187
188
189
def et0_hargreaves(self, tmin: float, tmax: float, day_of_year: int) -> float:
    """
    Calcule l'ETo par Hargreaves-Samani.

    :param tmin: Temp. min.
    :param tmax: Temp. max.
    :param day_of_year: Jour de l'année.
    :return: ETo en mm/jour.
    """
    self._ensure_components('hydrology')
    return self.hydrology.et0_hargreaves(tmin, tmax, day_of_year)