Skip to content
Merged
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
43 changes: 37 additions & 6 deletions src/winml/modelkit/commands/analyze.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
from ..session import DEVICE_TYPE_TO_DEVICE
from ..utils import cli as cli_utils
from ..utils.constants import (
DEVICE_PRIORITY,
EP_SUPPORTED_DEVICES,
SUPPORTED_DEVICES,
SUPPORTED_EPS,
Expand Down Expand Up @@ -643,6 +644,23 @@ def _select_best_exact_local_pair_for_device(
return local_candidates[0]


def _select_best_auto_local_pair(
supported_local_pairs: list[tuple[EPName, str]],
) -> tuple[EPName, str] | None:
"""Pick the best default target from exact local bindings."""
from ..session import available_eps_for_device

for device_name in DEVICE_PRIORITY:
best_local_pair = _select_best_exact_local_pair_for_device(
device_name.upper(),
supported_local_pairs,
available_eps_for_device(device_name),
)
if best_local_pair is not None:
return best_local_pair
return None


def _ep_name_device_display_name(ep_name: str, device_name: str) -> str:
"""Return EP/device label for table and summary display."""
return f"{ep_name} ({device_name.upper()})"
Expand Down Expand Up @@ -959,6 +977,11 @@ def analyze(
_filter_supported_local_ep_device_pairs(local_pair_list)
)
local_pairs = set(supported_local_pair_list)
default_auto_pair = (
_select_best_auto_local_pair(supported_local_pair_list)
if ep in (None, "auto") and device in (None, "auto")
else None
)

devices: list[str]
if device == "all":
Expand All @@ -976,6 +999,8 @@ def analyze(
f"{ep} has no supported local binding available on this system."
)
devices = [matching_local_pairs[0][1]]
elif default_auto_pair is not None:
devices = [default_auto_pair[1]]
else:
try:
resolved_device = resolve_device(
Expand Down Expand Up @@ -1014,14 +1039,20 @@ def analyze(
# context exists here -- but guard against an empty device list
# (e.g. a programmatic ``device=None`` call) so we exit cleanly
# instead of raising an unguarded IndexError on ``devices[0]``.
ref_device = devices[0] if devices else None
ref_device = default_auto_pair[1] if default_auto_pair is not None else None
ref_device = ref_device or (devices[0] if devices else None)
if default_auto_pair is not None:
best_local_pair = default_auto_pair
elif ref_device:
best_local_pair = _select_best_exact_local_pair_for_device(
ref_device,
supported_local_pair_list,
available_eps_for_device(ref_device),
)
else:
best_local_pair = None
if not ref_device:
raise click.UsageError("No device context available for EP auto-resolution.")
best_local_pair = _select_best_exact_local_pair_for_device(
ref_device,
supported_local_pair_list,
available_eps_for_device(ref_device),
)
if best_local_pair is None:
raise click.UsageError(
f"No execution provider is available for device '{ref_device}'."
Expand Down
55 changes: 55 additions & 0 deletions tests/unit/analyze/test_static_analyzer_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1459,6 +1459,61 @@ def test_concrete_ep_auto_device_requires_supported_local_binding(
assert "available on this system" in result.output.lower()
assert not mock_analyzer_class.called

@patch("winml.modelkit.analyze.ONNXStaticAnalyzer")
def test_auto_ep_auto_device_uses_device_priority_before_resolved_device_fallback(
self,
mock_analyzer_class: MagicMock,
runner: CliRunner,
tmp_path: Path,
mock_analyzer_result: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Default analyze target should use the strongest exact local binding."""
model_file = tmp_path / "test.onnx"
model_file.write_bytes(b"dummy")

multi_device_ep = next(
ep_name
for ep_name, devices in EP_SUPPORTED_DEVICES.items()
if {"npu", "gpu"}.issubset(devices)
)

monkeypatch.setattr(
"winml.modelkit.commands.analyze._get_local_ep_device_pairs",
lambda: [
(multi_device_ep, "GPU"),
(multi_device_ep, "NPU"),
("CPUExecutionProvider", "CPU"),
],
)
monkeypatch.setattr(
"winml.modelkit.session.resolve_device",
lambda target: type(target)(
ep=multi_device_ep,
device="gpu",
source=target.source,
),
)
monkeypatch.setattr(
"winml.modelkit.session.available_eps_for_device",
lambda device_name: (
[multi_device_ep] if str(device_name).lower() in {"gpu", "npu"} else []
),
)

mock_instance = Mock()
mock_instance.analyze.return_value = mock_analyzer_result
mock_analyzer_class.return_value = mock_instance

result = runner.invoke(analyze, ["--model", str(model_file)])
assert result.exit_code == 0

actual_calls = [
(call.kwargs["ep"], call.kwargs["device"])
for call in mock_instance.analyze.call_args_list
]
assert actual_calls == [(multi_device_ep, "NPU")]

@pytest.mark.parametrize(
"ranked_gpu_eps",
[
Expand Down
Loading