Feature/eval pipeline deploy

ml_pipelines/deployment/local/common.py

@@ -91,10 +91,27 @@ def save_eval_artifacts(
     plots.savefig(str(version_dir / "calibration_plot.png"))
 
 
+def get_all_available_train_versions(root_path: os.PathLike | str):
+    root_dir = Path(root_path)
+    return [d.stem for d in root_dir.iterdir() if d.is_dir()]
+
+
 def get_latest_version(root_path: os.PathLike, filename: str) -> str:
     root_dir = Path(root_path)
     versions: list[tuple[str, float]] = []
     for version_dir in root_dir.iterdir():
         st_mtime = (version_dir / filename).stat().st_mtime
         versions.append((version_dir.stem, st_mtime))
     return max(versions, key=lambda t: t[1])[0]
+
+
+def get_best_version(train_artifacts_root_path: os.PathLike):
+    train_dir = Path(train_artifacts_root_path)
+    with open(train_dir / "best_model") as f:
+        return f.read()
+
+
+def tag_best_version(train_version: str, train_artifacts_root_path: os.PathLike):
+    train_dir = Path(train_artifacts_root_path)
+    with open(train_dir / "best_model", "w") as f:
+        f.write(train_version)

ml_pipelines/deployment/local/eval.py

@@ -0,0 +1,81 @@
+import logging
+import os
+import sys
+from logging import Logger
+from typing import Union
+
+import typer
+
+from ml_pipelines.deployment.local.common import (
+    get_all_available_train_versions,
+    get_latest_version,
+    get_raw_data,
+    get_train_artifacts,
+    tag_best_version,
+)
+from ml_pipelines.pipeline.eval_pipeline import eval_pipeline
+
+
+def run_eval_comparison_pipeline(  # noqa: PLR0913
+    raw_data_version: str,
+    raw_data_root_path: os.PathLike,
+    train_versions: list[str],
+    train_artifacts_root_path: os.PathLike,
+    logger: Logger,
+):
+    logger.info(f"Running eval pipeline on model versions: {train_versions}.")
+    logger.info(f"Raw data version {raw_data_version}.")
+    raw_data = get_raw_data(raw_data_version, raw_data_root_path)
+    all_metrics = []
+    for v in train_versions:
+        train_artifacts = get_train_artifacts(
+            v, train_artifacts_root_path, load_data=False
+        )
+        metrics, _ = eval_pipeline(
+            train_artifacts["model"],
+            train_artifacts["feature_eng_params"],
+            raw_data,
+            logger,
+        )
+        all_metrics.append((v, metrics))
+    best_version = max(all_metrics, key=lambda t: t[1])[0]
+    logger.info(f"Tagging best version as {best_version}")
+    tag_best_version(best_version, train_artifacts_root_path)
+
+
+if __name__ == "__main__":
+    from dotenv import load_dotenv
+
+    load_dotenv()
+    RAW_DATA_ROOT_DIR = os.environ["RAW_DATA_ROOT_DIR"]
+    TRAIN_ARTIFACTS_ROOT_DIR = os.environ["TRAIN_ARTIFACTS_ROOT_DIR"]
+
+    def main(
+        raw_data_version: Union[str, None] = None,  # noqa: UP007
+        train_versions: Union[list[str], None] = None,  # noqa: UP007
+        raw_data_root_path: str = RAW_DATA_ROOT_DIR,
+        train_artifacts_root_path: str = TRAIN_ARTIFACTS_ROOT_DIR,
+    ):
+        logger = Logger(__file__)
+        logger.addHandler(logging.StreamHandler(sys.stdout))
+
+        if raw_data_version is None:
+            raw_data_version = get_latest_version(
+                raw_data_root_path,  # type: ignore
+                "raw_data.csv",
+            )
+
+        if not train_versions:
+            train_versions = get_all_available_train_versions(  # type: ignore
+                train_artifacts_root_path
+            )
+
+        run_eval_comparison_pipeline(  # noqa: PLR0913
+            raw_data_version=raw_data_version,
+            raw_data_root_path=raw_data_root_path,  # type: ignore
+            train_versions=train_versions,  # type: ignore
+            train_artifacts_root_path=train_artifacts_root_path,  # type: ignore
+            logger=logger,
+        )
+
+    typer.run(main)

ml_pipelines/deployment/local/serve.py

@@ -7,7 +7,10 @@
 import typer
 import uvicorn
 
-from ml_pipelines.deployment.local.common import get_latest_version, get_train_artifacts
+from ml_pipelines.deployment.local.common import (
+    get_best_version,
+    get_train_artifacts,
+)
 from ml_pipelines.logic.serve.serve import Point, create_fastapi_app
 
 
@@ -48,10 +51,11 @@ def main(
         logger.addHandler(logging.StreamHandler(sys.stdout))
 
         if train_version is None:
-            train_version = get_latest_version(
-                train_artifacts_root_path,  # type: ignore
-                "model.pickle",
-            )
+            train_version = get_best_version(train_artifacts_root_path)  # type: ignore
+            # train_version = get_latest_version(
+            #     train_artifacts_root_path,  # type: ignore
+            #     "model.pickle",
+            # )
 
         uvicorn_kwargs: dict = {}
         run_serve(  # noqa: PLR0913

ml_pipelines/deployment/local/train.py

@@ -36,7 +36,7 @@ def run_train_pipeline(  # noqa: PLR0913
     metrics, plots = eval_pipeline(
         train_artifacts["model"],
         train_artifacts["feature_eng_params"],
-        train_artifacts["raw_test_data"],
+        train_artifacts["raw_test_data"],  # type: ignore
         logger,
     )
     save_eval_artifacts(train_version, train_artifacts_root_path, metrics, plots)