refactor to results dir unique path to common utils, update anomaly dir

Author: codeloop

ads/opctl/operator/lowcode/anomaly/model/base_model.py

@@ -16,6 +16,7 @@
 
 from ads.opctl import logger
 
+from ...common.utils import get_unique_report_dir
 from ..operator_config import AnomalyOperatorConfig, AnomalyOperatorSpec
 from .anomaly_dataset import AnomalyDatasets, AnomalyOutput
 from ..const import OutputColumns, SupportedMetrics
@@ -266,17 +267,9 @@ def _save_report(
         """Saves resulting reports to the given folder."""
         import datapane as dp
 
-        if self.spec.output_directory:
-            output_dir = self.spec.output_directory.url
-        else:
-            output_dir = "tmp_operator_result"
-            logger.warn(
-                "Since the output directory was not specified, the output will be saved to {} directory.".format(
-                    output_dir
-                )
-            )
+        unique_output_dir = get_unique_report_dir(self.spec.output_directory)
 
-        if ObjectStorageDetails.is_oci_path(output_dir):
+        if ObjectStorageDetails.is_oci_path(unique_output_dir):
             storage_options = default_signer()
         else:
             storage_options = dict()
@@ -287,7 +280,7 @@ def _save_report(
             dp.save_report(report_sections, report_local_path)
             with open(report_local_path) as f1:
                 with fsspec.open(
-                    os.path.join(output_dir, self.spec.report_file_name),
+                    os.path.join(unique_output_dir, self.spec.report_file_name),
                     "w",
                     **default_signer(),
                 ) as f2:
@@ -297,15 +290,15 @@ def _save_report(
             inliers = anomaly_output.get_inliers(self.datasets.data)
             write_data(
                 data=inliers,
-                filename=os.path.join(output_dir, self.spec.inliers_filename),
+                filename=os.path.join(unique_output_dir, self.spec.inliers_filename),
                 format="csv",
                 storage_options=storage_options,
             )
 
         outliers = anomaly_output.get_outliers(self.datasets.data)
         write_data(
             data=outliers,
-            filename=os.path.join(output_dir, self.spec.outliers_filename),
+            filename=os.path.join(unique_output_dir, self.spec.outliers_filename),
             format="csv",
             storage_options=storage_options,
         )
@@ -314,15 +307,15 @@ def _save_report(
             write_data(
                 data=validation_metrics.rename_axis("metrics").reset_index(),
                 filename=os.path.join(
-                    output_dir, self.spec.validation_metrics_filename
+                    unique_output_dir, self.spec.validation_metrics_filename
                 ),
                 format="csv",
                 storage_options=storage_options,
             )
 
         logger.warn(
             f"The report has been successfully "
-            f"generated and placed to the: {output_dir}."
+            f"generated and placed to the: {unique_output_dir}."
         )
 
     @abstractmethod

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

@@ -17,6 +17,7 @@
 import yaml
 from typing import Union
 
+from ads.opctl import logger
 from ads.opctl.operator.lowcode.common.errors import (
     InputDataError,
     InvalidParameterError,
@@ -77,6 +78,37 @@ def write_data(data, filename, format, storage_options, index=False, **kwargs):
     )
 
 
+def get_unique_report_dir(output_dir: str) -> str:
+    """
+    Generate a unique directory path for the report output.
+    Args:
+    - output_dir (str): The requested output directory path.
+    - report_name (str): The name of the report.
+    Returns:
+    - str: The unique directory path for the report output.
+    """
+
+    if output_dir:
+            output_dir = output_dir.url
+            # set the unique directory path as the requested path by the user
+            unique_output_dir = output_dir
+    else:
+        output_dir = "results"
+
+        # If the directory exists, find the next unique directory name by appending an incrementing suffix
+        counter = 1
+        unique_output_dir = f"{output_dir}"
+        while os.path.exists(unique_output_dir):
+            unique_output_dir = f"{output_dir}_{counter}"
+            counter += 1
+        logger.warn(
+            "Since the output directory was not specified, the output will be saved to {} directory.".format(
+                unique_output_dir
+            )
+        )
+    return unique_output_dir
+
+
 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/model/base_model.py

@@ -44,6 +44,7 @@
     SupportedModels,
     SpeedAccuracyMode,
 )
+from ...common.utils import get_unique_report_dir
 from ..operator_config import ForecastOperatorConfig, ForecastOperatorSpec
 from ads.common.decorator.runtime_dependency import runtime_dependency
 from .forecast_datasets import ForecastDatasets, ForecastOutput
@@ -408,24 +409,7 @@ def _save_report(
         """Saves resulting reports to the given folder."""
         import datapane as dp
 
-        if self.spec.output_directory:
-            output_dir = self.spec.output_directory.url
-            # set the unique directory path as the requested path by the user
-            unique_output_dir = output_dir
-        else:
-            output_dir = "results"
-
-            # If the directory exists, find the next unique directory name by appending an incrementing suffix
-            counter = 1
-            unique_output_dir = f"{output_dir}"
-            while os.path.exists(unique_output_dir):
-                unique_output_dir = f"{output_dir}_{counter}"
-                counter += 1
-            logger.warn(
-                "Since the output directory was not specified, the output will be saved to {} directory.".format(
-                    unique_output_dir
-                )
-            )
+        unique_output_dir = get_unique_report_dir(self.spec.output_directory)
 
         if ObjectStorageDetails.is_oci_path(unique_output_dir):
             storage_options = default_signer()