Prints for frontend, deps for Docker; [skip ci]

Sebastijan-Dominis · Sebastijan-Dominis · commit 94cb3fca87dc · 2026-03-27T15:27:42.000+01:00
Added some prints for the frontend to show the user where the artifacts
are being saved. Updated requirements.txt, since Docker needed it.
Updated docker-compose.yml to have the used directories as volumes.
Some of them were missing earlier.
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -19,6 +19,11 @@ services:
       - ./experiments:/app/experiments
       - ./failure_management:/app/failure_management
       - ./orchestration_logs:/app/orchestration_logs
+      - ./model_registry:/app/model_registry
+      - ./predictions:/app/predictions
+      - ./monitoring:/app/monitoring
+      - ./docs:/app/docs
+      - ./assets:/app/assets
     ports:
       - "8000:8000"
       - "8050:8050"
diff --git a/ml/data/utils/persistence/save_data.py b/ml/data/utils/persistence/save_data.py
@@ -40,6 +40,9 @@ def save_data(df: pd.DataFrame, *, config: InterimConfig | ProcessedConfig, data
     try:
         if save_func == pd.DataFrame.to_parquet:
             save_func(df, data_path, compression=compression)
+            msg = f"Data saved successfully at {data_path} in parquet format with compression {compression}."
+            logger.info(msg)
+            print(msg)
         return data_path
     except Exception as e:
         msg = f"Error saving data to {data_path} with format {config.data.output.format} and compression {compression}. "
diff --git a/ml/feature_freezing/freeze_strategies/tabular/persistence.py b/ml/feature_freezing/freeze_strategies/tabular/persistence.py
@@ -67,6 +67,10 @@ def persist_feature_snapshot(
         compression=config.storage.compression
     )
 
+    msg = f"Persisted feature set snapshot to {data_path} using format {config.storage.format}"
+    logger.info(msg)
+    print(msg)
+
     return path, data_path
 
 def save_input_schema(path: Path, features: pd.DataFrame):
@@ -93,7 +97,9 @@ def save_input_schema(path: Path, features: pd.DataFrame):
     })
 
     schema.to_csv(schema_path, index=False)
-    logger.info(f"Input schema saved to {schema_path}")
+    msg = f"Input schema saved to {schema_path}"
+    logger.info(msg)
+    print(msg)
 
 def save_derived_schema(
     path: Path,
@@ -137,7 +143,9 @@ def save_derived_schema(
 
     derived_schema = pd.DataFrame(derived_features)
     derived_schema.to_csv(schema_path, index=False)
-    logger.info(f"Derived schema saved to {schema_path}")
+    msg = f"Derived schema saved to {schema_path}"
+    logger.info(msg)
+    print(msg)
 
 def create_metadata(
     *,
diff --git a/ml/io/persistence/save_metadata.py b/ml/io/persistence/save_metadata.py
@@ -40,6 +40,7 @@ def save_metadata(
     """
 
     metadata_file = target_dir / "metadata.json"
+    metadata_file = Path(metadata_file.as_posix())
 
     metadata_file.parent.mkdir(parents=True, exist_ok=True)
 
@@ -66,14 +67,17 @@ def save_metadata(
             os.fsync(tmp_file.fileno())
 
         os.replace(temp_path, metadata_file)
-        logger.info(f"Metadata successfully saved to {metadata_file}.")
+
+        msg = f"Metadata successfully saved to {metadata_file.as_posix()}."
+        logger.info(msg)
+        print(msg)
     except Exception as e:
         if temp_path and Path(temp_path).exists():
             try:
                 Path(temp_path).unlink()
             except OSError:
                 logger.warning("Failed to clean up temporary metadata file: %s", temp_path)
 
-        msg = f"Failed to save metadata to {metadata_file}"
+        msg = f"Failed to save metadata to {metadata_file.as_posix()}"
         logger.exception(msg)
         raise PersistenceError(msg) from e
diff --git a/ml/post_promotion/inference/persistence/store_predictions.py b/ml/post_promotion/inference/persistence/store_predictions.py
@@ -88,7 +88,9 @@ def store_predictions(
     pq.write_table(pa.Table.from_pandas(df), tmp_path)
     tmp_path.rename(file_path)
 
-    logger.info(f"Stored predictions at {file_path}")
+    msg = f"Predictions successfully stored at {file_path} with columns: {cols}"
+    logger.info(msg)
+    print(msg)
 
     return PredictionStoringReturn(
         file_path=file_path,
diff --git a/ml/promotion/persistence/registry.py b/ml/promotion/persistence/registry.py
@@ -77,6 +77,12 @@ def update_registry_and_archive(
             yaml.safe_dump(new_registry, f, sort_keys=False)
         os.replace(temp_registry_path, registry_path)
 
+        msg = f"Model registry successfully updated at {registry_path}."
+        if new_archive is not None:
+            msg += f" Previous production model archived at {archive_path}."
+        logger.info(msg)
+        print(msg)
+
         return new_registry
     except Exception as e:
         msg = f"Failed to update model registry and archive. Run info: {run_info}"
@@ -108,6 +114,9 @@ def persist_registry_diff(
     try:
         with open(diff_path, "w", encoding="utf-8") as f:
             yaml.safe_dump(diff, f, sort_keys=False)
+        msg = f"Registry diff successfully saved to {diff_path}."
+        logger.info(msg)
+        print(msg)
     except Exception as e:
         msg = f"Failed to persist registry diff to {diff_path}"
         logger.exception(msg)
diff --git a/ml/runners/evaluation/persistence/save_predictions.py b/ml/runners/evaluation/persistence/save_predictions.py
@@ -57,17 +57,23 @@ def save_predictions(prediction_dfs: PredictionArtifacts, target_dir: Path) -> P
     try:
         train_path = target_dir / "predictions_train.parquet"
         _write_parquet_atomic(prediction_dfs.train, train_path)
-        logger.info(f"Saved predictions for the train split to {train_path}")
+        msg = f"Saved predictions for the train split to {train_path}"
+        logger.info(msg)
+        print(msg)
         paths_raw["train_predictions_path"] = train_path.as_posix()
 
         val_path = target_dir / "predictions_val.parquet"
         _write_parquet_atomic(prediction_dfs.val, val_path)
-        logger.info(f"Saved predictions for the validation split to {val_path}")
+        msg = f"Saved predictions for the validation split to {val_path}"
+        logger.info(msg)
+        print(msg)
         paths_raw["val_predictions_path"] = val_path.as_posix()
 
         test_path = target_dir / "predictions_test.parquet"
         _write_parquet_atomic(prediction_dfs.test, test_path)
-        logger.info(f"Saved predictions for the test split to {test_path}")
+        msg = f"Saved predictions for the test split to {test_path}"
+        logger.info(msg)
+        print(msg)
         paths_raw["test_predictions_path"] = test_path.as_posix()
 
         paths = PredictionsPaths(**paths_raw)
diff --git a/ml/runners/explainability/persistence/save_metrics_csv.py b/ml/runners/explainability/persistence/save_metrics_csv.py
@@ -46,7 +46,10 @@ def save_metrics_csv(
             os.fsync(tmp_file.fileno())
 
         os.replace(temp_path, target_file)
-        logger.info(f'{name} successfully saved to {target_file}.')
+
+        msg = f"{name} successfully saved to {target_file}."
+        logger.info(msg)
+        print(msg)
     except Exception as e:
         if temp_path and Path(temp_path).exists():
             try:
diff --git a/ml/runners/shared/persistence/save_metrics.py b/ml/runners/shared/persistence/save_metrics.py
@@ -60,7 +60,10 @@ def save_metrics(metrics: dict[str, float] | dict[str, dict[str, float]], *, mod
             os.fsync(tmp_file.fileno())
 
         os.replace(temp_path, metrics_file)
-        logger.info(f"Metrics successfully saved to {metrics_file}.")
+
+        msg = f"Metrics successfully saved to {metrics_file}."
+        logger.info(msg)
+        print(msg)
         return str(metrics_file)
     except Exception as e:
         if temp_path and Path(temp_path).exists():
diff --git a/ml/runners/training/persistence/artifacts/save_model.py b/ml/runners/training/persistence/artifacts/save_model.py
@@ -38,7 +38,10 @@ def save_model(model, path: Path) -> Path:
 
         joblib.dump(model, temp_path)
         os.replace(temp_path, model_file)
-        logger.info(f"Model successfully saved to {model_file}.")
+
+        msg = f"Model successfully saved to {model_file}."
+        logger.info(msg)
+        print(msg)
         return model_file
     except Exception as e:
         if temp_path and temp_path.exists():
diff --git a/ml/runners/training/persistence/artifacts/save_pipeline.py b/ml/runners/training/persistence/artifacts/save_pipeline.py
@@ -40,7 +40,10 @@ def save_pipeline(pipeline: Pipeline, path: Path) -> Path:
 
         joblib.dump(pipeline, temp_path)
         os.replace(temp_path, pipeline_file)
-        logger.info(f"Pipeline successfully saved to {pipeline_file}.")
+
+        msg = f"Pipeline successfully saved to {pipeline_file}."
+        logger.info(msg)
+        print(msg)
         return pipeline_file
     except Exception as e:
         if temp_path and temp_path.exists():
diff --git a/ml/utils/runtime/save_runtime.py b/ml/utils/runtime/save_runtime.py
@@ -49,7 +49,9 @@ def save_runtime_snapshot(
             os.fsync(tmp_file.fileno())
 
         os.replace(temp_path, snapshot_path)
-        logger.info("Runtime snapshot successfully saved to %s", snapshot_path)
+        msg = f"Runtime snapshot successfully saved to {snapshot_path}"
+        logger.info(msg)
+        print(msg)
     except Exception as e:
         if temp_path and Path(temp_path).exists():
             try:
diff --git a/pipelines/orchestration/data/execute_all_data_preprocessing.py b/pipelines/orchestration/data/execute_all_data_preprocessing.py
@@ -126,7 +126,9 @@ def main() -> int:
 
                         logger.info(f"Starting register_raw_snapshot for {data_name} {data_version} snapshot {snapshot_id}")
                         run_cmd(cmd)
-                        logger.info(f"Completed register_raw_snapshot for {data_name} {data_version} snapshot {snapshot_id}")
+                        msg = f"Completed register_raw_snapshot for {data_name} {data_version} snapshot {snapshot_id}"
+                        logger.info(msg)
+                        print(msg)
 
         interim_config_root = Path("configs/data/interim")
 
@@ -165,7 +167,9 @@ def main() -> int:
 
                     logger.info(f"Starting build_interim_dataset for {data_name} {version}")
                     run_cmd(cmd)
-                    logger.info(f"Completed build_interim_dataset for {data_name} {version}")
+                    msg = f"Completed build_interim_dataset for {data_name} {version}"
+                    logger.info(msg)
+                    print(msg)
 
         processed_config_root = Path("configs/data/processed")
 
@@ -204,7 +208,9 @@ def main() -> int:
 
                     logger.info(f"Starting build_processed_dataset for {data_name} {version}")
                     run_cmd(cmd)
-                    logger.info(f"Completed build_processed_dataset for {data_name} {version}")
+                    msg = f"Completed build_processed_dataset for {data_name} {version}"
+                    logger.info(msg)
+                    print(msg)
 
         log_completion(start_time=start_time, message="Full data preprocessing run completed successfully.")
         return 0
diff --git a/pipelines/orchestration/experiments/execute_all_experiments_with_latest.py b/pipelines/orchestration/experiments/execute_all_experiments_with_latest.py
@@ -18,7 +18,6 @@
 from ml.io.formatting.iso_no_colon import iso_no_colon
 from ml.io.formatting.str_to_bool import str_to_bool
 from ml.logging_config import setup_logging
-
 from pipelines.orchestration.common.orchestration_logging import log_completion
 
 MODEL_SPECS_DIR = Path("configs/model_specs")
@@ -164,7 +163,9 @@ def run_model(
     if result.returncode != 0:
         log_completion(start_time, f"Experiments run failed with model problem={problem}, segment={segment}, version={version} with return code {result.returncode}")
     else:
-        log_completion(model_start_time, f"Experiment for model problem={problem}, segment={segment}, version={version} completed successfully")
+        msg = f"Experiment for model problem={problem}, segment={segment}, version={version} completed successfully"
+        print(msg)
+        log_completion(model_start_time, msg)
     return result.returncode
 
 def main() -> int:
diff --git a/pipelines/orchestration/features/freeze_all_feature_sets.py b/pipelines/orchestration/features/freeze_all_feature_sets.py
@@ -133,7 +133,9 @@ def main() -> int:
             try:
                 logger.info(f"Freezing feature set '{feature_set_name}' version '{feature_set_version}'...")
                 subprocess.run(cmd, check=True, capture_output=True, text=True, encoding="utf-8")
-                logger.info(f"Feature set '{feature_set_name}' succeeded.")
+                msg = f"Feature set '{feature_set_name}' succeeded."
+                logger.info(msg)
+                print(msg)
                 successes_count += 1
             except subprocess.CalledProcessError as e:
                 first_line = e.stderr.splitlines()[0] if e.stderr else ""
diff --git a/pipelines/orchestration/master/run_all_workflows.py b/pipelines/orchestration/master/run_all_workflows.py
@@ -146,7 +146,9 @@ def main() -> int:
             log_completion(start_time, f"Run all workflows failed at step: {step_name}")
             return code
 
-    log_completion(start_time, "Run all workflows completed successfully.")
+    msg = "Run all workflows completed successfully."
+    log_completion(start_time, msg)
+    print(msg)
     return 0
 
 
diff --git a/requirements.txt b/requirements.txt
@@ -5,6 +5,7 @@ dash-bootstrap-components==2.0.4
 fastapi==0.135.1
 uvicorn==0.41.0
 python-dotenv==1.2.2
+networkx==3.6.1
 
 # NOTE: sdv is only needed if you want to generate fake data. If you don't need that, you can safely ignore this dependency and the related code in the project - scripts/generators/generate_fake_data.py.
 # sdv==1.35.0
