Support INTEL_VARIANT_PROVIDER_FORCE_DEVICE_IP

Signed-off-by: Dmitry Rogozhkin <dmitry.v.rogozhkin@intel.com>

Author: dvrogozh

README.md

@@ -53,6 +53,12 @@ powershell -c { $env:INSTALLER_DOWNLOAD_URL = 'https://wheelnext.astral.sh/v0.0.
    * Each value (`<ip>`) in the list represents human readable form of
      Intel hardware device IP (GMDID) quariable via Level Zero [ZE_extension_device_ip_version]
 
+## Environment variables
+
+`INTEL_VARIANT_PROVIDER_FORCE_DEVICE_IP` allows to override GPU architecture detection
+performed by plugin. **DISCLAIMER:** this is debug and test purpose variable as it can
+lead to a non-functional installation.
+
 ## WheelNext package index
 
 [WheelNext] initiative supports aggregated index of the packages which are enabled with variant providers.
@@ -75,7 +81,7 @@ namespace = ["intel"]
 
 [variant.providers.intel]
 requires = ["intel_variant_provider"]
-enable-if = "platform_system == 'Linux'"
+enable-if = "platform_system == 'Linux' or platform_system == 'Windows'"
 plugin-api = "intel_variant_provider.plugin:IntelVariantPlugin"
 ```
 

intel_variant_provider/devices.py

@@ -76,7 +76,7 @@
     },
 }
 
-def get_all_ips():
+def get_all_known_ips():
     return list(_intel_devips.keys())
 
 # See: https://github.com/intel/compute-runtime/blob/25.27.34303.6/shared/source/helpers/hw_ip_version.h

intel_variant_provider/plugin.py

@@ -24,49 +24,62 @@ class VariantFeatureConfig:
     multi_value: bool = False
 
 
+def getAllUniqueDeviceIPs():
+    pci_vendor_id_intel = 0x8086
+    devips = []
+    try:
+        desc = c_ze_init_driver_type_desc_t()
+        desc.flags = ZE_INIT_DRIVER_TYPE_FLAG_GPU
+        drivers = zeInitDrivers(desc)
+
+        for driver in drivers:
+            devices = zeDeviceGet(driver)
+            for device in devices:
+                devip = c_ze_device_ip_version_ext_t()
+                props = zeDeviceGetProperties(device, [devip])
+                if props.vendorId == pci_vendor_id_intel:
+                    devip = IntelDeviceIp(devip.ipVersion)
+                    for ip in devip.get_all_compat_ips():
+                        # We must return list of unique IPs as a requirement
+                        # of variantlib.
+                        if ip not in devips:
+                            devips.append(ip)
+    except Exception as e:
+        warnings.warn(f"Intel driver stack not installed or malfunctions: {e}", UserWarning, stacklevel=1)
+    return devips
+
+
 class IntelVariantPlugin:
     namespace = "intel"
     dynamic = False
 
     @classmethod
     @cache
     def generate_all_device_ips(cls) -> list[str] | None:
-        pci_vendor_id_intel = 0x8086
-
         if platform.system() not in ["Linux", "Windows"]:
             warnings.warn(f"Unsupported OS: {system}", UserWarning, stacklevel=1)
             return []
 
-        try:
-            desc = c_ze_init_driver_type_desc_t()
-            desc.flags = ZE_INIT_DRIVER_TYPE_FLAG_GPU
-            drivers = zeInitDrivers(desc)
-
-            devips = []
-            for driver in drivers:
-                devices = zeDeviceGet(driver)
-                for device in devices:
-                    devip = c_ze_device_ip_version_ext_t()
-                    props = zeDeviceGetProperties(device, [devip])
-                    if props.vendorId == pci_vendor_id_intel:
-                        devip = IntelDeviceIp(devip.ipVersion)
-                        for ip in devip.get_all_compat_ips():
-                            # Filter out devices which IPs are not explicitly
-                            # known to plugin. This gives consistency with the
-                            # check in validate_property().
-                            if ip not in get_all_ips():
-                                warnings.warn(f"Intel device with {ip} device IP is filtered out as not known to plugin)")
-                                continue
-                            # We must return list of unique IPs as a requirement
-                            # of variantlib.
-                            if ip not in devips:
-                                devips.append(ip)
-            if not devips:
-                warnings.warn("No Intel GPU detected", UserWarning, stacklevel=1)
-            return devips
-        except Exception as e:
-            warnings.warn(f"Intel driver stack not installed or malfunctions: {e}", UserWarning, stacklevel=1)
-            return []
+        devip = os.getenv("INTEL_VARIANT_PROVIDER_FORCE_DEVICE_IP")
+        if devip:
+            unique_devips = [devip]
+        else:
+            unique_devips = getAllUniqueDeviceIPs()
+
+        devips = []
+        known_ips = get_all_known_ips()
+        for ip in unique_devips:
+            # Filter out devices which IPs are not explicitly
+            # known to plugin. This gives consistency with the
+            # check in validate_property().
+            if ip not in known_ips:
+                warnings.warn(f"Intel device with {ip} device IP is filtered out as not known to plugin)")
+            else:
+                devips.append(ip)
+
+        if not devips:
+            warnings.warn("No Intel GPU detected", UserWarning, stacklevel=1)
+        return devips
 
     @classmethod
     def get_supported_configs(cls) -> list[VariantFeatureConfig]:
@@ -88,7 +101,7 @@ def get_all_configs(cls) -> list[VariantFeatureConfig]:
         return [
             VariantFeatureConfig(
                 name="device_ip",
-                values=get_all_ips(),
+                values=get_all_known_ips(),
                 multi_value=True,
             ),
         ]

tests/test_plugin.py

@@ -6,14 +6,19 @@
 # to internal module variables (such as _g_zelib).
 import intel_variant_provider.ze as ze
 from intel_variant_provider.devices import _intel_devips
-from intel_variant_provider.plugin import IntelVariantPlugin
+from intel_variant_provider.plugin import IntelVariantPlugin, VariantFeatureConfig
 from intel_variant_provider.ze import *
 
 
 @pytest.fixture
 def plugin() -> IntelVariantPlugin:
     return IntelVariantPlugin()
 
+@pytest.fixture(autouse=True)
+def clear_cache():
+    IntelVariantPlugin.generate_all_device_ips.cache_clear()
+
+
 def test_get_all_configs(plugin):
     configs = plugin.get_all_configs()
     assert len(configs) == 1
@@ -40,3 +45,15 @@ def test_no_L0_library(self, plugin, mocker):
         mocker.patch(self.cdll_name, side_effect=OSError("No such file"))
         with pytest.warns(UserWarning):
             assert not plugin.get_supported_configs()
+
+    def test_force_existing_ip(self, plugin, monkeypatch):
+        monkeypatch.setenv("INTEL_VARIANT_PROVIDER_FORCE_DEVICE_IP", "12.60.7")
+        assert plugin.get_supported_configs() == [
+            VariantFeatureConfig(name="device_ip", values=["12.60.7"], multi_value=True)
+        ]
+
+    def test_force_invalid_ip(self, plugin, monkeypatch):
+        monkeypatch.setenv("INTEL_VARIANT_PROVIDER_FORCE_DEVICE_IP", "12.99.99")
+        with pytest.warns(UserWarning):
+            assert not plugin.get_supported_configs()
+