ajout du upload des predictions sur s3

airflow/dags/tasks/run_model.py

@@ -60,8 +60,30 @@ def run_model(**context):
     # clip the predictions to be above 0
     clipped_predictions = raw_predictions.clip(0, 1e6).tolist()
 
-    logger.info(f"✅ Clipped Predictions : {clipped_predictions}")
-    return clipped_predictions
+    
+    # Les noms de colonnes
+    columns = ["O3", "NOx", "PM10", "PM25"]
+
+    # Conversion en dictionnaire
+    formatted_result = dict(zip(columns, clipped_predictions[0]))
+    # Conversion en JSON
+    filename = f"{datetime.now().strftime('%Y%m%d-%H%M%S')}_prediction_data.json"
+    local_path = f"/tmp/{filename}"
+    s3_key = f"datasets/output/{filename}"
+
+    with open(local_path, "w") as f:
+        json.dump(formatted_result, f)
+
+    # Upload du resultat sur S3
+    if S3Hook is not None:
+        s3_hook = S3Hook(aws_conn_id="aws_default")
+        s3_hook.load_file(filename=local_path, key=s3_key, bucket_name=s3_bucket, replace=True)
+        logger.info(f"✅ Uploaded to S3: {s3_key}")
+    else:
+        logger.info("ℹ️ S3Hook non disponible (hors Airflow). Fichier seulement écrit en local.")
+
+    logger.info(f"✅ Clipped Predictions : {filename}")
+    return filename
 
 
 # ✅ Fonction main pour tester localement (verifier que le fichier model est présent)