diff --git a/src/winml/modelkit/config/build.py b/src/winml/modelkit/config/build.py index e663e367f..020c58033 100644 --- a/src/winml/modelkit/config/build.py +++ b/src/winml/modelkit/config/build.py @@ -352,7 +352,7 @@ def _resolve_policy_target(device: str, ep: str | None) -> tuple[str, str | None ): continue try: - if not EP_CATALOG.is_compatible(spec.ep): + if not EP_CATALOG.is_compatible(spec.ep, spec.device): continue except RuntimeError as e: detection_error = e diff --git a/src/winml/modelkit/ep_path.py b/src/winml/modelkit/ep_path.py index 8f8e13b3e..5cd374ea3 100644 --- a/src/winml/modelkit/ep_path.py +++ b/src/winml/modelkit/ep_path.py @@ -53,10 +53,12 @@ from importlib import metadata from pathlib import Path from types import MappingProxyType -from typing import Any, Final +from typing import Any, Final, cast from packaging.version import InvalidVersion, Version +from .utils.constants import DEVICE_PRIORITY, EP_SUPPORTED_DEVICES, DeviceType + logger = logging.getLogger(__name__) @@ -136,17 +138,28 @@ def ep_for_dll(self, dll: str) -> str | None: """Reverse lookup: DLL filename -> canonical EP name. ``None`` if unknown.""" return self._by_dll.get(dll) - def is_compatible(self, ep: str) -> bool: + def is_compatible(self, ep: str, device_type: DeviceType | None = None) -> bool: """Return True iff ``ep`` has compatible hardware on this machine. Empty / missing vendor requirement -> always compatible. Otherwise compatible iff at least one required vendor substring - appears (case-insensitively) in any detected vendor string. + appears (case-insensitively) in a supported hardware class. When + ``device_type`` is provided, only that class is considered. """ entry = self._by_name.get(ep) - if entry is None or not entry.vendor_requirements: + if entry is None: + return True + supported_devices = EP_SUPPORTED_DEVICES.get(cast("Any", ep), DEVICE_PRIORITY) + device_types: tuple[DeviceType, ...] + if device_type is not None: + if device_type not in supported_devices: + return False + device_types = (device_type,) + else: + device_types = supported_devices + if not entry.vendor_requirements: return True - detected = _get_detected_vendors() + detected = _get_detected_vendors(device_types) return any(req.lower() in v.lower() for req in entry.vendor_requirements for v in detected) def all_eps(self) -> tuple[str, ...]: @@ -192,39 +205,50 @@ def all_eps(self) -> tuple[str, ...]: @functools.cache -def _get_detected_vendors() -> frozenset[str]: - """Return the union of vendor identification strings from sysinfo. - - Aggregates ``manufacturer`` and ``name`` across detected GPUs and - NPUs. Both fields are included because Windows reports vendor - inconsistently — sometimes the manufacturer is the IHV - (``"Qualcomm Incorporated"``), sometimes a parent company - (``"Microsoft Corporation"`` for OEM-rebranded devices). - - Cached process-wide; tests reset via ``_get_detected_vendors.cache_clear()``. - Raises ``RuntimeError`` if hardware detection fails — preventing - ``functools.cache`` from pinning an empty-set fallback that would - silently make every hardware-gated EP appear incompatible. - """ +def _get_detected_vendors_for_device(device_type: DeviceType) -> frozenset[str]: + """Detect and cache vendor strings for one hardware class.""" try: - from .sysinfo.hardware import GPU, NPU + from .sysinfo.hardware import CPU, GPU, NPU except ImportError as e: raise RuntimeError(f"Hardware detection unavailable: {e}") from e + hardware_getters: dict[DeviceType, tuple[str, Callable[[], Iterable[Any]]]] = { + "cpu": ("CPU", CPU.get_all), + "gpu": ("GPU", GPU.get_all), + "npu": ("NPU", NPU.get_all), + } + class_name, get_all = hardware_getters[device_type] strings: set[str] = set() - for cls in (GPU, NPU): - try: - for hw in cls.get_all(): - for attr in ("manufacturer", "name"): - value = getattr(hw, attr, None) - if value: - strings.add(str(value)) - except Exception as e: # noqa: PERF203 - raise RuntimeError(f"{cls.__name__}.get_all() failed: {e}") from e - + try: + for hw in get_all(): + for attr in ("manufacturer", "name"): + value = getattr(hw, attr, None) + if value: + strings.add(str(value)) + except Exception as e: + raise RuntimeError(f"{class_name}.get_all() failed: {e}") from e return frozenset(strings) +def _get_detected_vendors( + device_types: tuple[DeviceType, ...] = DEVICE_PRIORITY, +) -> frozenset[str]: + """Return cached vendor strings for the selected hardware classes. + + Both ``manufacturer`` and ``name`` are included because Windows reports + vendor information inconsistently. Tests reset the shared inventory via + ``_get_detected_vendors.cache_clear()``. + """ + return frozenset( + vendor + for device_type in device_types + for vendor in _get_detected_vendors_for_device(device_type) + ) + + +_get_detected_vendors.cache_clear = _get_detected_vendors_for_device.cache_clear # type: ignore[attr-defined] + + # --------------------------------------------------------------------------- # Architecture resolver helpers. # --------------------------------------------------------------------------- diff --git a/src/winml/modelkit/session/ep_device.py b/src/winml/modelkit/session/ep_device.py index 1321aa2c0..ce2672d83 100644 --- a/src/winml/modelkit/session/ep_device.py +++ b/src/winml/modelkit/session/ep_device.py @@ -44,6 +44,7 @@ EP_ALIASES, EP_NAMES, EP_SUPPORTED_DEVICES, + DeviceType, EPName, normalize_ep_name, ) @@ -354,7 +355,7 @@ class EPDeviceSpec: """ ep: EPName - device: str + device: DeviceType default_provider_options: Mapping[str, str] = field(default_factory=dict) provider_option_hints: Mapping[str, str] = field(default_factory=dict) @@ -503,7 +504,7 @@ def default_ep_for_device(device: str) -> str | None: if s.device == device and _is_policy_supported_spec(s) and s.ep in eps # L0: discovered - and EP_CATALOG.is_compatible(s.ep) # L2: vendor-compatible + and EP_CATALOG.is_compatible(s.ep, s.device) # L2: vendor-compatible ), None, ) @@ -619,7 +620,7 @@ def auto_detect_device() -> str: spec.device != dev or not _is_policy_supported_spec(spec) or spec.ep not in available_eps - or not EP_CATALOG.is_compatible(spec.ep) + or not EP_CATALOG.is_compatible(spec.ep, spec.device) ): continue try: @@ -699,7 +700,7 @@ def resolve_device(target: EPDeviceTarget) -> EPDeviceTarget: continue else: try: - if not EP_CATALOG.is_compatible(spec.ep): + if not EP_CATALOG.is_compatible(spec.ep, spec.device): continue except RuntimeError as e: logger.warning( @@ -771,7 +772,7 @@ def resolve_device(target: EPDeviceTarget) -> EPDeviceTarget: ): continue try: - if not EP_CATALOG.is_compatible(spec.ep): + if not EP_CATALOG.is_compatible(spec.ep, spec.device): continue except RuntimeError as e: if not vendor_detection_failed: diff --git a/tests/cli/test_main.py b/tests/cli/test_main.py index 839594786..99d8fcba5 100644 --- a/tests/cli/test_main.py +++ b/tests/cli/test_main.py @@ -299,7 +299,7 @@ def _isolate_for_e2e(self, monkeypatch): monkeypatch.setattr( _ep, "_get_detected_vendors", - lambda: frozenset({"Qualcomm Inc"}), + lambda *_device_types: frozenset({"Qualcomm Inc"}), ) def test_json_shape_has_all_required_fields(self, runner: CliRunner) -> None: diff --git a/tests/unit/config/test_build.py b/tests/unit/config/test_build.py index 81bb3f653..fe9c494e5 100644 --- a/tests/unit/config/test_build.py +++ b/tests/unit/config/test_build.py @@ -4093,7 +4093,7 @@ def test_auto_cpu_survives_vendor_probe_failure(self) -> None: from winml.modelkit.ep_path import EPCatalog from winml.modelkit.session import WinMLEPRegistry - def _compatible(ep: str) -> bool: + def _compatible(ep: str, _device_type: str | None = None) -> bool: if ep == "OpenVINOExecutionProvider": raise RuntimeError("WMI unavailable") return True diff --git a/tests/unit/ep_path/test_compat.py b/tests/unit/ep_path/test_compat.py index 1592fc464..803d3d22c 100644 --- a/tests/unit/ep_path/test_compat.py +++ b/tests/unit/ep_path/test_compat.py @@ -6,7 +6,7 @@ Covers: - ``EP_CATALOG`` vendor requirement contents. - - ``_get_detected_vendors()`` aggregation across GPU/NPU. + - ``_get_detected_vendors()`` aggregation across CPU/GPU/NPU. - ``EP_CATALOG.is_compatible()`` matching rules. - ``is_compatible()`` method on every EPSource subclass. """ @@ -93,7 +93,7 @@ def test_empty_requirement_always_compatible( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset(), + lambda *_device_types: frozenset(), ) assert EP_CATALOG.is_compatible("CPUExecutionProvider") is True assert EP_CATALOG.is_compatible("DmlExecutionProvider") is True @@ -107,7 +107,7 @@ def test_unknown_ep_defaults_compatible( # is not silently hidden in `--list-ep`. monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset(), + lambda *_device_types: frozenset(), ) assert EP_CATALOG.is_compatible("FutureEpNotInTable") is True @@ -116,9 +116,30 @@ def test_qualcomm_substring_match( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"Qualcomm Technologies, Inc."}), + lambda *_device_types: frozenset({"Qualcomm Technologies, Inc."}), ) assert EP_CATALOG.is_compatible("QNNExecutionProvider") is True + + def test_custom_catalog_falls_back_to_all_device_classes( + self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch + ) -> None: + from winml.modelkit.ep_path import EPCatalog + + catalog = EPCatalog( + [ + EPCatalog.Row( + name="CustomExecutionProvider", + dll_name="custom.dll", + vendor_requirements=frozenset({"Example Vendor"}), + ) + ] + ) + monkeypatch.setattr( + "winml.modelkit.ep_path._get_detected_vendors", + lambda *_device_types: frozenset({"Example Vendor CPU"}), + ) + + assert catalog.is_compatible("CustomExecutionProvider") is True assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider") is False assert EP_CATALOG.is_compatible("NvTensorRTRTXExecutionProvider") is False @@ -129,7 +150,7 @@ def test_intel_substring_match_case_insensitive( # — substring lowercase match accepts any. monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"intel(r) corporation"}), + lambda *_device_types: frozenset({"intel(r) corporation"}), ) assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider") is True @@ -138,7 +159,7 @@ def test_amd_matches_both_vitisai_and_migraphx( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"AMD Radeon Graphics"}), + lambda *_device_types: frozenset({"AMD Radeon Graphics"}), ) assert EP_CATALOG.is_compatible("VitisAIExecutionProvider") is True assert EP_CATALOG.is_compatible("MIGraphXExecutionProvider") is True @@ -148,7 +169,7 @@ def test_no_vendor_detected_means_constrained_eps_incompatible( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset(), + lambda *_device_types: frozenset(), ) assert EP_CATALOG.is_compatible("QNNExecutionProvider") is False assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider") is False @@ -163,7 +184,9 @@ def test_partial_match_within_long_vendor_string( # Substring matching handles this. monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"Snapdragon(R) X Elite - Qualcomm(R) Hexagon(TM) NPU"}), + lambda *_device_types: frozenset( + {"Snapdragon(R) X Elite - Qualcomm(R) Hexagon(TM) NPU"} + ), ) assert EP_CATALOG.is_compatible("QNNExecutionProvider") is True @@ -174,11 +197,14 @@ def test_partial_match_within_long_vendor_string( class TestGetDetectedVendors: - """``_get_detected_vendors`` aggregates GPU.manufacturer/name + NPU.manufacturer/name.""" + """``_get_detected_vendors`` aggregates requested hardware classes.""" - def test_aggregates_gpu_and_npu( + def test_aggregates_cpu_gpu_and_npu( self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch ) -> None: + cpu = MagicMock() + cpu.manufacturer = "GenuineIntel" + cpu.name = "Intel Core Ultra" gpu = MagicMock() gpu.manufacturer = "NVIDIA Corporation" gpu.name = "NVIDIA RTX 4090" @@ -186,15 +212,54 @@ def test_aggregates_gpu_and_npu( npu.manufacturer = "Intel Corporation" npu.name = "Intel AI Boost" + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.CPU.get_all", lambda: [cpu]) monkeypatch.setattr("winml.modelkit.sysinfo.hardware.GPU.get_all", lambda: [gpu]) monkeypatch.setattr("winml.modelkit.sysinfo.hardware.NPU.get_all", lambda: [npu]) result = _get_detected_vendors() + assert "GenuineIntel" in result + assert "Intel Core Ultra" in result assert "NVIDIA Corporation" in result assert "NVIDIA RTX 4090" in result assert "Intel Corporation" in result assert "Intel AI Boost" in result + def test_compatibility_uses_only_ep_supported_device_classes( + self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch + ) -> None: + cpu = MagicMock(manufacturer="GenuineIntel", name="Intel Core i9") + gpu = MagicMock(manufacturer="NVIDIA", name="NVIDIA RTX 4080") + get_cpus = MagicMock(return_value=[cpu]) + get_gpus = MagicMock(return_value=[gpu]) + get_npus = MagicMock(return_value=[]) + + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.CPU.get_all", get_cpus) + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.GPU.get_all", get_gpus) + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.NPU.get_all", get_npus) + + assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider") is True + assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider", "cpu") is True + assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider", "gpu") is False + assert EP_CATALOG.is_compatible("OpenVINOExecutionProvider", "npu") is False + assert EP_CATALOG.is_compatible("NvTensorRTRTXExecutionProvider") is True + assert EP_CATALOG.is_compatible("QNNExecutionProvider") is False + assert get_cpus.call_count == 1 + assert get_gpus.call_count == 1 + assert get_npus.call_count == 1 + + def test_cpu_vendor_does_not_enable_accelerator_only_eps( + self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch + ) -> None: + cpu = MagicMock(manufacturer="AuthenticAMD", name="AMD Ryzen") + gpu = MagicMock(manufacturer="NVIDIA", name="NVIDIA RTX 4080") + + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.CPU.get_all", lambda: [cpu]) + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.GPU.get_all", lambda: [gpu]) + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.NPU.get_all", list) + + assert EP_CATALOG.is_compatible("MIGraphXExecutionProvider") is False + assert EP_CATALOG.is_compatible("VitisAIExecutionProvider") is False + def test_handles_missing_attribute( self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -203,34 +268,40 @@ def test_handles_missing_attribute( gpu = MagicMock(spec=["manufacturer"]) gpu.manufacturer = "AMD" + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.CPU.get_all", list) monkeypatch.setattr("winml.modelkit.sysinfo.hardware.GPU.get_all", lambda: [gpu]) monkeypatch.setattr("winml.modelkit.sysinfo.hardware.NPU.get_all", list) result = _get_detected_vendors() assert result == frozenset({"AMD"}) + @pytest.mark.parametrize("hardware_class", ["CPU", "GPU", "NPU"]) def test_get_all_failure_raises( - self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch + self, + reset_vendor_cache: None, + monkeypatch: pytest.MonkeyPatch, + hardware_class: str, ) -> None: - # If GPU.get_all raises (WMI failure), the whole detection fails with RuntimeError. + # If any requested class fails, detection fails instead of caching partial results. # (Old behavior was to swallow the error and continue; new behavior is to raise # so functools.cache doesn't pin a false "no hardware" result.) - npu = MagicMock() - npu.manufacturer = "Qualcomm" - npu.name = "Qualcomm Hexagon" - def raise_wmi() -> list: raise RuntimeError("WMI down") - monkeypatch.setattr("winml.modelkit.sysinfo.hardware.GPU.get_all", raise_wmi) - monkeypatch.setattr("winml.modelkit.sysinfo.hardware.NPU.get_all", lambda: [npu]) + for class_name in ("CPU", "GPU", "NPU"): + implementation = raise_wmi if class_name == hardware_class else list + monkeypatch.setattr( + f"winml.modelkit.sysinfo.hardware.{class_name}.get_all", + implementation, + ) - with pytest.raises(RuntimeError, match=r"GPU\.get_all"): + with pytest.raises(RuntimeError, match=rf"{hardware_class}\.get_all"): _get_detected_vendors() def test_no_hardware_returns_empty( self, reset_vendor_cache: None, monkeypatch: pytest.MonkeyPatch ) -> None: + monkeypatch.setattr("winml.modelkit.sysinfo.hardware.CPU.get_all", list) monkeypatch.setattr("winml.modelkit.sysinfo.hardware.GPU.get_all", list) monkeypatch.setattr("winml.modelkit.sysinfo.hardware.NPU.get_all", list) assert _get_detected_vendors() == frozenset() @@ -249,7 +320,7 @@ def test_pypi_source_compatible( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"Qualcomm Inc"}), + lambda *_device_types: frozenset({"Qualcomm Inc"}), ) src = PyPISource( distribution="onnxruntime-qnn", @@ -264,7 +335,7 @@ def test_pypi_source_incompatible( # OpenVINO PyPI on a Snapdragon-only box. monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"Qualcomm Inc"}), + lambda *_device_types: frozenset({"Qualcomm Inc"}), ) src = PyPISource( distribution="onnxruntime-ep-openvino", @@ -278,7 +349,7 @@ def test_filesystem_source_uses_dll_patterns_keys( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"AMD"}), + lambda *_device_types: frozenset({"AMD"}), ) src = DirectorySource( root=Path("ignored"), @@ -291,7 +362,7 @@ def test_winml_catalog_source_compatible( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"NVIDIA Corp"}), + lambda *_device_types: frozenset({"NVIDIA Corp"}), ) src = WinMLCatalogSource( catalog_name="NvTensorRTRTXExecutionProvider", @@ -304,7 +375,7 @@ def test_msix_package_source_incompatible( ) -> None: monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"Qualcomm"}), + lambda *_device_types: frozenset({"Qualcomm"}), ) src = MSIXPackageSource( family_name_prefix="...OpenVINO.EP._...", @@ -321,7 +392,7 @@ def test_multi_ep_source_all_must_match( # EP — but the contract should be strict.) monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"AMD"}), + lambda *_device_types: frozenset({"AMD"}), ) # AMD-only box: VitisAI ok, but QNN and OpenVINO not. src = DirectorySource( @@ -387,7 +458,7 @@ def test_iter_eps_drives_is_compatible( # actual driver (not a hardcoded path through self.eps). monkeypatch.setattr( "winml.modelkit.ep_path._get_detected_vendors", - lambda: frozenset({"AMD"}), + lambda *_device_types: frozenset({"AMD"}), ) ok_src = DirectorySource( root=Path("ignored"), diff --git a/tests/unit/ep_path/test_nuget_source.py b/tests/unit/ep_path/test_nuget_source.py index f315cbc52..0e2189b97 100644 --- a/tests/unit/ep_path/test_nuget_source.py +++ b/tests/unit/ep_path/test_nuget_source.py @@ -263,7 +263,7 @@ def test_is_compatible_matches_hardware( monkeypatch.setattr( _ep, "_get_detected_vendors", - lambda: frozenset({"Intel(R) Corporation"}), + lambda *_device_types: frozenset({"Intel(R) Corporation"}), ) ov_src = NuGetSource( distribution="Intel.ML.OnnxRuntime.EP.OpenVINO", diff --git a/tests/unit/session/test_ep_device.py b/tests/unit/session/test_ep_device.py index 727a6a135..b6e175c6b 100644 --- a/tests/unit/session/test_ep_device.py +++ b/tests/unit/session/test_ep_device.py @@ -332,7 +332,7 @@ def test_resolve_auto_ep_skips_failed_vendor_probe_for_cpu_fallback() -> None: ) registry.auto_device.return_value = object() - def check_compatibility(ep_name: str) -> bool: + def check_compatibility(ep_name: str, _device_type: str | None = None) -> bool: if ep_name == "OpenVINOExecutionProvider": raise RuntimeError("WMI unavailable") return True @@ -901,7 +901,7 @@ class slot. Unknown EPs default to True (forward-compat — matches the catalog's behavior for EPs without vendor_requirements). """ - def fake_is_compatible(self, ep_name: str) -> bool: + def fake_is_compatible(self, ep_name: str, _device_type: str | None = None) -> bool: return compatible_map.get(ep_name, True) return (