make common run_pipeline()

Author: kshpv

tests/post_training/data/ptq_reference_data.yaml

@@ -18,7 +18,7 @@ hf/hf-internal-testing/tiny-random-GPTNeoXForCausalLM_statefull_backend_OPTIMUM:
   metric_value: null
 hf/hf-internal-testing/tiny-random-GPTNeoXForCausalLM_stateless_backend_OPTIMUM:
   metric_value: null
-  xfail_reason: "Issue-161969"
+  exception_xfail_reason: "Issue-161969"
 hf/hf-internal-testing/tiny-random-gpt2_backend_FP32:
   metric_value: null
 hf/hf-internal-testing/tiny-random-gpt2_backend_OPTIMUM:

tests/post_training/experimental/sparsify_activations/pipelines.py

@@ -35,8 +35,6 @@
 from tests.post_training.pipelines.base import LIMIT_LENGTH_OF_STATUS
 from tests.post_training.pipelines.base import PT_BACKENDS
 from tests.post_training.pipelines.base import BackendType
-from tests.post_training.pipelines.base import ErrorReason
-from tests.post_training.pipelines.base import ErrorReport
 from tests.post_training.pipelines.base import NumCompressNodes
 from tests.post_training.pipelines.base import RunInfo
 from tests.post_training.pipelines.image_classification_timm import ImageClassificationTimm
@@ -171,16 +169,6 @@ def _compress(self):
                 **self.compression_params["sparsify_activations"],
             )
 
-    def _validate(self):
-        errors = super()._validate()
-        ref_num_sparse_activations = self.reference_data.get("num_sparse_activations", 0)
-        num_sparse_activations = self.run_info.num_compress_nodes.num_sparse_activations
-        if num_sparse_activations != ref_num_sparse_activations:
-            status_msg = f"Regression: The number of sparse activations is {num_sparse_activations}, \
-                which differs from reference {ref_num_sparse_activations}."
-            errors.append(ErrorReport(ErrorReason.NUM_COMPRESSED, status_msg))
-        return errors
-
 
 class LMSparsifyActivations(SAPipelineMixin, LMWeightCompression):
     DEFAULT_SUBSET_SIZE = 32

tests/post_training/experimental/sparsify_activations/test_sparsify_activations_conformance.py

@@ -9,9 +9,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
-import time
-import traceback
 from collections import OrderedDict
 from pathlib import Path
 from typing import Dict, Optional
@@ -23,8 +20,6 @@
 from tests.post_training.experimental.sparsify_activations.model_scope import SPARSIFY_ACTIVATIONS_TEST_CASES
 from tests.post_training.experimental.sparsify_activations.pipelines import SARunInfo
 from tests.post_training.pipelines.base import BackendType
-from tests.post_training.pipelines.base import BaseTestPipeline
-from tests.post_training.test_quantize_conformance import create_short_run_info
 from tests.post_training.test_quantize_conformance import fixture_batch_size  # noqa: F401
 from tests.post_training.test_quantize_conformance import fixture_data  # noqa: F401
 from tests.post_training.test_quantize_conformance import fixture_extra_columns  # noqa: F401
@@ -34,8 +29,7 @@
 from tests.post_training.test_quantize_conformance import fixture_run_fp32_backend  # noqa: F401
 from tests.post_training.test_quantize_conformance import fixture_run_torch_cuda_backend  # noqa: F401
 from tests.post_training.test_quantize_conformance import fixture_subset_size  # noqa: F401
-from tests.post_training.test_quantize_conformance import maybe_skip_test_case
-from tests.post_training.test_quantize_conformance import write_logs
+from tests.post_training.test_quantize_conformance import run_pipeline
 
 
 @pytest.fixture(scope="session", name="sparsify_activations_reference_data")
@@ -59,39 +53,6 @@ def fixture_sparsify_activations_report_data(output_dir):
         df.to_csv(output_dir / "results.csv", index=False)
 
 
-def create_pipeline_kwargs(
-    test_model_param: Dict,
-    subset_size,
-    test_case_name: str,
-    reference_data: Dict[str, Dict],
-    fp32_model_params: Dict[str, Dict],
-):
-    if subset_size:
-        if "compression_params" not in test_model_param:
-            test_model_param["compression_params"] = {}
-        test_model_param["compression_params"]["subset_size"] = subset_size
-
-    print("\n")
-    print(f"Model: {test_model_param['reported_name']}")
-    print(f"Backend: {test_model_param['backend']}")
-    print(f"Comprssion params: {test_model_param['compression_params']}")
-
-    # Get target fp32 metric value
-    model_id = test_model_param["model_id"]
-    fp32_test_case_name = fp32_model_params[model_id]["reported_name"] + f"_backend_{BackendType.FP32.value}"
-    test_reference = reference_data[test_case_name]
-    test_reference["metric_value_fp32"] = reference_data[fp32_test_case_name]["metric_value"]
-
-    return {
-        "reported_name": test_model_param["reported_name"],
-        "model_id": test_model_param["model_id"],
-        "backend": test_model_param["backend"],
-        "compression_params": test_model_param["compression_params"],
-        "params": test_model_param.get("params"),
-        "reference_data": test_reference,
-    }
-
-
 @pytest.mark.parametrize("test_case_name", SPARSIFY_ACTIVATIONS_TEST_CASES.keys())
 def test_sparsify_activations(
     sparsify_activations_reference_data: dict,
@@ -108,55 +69,26 @@ def test_sparsify_activations(
     capsys: pytest.CaptureFixture,
     extra_columns: bool,
 ):
-    pipeline = None
-    err_msg = None
-    test_model_param = None
-    start_time = time.perf_counter()
-    try:
-        if test_case_name not in sparsify_activations_reference_data:
-            msg = f"{test_case_name} is not defined in `sparsify_activations_reference_data` fixture"
-            raise RuntimeError(msg)
-        test_model_param = SPARSIFY_ACTIVATIONS_TEST_CASES[test_case_name]
-        maybe_skip_test_case(test_model_param, run_fp32_backend, run_torch_cuda_backend, batch_size)
-        fp32_model_params = {
-            tc["model_id"]: tc for tc in SPARSIFY_ACTIVATIONS_TEST_CASES.values() if tc["backend"] == BackendType.FP32
-        }
-        pipeline_cls = test_model_param["pipeline_cls"]
-        pipeline_kwargs = create_pipeline_kwargs(
-            test_model_param, subset_size, test_case_name, sparsify_activations_reference_data, fp32_model_params
-        )
-        calibration_batch_size = batch_size or test_model_param.get("batch_size", 1)
-        pipeline_kwargs.update(
-            {
-                "output_dir": output_dir,
-                "data_dir": data_dir,
-                "no_eval": no_eval,
-                "run_benchmark_app": run_benchmark_app,
-                "batch_size": calibration_batch_size,
-            }
-        )
-        pipeline: BaseTestPipeline = pipeline_cls(**pipeline_kwargs)
-        pipeline.run()
-    except Exception as e:
-        err_msg = str(e)
-        traceback.print_exc()
-
-    if pipeline is not None:
-        pipeline.cleanup_cache()
-        run_info = pipeline.run_info
-        if err_msg:
-            run_info.status = f"{run_info.status} | {err_msg}" if run_info.status else err_msg
-
-        captured = capsys.readouterr()
-        write_logs(captured, pipeline)
-
-        if extra_columns:
-            pipeline.collect_data_from_stdout(captured.out)
-    else:
-        run_info = create_short_run_info(test_model_param, err_msg, test_case_name)
-
-    run_info.time_total = time.perf_counter() - start_time
-    sparsify_activations_result_data[test_case_name] = run_info
-
-    if err_msg:
-        pytest.fail(err_msg)
+    fp32_model_params = {
+        tc["model_id"]: tc for tc in SPARSIFY_ACTIVATIONS_TEST_CASES.values() if tc["backend"] == BackendType.FP32
+    }
+    run_pipeline(
+        test_case_name,
+        sparsify_activations_reference_data,
+        SPARSIFY_ACTIVATIONS_TEST_CASES,
+        sparsify_activations_result_data,
+        output_dir,
+        data_dir,
+        no_eval,
+        batch_size,
+        run_fp32_backend,
+        run_torch_cuda_backend,
+        subset_size,
+        run_benchmark_app,
+        False,  # torch_compile_validation is not used in SA
+        capsys,
+        extra_columns,
+        False,  # memory_monitor is not used in SA
+        None,  # use_avx2 is not used in SA
+        fp32_model_params,
+    )

tests/post_training/pipelines/base.py

@@ -19,9 +19,8 @@
 from datetime import timedelta
 from enum import Enum
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, Optional
 
-import numpy as np
 import onnx
 import openvino as ov
 import torch
@@ -42,6 +41,7 @@
 class ErrorReason(Enum):
     METRICS = "metrics"
     NUM_COMPRESSED = "num_compressed"
+    EXCEPTION = "exception"
 
 
 @dataclass
@@ -293,31 +293,12 @@ def get_num_compressed(self) -> None:
     def run_bench(self) -> None:
         """Run a benchmark to collect performance statistics."""
 
-    def _validate(self) -> List[ErrorReport]:
+    def _validate(self) -> None:
         """
         Validates some test criteria.
         returns:
             A list of error reports generated during validation.
         """
-        return []
-
-    def _process_errors(self, errors) -> str:
-        """
-        Processes a list of error reports and updates the run status.
-
-        :param errors: A list of error reports.
-        :return: A string representing the concatenated statuses of the processed errors.
-        """
-        xfails, msg_list = [], []
-        for report in errors:
-            xfail_reason = report.reason.value + XFAIL_SUFFIX
-            if xfail_reason in self.reference_data:
-                xfails.append(f"XFAIL: {self.reference_data[xfail_reason]} - {report.msg}")
-            else:
-                msg_list.append(report.msg)
-        if msg_list:
-            raise ValueError("\n".join(msg_list))
-        self.run_info.status = "\n".join(xfails)
 
     def prepare(self):
         """
@@ -339,29 +320,7 @@ def validate(self) -> None:
             print("Validation skipped")
             return
         print("Validation...")
-
-        errors = self._validate()
-
-        metric_value = self.run_info.metric_value
-        metric_reference = self.reference_data.get("metric_value")
-        metric_value_fp32 = self.reference_data.get("metric_value_fp32")
-
-        if metric_value is not None and metric_value_fp32 is not None:
-            self.run_info.metric_diff = round(self.run_info.metric_value - self.reference_data["metric_value_fp32"], 5)
-
-        if (
-            metric_value is not None
-            and metric_reference is not None
-            and not np.isclose(metric_value, metric_reference, atol=self.reference_data.get("atol", 0.001))
-        ):
-            status_msg = None
-            if metric_value < metric_reference:
-                status_msg = f"Regression: Metric value is less than reference {metric_value} < {metric_reference}"
-            if metric_value > metric_reference:
-                status_msg = f"Improvement: Metric value is better than reference {metric_value} > {metric_reference}"
-            if status_msg:
-                errors.append(ErrorReport(ErrorReason.METRICS, status_msg))
-        self._process_errors(errors)
+        self._validate()
 
     def run(self) -> None:
         """

tests/post_training/pipelines/image_classification_base.py

@@ -43,7 +43,6 @@ def _validate_ov(
         references: np.ndarray,
         dataset_size: int,
     ):
-
         core = ov.Core()
         if os.environ.get("INFERENCE_NUM_THREADS"):
             # Set CPU_THREADS_NUM for OpenVINO inference
@@ -106,4 +105,3 @@ def _validate(self) -> List[ErrorReport]:
 
         self.run_info.metric_name = "Acc@1"
         self.run_info.metric_value = acc_top1
-        return []

tests/post_training/pipelines/lm_weight_compression.py

@@ -14,7 +14,7 @@
 import shutil
 import time
 from dataclasses import dataclass
-from typing import Dict, List, Optional
+from typing import Dict, Optional
 
 import numpy as np
 import openvino as ov
@@ -31,8 +31,6 @@
 from tests.cross_fw.shared.paths import TEST_ROOT
 from tests.post_training.pipelines.base import BackendType
 from tests.post_training.pipelines.base import BaseTestPipeline
-from tests.post_training.pipelines.base import ErrorReason
-from tests.post_training.pipelines.base import ErrorReport
 from tests.post_training.pipelines.base import StatsFromOutput
 from tools.memory_monitor import MemoryType
 from tools.memory_monitor import MemoryUnit
@@ -269,8 +267,7 @@ def _compress(self):
             **self.compression_params,
         )
 
-    def _validate(self) -> List[ErrorReport]:
-        errors = []
+    def _validate(self) -> None:
         is_stateful = self.params.get("is_stateful", False)
         core = ov.Core()
 
@@ -315,18 +312,3 @@ def _validate(self) -> List[ErrorReport]:
         similarity = all_metrics["similarity"][0]
         self.run_info.metric_name = "Similarity"
         self.run_info.metric_value = round(similarity, 5)
-
-        num_int4_reference = self.reference_data.get("num_int4")
-        num_int8_reference = self.reference_data.get("num_int8")
-
-        num_int4_value = self.run_info.num_compress_nodes.num_int4
-        num_int8_value = self.run_info.num_compress_nodes.num_int8
-
-        template = "Regression: The number of int{} ops is different than reference {} != {}"
-        if num_int4_reference != num_int4_value:
-            status_msg = template.format(4, num_int4_reference, num_int4_value)
-            errors.append(ErrorReport(ErrorReason.NUM_COMPRESSED, status_msg))
-        if num_int8_reference != num_int8_value:
-            status_msg = template.format(8, num_int8_reference, num_int8_value)
-            errors.append(ErrorReport(ErrorReason.NUM_COMPRESSED, status_msg))
-        return errors