Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/winml/modelkit/config/build.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
84 changes: 54 additions & 30 deletions src/winml/modelkit/ep_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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, ...]:
Expand Down Expand Up @@ -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.
# ---------------------------------------------------------------------------
Expand Down
11 changes: 6 additions & 5 deletions src/winml/modelkit/session/ep_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
EP_ALIASES,
EP_NAMES,
EP_SUPPORTED_DEVICES,
DeviceType,
EPName,
normalize_ep_name,
)
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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,
)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion tests/cli/test_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion tests/unit/config/test_build.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading