update pour jouer en local et sur airflow

airflow/dags/tasks/run_model.py

@@ -27,8 +27,11 @@ def run_model(**context):
     logger.info("📡 Launch model on data")
     model_filename = "linear_model_2025_07_10_16_28_59.pkl"
     
+    # Nom du fichier result de la prediction 
+    filename = f"{datetime.now().strftime('%Y%m%d-%H%M%S')}_prediction_data.json"
+    logger.info(context)
     #Je suis dans Airflow
-    if S3Hook is not None and context is not None:
+    if S3Hook is not None and context:
         s3_bucket = get_config("S3BucketName")
         s3_path = 'models/'
         # Connect to our S3 bucket and download the model file
@@ -39,18 +42,21 @@ def run_model(**context):
         #input_data_model = {"Pressure": 1021, "Temperature": 22.73, "Wind Speed": 4.12, "Humidity": 59, "Traffic Status": 0.23}
         print(type(input_data_model))
         logger.info(f"✅ input data model : {input_data_model}")
+        local_path = f"/opt/airflow/data/{filename}"
+        print(local_path)
     else:
         #recuperer le model dans git ou ailleurs
         logger.info(f"✅ Model local used en local: {model_filename}")
         filename_model = '../data/'+model_filename
         # test the model with random values
         input_data_model = {
-            "Pressure": 999,
-            "Temperature": 22,
-            "Wind Speed": 10,
-            "Humidity": 50,
-            "Traffic Status": 0,
+            "Pressure":1020,
+            "Temperature":25,
+            "Wind Speed":3.09,
+            "Humidity":53,
+            "Traffic Status":0.44
         }
+        local_path = f"../data/{filename}"
 
     with open(filename_model, "rb") as f:
         model = pickle.load(f)
@@ -62,20 +68,20 @@ def run_model(**context):
 
     
     # Les noms de colonnes
-    columns = ["O3", "NOx", "PM10", "PM25"]
+    columns = ["NOX", "PM10", "PM25", "O3"]
 
     # 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}"
+    
+    # Path vers le fichier sur le s3
     s3_key = f"datasets/output/{filename}"
 
+    # Conversion du fichier JSON
     with open(local_path, "w") as f:
         json.dump(formatted_result, f)
 
     # Upload du resultat sur S3
-    if S3Hook is not None:
+    if S3Hook is not None and context:
         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}")