import pandas as pd import io import os import boto3 import datetime import json import joblib import requests import boto3 import numpy as np from datetime import date, timedelta, timezone from geopy.distance import geodesic from sqlalchemy import create_engine, text from dotenv import load_dotenv load_dotenv('.secrets') def run_features_and_fusion(): os.environ['AWS_ACCESS_KEY_ID'] = os.getenv('AWS_ACCESS_KEY_ID') os.environ['AWS_SECRET_ACCESS_KEY'] = os.getenv('AWS_SECRET_ACCESS_KEY') os.environ['MLFLOW_DEFAULT_ARTIFACT_ROOT'] = os.getenv('MLFLOW_DEFAULT_ARTIFACT_ROOT') os.environ['S3_BUCKET'] = os.getenv('S3_BUCKET') s3 = boto3.client("s3") db_user = os.getenv("DB_USER") db_password = os.getenv("DB_PASSWORD") db_host = os.getenv("DB_HOST") db_name = os.getenv("DB_NAME") engine = create_engine(f"postgresql+psycopg2://{db_user}:{db_password}@{db_host}/{db_name}") query = ("""SELECT * FROM meteoday WHERE date >= CURRENT_DATE - INTERVAL '1825 DAYS'""") df = pd.read_sql(query, engine) df = pd.DataFrame(df) df = df.copy() print(df.info()) print('done ....') df["poste"] = df["poste"].astype('Float64') print(df.head(2)) print('done ....') print('1-nombe de date null :',df['date'].isnull().sum()) bucket = os.getenv('S3_BUCKET') s3.download_file(bucket,"dataset/liste_station.csv", "liste_station.csv" ) df_station = pd.read_csv("liste_station.csv") list_station = df_station["0"].unique().tolist() print('done ....') print(list_station) # on ne garde que les stations présentent dans la liste station météo corse mask = df['poste'].isin(list_station) df= df[mask] print('done ....') print(df.head()) df['date'] = pd.to_datetime(df['date'], format='ISO8601') # Traitements de colonnes innutiles drop_cols = list(set([ 'PMERM','PMERMIN','QPMERMIN','FF2M','QFF2M','FXI2','QFXI2','DXI2','QDXI2','HXI2','QHXI2','DXI3S','QDXI3S','DHUMEC','QDHUMEC','INST','QINST','GLOT','QGLOT','DIFT','QDIFT','DIRT','QDIRT','SIGMA','QSIGMA','INFRART','QINFRART','UV_INDICEX','QUV_INDICEX','NB300','QNB300','BA300','QBA300','NEIG','QNEIG','BROU','QBROU','GRESIL','GRELE','QGRELE','ROSEE','QROSEE','VERGLAS','QVERGLAS','SOLNEIGE','QSOLNEIGE','GELEE','QGELEE','FUMEE','QFUMEE','UV','QUV','TMERMAX','QTMERMAX','TMERMIN','QTMERMIN','HNEIGEF','QHNEIGEF','NEIGETOTX','QNEIGETOTX','NEIGETOT06','QNEIGETOT06','QRR','QDRR','QTN','QHTN','QTX','QHTX','QTM','QTMNX','QTNSOL','QTN50','DG','QDG','QTAMPLI','QTNTXM','QPMERM','QFFM','QFXI','QDXI','QHXI','QFXY','QDXY','QHXY','QFXI3S','QHXI3S','QUN','QHUN','QUX','QHUX','QDHUMI40','QDHUMI80','QTSVM','QUM','QORAG','QGRESIL','QBRUME','ECLAIR','QECLAIR','QETPMON','QETPGRILLE' ])) drop_cols = [d.lower() for d in drop_cols] df = df.drop(columns=drop_cols, axis=1) # convertion des colonnes object en float64 for column in df.columns: if column != "date" and df[column].dtype == 'object': df[column] = df[column].astype('Float64') # fonction de moyenne lissante df['moyenne_precipitations_mois'] = (df['rr'].rolling(window=31, min_periods=10).mean().round(2)) df['moyenne_vitesse_vent_mois'] = (df['ffm'].rolling(window=31, min_periods=10).mean().round(2)) df['moyenne_temperature_mois'] = (df['tn'].rolling(window=31, min_periods=10).mean().round(2)) # Fonction pour compter les jours consécutifs sans pluie def compter_jours_sans_pluie(groupe): compteur = 0 jours_sans_pluie = [] for rr in groupe['rr']: if pd.isna(rr): jours_sans_pluie.append(np.nan) elif rr == 0: compteur += 1 jours_sans_pluie.append(compteur) else: compteur = 0 jours_sans_pluie.append(compteur) return pd.Series(jours_sans_pluie, index=groupe.index) # Fonction pour compter les jours consécutifs avec TX > 30 def compter_jours_chauds(groupe): compteur = 0 jours_chauds = [] for tx in groupe['tx']: if pd.isna(tx): jours_chauds.append(np.nan) elif tx > 30: compteur += 1 jours_chauds.append(compteur) else: compteur = 0 jours_chauds.append(compteur) return pd.Series(jours_chauds, index=groupe.index) # Application des 2 fonctions df= df.sort_values(['poste', 'date']) df['jours_tx_sup_30'] = ( df.groupby('poste')['tx'] .transform(lambda s: compter_jours_chauds( pd.DataFrame({'tx': s}, index=s.index) )) .astype(float) ) df['jours_sans_pluie'] = ( df.groupby('poste')['rr'] .transform(lambda s: compter_jours_sans_pluie( pd.DataFrame({'rr': s}, index=s.index) )) .astype(float) ) df["etpgrille_7j"] = df.groupby("poste")["etpgrille"].transform(lambda x: x.rolling(7, min_periods=1).mean()) df['poste'] = df['poste'].astype('Int64') print('done ....') print('2-nombe de date null :',df['date'].isnull().sum()) # appelle du fichier cordonnées gps des stations météo bucket = os.getenv('S3_BUCKET') s3.download_file(bucket, "dataset/coord_stations.csv", "coordonnée_station.csv") df_coord_station = pd.read_csv("coordonnée_station.csv", sep=',') df_coord_station.drop(columns={'Unnamed: 0'}, inplace=True) df_coord_station.columns = df_coord_station.columns.str.lower() # on ne garde que les stations présentent dans la liste station météo corse list_station2 = df["poste"].unique().tolist() mask = df_coord_station['poste'].isin(list_station2) df_coord_station2= df_coord_station[mask] print('done ....') print(df_coord_station2.head(2)) # on crée notre fichier météo avec les coordonnées gps des stations df_meteo=pd.merge(df_coord_station2, df, on="poste", how='left') print(df_meteo.head(2)) print('done ....') print('3-nombe de date null :',df_meteo['date'].isnull().sum()) # on nettoie/supprime et modifie les colonnes df_meteo['date'] = pd.to_datetime(df_meteo['date'], format='ISO8601') df_meteo['nom'] = df_meteo['nom'].astype('string') df_stations_unique = df_meteo[['nom', 'lat', 'lon','poste']].drop_duplicates() print('done ....') print(df_stations_unique.head(2)) # charge depuis le S3 du fichier incendie corse s3.download_file(bucket, "dataset/historique_incendie_corse_cleaned_2025.csv", "historique_incendie.csv") df_feux = pd.read_csv("historique_incendie.csv", sep=',') df_feux.columns = df_feux.columns.str.lower() def find_nearest_station(row, stations): p_feu = (row["latitude_feu"], row["longitude_feu"]) distances = stations.apply( lambda s: geodesic(p_feu, (s["lat"], s["lon"])).km, axis=1 ) idx = distances.idxmin() post = stations.loc[idx, "poste"] return pd.Series([idx, distances[idx], post]) df_feux[["station_index", "dist_km",'poste']] = df_feux.apply( find_nearest_station, stations=df_stations_unique, axis=1 ) df_fire = df_feux[['date', 'poste', 'feux', 'ville', 'latitude_feu', 'longitude_feu', 'surface_parcourue_m2', 'surface_forêt_m2', 'surface_maquis_garrigues_m2', 'autres_surfaces_naturelles_hors_forêt_m2', 'surfaces_agricoles_m2', 'autres_surfaces_m2', 'surface_autres_terres_boisées_m2', 'surfaces_non_boisées_naturelles_m2', 'surfaces_non_boisées_artificialisées_m2', 'surfaces_non_boisées_m2', 'type_de_peuplement', 'nature']] df_final = pd.merge(df_stations_unique, df_fire, on=['poste'], how='left') print('done ....') print('4-nombe de date null :',df_final['date'].isnull().sum()) df_final['date'] = pd.to_datetime(df_final['date']) df_meteo['date'] = pd.to_datetime(df_meteo['date']) df_final.drop(columns={'lat', 'lon','poste'}, inplace=True) df_feux_corse = pd.merge(df_meteo, df_final, on=('date','nom'), how='left') print('done ....') print('5-nombe de date null :',df_feux_corse['date'].isnull().sum()) df_feux_corse = df_feux_corse.sort_values(["poste", "date"]) # Suppression des doublons crée lors du merge, on garde la première valeur df_feux_corse = df_feux_corse.drop_duplicates(subset=["poste","date"], keep='first') def days_until_next_fire(g): # index des feux fire_idx = g.index[g["feux"] == 1] # tableau vide days = pd.Series(index=g.index, dtype=float) for i in range(len(g)): current_date = g.loc[g.index[i], "date"] # dates de feux dans le futur future_fires = g.loc[fire_idx, "date"] future_fires = future_fires[future_fires >= current_date] if len(future_fires) == 0: days[g.index[i]] = None # pas de feu après else: days[g.index[i]] = (future_fires.iloc[0] - current_date).days return days df_feux_corse["décompte"] = df_feux_corse.groupby("poste").apply(days_until_next_fire).reset_index(level=0, drop=True) df_feux_corse['feux'] = df_feux_corse.feux.fillna(0).astype(int) df_feux_corse= df_feux_corse.sort_values(['nom', 'date']) # Création de la colonne évènement pour indiquer si un feu a eu lieu df_feux_corse['évènement'] = df_feux_corse['feux'] == 1 # Nouvelle colonne initialisée à NaN df_feux_corse["compteur_jours_vers_prochain_feu"] = pd.NA # Traitement par ville for nom, groupe in df_feux_corse.groupby("nom"): groupe = groupe.sort_values("date") indices_feux = groupe[groupe["évènement"] == True].index.tolist() for i in range(len(indices_feux) - 1): debut = indices_feux[i] fin = indices_feux[i + 1] # Remplir les jours entre les deux feux avec un compteur croissant for j, idx in enumerate(range(debut, fin)): df_feux_corse.loc[idx, "compteur_jours_vers_prochain_feu"] = j # # nombre de jour sans feu + log et carré df_feux_corse['compteur_feu_log'] = df_feux_corse['compteur_jours_vers_prochain_feu'].apply(lambda x: np.log1p(x) if pd.notnull(x) else np.nan) df_feux_corse['année'] = df_feux_corse['date'].dt.year df_feux_corse['mois'] = df_feux_corse['date'].dt.month # Trier par ville et par date df_feux_corse = df_feux_corse.sort_values(['nom', 'date']) # Création de la colone Feu prévu pour le modèle Survival df_feux_corse.loc[:, "feu_prévu"] = df_feux_corse["décompte"].notna().astype(int) df_feux_corse['feu_prévu'] = df_feux_corse['feu_prévu'].astype(bool) df_feux_corse['date'] = pd.to_datetime(df_feux_corse['date']).dt.date print('done ....') print('visu du format date de feux corse:', df_feux_corse['date'].tail(),'.........', df_feux_corse['date'].info()) yesterday = (datetime.date.today() - datetime.timedelta(days=1)) print('done ....') print("date d'hier:", yesterday) df_push = df_feux_corse[df_feux_corse["date"] == yesterday] print('done ....') print(df_push.head()) print('6-nombe de date null :',df_feux_corse['date'].isnull().sum(), 'nombre de post null :', df_feux_corse['poste'].isnull().sum()) print("--------------------Opération terminée-------------------") df_push.to_sql('data_prediction', engine, if_exists="append", index=False) print("-----------Envoie sur la database neon effectué----------")