Computation of compression parameters via OpenVINO models

docs/api/source/conf.py

@@ -145,6 +145,7 @@ def collect_api_entities() -> APIInfo:
     "nncf.tensor.functions.torch_linalg",
     "nncf.tensor.functions.torch_io",
     "nncf.tensor.functions.numpy_io",
+    "nncf.tensor.functions.ov_numeric",
 ]
 
 with mock(mock_modules):

nncf/common/logging/logger.py

@@ -11,11 +11,41 @@
 
 import logging
 import sys
-from typing import Set
+from functools import lru_cache
+from typing import cast
+
+
+class NNCFLogger(logging.Logger):
+    def __init__(self, name: str, level: int = logging.NOTSET):
+        super().__init__(name, level)
+
+    @lru_cache(None)
+    def _log_once(self, level: int, msg: str) -> None:
+        self.log(level, msg)
+
+    def debug_once(self, msg: str) -> None:
+        """
+        Log a message at the DEBUG level, ensuring the message is logged only once.
+        """
+        self._log_once(logging.DEBUG, msg)
+
+    def info_once(self, msg: str) -> None:
+        """
+        Log a message at the INFO level, ensuring the message is logged only once.
+        """
+        self._log_once(logging.INFO, msg)
+
+    def warning_once(self, msg: str) -> None:
+        """
+        Log a message at the WARNING level, ensuring the message is logged only once.
+        """
+        self._log_once(logging.WARNING, msg)
+
 
 NNCF_LOGGER_NAME = "nncf"
 
-nncf_logger = logging.getLogger(NNCF_LOGGER_NAME)
+logging.setLoggerClass(NNCFLogger)
+nncf_logger = cast(NNCFLogger, logging.getLogger(NNCF_LOGGER_NAME))
 nncf_logger.propagate = False
 
 stdout_handler = logging.StreamHandler(sys.stdout)
@@ -60,16 +90,6 @@ def disable_logging() -> None:
     nncf_logger.handlers = []
 
 
-class DuplicateFilter:
-    def __init__(self) -> None:
-        self.msgs: Set[str] = set()
-
-    def filter(self, rec: logging.LogRecord) -> bool:
-        retval = rec.msg not in self.msgs
-        self.msgs.add(rec.msg)
-        return retval
-
-
 NNCFDeprecationWarning = FutureWarning
 
 

nncf/common/utils/backend.py

@@ -16,6 +16,13 @@
 
 TModel = TypeVar("TModel")
 
+try:
+    import openvino  # type: ignore # noqa: F401
+
+    _OPENVINO_AVAILABLE = True
+except ImportError:
+    _OPENVINO_AVAILABLE = False
+
 
 class BackendType(Enum):
     TORCH = "Torch"
@@ -159,3 +166,11 @@ def copy_model(model: TModel) -> TModel:
         model = TFModelTransformer(model).transform(TFTransformationLayout())
         return model
     return deepcopy(model)
+
+
+def is_openvino_available() -> bool:
+    """
+    Check if OpenVINO is available.
+    :return: True if openvino package is installed, False otherwise.
+    """
+    return _OPENVINO_AVAILABLE

nncf/common/utils/caching.py

@@ -0,0 +1,102 @@
+# Copyright (c) 2025 Intel Corporation
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#      http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import copy
+import inspect
+from contextlib import contextmanager
+from functools import wraps
+from typing import Any, Callable, Dict, Iterator, TypeVar, cast
+
+
+class ResultsCache:
+    """
+    A container for results decorated with @cache_results decorator.
+    """
+
+    def __init__(self) -> None:
+        self._enabled = True
+        # Stores the results of the decorated function
+        self._cache: Dict[Any, Any] = {}
+        # Stores the number of times the cached result was accessed
+        self._access_count: Dict[Any, int] = {}
+
+    def enable(self) -> None:
+        self._enabled = True
+
+    def disable(self) -> None:
+        self._enabled = False
+
+    def enabled(self) -> bool:
+        return self._enabled
+
+    def access_count(self) -> Dict[Any, int]:
+        return copy.deepcopy(self._access_count)
+
+    def clear(self) -> None:
+        self._cache.clear()
+        self._access_count.clear()
+
+    def __getitem__(self, key: Any) -> Any:
+        self._access_count[key] += 1
+        return self._cache[key]
+
+    def __setitem__(self, key: Any, value: Any) -> None:
+        self._access_count[key] = 0
+        self._cache[key] = value
+
+    def __contains__(self, key: Any) -> bool:
+        return key in self._cache
+
+
+TFunc = TypeVar("TFunc", bound=Callable[..., Any])
+
+
+def cache_results(cache: ResultsCache) -> Callable[[TFunc], TFunc]:
+    """
+    Decorator to cache results of a function. When decorated function is called with the same set of arguments, it
+    will return the cached result instead of recomputing it. If it was the first call with such set of arguments, the
+    result will be computed and stored in the cache. The cache is stored in the `cache` object. Function arguments
+    must be hashable.
+
+    :param cache: A cache container where results will be stored.
+    """
+
+    def decorator(func: TFunc) -> TFunc:
+        @wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            if not cache.enabled():
+                return func(*args, **kwargs)
+            sig = inspect.signature(func)
+            new_kwargs = {name: arg for name, arg in zip(sig.parameters, args)}
+            new_kwargs.update(kwargs)
+            cache_key = (func.__name__, frozenset(new_kwargs.items()))
+            if cache_key in cache:
+                return cache[cache_key]
+            result = func(*args, **kwargs)
+            cache[cache_key] = result
+            return result
+
+        return cast(TFunc, wrapper)
+
+    return decorator
+
+
+@contextmanager
+def disable_results_caching(cache: ResultsCache) -> Iterator[None]:
+    """
+    Context manager to disable caching of results for a block of code.
+    :param cache: A cache container where results are stored.
+    """
+    if cache.enabled():
+        cache.disable()
+        yield
+        cache.enable()
+    else:
+        yield

nncf/openvino/graph/node_utils.py

@@ -13,6 +13,7 @@
 
 import numpy as np
 import openvino.runtime as ov
+import openvino.runtime.op as op
 import openvino.runtime.opset13 as opset
 
 import nncf
@@ -41,6 +42,8 @@
 from nncf.openvino.graph.metatypes.openvino_metatypes import OVMatMulMetatype
 from nncf.openvino.graph.metatypes.openvino_metatypes import OVOpMetatype
 from nncf.openvino.graph.metatypes.openvino_metatypes import get_node_metatype
+from nncf.tensor import Tensor
+from nncf.tensor import TensorBackend
 
 InplaceInsertionFnType = Callable[[ov.Node, int, str], ov.Node]
 
@@ -97,26 +100,27 @@ def get_number_if_op(model: ov.Model) -> int:
     """
 
     def cnt_if_op(model: ov.Model, cnt: int) -> int:
-        for op in model.get_ops():
-            if get_node_metatype(op) == OVIfMetatype:
+        for model_op in model.get_ops():
+            if get_node_metatype(model_op) == OVIfMetatype:
                 cnt += 1
-                cnt = cnt_if_op(op.get_function(0), cnt)
-                cnt = cnt_if_op(op.get_function(1), cnt)
+                cnt = cnt_if_op(model_op.get_function(0), cnt)
+                cnt = cnt_if_op(model_op.get_function(1), cnt)
         return cnt
 
     return cnt_if_op(model, 0)
 
 
-def get_const_value(const_node: ov.Node) -> np.ndarray:
+def get_const_value(const_node: ov.Node, cast_bf16_to_fp32: bool = True) -> np.ndarray:
     """
     Returns the constant tensor for the node.
     This method is applicable only for the floating-point constant data.
 
     :param const_node: OpenVINO node.
+    :param cast_bf16_to_fp32: Whether to cast bf16 node data to fp32 or not. If False and the node contains bf16 data,
+        the resulting bf16 value will be returned encoded inside a numpy.float16 array.
     :return: The constant value.
     """
-    if const_node.get_element_type() == ov.Type.bf16:
-        # Fixed FP32 data type as the result for BF16 constant
+    if const_node.get_element_type() == ov.Type.bf16 and cast_bf16_to_fp32:
         return const_node.get_data(dtype=np.float32)
     return const_node.data
 
@@ -635,3 +639,42 @@ def get_activation_channel_axis(node: NNCFNode, port_id: int, input_shape: Tuple
         channel_axis = activations_layout.index(OVLayoutElem.C_IN)
 
     return channel_axis
+
+
+def convert_op(node: ov.Node, target_dtype: ov.Type) -> ov.Node:
+    """
+    Return a subgraph which converts the given node output to the target data type. If the output is already in the
+    target data type then the given node is returned.
+
+    :param node: The input node to convert.
+    :param target_dtype: The target data type to convert the input node to.
+    :return: The converted node.
+    """
+    if node.get_element_type() == target_dtype:
+        return node
+    return opset.convert(node, target_dtype)
+
+
+def non_convertable_divide_op(a: ov.Node, b: ov.Node) -> ov.Node:
+    """
+    Creates a "non-convertable" divide operation. It won't be converted to a*(1/b).
+    """
+    divide_node = a / b
+    divide_node.get_rt_info()["nonconvertable_divide_0"] = True
+    return divide_node
+
+
+def create_ov_const_from_tensor(x: Tensor, dtype: ov.Type, name: Optional[str] = None) -> op.Constant:
+    """
+    Create an OpenVINO Constant node from the given tensor.
+    :param x: Data tensor. Supports NumPy and OV tensor backends. If x backend is OV, the constant node is created
+        directly from underlying OV tensor.
+    :param dtype: Data type of the constant.
+    :param name: Optional name of the constant.
+    :return: OpenVINO Constant node.
+    """
+    if x.backend == TensorBackend.ov:
+        assert x.data.get_element_type() == dtype
+        return opset.constant(x.data, name=name, shared_memory=True)
+    const = opset.constant(x.data, dtype=dtype, name=name)
+    return const

nncf/openvino/optimized_functions/__init__.py

@@ -0,0 +1,16 @@
+# Copyright (c) 2025 Intel Corporation
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#      http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from nncf.openvino.optimized_functions.functions import astype as astype
+from nncf.openvino.optimized_functions.functions import do_int_quantization as do_int_quantization
+from nncf.openvino.optimized_functions.functions import quantize_dequantize_weight as quantize_dequantize_weight
+from nncf.openvino.optimized_functions.models import OVModelParameters as OVModelParameters
+from nncf.openvino.optimized_functions.models import clear_ov_model_cache as clear_ov_model_cache