Added training DAG

dags_airflow_ale_trainmod.py

@@ -11,6 +11,9 @@
 #baseapiurl="http://gflapi.glitchflow.svc.cluster.local:8000/"
 #apifrzendp="train"
 
+WAITMSG="WAITING"
+FRZMSG="FROZEN"
+
 dag = DAG(
     dag_id="Trainpipe",
     start_date=airflow.utils.dates.days_ago(3),
@@ -19,6 +22,22 @@
 
 )
 
+
+def pick_branch(**context):
+
+    jsmsg=context["task_instance"].xcom_pull(
+        task_ids="send_frz_sign", key="return_value"
+    )
+
+    respob=json.loads(jsmsg)
+
+    if(respob["resp"]==WAITMSG):
+        return "next_sensor"
+    else:
+        return "next_metrics"     
+
+
+
 IniTrain = DummyOperator(task_id="start_training", dag=dag)
 
 sign_train = SimpleHttpOperator(
@@ -29,11 +48,21 @@
     data=json.dumps({"user":"airflow","token":"airflow"}),
     headers={"Content-Type": "application/json"},
 
+
     retries=3,
     retry_delay=timedelta(minutes=5),
     dag=dag,
 )
 
-Next = DummyOperator(task_id="next", dag=dag)
+chech_train_resp=BranchPythonOperator(
+    task_id="check_frz_sign",
+    python_callable=pick_branch,
+)
+
+next_sens = DummyOperator(task_id="next_sensor", dag=dag)
+next_metrics = DummyOperator(task_id="next_sensor", dag=dag)
+
+
+
 
-IniTrain>>sign_train>>Next
+IniTrain>>sign_train