added option to create model deployment for what if analysis

Author: prasankh

ads/opctl/operator/lowcode/common/utils.py

@@ -12,6 +12,7 @@
 
 import fsspec
 import oracledb
+import json
 import pandas as pd
 
 from ads.common.object_storage_details import ObjectStorageDetails
@@ -135,6 +136,15 @@ def write_data(data, filename, format, storage_options=None, index=False, **kwar
     )
 
 
+def write_simple_json(data, path):
+    if ObjectStorageDetails.is_oci_path(path):
+        storage_options = default_signer()
+    else:
+        storage_options = {}
+    with fsspec.open(path, mode="w", **storage_options) as f:
+        json.dump(data, f, indent=4)
+
+
 def merge_category_columns(data, target_category_columns):
     result = data.apply(
         lambda x: "__".join([str(x[col]) for col in target_category_columns]), axis=1

ads/opctl/operator/lowcode/forecast/__main__.py

@@ -33,6 +33,9 @@ def operate(operator_config: ForecastOperatorConfig) -> None:
     if spec.what_if_analysis and datasets.additional_data:
         mdm = ModelDeploymentManager(spec, datasets.additional_data)
         mdm.save_to_catalog()
+        if spec.what_if_analysis.model_deployment:
+            mdm.create_deployment()
+        mdm.save_deployment_info()
 
 
 def verify(spec: Dict, **kwargs: Dict) -> bool:

ads/opctl/operator/lowcode/forecast/operator_config.py

@@ -18,13 +18,34 @@
 
 from .const import SpeedAccuracyMode, SupportedMetrics, SupportedModels
 
+@dataclass
+class AutoScaling(DataClassSerializable):
+    """Class representing simple autoscaling policy"""
+    minimum_instance: int = 1
+    maximum_instance: int = None
+    cool_down_in_seconds: int = 600
+    scale_in_threshold: int = 10
+    scale_out_threshold: int = 80
+    scaling_metric: str = "CPU_UTILIZATION"
+
+@dataclass(repr=True)
+class ModelDeploymentServer(DataClassSerializable):
+    """Class representing model deployment server specification for whatif-analysis."""
+    display_name: str = None
+    initial_shape: str = None
+    description: str = None
+    log_group: str = None
+    log_id: str = None
+    auto_scaling: AutoScaling = field(default_factory=AutoScaling)
+
 
 @dataclass(repr=True)
 class WhatIfAnalysis(DataClassSerializable):
     """Class representing operator specification for whatif-analysis."""
-    model_name: str = None
+    model_display_name: str = None
     compartment_id: str = None
     project_id: str = None
+    model_deployment: ModelDeploymentServer = field(default_factory=ModelDeploymentServer)
 
 
 @dataclass(repr=True)

ads/opctl/operator/lowcode/forecast/schema.yaml

@@ -344,7 +344,52 @@ spec:
       type: dict
       required: false
       schema:
-        model_name:
+        model_deployment:
+          type: dict
+          required: false
+          meta: "If model_deployment_id is not specified, a new model deployment is created; otherwise, the model is linked to the specified model deployment."
+          schema:
+            model_deployment_id:
+              type: string
+              required: false
+            display_name:
+              type: string
+              required: false
+            initial_shape:
+              type: string
+              required: false
+            description:
+              type: string
+              required: false
+            log_group:
+              type: string
+              required: true
+            log_id:
+              type: string
+              required: true
+            auto_scaling:
+              type: dict
+              required: false
+              schema:
+                minimum_instance:
+                  type: integer
+                  required: true
+                maximum_instance:
+                  type: integer
+                  required: true
+                scale_in_threshold:
+                  type: integer
+                  required: true
+                scale_out_threshold:
+                  type: integer
+                  required: true
+                scaling_metric:
+                  type: string
+                  required: true
+                cool_down_in_seconds:
+                  type: integer
+                  required: true
+        model_display_name:
           type: string
           required: true
         project_id:

ads/opctl/operator/lowcode/forecast/whatifserve/deployment_manager.py

@@ -8,17 +8,25 @@
 import shutil
 import sys
 import tempfile
+import oci
 
 import pandas as pd
 from joblib import dump
 
 from ads.opctl import logger
 from ads.common.model_export_util import prepare_generic_model
-from ads.opctl.operator.lowcode.common.utils import write_data, call_pandas_fsspec
-
+from ads.opctl.operator.lowcode.common.utils import write_data, write_simple_json
+from ads.opctl.operator.lowcode.common.utils import default_signer
 from ..model.forecast_datasets import AdditionalData
 from ..operator_config import ForecastOperatorSpec
 
+from oci.data_science import DataScienceClient, DataScienceClientCompositeOperations
+
+from oci.data_science.models import ModelConfigurationDetails, InstanceConfiguration, \
+    FixedSizeScalingPolicy, CategoryLogDetails, LogDetails, \
+    SingleModelDeploymentConfigurationDetails, CreateModelDeploymentDetails
+from ads.common.object_storage_details import ObjectStorageDetails
+
 
 class ModelDeploymentManager:
     def __init__(self, spec: ForecastOperatorSpec, additional_data: AdditionalData, previous_model_version=None):
@@ -27,12 +35,19 @@ def __init__(self, spec: ForecastOperatorSpec, additional_data: AdditionalData,
         self.horizon = spec.horizon
         self.additional_data = additional_data.get_dict_by_series()
         self.model_obj = {}
-        self.display_name = spec.what_if_analysis.model_name
-        self.project_id = spec.what_if_analysis.project_id
-        self.compartment_id = spec.what_if_analysis.compartment_id
+        self.display_name = spec.what_if_analysis.model_display_name
+        self.project_id = spec.what_if_analysis.project_id if spec.what_if_analysis.project_id \
+            else os.environ.get('PROJECT_OCID')
+        self.compartment_id = spec.what_if_analysis.compartment_id if spec.what_if_analysis.compartment_id \
+            else os.environ.get('NB_SESSION_COMPARTMENT_OCID')
+        if self.project_id is None or self.compartment_id is None:
+            raise ValueError("Either project_id or compartment_id cannot be None.")
         self.path_to_artifact = f"{self.spec.output_directory.url}/artifacts/"
         self.pickle_file_path = f"{self.spec.output_directory.url}/model.pkl"
         self.model_version = previous_model_version + 1 if previous_model_version else 1
+        self.catalog_id = None
+        self.test_mode = os.environ.get("TEST_MODE", False)
+        self.deployment_info = {}
 
     def _satiny_test(self):
         """
@@ -51,8 +66,7 @@ def _satiny_test(self):
             date_col_format = self.spec.datetime_column.format
             sample_prediction_data[date_col_name] = sample_prediction_data[date_col_name].dt.strftime(date_col_format)
             sample_prediction_data.to_csv(temp_file.name, index=False)
-            additional_data_uri = "additional_data"
-            input_data = {additional_data_uri: {"url": temp_file.name}}
+            input_data = {"additional_data": {"url": temp_file.name}}
             prediction_test = predict(input_data, _)
             logger.info(f"prediction test completed with result :{prediction_test}")
 
@@ -80,8 +94,11 @@ def save_to_catalog(self):
 
         artifact_dict = {"spec": self.spec.to_dict(), "models": self.model_obj}
         dump(artifact_dict, os.path.join(self.path_to_artifact, "model.joblib"))
-        artifact = prepare_generic_model(self.path_to_artifact, function_artifacts=False, force_overwrite=True,
-                                         data_science_env=True)
+        artifact = prepare_generic_model(
+            self.path_to_artifact,
+            function_artifacts=False,
+            force_overwrite=True,
+            data_science_env=True)
 
         self._copy_score_file()
         self._satiny_test()
@@ -92,29 +109,115 @@ def save_to_catalog(self):
             series = self.additional_data.keys()
         description = f"The object contains {len(series)} {self.model_name} models"
 
-        catalog_id = "None"
-        if not os.environ.get("TEST_MODE", False):
-            catalog_entry = artifact.save(display_name=self.display_name,
-                                          compartment_id=self.compartment_id,
-                                          project_id=self.project_id,
-                                          description=description)
-            catalog_id = catalog_entry.id
-
+        if not self.test_mode:
+            catalog_entry = artifact.save(
+                display_name=self.display_name,
+                compartment_id=self.compartment_id,
+                project_id=self.project_id,
+                description=description)
+            self.catalog_id = catalog_entry.id
 
         logger.info(f"Saved {self.model_name} version-v{self.model_version} to model catalog"
-              f" with catalog id : {catalog_id}")
+                    f" with catalog id : {self.catalog_id}")
 
-        catalog_mapping = {"catalog_id": catalog_id, "series": list(series)}
+        self.deployment_info = {"catalog_id": self.catalog_id, "series": list(series)}
 
-        write_data(
-            data=pd.DataFrame([catalog_mapping]),
-            filename=os.path.join(
-                self.spec.output_directory.url, "model_ids.csv"
-            ),
-            format="csv"
+    def create_deployment(self):
+        """Create a model deployment serving"""
+
+        # create new model deployment
+        initial_shape = self.spec.what_if_analysis.model_deployment.initial_shape
+        name = self.spec.what_if_analysis.model_deployment.display_name
+        description = self.spec.what_if_analysis.model_deployment.description
+        auto_scaling_config = self.spec.what_if_analysis.model_deployment.auto_scaling
+
+        # if auto_scaling_config is defined
+        if auto_scaling_config:
+            scaling_policy = oci.data_science.models.AutoScalingPolicy(
+                policy_type="AUTOSCALING",
+                auto_scaling_policies=[
+                    oci.data_science.models.ThresholdBasedAutoScalingPolicyDetails(
+                        auto_scaling_policy_type="THRESHOLD",
+                        rules=[
+                            oci.data_science.models.PredefinedMetricExpressionRule(
+                                metric_expression_rule_type="PREDEFINED_EXPRESSION",
+                                metric_type=auto_scaling_config.scaling_metric,
+                                scale_in_configuration=oci.data_science.models.PredefinedExpressionThresholdScalingConfiguration(
+                                    scaling_configuration_type="THRESHOLD",
+                                    threshold=auto_scaling_config.scale_in_threshold
+                                ),
+                                scale_out_configuration=oci.data_science.models.PredefinedExpressionThresholdScalingConfiguration(
+                                    scaling_configuration_type="THRESHOLD",
+                                    threshold=auto_scaling_config.scale_out_threshold
+                                )
+                            )],
+                        maximum_instance_count=auto_scaling_config.maximum_instance,
+                        minimum_instance_count=auto_scaling_config.minimum_instance,
+                        initial_instance_count=auto_scaling_config.minimum_instance)],
+                cool_down_in_seconds=auto_scaling_config.cool_down_in_seconds,
+                is_enabled=True)
+            logger.info(f"Using autoscaling {auto_scaling_config.scaling_metric} for creating MD")
+        else:
+            scaling_policy = FixedSizeScalingPolicy(instance_count=1)
+            logger.info("Using fixed size policy for creating MD")
+
+        model_configuration_details_object = ModelConfigurationDetails(
+            model_id=self.catalog_id,
+            instance_configuration=InstanceConfiguration(
+                instance_shape_name=initial_shape),
+            scaling_policy=scaling_policy,
+            bandwidth_mbps=20)
+
+        single_model_config = SingleModelDeploymentConfigurationDetails(
+            deployment_type='SINGLE_MODEL',
+            model_configuration_details=model_configuration_details_object
         )
-        return catalog_id
 
-    def create_deployment(self, deployment_config):
-        """Create a model deployment serving"""
-        pass
+        log_group = self.spec.what_if_analysis.model_deployment.log_group
+        log_id = self.spec.what_if_analysis.model_deployment.log_id
+
+        logs_configuration_details_object = CategoryLogDetails(
+            access=LogDetails(log_group_id=log_group,
+                              log_id=log_id),
+            predict=LogDetails(log_group_id=log_group,
+                               log_id=log_id))
+
+        model_deploy_configuration = CreateModelDeploymentDetails(
+            display_name=name,
+            description=description,
+            project_id=self.project_id,
+            compartment_id=self.compartment_id,
+            model_deployment_configuration_details=single_model_config,
+            category_log_details=logs_configuration_details_object)
+
+        if not self.test_mode:
+            auth = oci.auth.signers.get_resource_principals_signer()
+            data_science = DataScienceClient({}, signer=auth)
+            data_science_composite = DataScienceClientCompositeOperations(data_science)
+            model_deployment = data_science_composite.create_model_deployment_and_wait_for_state(
+                model_deploy_configuration,
+                wait_for_states=[
+                    "SUCCEEDED", "FAILED"])
+            self.deployment_info['model_deployment_id'] = model_deployment.data.id
+            logger.info(f"deployment metadata :{model_deployment.data}")
+            md = data_science.get_model_deployment(model_deployment_id=model_deployment.data.resources[0].identifier)
+            endpoint_url = md.data.model_deployment_url
+            self.deployment_info['model_deployment_endpoint'] = f"{endpoint_url}/predict"
+
+    def save_deployment_info(self):
+        output_dir = self.spec.output_directory.url
+        if ObjectStorageDetails.is_oci_path(output_dir):
+            storage_options = default_signer()
+        else:
+            storage_options = {}
+        write_data(
+            data=pd.DataFrame.from_dict(self.deployment_info),
+            filename=os.path.join(output_dir, "deployment_info.json"),
+            format="json",
+            storage_options=storage_options,
+            index=False,
+            indent=4,
+            orient="records"
+        )
+        write_simple_json(self.deployment_info, os.path.join(output_dir, "deployment_info.json"))
+        logger.info(f"Saved deployment info to {output_dir}")

ads/opctl/operator/lowcode/forecast/whatifserve/score.py

@@ -163,7 +163,7 @@ def get_forecast(future_df, model_name, series_id, model_object, date_col, targe
     elif model_name == NEURALPROPHET and series_id in model_object:
         model = model_object[series_id]
         model.restore_trainer()
-        accepted_regressors = list(model.config_regressors.keys())
+        accepted_regressors = list(model.config_regressors.regressors.keys())
         data = future_df.rename(columns={date_col_name: 'ds', target_column: 'y'})
         future = data[accepted_regressors + ["ds"]].reset_index(drop=True)
         future["y"] = None