updates

dags/config.py

@@ -8,7 +8,8 @@
 
 TRAIN_DATA_PATH = os.path.join(DATA_FOLDER, 'la-haute-borne-data-2017-2020.csv')
 # TODO change, should point to generator
-GENERATED_DATA_PATH = os.path.join(DATA_FOLDER, 'la-haute-borne-data-2017-2020.csv')
+#GENERATED_DATA_PATH = os.path.join(DATA_FOLDER, 'la-haute-borne-data-2017-2020.csv')
+GENERATED_DATA_PATH = os.path.join(GENERATED_DATA_FOLDER, 'latest.csv')
 
 FEATURES_PATH = os.path.join(DATA_FOLDER, 'prepared_features.parquet')
 

dags/predict.py

@@ -7,8 +7,9 @@
 
 sys.path.insert(0, os.path.abspath(os.path.dirname(os.path.dirname(__file__))))  # So that airflow can find config files
 
-from dags.config import GENERATED_DATA_PATH, DATA_FOLDER
+from dags.config import GENERATED_DATA_PATH, DATA_FOLDER, MODEL_REGISTRY_FOLDER, PREDICTIONS_FOLDER
 from formation_indus_ds_avancee.feature_engineering import prepare_features_with_io
+from formation_indus_ds_avancee.train_and_predict import predict_with_io
 
 
 @dag(default_args={'owner': 'airflow'}, schedule=timedelta(minutes=2),
@@ -22,12 +23,22 @@ def prepare_features_with_io_task():
                                  training_mode=False)
         return features_path
 
-    # Start completing predict task
-    # predict = PythonOperator()
-    # End completing predict task
+    @task
+    def predict_with_io_task(features_path):
+        model_path = os.path.join(MODEL_REGISTRY_FOLDER, 'model.joblib')
+        predictions_folder = PREDICTIONS_FOLDER
+        predict_with_io(features_path, model_path, predictions_folder)
 
-    # feature_path = prepare_features_with_io_task()
-    # predict_with_io_task(feature_path=feature_path)
+    features_path = prepare_features_with_io_task()
+    predict_with_io_task(features_path=features_path)
 
 
 predict_dag = predict()
+
+dag = DAG(
+
+)
+
+task1 = PythonOpeartor(
+
+)