change based on the comments

Author: Ziqun Ye

ads/opctl/backend/base.py

@@ -8,22 +8,24 @@
 from typing import Dict
 
 from ads.common.auth import create_signer
+from ads.common.oci_client import OCIClientFactory
 
 
 class Backend:
     """Interface for backend"""
 
     def __init__(self, config: Dict) -> None:
-        
         self.config = config
-        self.oci_auth = create_signer(
-            config["execution"].get("auth"),
-            config["execution"].get("oci_config", None),
-            config["execution"].get("oci_profile", None),
-        )
         self.auth_type = config["execution"].get("auth")
         self.profile = config["execution"].get("oci_profile", None)
+        self.oci_config = config["execution"].get("oci_config", None)
 
+        self.oci_auth = create_signer(
+            self.auth_type,
+            self.oci_config,
+            self.profile,
+        )
+        self.client = OCIClientFactory(**self.oci_auth).data_science
 
     @abstractmethod
     def run(self) -> Dict:
@@ -98,10 +100,10 @@ def run_diagnostics(self):
 
     def predict(self) -> None:
         """
-        Deactivate a remote service.
+        Run model predict.
 
         Returns
         -------
         None
         """
-        raise NotImplementedError("`predict` has not been implemented yet.")
+        raise NotImplementedError("`predict` has not been implemented yet.")

ads/opctl/backend/local.py

@@ -15,33 +15,45 @@
 from oci.data_science.models import PipelineStepRun
 
 from ads.common.auth import create_signer
-from ads.common.decorator.runtime_dependency import (OptionalDependency,
-                                                     runtime_dependency)
-from ads.common.oci_client import OCIClientFactory
+from ads.common.decorator.runtime_dependency import (
+    OptionalDependency,
+    runtime_dependency,
+)
+
 from ads.model.model_metadata import ModelCustomMetadata
 from ads.model.runtime.runtime_info import RuntimeInfo
 from ads.opctl import logger
 from ads.opctl.backend.base import Backend
 from ads.opctl.conda.cmds import _install
 from ads.opctl.config.resolver import ConfigResolver
-from ads.opctl.constants import (DEFAULT_IMAGE_CONDA_DIR,
-                                 DEFAULT_IMAGE_HOME_DIR,
-                                 DEFAULT_IMAGE_SCRIPT_DIR,
-                                 DEFAULT_MODEL_FOLDER,
-                                 DEFAULT_NOTEBOOK_SESSION_CONDA_DIR,
-                                 DEFAULT_NOTEBOOK_SESSION_SPARK_CONF_DIR,
-                                 ML_JOB_GPU_IMAGE, ML_JOB_IMAGE)
+from ads.opctl.constants import (
+    DEFAULT_IMAGE_CONDA_DIR,
+    DEFAULT_IMAGE_HOME_DIR,
+    DEFAULT_IMAGE_SCRIPT_DIR,
+    DEFAULT_MODEL_FOLDER,
+    DEFAULT_NOTEBOOK_SESSION_CONDA_DIR,
+    DEFAULT_NOTEBOOK_SESSION_SPARK_CONF_DIR,
+    ML_JOB_GPU_IMAGE,
+    ML_JOB_IMAGE,
+    DEFAULT_MODEL_DEPLOYMENT_FOLDER,
+)
 from ads.opctl.distributed.cmds import load_ini, local_run
 from ads.opctl.model.cmds import _download_model
-from ads.opctl.spark.cmds import (generate_core_site_properties,
-                                  generate_core_site_properties_str)
-from ads.opctl.utils import (build_image, get_docker_client,
-                             is_in_notebook_session, run_command,
-                             run_container)
+from ads.opctl.spark.cmds import (
+    generate_core_site_properties,
+    generate_core_site_properties_str,
+)
+from ads.opctl.utils import (
+    build_image,
+    get_docker_client,
+    is_in_notebook_session,
+    run_command,
+    run_container,
+)
 from ads.pipeline.ads_pipeline import Pipeline, PipelineStep
 
 
-class CondaPackNotFound(Exception):   # pragma: no cover
+class CondaPackNotFound(Exception):  # pragma: no cover
     pass
 
 
@@ -55,17 +67,7 @@ def __init__(self, config: Dict) -> None:
         config: dict
             dictionary of configurations
         """
-        self.config = config
-        self.auth_type = config["execution"].get("auth")
-        self.profile = config["execution"].get("oci_profile", None)
-        self.oci_config = config["execution"].get("oci_config", None)
-        
-        self.oci_auth = create_signer(
-            self.auth_type,
-            self.oci_config,
-            self.profile ,
-        )
-        self.client = OCIClientFactory(**self.oci_auth).data_science
+        super().__init__(config=config)
 
     def run(self):
         if self.config.get("version") == "v1.0":
@@ -190,7 +192,13 @@ def init_vscode_container(self) -> None:
                 f.write(json.dumps(dev_container, indent=2))
             print(f"File {os.path.abspath('.devcontainer.json')} created.")
 
-    def _run_with_conda_pack(self, bind_volumes: Dict, extra_cmd: str="", install: bool=False, conda_uri: str="") -> int:
+    def _run_with_conda_pack(
+        self,
+        bind_volumes: Dict,
+        extra_cmd: str = "",
+        install: bool = False,
+        conda_uri: str = "",
+    ) -> int:
         env_vars = self.config["execution"].get("env_vars", {})
         slug = self.config["execution"]["conda_slug"]
         image = self.config["execution"].get("image", None)
@@ -214,7 +222,7 @@ def _run_with_conda_pack(self, bind_volumes: Dict, extra_cmd: str="", install: b
                 image, slug, command, bind_volumes, env_vars
             )
 
-    def _build_command_for_conda_run(self, extra_cmd: str="") -> str:
+    def _build_command_for_conda_run(self, extra_cmd: str = "") -> str:
         if ConfigResolver(self.config)._is_ads_operator():
             if is_in_notebook_session():
                 curr_dir = os.path.dirname(os.path.abspath(__file__))
@@ -285,7 +293,7 @@ def _run_with_image(self, bind_volumes: Dict) -> int:
         if self.config["execution"].get("source_folder", None):
             bind_volumes.update(self._mount_source_folder_if_exists(bind_volumes))
         bind_volumes.update(self.config["execution"]["volumes"])
-        
+
         return run_container(image, bind_volumes, env_vars, command, entrypoint)
 
     def _run_with_image_v1(self, bind_volumes: Dict) -> int:
@@ -308,15 +316,22 @@ def _run_with_image_v1(self, bind_volumes: Dict) -> int:
         )
 
     def _check_conda_pack_and_install_if_applicable(
-        self, slug: str, bind_volumes: Dict, env_vars: Dict, install: bool=False, conda_uri: str = None
+        self,
+        slug: str,
+        bind_volumes: Dict,
+        env_vars: Dict,
+        install: bool = False,
+        conda_uri: str = None,
     ) -> Dict:
-        conda_pack_folder = os.path.abspath(os.path.expanduser(self.config['execution']["conda_pack_folder"]))
-        conda_pack_path = os.path.join(
-            conda_pack_folder, slug
+        conda_pack_folder = os.path.abspath(
+            os.path.expanduser(self.config["execution"]["conda_pack_folder"])
         )
+        conda_pack_path = os.path.join(conda_pack_folder, slug)
         if not os.path.exists(conda_pack_path):
             if install:
-                logger.info(f"Downloading the conda pack {slug} to this conda pack {conda_pack_folder}. If this conda pack is already installed locally in a different location, pass in `conda_pack_folder` to avoid downloading it again.")
+                logger.info(
+                    f"Downloading a `{slug}` to the `{conda_pack_folder}`. If this conda pack is already installed locally in a different location, pass in `conda_pack_folder` to avoid downloading it again."
+                )
                 _install(
                     conda_uri=conda_uri,
                     conda_pack_folder=conda_pack_folder,
@@ -648,7 +663,7 @@ def __init__(self, config: Dict) -> None:
             dictionary of configurations
         """
         super().__init__(config)
-        
+
     def predict(self) -> None:
         """
         Conducts local verify.
@@ -661,29 +676,63 @@ def predict(self) -> None:
         artifact_directory = self.config["execution"].get("artifact_directory")
         ocid = self.config["execution"].get("ocid")
         data = self.config["execution"].get("payload")
-        model_folder = os.path.expanduser(self.config["execution"].get("model_save_folder", DEFAULT_MODEL_FOLDER))
+        model_folder = os.path.expanduser(
+            self.config["execution"].get("model_save_folder", DEFAULT_MODEL_FOLDER)
+        )
         artifact_directory = artifact_directory or os.path.join(model_folder, str(ocid))
-        if ocid and (not os.path.exists(artifact_directory) or len(os.listdir(artifact_directory)) == 0):
+        if ocid and (
+            not os.path.exists(artifact_directory)
+            or len(os.listdir(artifact_directory)) == 0
+        ):
             region = self.config["execution"].get("region", None)
             bucket_uri = self.config["execution"].get("bucket_uri", None)
             timeout = self.config["execution"].get("timeout", None)
-            logger.info(f"No cached model found. Downloading the model {ocid} to {artifact_directory}. If you already have a copy of the model, specify `artifact_directory` instead of `ocid`. You can specify `model_save_folder` to decide where to store the model artifacts.")
-            
-            _download_model(ocid=ocid, artifact_directory=artifact_directory, region=region, bucket_uri=bucket_uri, timeout=timeout)
+            logger.info(
+                f"No cached model found. Downloading the model {ocid} to {artifact_directory}. If you already have a copy of the model, specify `artifact_directory` instead of `ocid`. You can specify `model_save_folder` to decide where to store the model artifacts."
+            )
 
+            _download_model(
+                oci_auth=self.oci_auth,
+                ocid=ocid,
+                artifact_directory=artifact_directory,
+                region=region,
+                bucket_uri=bucket_uri,
+                timeout=timeout,
+                force_overwrite=True,
+            )
+        conda_slug, conda_path = None, None
         if ocid:
             conda_slug, conda_path = self._get_conda_info_from_custom_metadata(ocid)
         if not conda_path:
-            if not os.path.exists(artifact_directory) or len(os.listdir(artifact_directory)) == 0:
-                raise ValueError(f"`artifact_directory` {artifact_directory} does not exist or is empty.")
-            conda_slug, conda_path = self._get_conda_info_from_runtime(artifact_dir=artifact_directory)
+            if (
+                not os.path.exists(artifact_directory)
+                or len(os.listdir(artifact_directory)) == 0
+            ):
+                raise ValueError(
+                    f"`artifact_directory` {artifact_directory} does not exist or is empty."
+                )
+            conda_slug, conda_path = self._get_conda_info_from_runtime(
+                artifact_dir=artifact_directory
+            )
         if not conda_path or not conda_slug:
             raise ValueError("Conda information cannot be detected.")
-        compartment_id = self.config["execution"].get("compartment_id", self.config["infrastructure"].get("compartment_id"))
-        project_id = self.config["execution"].get("project_id", self.config["infrastructure"].get("project_id"))
+        compartment_id = self.config["execution"].get(
+            "compartment_id", self.config["infrastructure"].get("compartment_id")
+        )
+        project_id = self.config["execution"].get(
+            "project_id", self.config["infrastructure"].get("project_id")
+        )
         if not compartment_id or not project_id:
             raise ValueError("`compartment_id` and `project_id` must be provided.")
-        extra_cmd = "/opt/ds/model/deployed_model/ " + data + " " + compartment_id + " " + project_id
+        extra_cmd = (
+            DEFAULT_MODEL_DEPLOYMENT_FOLDER
+            + " "
+            + data
+            + " "
+            + compartment_id
+            + " "
+            + project_id
+        )
         bind_volumes = {}
         if not is_in_notebook_session():
             bind_volumes = {
@@ -693,16 +742,20 @@ def predict(self) -> None:
             }
             dir_path = os.path.dirname(os.path.realpath(__file__))
             script = "script.py"
-            self.config["execution"]["source_folder"] = os.path.abspath(os.path.join(dir_path, ".."))
+            self.config["execution"]["source_folder"] = os.path.abspath(
+                os.path.join(dir_path, "..")
+            )
             self.config["execution"]["entrypoint"] = script
-            bind_volumes[artifact_directory] = {"bind": "/opt/ds/model/deployed_model/"}
+            bind_volumes[artifact_directory] = {"bind": DEFAULT_MODEL_DEPLOYMENT_FOLDER}
         if self.config["execution"].get("conda_slug", conda_slug):
             self.config["execution"]["image"] = ML_JOB_IMAGE
             if not self.config["execution"].get("conda_slug"):
                 self.config["execution"]["conda_slug"] = conda_slug
                 self.config["execution"]["slug"] = conda_slug
             self.config["execution"]["conda_path"] = conda_path
-            exit_code = self._run_with_conda_pack(bind_volumes, extra_cmd, install=True, conda_uri=conda_path)
+            exit_code = self._run_with_conda_pack(
+                bind_volumes, extra_cmd, install=True, conda_uri=conda_path
+            )
         else:
             raise ValueError("Either conda pack info or image should be specified.")
 
@@ -722,14 +775,16 @@ def _get_conda_info_from_custom_metadata(self, ocid):
             conda slug and conda path.
         """
         response = self.client.get_model(ocid)
-        custom_metadata = ModelCustomMetadata._from_oci_metadata(response.data.custom_metadata_list)
+        custom_metadata = ModelCustomMetadata._from_oci_metadata(
+            response.data.custom_metadata_list
+        )
         conda_slug, conda_path = None, None
         if "CondaEnvironmentPath" in custom_metadata.keys:
-            conda_path = custom_metadata['CondaEnvironmentPath'].value
+            conda_path = custom_metadata["CondaEnvironmentPath"].value
         if "SlugName" in custom_metadata.keys:
-            conda_slug = custom_metadata['SlugName'].value
+            conda_slug = custom_metadata["SlugName"].value
         return conda_slug, conda_path
-    
+
     @staticmethod
     def _get_conda_info_from_runtime(artifact_dir):
         """
@@ -742,7 +797,10 @@ def _get_conda_info_from_runtime(artifact_dir):
         """
         runtime_yaml_file = os.path.join(artifact_dir, "runtime.yaml")
         runtime_info = RuntimeInfo.from_yaml(uri=runtime_yaml_file)
-        conda_slug = runtime_info.model_deployment.inference_conda_env.inference_env_slug
-        conda_path = runtime_info.model_deployment.inference_conda_env.inference_env_path
+        conda_slug = (
+            runtime_info.model_deployment.inference_conda_env.inference_env_slug
+        )
+        conda_path = (
+            runtime_info.model_deployment.inference_conda_env.inference_env_path
+        )
         return conda_slug, conda_path
-