minor updates and refinements

Author: prasankh

ads/opctl/operator/lowcode/forecast/model/forecast_datasets.py

@@ -167,8 +167,7 @@ def get_data_multi_indexed(self):
 self.historical_data.data,
 self.additional_data.data,
 ],
-axis=1,
-join='inner'
+axis=1
 )
 
 def get_data_by_series(self, include_horizon=True):

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

@@ -348,9 +348,9 @@ spec:
 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."
+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:
+id:
 type: string
 required: false
 display_name:

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

@@ -53,22 +53,30 @@ def _sanity_test(self):
 """
 Function perform sanity test for saved artifact
 """
-sys.path.insert(0, f"{self.path_to_artifact}")
-from score import load_model, predict
-_ = load_model()
-
-# Write additional data to tmp file and perform sanity check
-with tempfile.NamedTemporaryFile(suffix='.csv') as temp_file:
-one_series = next(iter(self.additional_data))
-sample_prediction_data = self.additional_data[one_series].tail(self.horizon)
-sample_prediction_data[self.spec.target_category_columns[0]] = one_series
-date_col_name = self.spec.datetime_column.name
-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)
-input_data = {"additional_data": {"url": temp_file.name}}
-prediction_test = predict(input_data, _)
-logger.info(f"prediction test completed with result :{prediction_test}")
+org_sys_path = sys.path[:]
+try:
+sys.path.insert(0, f"{self.path_to_artifact}")
+from score import load_model, predict
+_ = load_model()
+
+# Write additional data to tmp file and perform sanity check
+with tempfile.NamedTemporaryFile(suffix='.csv') as temp_file:
+one_series = next(iter(self.additional_data))
+sample_prediction_data = self.additional_data[one_series].tail(self.horizon)
+sample_prediction_data[self.spec.target_category_columns[0]] = one_series
+date_col_name = self.spec.datetime_column.name
+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)
+input_data = {"additional_data": {"url": temp_file.name}}
+prediction_test = predict(input_data, _)
+logger.info(f"prediction test completed with result :{prediction_test}")
+except Exception as e:
+logger.error(f"An error occurred during the sanity test: {e}")
+raise
+finally:
+sys.path = org_sys_path
 
 def _copy_score_file(self):
 """

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

@@ -13,6 +13,7 @@
 import ads
 from ads.opctl.operator.lowcode.common.utils import load_data
 from ads.opctl.operator.common.operator_config import InputData
+from ads.opctl.operator.lowcode.forecast.const import SupportedModels
 
 ads.set_auth("resource_principal")
 
@@ -26,12 +27,6 @@
 Inference script. This script is used for prediction by scoring server when schema is known.
 """
 
-AUTOTS = "autots"
-ARIMA = "arima"
-PROPHET = "prophet"
-NEURALPROPHET = "neuralprophet"
-AUTOMLX = "automlx"
-
 
 @lru_cache(maxsize=10)
 def load_model():
@@ -142,7 +137,7 @@ def get_forecast(future_df, model_name, series_id, model_object, date_col, targe
 future_df[date_col_name] = pd.to_datetime(
 future_df[date_col_name], format=date_col_format
 )
-if model_name == AUTOTS:
+if model_name == SupportedModels.AutoTS:
 series_id_col = "Series"
 full_data_indexed = future_df.rename(columns={target_cat_col: series_id_col})
 additional_regressors = list(
@@ -155,12 +150,12 @@ def get_forecast(future_df, model_name, series_id, model_object, date_col, targe
 )
 pred_obj = model_object.predict(future_regressor=future_reg)
 return pred_obj.forecast[series_id].tolist()
-elif model_name == PROPHET and series_id in model_object:
+elif model_name == SupportedModels.Prophet and series_id in model_object:
 model = model_object[series_id]
 processed = future_df.rename(columns={date_col_name: 'ds', target_column: 'y'})
 forecast = model.predict(processed)
 return forecast['yhat'].tolist()
-elif model_name == NEURALPROPHET and series_id in model_object:
+elif model_name == SupportedModels.NeuralProphet and series_id in model_object:
 model = model_object[series_id]
 model.restore_trainer()
 accepted_regressors = list(model.config_regressors.regressors.keys())
@@ -169,7 +164,7 @@ def get_forecast(future_df, model_name, series_id, model_object, date_col, targe
 future["y"] = None
 forecast = model.predict(future)
 return forecast['yhat1'].tolist()
-elif model_name == ARIMA and series_id in model_object:
+elif model_name == SupportedModels.Arima and series_id in model_object:
 model = model_object[series_id]
 future_df = future_df.set_index(date_col_name)
 x_pred = future_df.drop(target_cat_col, axis=1)
@@ -180,7 +175,7 @@ def get_forecast(future_df, model_name, series_id, model_object, date_col, targe
 )
 yhat_clean = pd.DataFrame(yhat, index=yhat.index, columns=["yhat"])
 return yhat_clean['yhat'].tolist()
-elif model_name == AUTOMLX and series_id in model_object:
+elif model_name == SupportedModels.AutoMLX and series_id in model_object:
 # automlx model
 model = model_object[series_id]
 x_pred = future_df.drop(target_cat_col, axis=1)