Fixes with API refactor for object based execution

mlpipeline/_pipeline_subprocess.py

@@ -202,7 +202,7 @@ def _experiment_main_loop(current_experiment, version_name_s, clean_experiment_d
                 test__eval_steps = 1 if test__eval_steps is not None else None
                 train_eval_steps = 1 if train_eval_steps is not None else None
 
-            _save_training_time(current_experiment, version_name)
+            _save_training_time(current_experiment, version_name, config)
 
             try:
                 input_fn = dataloader.get_train_input(mode=ExecutionModeKeys.TRAIN)
@@ -312,7 +312,7 @@ def _experiment_main_loop(current_experiment, version_name_s, clean_experiment_d
             _add_to_and_return_result_string("DATALOADER	 SUMMERY:")
             _add_to_and_return_result_string(dataloader.summery)
             if record_training and not config.no_log:
-                _save_results_to_file(_add_to_and_return_result_string(), current_experiment)
+                _save_results_to_file(_add_to_and_return_result_string(), current_experiment, config)
 
         except Exception as e:
             mlflow.end_run(mlflow.entities.RunStatus.to_string(mlflow.entities.RunStatus.FAILED))
@@ -462,36 +462,36 @@ def _get_experiment(file_path,
     return experiment, returning_version, clean_experiment_dir
 
 
-def _save_training_time(experiment, version_):
-    if CONFIG.experiment_mode == ExperimentModeKeys.TEST:
+def _save_training_time(experiment, version_, config):
+    if config.experiment_mode == ExperimentModeKeys.TEST:
         return
     name = experiment.name
-    with open(CONFIG.training_history_log_file, "a") as log_file:
+    with open(config.training_history_log_file, "a") as log_file:
         time = datetime.now().timestamp()
-        CONFIG.executed_experiments[name].version.addExecutingVersion(version_, time)
+        config.executed_experiments[name].version.addExecutingVersion(version_, time)
         log("Executing version: {0}".format(
-            CONFIG.executed_experiments[experiment.name].version.executing_version),
+            config.executed_experiments[experiment.name].version.executing_version),
             log_to_file=False)
         log_file.write("{0}::{1}::{2}\n".format(name,
-                                                CONFIG.executed_experiments[name].version.executing_version,
+                                                config.executed_experiments[name].version.executing_version,
                                                 time))
 
 
-def _save_results_to_file(resultString, experiment):
+def _save_results_to_file(resultString, experiment, config):
     modified_dt = datetime.isoformat(datetime.fromtimestamp(
-        CONFIG.executed_experiments[experiment.name].modified_time))
+        config.executed_experiments[experiment.name].modified_time))
     result_dt = datetime.now().isoformat()
 
-    with open(CONFIG.output_file, 'a', encoding="utf-8") as outfile:
+    with open(config.output_file, 'a', encoding="utf-8") as outfile:
         outfile.write("\n[{0}]:ml-pipline: output: \n".format(result_dt))
         outfile.write(resultString)
-    with open(CONFIG.history_file, 'a', encoding="utf-8") as hist_file:
+    with open(config.history_file, 'a', encoding="utf-8") as hist_file:
         hist_file.write("{0}::{1}::{2}\n".format(
             experiment.name,
-            CONFIG.executed_experiments[experiment.name].modified_time,
-            CONFIG.executed_experiments[experiment.name].version.executing_version))
+            config.executed_experiments[experiment.name].modified_time,
+            config.executed_experiments[experiment.name].version.executing_version))
 
-    CONFIG.executed_experiments[experiment.name].version.moveExecutingToExecuted()
+    config.executed_experiments[experiment.name].version.moveExecutingToExecuted()
 
 
 def _execute_exeperiment(file_path,

mlpipeline/api.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 from mlpipeline._pipeline import (_mlpipeline_main_loop, _init_pipeline)
 from mlpipeline._pipeline_subprocess import (_execute_exeperiment,
                                              _get_experiment_dir,
@@ -37,11 +38,22 @@ def mlpipeline_execute_exeperiment(experiment,
                                    blacklist_versions=None,
                                    pipeline_config=None):
     if pipeline_config is None:
-        pipeline_config = PipelineConfig()
-    pipeline_config.output_file = os.path.join(pipeline_config.experiments_dir, "output")
-    pipeline_config.history_file = os.path.join(pipeline_config.experiments_dir, "history")
-    pipeline_config.training_history_log_file = os.path.join(pipeline_config.experiments_dir, "t_history")
-    pipeline_config.log_file = os.path.join(pipeline_config.experiments_dir, "log")
+        pipeline_config = PipelineConfig(experiments_dir="", experiments_outputs_dir="outputs")
+    pipeline_config.output_file = Path(os.path.join(pipeline_config.experiments_outputs_dir, "output"))
+    pipeline_config.history_file = Path(os.path.join(pipeline_config.experiments_outputs_dir, "history"))
+    pipeline_config.training_history_log_file = Path(os.path.join(pipeline_config.experiments_outputs_dir, "t_history"))
+    pipeline_config.log_file = Path(os.path.join(pipeline_config.experiments_outputs_dir, "log"))
+
+    pipeline_config.output_file.parent.mkdir(parents=True, exist_ok=True)
+    pipeline_config.history_file.parent.mkdir(parents=True, exist_ok=True)
+    pipeline_config.training_history_log_file.parent.mkdir(parents=True, exist_ok=True)
+    pipeline_config.log_file.parent.mkdir(parents=True, exist_ok=True)
+
+    pipeline_config.output_file.touch()
+    pipeline_config.history_file.touch()
+    pipeline_config.training_history_log_file.touch()
+    pipeline_config.log_file.touch()
+
     pipeline_config.logger = set_logger(experiment_mode=experiment_mode,
                                         no_log=False,
                                         log_file=pipeline_config.log_file)