use some OOP

Author: kshpv

tests/post_training/experimental/sparsify_activations/pipelines.py

@@ -32,7 +32,9 @@
 from nncf.torch.quantization.layers import INT8SymmetricWeightsDecompressor
 from tests.post_training.pipelines.base import PT_BACKENDS
 from tests.post_training.pipelines.base import BackendType
-from tests.post_training.pipelines.base import PTQTestPipeline
+from tests.post_training.pipelines.base import ErrorReason
+from tests.post_training.pipelines.base import ErrorReport
+from tests.post_training.pipelines.base import RunInfo
 from tests.post_training.pipelines.image_classification_timm import ImageClassificationTimm
 from tests.post_training.pipelines.lm_weight_compression import LMWeightCompression
 from tests.post_training.pipelines.lm_weight_compression import WCTimeStats
@@ -52,7 +54,18 @@ class SATimeStats(WCTimeStats):
     REGEX_PREFIX = [*WCTimeStats.REGEX_PREFIX, SparsifyActivationsAlgoBackend.CALIBRATION_TRACKING_DESC]
 
 
-class SAPipelineMixin(PTQTestPipeline):
+@dataclass
+class SARunInfo(RunInfo):
+    def get_result_dict(self):
+        result = super().get_result_dict()
+        result["Num FQ"] = self.num_compress_nodes.num_fq_nodes
+        result["Num int4"] = self.num_compress_nodes.num_int4
+        result["Num int8"] = self.num_compress_nodes.num_int8
+        result["Num sparse activations"] = self.num_compress_nodes.num_sparse_activations
+        return result
+
+
+class SAPipelineMixin(LMWeightCompression):
     """
     Common methods in the test pipeline for Sparsify Activations.
     """
@@ -88,8 +101,24 @@ def get_num_compressed(self) -> None:
         model = ie.read_model(model=self.path_compressed_ir)
         self.run_info.num_compress_nodes.num_sparse_activations = count_sparsifier_patterns_in_ov(model)
 
+    def collect_errors(self) -> List[ErrorReport]:
+        errors = super().collect_errors()
+        run_info = self.run_info
+        reference_data = self.reference_data
+
+        ref_num_sparse_activations = reference_data.get("num_sparse_activations")
+        num_sparse_activations = run_info.num_compress_nodes.num_sparse_activations
+
+        if ref_num_sparse_activations is not None and num_sparse_activations != ref_num_sparse_activations:
+            status_msg = (
+                f"Regression: The number of sparse activations is {num_sparse_activations}, "
+                f"which differs from reference {ref_num_sparse_activations}."
+            )
+            errors.append(ErrorReport(ErrorReason.NUM_COMPRESSED, status_msg))
+        return errors
+
 
-class LMSparsifyActivations(SAPipelineMixin, LMWeightCompression):
+class LMSparsifyActivations(SAPipelineMixin):
     DEFAULT_SUBSET_SIZE = 32
 
     def prepare_model(self):

tests/post_training/pipelines/base.py

@@ -19,8 +19,9 @@
 from datetime import timedelta
 from enum import Enum
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
+import numpy as np
 import onnx
 import openvino as ov
 import torch
@@ -179,7 +180,8 @@ def format_memory_usage(memory):
             return None
         return int(memory)
 
-    def get_result_dict(self):
+    def get_result_dict(self) -> Dict[str, str]:
+        """Returns a dictionary with the results of the run."""
         ram_data = {}
         if self.compression_memory_usage_rss is None and self.compression_memory_usage_system is None:
             ram_data["RAM MiB"] = self.format_memory_usage(self.compression_memory_usage)
@@ -194,10 +196,6 @@ def get_result_dict(self):
             "Metric name": self.metric_name,
             "Metric value": self.metric_value,
             "Metric diff": self.metric_diff,
-            "Num FQ": self.num_compress_nodes.num_fq_nodes,
-            "Num int4": self.num_compress_nodes.num_int4,
-            "Num int8": self.num_compress_nodes.num_int8,
-            "Num sparse activations": self.num_compress_nodes.num_sparse_activations,
             "Compr. time": self.format_time(self.time_compression),
             **self.stats_from_output.get_stats(),
             "Total time": self.format_time(self.time_total),
@@ -209,6 +207,15 @@ def get_result_dict(self):
         return result
 
 
+@dataclass
+class PTQRunInfo(RunInfo):
+    def get_result_dict(self):
+        result = super().get_result_dict()
+        result["Num FQ"] = self.num_compress_nodes.num_fq_nodes
+        result["Num int8"] = self.num_compress_nodes.num_int8
+        return result
+
+
 class BaseTestPipeline(ABC):
     """
     Base class to test compression algorithms.
@@ -286,9 +293,28 @@ def compress(self) -> None:
     def save_compressed_model(self) -> None:
         """Save compressed model to IR."""
 
-    @abstractmethod
     def get_num_compressed(self) -> None:
         """Get number of the compressed nodes in the compressed IR."""
+        ie = ov.Core()
+        model = ie.read_model(model=self.path_compressed_ir)
+
+        num_fq = 0
+        num_int4 = 0
+        num_int8 = 0
+        for node in model.get_ops():
+            node_type = node.type_info.name
+            if node_type == "FakeQuantize":
+                num_fq += 1
+
+            for i in range(node.get_output_size()):
+                if node.get_output_element_type(i).get_type_name() in ["i8", "u8"]:
+                    num_int8 += 1
+                if node.get_output_element_type(i).get_type_name() in ["i4", "u4", "nf4"]:
+                    num_int4 += 1
+
+        self.run_info.num_compress_nodes.num_int8 = num_int8
+        self.run_info.num_compress_nodes.num_int4 = num_int4
+        self.run_info.num_compress_nodes.num_fq_nodes = num_fq
 
     @abstractmethod
     def run_bench(self) -> None:
@@ -334,6 +360,61 @@ def run(self) -> None:
         self.validate()
         self.run_bench()
 
+    def collect_errors(self) -> List[ErrorReport]:
+        """
+        Collects errors based on the pipeline's run information.
+
+        :param pipeline: The pipeline object containing run information.
+        :return: List of error reports.
+        """
+        errors = []
+
+        run_info = self.run_info
+        reference_data = self.reference_data
+
+        metric_value = run_info.metric_value
+        metric_reference = reference_data.get("metric_value")
+        metric_value_fp32 = reference_data.get("metric_value_fp32")
+
+        if metric_value is not None and metric_value_fp32 is not None:
+            run_info.metric_diff = round(metric_value - metric_value_fp32, 5)
+
+        if metric_value is not None and metric_reference is not None:
+            atol = reference_data.get("atol", 0.001)
+            if not np.isclose(metric_value, metric_reference, atol=atol):
+                status_msg = (
+                    f"Regression: Metric value is less than reference {metric_value} < {metric_reference}"
+                    if metric_value < metric_reference
+                    else f"Improvement: Metric value is better than reference {metric_value} > {metric_reference}"
+                )
+                errors.append(ErrorReport(ErrorReason.METRICS, status_msg))
+
+        return errors
+
+    def update_status(self, error_reports: List[ErrorReport]) -> List[str]:
+        """
+        Updates status of the pipeline based on the errors encountered during the run.
+
+        :param pipeline: The pipeline object containing run information.
+        :param error_reports: List of errors encountered during the run.
+        :return: List of unexpected errors.
+        """
+        self.run_info.status = ""  # Successful status
+        xfails, unexpected_errors = [], []
+
+        for report in error_reports:
+            xfail_reason = report.reason.value + XFAIL_SUFFIX
+            if _is_error_xfailed(report, xfail_reason, self.reference_data):
+                xfails.append(_get_xfail_message(report, xfail_reason, self.reference_data))
+            else:
+                unexpected_errors.append(report.msg)
+
+        if xfails:
+            self.run_info.status = "\n".join(xfails)
+        if unexpected_errors:
+            self.run_info.status = "\n".join(unexpected_errors)
+        return unexpected_errors
+
 
 class PTQTestPipeline(BaseTestPipeline):
     """
@@ -421,28 +502,6 @@ def save_compressed_model(self) -> None:
             apply_moc_transformations(self.compressed_model, cf=True)
             ov.serialize(self.compressed_model, str(self.path_compressed_ir))
 
-    def get_num_compressed(self) -> None:
-        ie = ov.Core()
-        model = ie.read_model(model=self.path_compressed_ir)
-
-        num_fq = 0
-        num_int4 = 0
-        num_int8 = 0
-        for node in model.get_ops():
-            node_type = node.type_info.name
-            if node_type == "FakeQuantize":
-                num_fq += 1
-
-            for i in range(node.get_output_size()):
-                if node.get_output_element_type(i).get_type_name() in ["i8", "u8"]:
-                    num_int8 += 1
-                if node.get_output_element_type(i).get_type_name() in ["i4", "u4", "nf4"]:
-                    num_int4 += 1
-
-        self.run_info.num_compress_nodes.num_int8 = num_int8
-        self.run_info.num_compress_nodes.num_int4 = num_int4
-        self.run_info.num_compress_nodes.num_fq_nodes = num_fq
-
     def run_bench(self) -> None:
         """
         Run benchmark_app to collect performance statistics.
@@ -476,3 +535,32 @@ def collect_data_from_stdout(self, stdout: str):
         stats = PTQTimeStats()
         stats.fill(stdout)
         self.run_info.stats_from_output = stats
+
+
+def _get_exception_type_name(report: ErrorReport) -> str:
+    return report.msg.split("|")[0].replace("Exception Type: ", "")
+
+
+def _get_exception_error_message(report: ErrorReport) -> str:
+    return report.msg.split("|")[1]
+
+
+def _are_exceptions_matched(report: ErrorReport, reference_exception: Dict[str, str]) -> bool:
+    return reference_exception["error_message"] == _get_exception_error_message(report) and reference_exception[
+        "type"
+    ] == _get_exception_type_name(report)
+
+
+def _is_error_xfailed(report: ErrorReport, xfail_reason: str, reference_data: Dict[str, Dict[str, str]]) -> bool:
+    if xfail_reason not in reference_data:
+        return False
+
+    if report.reason == ErrorReason.EXCEPTION:
+        return _are_exceptions_matched(report, reference_data[xfail_reason])
+    return True
+
+
+def _get_xfail_message(report: ErrorReport, xfail_reason: str, reference_data: Dict[str, Dict[str, str]]) -> str:
+    if report.reason == ErrorReason.EXCEPTION:
+        return f"XFAIL: {reference_data[xfail_reason]['message']} - {report.msg}"
+    return f"XFAIL: {xfail_reason} - {report.msg}"

tests/post_training/pipelines/lm_weight_compression.py

@@ -14,7 +14,8 @@
 import shutil
 import time
 from dataclasses import dataclass
-from typing import Dict, Optional
+from pathlib import Path
+from typing import Dict, List, Optional
 
 import numpy as np
 import openvino as ov
@@ -30,7 +31,11 @@
 import nncf
 from tests.cross_fw.shared.paths import TEST_ROOT
 from tests.post_training.pipelines.base import BackendType
-from tests.post_training.pipelines.base import PTQTestPipeline
+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 NumCompressNodes
+from tests.post_training.pipelines.base import RunInfo
 from tests.post_training.pipelines.base import StatsFromOutput
 from tools.memory_monitor import MemoryType
 from tools.memory_monitor import MemoryUnit
@@ -71,11 +76,53 @@ def get_stats(self) -> Dict[str, str]:
         return dict(zip(self.STAT_NAMES, VARS))
 
 
-class LMWeightCompression(PTQTestPipeline):
+@dataclass
+class WCRunInfo(RunInfo):
+    def get_result_dict(self):
+        result = super().get_result_dict()
+        result["Num int4"] = self.num_compress_nodes.num_int4
+        result["Num int8"] = self.num_compress_nodes.num_int8
+        return result
+
+
+class LMWeightCompression(BaseTestPipeline):
     """Pipeline for casual language models from Hugging Face repository"""
 
     OV_MODEL_NAME = "openvino_model.xml"
 
+    def __init__(
+        self,
+        reported_name: str,
+        model_id: str,
+        backend: BackendType,
+        compression_params: dict,
+        output_dir: Path,
+        data_dir: Path,
+        reference_data: dict,
+        no_eval: bool,
+        run_benchmark_app: bool,
+        torch_compile_validation: bool = False,
+        params: dict = None,
+        batch_size: int = 1,
+        memory_monitor: bool = False,
+    ) -> None:
+        super().__init__(
+            reported_name,
+            model_id,
+            backend,
+            compression_params,
+            output_dir,
+            data_dir,
+            reference_data,
+            no_eval,
+            run_benchmark_app,
+            torch_compile_validation,
+            params,
+            batch_size,
+            memory_monitor,
+        )
+        self.run_info = WCRunInfo(model=reported_name, backend=self.backend, num_compress_nodes=NumCompressNodes())
+
     def prepare_model(self) -> None:
         is_stateful = self.params.get("is_stateful", False)
 
@@ -291,3 +338,29 @@ def _validate(self) -> None:
         similarity = all_metrics["similarity"][0]
         self.run_info.metric_name = "Similarity"
         self.run_info.metric_value = round(similarity, 5)
+
+    def collect_errors(self) -> List[ErrorReport]:
+        errors = super().collect_errors()
+        run_info = self.run_info
+        reference_data = self.reference_data
+
+        num_int4_reference = reference_data.get("num_int4")
+        num_int8_reference = reference_data.get("num_int8")
+        num_int4_value = run_info.num_compress_nodes.num_int4
+        num_int8_value = run_info.num_compress_nodes.num_int8
+
+        if num_int4_reference is not None and num_int4_reference != num_int4_value:
+            status_msg = (
+                "Regression: The number of int4 ops is different "
+                f"than reference {num_int4_reference} != {num_int4_value}"
+            )
+            errors.append(ErrorReport(ErrorReason.NUM_COMPRESSED, status_msg))
+
+        if num_int8_reference is not None and num_int8_reference != num_int8_value:
+            status_msg = (
+                "Regression: The number of int8 ops is different "
+                f"than reference {num_int8_reference} != {num_int8_value}"
+            )
+            errors.append(ErrorReport(ErrorReason.NUM_COMPRESSED, status_msg))
+
+        return errors