diff --git a/src/winml/modelkit/commands/perf.py b/src/winml/modelkit/commands/perf.py index aae0ce7a7..f8a0fca4d 100644 --- a/src/winml/modelkit/commands/perf.py +++ b/src/winml/modelkit/commands/perf.py @@ -75,6 +75,24 @@ RUNTIME_NAMES: tuple[RuntimeName, ...] = get_args(RuntimeName) +def _detail_fallback_guidance(reason: str | None) -> str: + """Return actionable guidance for a structured detail-trace fallback.""" + guidance = { + "qnn_log_missing": "the QNN optrace log was not produced", + "schematic_missing": ( + "the compiled EPContext has no optrace schematic; rerun detail " + "profiling from the raw ONNX so WinML can compile it with optrace enabled" + ), + "sdk_missing": "the QNN SDK was not found; set QNN_SDK_ROOT to enable QHAS", + "viewer_failed": "the QHAS viewer did not produce an output", + "qhas_output_missing": "the requested QHAS output was not found", + "qhas_parse_failed": "the QHAS output could not be parsed", + } + if reason is None: + return "QHAS post-processing was unavailable" + return guidance.get(reason, "QHAS post-processing was unavailable") + + class _NativeWarningFilteredPerfContext: """Filter native warnings from session.perf enter/exit without wrapping the loop.""" @@ -3235,9 +3253,9 @@ def perf( "EPContext model, but the benchmark ran the original ONNX model." ) sys.exit(4) + detail = _detail_fallback_guidance(trace_result.fallback_reason) console.print( - "[yellow]Notice:[/yellow] Detail mode degraded to basic CSV " - "(QHAS unavailable; set QNN_SDK_ROOT to enable)." + f"[yellow]Notice:[/yellow] Detail mode degraded to basic CSV ({detail})." ) if json_mode: diff --git a/src/winml/modelkit/session/monitor/op_metrics.py b/src/winml/modelkit/session/monitor/op_metrics.py index 4761a3b08..9866763af 100644 --- a/src/winml/modelkit/session/monitor/op_metrics.py +++ b/src/winml/modelkit/session/monitor/op_metrics.py @@ -32,6 +32,16 @@ #: serialization are unaffected. TraceStatus = Literal["ok", "no_data", "parse_failed", "basic_fallback", "not_run"] +#: Machine-readable reason why a requested detail trace degraded to basic data. +TraceFallbackReason = Literal[ + "qnn_log_missing", + "schematic_missing", + "sdk_missing", + "viewer_failed", + "qhas_output_missing", + "qhas_parse_failed", +] + @dataclass class OperatorMetrics: @@ -145,9 +155,15 @@ class OpTraceResult: # Status of the trace. See :data:`TraceStatus` for the closed set of # legal values; static type checkers enforce the alias. status: TraceStatus = "ok" + # Populated when status == "basic_fallback". + fallback_reason: TraceFallbackReason | None = None # Populated when status == "parse_failed". error: str | None = None + def __post_init__(self) -> None: + if self.fallback_reason is not None and self.status != "basic_fallback": + raise ValueError("fallback_reason requires status='basic_fallback'") + def to_dict(self) -> dict[str, Any]: """Serialize to structured dict. @@ -170,6 +186,7 @@ def to_dict(self) -> dict[str, Any]: "artifacts": self.artifacts, # ---- Additive ---- "status": self.status, + "fallback_reason": self.fallback_reason, "error": self.error, } diff --git a/src/winml/modelkit/session/monitor/qnn/viewer.py b/src/winml/modelkit/session/monitor/qnn/viewer.py index 79ede9ee1..5fa2a41f6 100644 --- a/src/winml/modelkit/session/monitor/qnn/viewer.py +++ b/src/winml/modelkit/session/monitor/qnn/viewer.py @@ -21,8 +21,9 @@ import logging import os import subprocess +from dataclasses import dataclass from pathlib import Path -from typing import Any +from typing import Any, Literal logger = logging.getLogger(__name__) @@ -50,6 +51,18 @@ _QHAS_SUMMARY_SUFFIX = "_qnn_htp_analysis_summary.json" +@dataclass(frozen=True) +class QHASViewerResult: + """Detailed outcome of QHAS viewer preparation and execution.""" + + path: Path | None + failure_reason: Literal["viewer_failed", "qhas_output_missing"] | None + + def __post_init__(self) -> None: + if (self.path is None) == (self.failure_reason is None): + raise ValueError("QHAS viewer result requires exactly one outcome") + + def find_qnn_sdk() -> Path | None: """Auto-detect a QNN SDK from the environment or documented common roots. @@ -175,57 +188,76 @@ def run_qhas_viewer( ------- Path to the generated QNN HTP analysis summary JSON, or ``None`` on failure. """ - viewer = _find_viewer_exe(sdk_root) - if viewer is None: - logger.warning( - "qnn-profile-viewer not found; set QNN_SDK_ROOT to enable detail mode " - "(falling back to basic CSV)" - ) - return None - reader = _find_optrace_reader(viewer) - if reader is None: - logger.warning( - "%s not found for qnn-profile-viewer at %s; falling back to basic CSV", - _OPTRACE_READER_NAME, - viewer, - ) - return None + return run_qhas_viewer_result( + qnn_log, + schematic, + output, + config, + sdk_root=sdk_root, + ).path - if not schematic.is_file(): - logger.warning("Schematic file not found: %s", schematic) - return None - - # Write the config next to the output and bind it to that run's artifact stem. - cfg = config if config is not None else _DEFAULT_CONFIG - config_path = output.with_name(f"{output.stem}_optrace_config.json") - config_path.write_text(json.dumps(cfg, indent=2), encoding="utf-8") - - cmd = [ - str(viewer), - "--input_log", - str(qnn_log), - "--output", - str(output), - "--reader", - str(reader), - "--schematic", - str(schematic), - "--config", - str(config_path), - ] - logger.info("Running QHAS viewer: %s", " ".join(cmd)) +def run_qhas_viewer_result( + qnn_log: Path, + schematic: Path, + output: Path, + config: dict[str, Any] | None = None, + *, + sdk_root: Path | None = None, +) -> QHASViewerResult: + """Run QHAS viewer and distinguish execution from missing-output failures.""" try: + viewer = _find_viewer_exe(sdk_root) + if viewer is None: + logger.warning( + "qnn-profile-viewer not found; set QNN_SDK_ROOT to enable detail mode " + "(falling back to basic CSV)" + ) + return QHASViewerResult(path=None, failure_reason="viewer_failed") + reader = _find_optrace_reader(viewer) + if reader is None: + logger.warning( + "%s not found for qnn-profile-viewer at %s; falling back to basic CSV", + _OPTRACE_READER_NAME, + viewer, + ) + return QHASViewerResult(path=None, failure_reason="viewer_failed") + + if not schematic.is_file(): + logger.warning("Schematic file not found: %s", schematic) + return QHASViewerResult(path=None, failure_reason="viewer_failed") + + cfg = config if config is not None else _DEFAULT_CONFIG + config_path = output.with_name(f"{output.stem}_optrace_config.json") + config_path.write_text(json.dumps(cfg, indent=2), encoding="utf-8") + + cmd = [ + str(viewer), + "--input_log", + str(qnn_log), + "--output", + str(output), + "--reader", + str(reader), + "--schematic", + str(schematic), + "--config", + str(config_path), + ] + logger.info("Running QHAS viewer: %s", " ".join(cmd)) subprocess.run(cmd, check=True, capture_output=True, text=True) # noqa: S603 except subprocess.CalledProcessError as exc: logger.error("QHAS viewer failed: %s", exc.stderr) - return None - except FileNotFoundError: - logger.error("qnn-profile-viewer executable not found at %s", viewer) - return None + return QHASViewerResult(path=None, failure_reason="viewer_failed") + except (OSError, TypeError, ValueError) as exc: + logger.error("QHAS viewer preparation or execution failed: %s", exc) + return QHASViewerResult(path=None, failure_reason="viewer_failed") summary_output = output.with_name(f"{output.stem}{_QHAS_SUMMARY_SUFFIX}") - if summary_output.is_file(): - return summary_output + try: + if summary_output.is_file(): + return QHASViewerResult(path=summary_output, failure_reason=None) + except OSError as exc: + logger.warning("Could not inspect QHAS analysis summary %s: %s", summary_output, exc) logger.warning("QHAS viewer did not produce analysis summary: %s", summary_output) - return None + return QHASViewerResult(path=None, failure_reason="qhas_output_missing") diff --git a/src/winml/modelkit/session/monitor/qnn_monitor.py b/src/winml/modelkit/session/monitor/qnn_monitor.py index 5eb39c935..59bfebbdb 100644 --- a/src/winml/modelkit/session/monitor/qnn_monitor.py +++ b/src/winml/modelkit/session/monitor/qnn_monitor.py @@ -30,9 +30,14 @@ from ...onnx.epcontext import select_main_epcontext_partition_name from ._onnx_metadata import _load_onnx_operator_data from .ep_monitor import WinMLEPMonitor -from .op_metrics import OperatorMetrics, OpTraceResult, TraceStatus +from .op_metrics import ( + OperatorMetrics, + OpTraceResult, + TraceFallbackReason, + TraceStatus, +) from .qnn._internal import _TOKEN_SUFFIX, parse_qhas, parse_qnn_profiling_csv -from .qnn.viewer import find_qnn_sdk, run_qhas_viewer +from .qnn.viewer import find_qnn_sdk, run_qhas_viewer_result if TYPE_CHECKING: @@ -628,9 +633,10 @@ def _metadata_mean(field: str) -> float: } status: TraceStatus = "ok" + fallback_reason: TraceFallbackReason | None = None # Detail mode: attempt QHAS post-processing. if self._level == "detail": - qhas_summary, qhas_operators, qhas_path = self._try_qhas( + qhas_summary, qhas_operators, qhas_path, fallback_reason = self._try_qhas( artifacts, qhas_override=qhas_override ) if qhas_path is not None and qhas_operators is not None: @@ -653,17 +659,23 @@ def _metadata_mean(field: str) -> float: num_samples=len(samples), artifacts=artifacts, status=status, + fallback_reason=fallback_reason, ) def _try_qhas( self, artifacts: dict[str, str], qhas_override: Path | None = None, - ) -> tuple[dict[str, Any] | None, list[OperatorMetrics] | None, Path | None]: + ) -> tuple[ + dict[str, Any] | None, + list[OperatorMetrics] | None, + Path | None, + TraceFallbackReason | None, + ]: """Attempt QHAS post-processing. - Returns ``(summary, operators, qhas_path)`` on success, or - ``(None, None, None)`` on any failure. Never raises. + Returns ``(summary, operators, qhas_path, fallback_reason)``. The + reason is ``None`` on success and a stable code on failure. Never raises. Per C-5 / FR-12 this method does NOT call :func:`os.chdir`. Live-path QNN logs are bound by the profiling CSV stem: ORT writes @@ -680,34 +692,55 @@ def _try_qhas( """ if qhas_override is not None: # Offline path: caller supplied the QHAS JSON; parse directly. - if not qhas_override.is_file(): + try: + qhas_available = qhas_override.is_file() + except OSError as exc: + logger.info("QNNMonitor: qhas_override %s is unavailable: %s", qhas_override, exc) + return None, None, None, "qhas_output_missing" + if not qhas_available: logger.info("QNNMonitor: qhas_override %s is not a file", qhas_override) - return None, None, None + return None, None, None, "qhas_output_missing" result_path = qhas_override else: # Live path: locate inputs and shell out to the QHAS viewer. - qnn_log = self._select_fresh_qnn_log() + try: + qnn_log = self._select_fresh_qnn_log() + except OSError as exc: + logger.info("QNNMonitor: QNN log metadata unavailable: %s", exc) + return None, None, None, "qnn_log_missing" if qnn_log is None: logger.info("QNNMonitor: no *_qnn.log found for QHAS") - return None, None, None + return None, None, None, "qnn_log_missing" # Find the schematic by EPContext partition metadata (never chdir). schematic = self._find_schematic() if schematic is None: logger.info("QNNMonitor: no *_schematic.bin found for QHAS") - return None, None, None + return None, None, None, "schematic_missing" - sdk_root = find_qnn_sdk() + try: + sdk_root = find_qnn_sdk() + except OSError as exc: + logger.info("QNNMonitor: QNN SDK discovery failed: %s", exc) + return None, None, None, "sdk_missing" if sdk_root is None: logger.info("QNNMonitor: QNN SDK not located; skipping QHAS") - return None, None, None + return None, None, None, "sdk_missing" qhas_output = self._qhas_output_path() - viewer_output = run_qhas_viewer(qnn_log, schematic, qhas_output, sdk_root=sdk_root) - if viewer_output is None or not viewer_output.is_file(): - logger.info("QNNMonitor: QHAS viewer produced no output") - return None, None, None - result_path = viewer_output + viewer_result = run_qhas_viewer_result( + qnn_log, + schematic, + qhas_output, + sdk_root=sdk_root, + ) + if viewer_result.path is None: + logger.info( + "QNNMonitor: QHAS viewer unavailable (%s)", + viewer_result.failure_reason, + ) + return None, None, None, viewer_result.failure_reason + result_path = viewer_result.path artifacts["schematic"] = str(schematic) @@ -716,7 +749,7 @@ def _try_qhas( parsed = parse_qhas(qhas_data) except Exception as exc: logger.warning("QNNMonitor: QHAS JSON parse failed: %s", exc) - return None, None, None + return None, None, None, "qhas_parse_failed" # QHAS is inherently a single-snapshot summary (no per-sample # breakdown), so ``samples_us`` carries one entry equal to the @@ -747,12 +780,16 @@ def _try_qhas( ) for op in parsed.get("operators", []) ] - return parsed.get("summary"), operators, result_path + return parsed.get("summary"), operators, result_path, None def _snapshot_qnn_log_signatures(self) -> dict[Path, tuple[int, int, int, int, int]]: """Capture this run's QNN log metadata at monitor entry.""" candidate = self._qnn_log_path() - signature = self._artifact_signature(candidate) + try: + signature = self._artifact_signature(candidate) + except OSError as exc: + logger.info("QNNMonitor: unable to snapshot QNN log metadata: %s", exc) + signature = None signatures: dict[Path, tuple[int, int, int, int, int]] = {} if signature is not None: signatures[candidate.resolve()] = signature diff --git a/tests/unit/commands/test_perf_optracing.py b/tests/unit/commands/test_perf_optracing.py index c33466045..abdf76a3d 100644 --- a/tests/unit/commands/test_perf_optracing.py +++ b/tests/unit/commands/test_perf_optracing.py @@ -923,6 +923,7 @@ def test_basic_fallback_status_exits_0_with_notice(self, tmp_path: Path): device="npu", tracing_level="detail", status="basic_fallback", + fallback_reason="schematic_missing", ) mock_ctx = MagicMock() mock_ctx.monitor.result = trace @@ -949,7 +950,28 @@ def test_basic_fallback_status_exits_0_with_notice(self, tmp_path: Path): ) assert result.exit_code == 0, f"Expected exit 0, got {result.exit_code}: {result.output}" - assert "degraded" in result.output.lower() or "notice" in result.output.lower() + assert "schematic" in result.output.lower() + assert "raw onnx" in result.output.lower() + assert "qnn_sdk_root" not in result.output.lower() + + @pytest.mark.parametrize( + ("reason", "expected"), + [ + ("qnn_log_missing", "optrace log"), + ("schematic_missing", "raw ONNX"), + ("sdk_missing", "QNN_SDK_ROOT"), + ("viewer_failed", "viewer"), + ("qhas_output_missing", "output was not found"), + ("qhas_parse_failed", "could not be parsed"), + (None, "post-processing was unavailable"), + ], + ) + def test_detail_fallback_guidance_is_reason_specific( + self, reason: str | None, expected: str + ) -> None: + from winml.modelkit.commands.perf import _detail_fallback_guidance + + assert expected in _detail_fallback_guidance(reason) def test_basic_fallback_status_rejects_raw_running_model(self, tmp_path: Path): """Detail tracing cannot degrade successfully when ORT ran the raw model.""" diff --git a/tests/unit/session/monitor/qnn/test_viewer.py b/tests/unit/session/monitor/qnn/test_viewer.py index 87cbd14f4..e597a6189 100644 --- a/tests/unit/session/monitor/qnn/test_viewer.py +++ b/tests/unit/session/monitor/qnn/test_viewer.py @@ -186,3 +186,99 @@ def _fake_run(cmd: list[str], **_kwargs): assert run_qhas_viewer(qnn_log, schematic, output, sdk_root=sdk_root) is None assert "did not produce analysis summary" in caplog.text + + +def test_run_qhas_viewer_result_reports_config_write_failure(monkeypatch, tmp_path): + """Viewer preparation failures remain detail fallbacks, not parse failures.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn.viewer import run_qhas_viewer_result + + sdk_root = tmp_path / "sdk" + viewer_path = sdk_root / "bin" / "x64" / "qnn-profile-viewer.exe" + viewer_path.parent.mkdir(parents=True) + viewer_path.write_bytes(b"") + reader = sdk_root / "lib" / viewer_path.parent.name / "QnnHtpOptraceProfilingReader.dll" + reader.parent.mkdir(parents=True) + reader.write_bytes(b"") + qnn_log = tmp_path / "profile.log" + schematic = tmp_path / "schematic.bin" + qnn_log.write_bytes(b"") + schematic.write_bytes(b"") + + def _fail_write_text(self: Path, *_args, **_kwargs): + raise PermissionError(f"read-only output: {self}") + + monkeypatch.setattr(Path, "write_text", _fail_write_text) + monkeypatch.setattr(subprocess, "run", pytest.fail) + + result = run_qhas_viewer_result( + qnn_log, + schematic, + tmp_path / "qhas_output.json", + sdk_root=sdk_root, + ) + + assert result.path is None + assert result.failure_reason == "viewer_failed" + + +def test_run_qhas_viewer_result_reports_config_serialization_failure(monkeypatch, tmp_path): + """Invalid custom viewer config remains a structured detail fallback.""" + from winml.modelkit.session.monitor.qnn.viewer import run_qhas_viewer_result + + sdk_root = tmp_path / "sdk" + viewer_path = sdk_root / "bin" / "x64" / "qnn-profile-viewer.exe" + viewer_path.parent.mkdir(parents=True) + viewer_path.write_bytes(b"") + reader = sdk_root / "lib" / viewer_path.parent.name / "QnnHtpOptraceProfilingReader.dll" + reader.parent.mkdir(parents=True) + reader.write_bytes(b"") + qnn_log = tmp_path / "profile.log" + schematic = tmp_path / "schematic.bin" + qnn_log.write_bytes(b"") + schematic.write_bytes(b"") + monkeypatch.setattr(subprocess, "run", pytest.fail) + + result = run_qhas_viewer_result( + qnn_log, + schematic, + tmp_path / "qhas_output.json", + config={"not_json": object()}, + sdk_root=sdk_root, + ) + + assert result.path is None + assert result.failure_reason == "viewer_failed" + + +def test_run_qhas_viewer_result_distinguishes_missing_summary(monkeypatch, tmp_path): + """A successful viewer process without its summary is output_missing.""" + from winml.modelkit.session.monitor.qnn.viewer import run_qhas_viewer_result + + sdk_root = tmp_path / "sdk" + viewer_path = sdk_root / "bin" / "x64" / "qnn-profile-viewer.exe" + viewer_path.parent.mkdir(parents=True) + viewer_path.write_bytes(b"") + reader = sdk_root / "lib" / viewer_path.parent.name / "QnnHtpOptraceProfilingReader.dll" + reader.parent.mkdir(parents=True) + reader.write_bytes(b"") + qnn_log = tmp_path / "profile.log" + schematic = tmp_path / "schematic.bin" + qnn_log.write_bytes(b"") + schematic.write_bytes(b"") + monkeypatch.setattr( + subprocess, + "run", + lambda cmd, **_kwargs: subprocess.CompletedProcess(cmd, 0, "", ""), + ) + + result = run_qhas_viewer_result( + qnn_log, + schematic, + tmp_path / "qhas_output.json", + sdk_root=sdk_root, + ) + + assert result.path is None + assert result.failure_reason == "qhas_output_missing" diff --git a/tests/unit/session/monitor/test_op_metrics.py b/tests/unit/session/monitor/test_op_metrics.py index 251dc5fe2..29bfdbfd2 100644 --- a/tests/unit/session/monitor/test_op_metrics.py +++ b/tests/unit/session/monitor/test_op_metrics.py @@ -8,6 +8,8 @@ import json +import pytest + from winml.modelkit.session.monitor.op_metrics import ( OperatorMetrics, OpTraceResult, @@ -146,6 +148,46 @@ def test_to_dict_status_only_accepts_known_values_per_typing() -> None: assert r.to_dict()["status"] == status +def test_to_dict_serializes_fallback_reason() -> None: + """Degraded detail traces expose a stable machine-readable reason.""" + r = OpTraceResult( + model=None, + device="npu", + tracing_level="detail", + status="basic_fallback", + fallback_reason="schematic_missing", + ) + + serialized = r.to_dict() + + assert serialized["fallback_reason"] == "schematic_missing" + assert serialized["error"] is None + + +def test_non_fallback_status_rejects_fallback_reason() -> None: + """A reason cannot be attached to a success or hard-failure status.""" + with pytest.raises(ValueError, match="fallback_reason"): + OpTraceResult( + model=None, + device="npu", + tracing_level="detail", + status="ok", + fallback_reason="sdk_missing", + ) + + +def test_legacy_basic_fallback_without_reason_remains_valid() -> None: + """Older direct callers may omit the additive reason field.""" + result = OpTraceResult( + model=None, + device="npu", + tracing_level="detail", + status="basic_fallback", + ) + + assert result.fallback_reason is None + + def test_trace_status_alias_importable() -> None: """``TraceStatus`` must be importable as a public symbol from op_metrics.""" from winml.modelkit.session.monitor.op_metrics import TraceStatus # noqa: F401 diff --git a/tests/unit/session/monitor/test_qnn_monitor.py b/tests/unit/session/monitor/test_qnn_monitor.py index d6e9d73dd..a0792b429 100644 --- a/tests/unit/session/monitor/test_qnn_monitor.py +++ b/tests/unit/session/monitor/test_qnn_monitor.py @@ -1071,6 +1071,7 @@ def test_find_schematic_malformed_main_context_falls_back_to_basic(tmp_path): def test_try_qhas_uses_csv_bound_artifacts(tmp_path, monkeypatch): + from winml.modelkit.session.monitor.qnn.viewer import QHASViewerResult from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor context_model = tmp_path / "model_ctx.onnx" @@ -1103,11 +1104,14 @@ def _run_qhas_viewer(log: Path, selected_schematic: Path, output: Path, *, sdk_r lambda: tmp_path, ) monkeypatch.setattr( - "winml.modelkit.session.monitor.qnn_monitor.run_qhas_viewer", - _run_qhas_viewer, + "winml.modelkit.session.monitor.qnn_monitor.run_qhas_viewer_result", + lambda *args, **kwargs: QHASViewerResult( + path=_run_qhas_viewer(*args, **kwargs), + failure_reason=None, + ), ) - summary, operators, result_path = monitor._try_qhas({}) + summary, operators, result_path, fallback_reason = monitor._try_qhas({}) assert seen_logs == [qnn_log] assert seen_schematics == [schematic] @@ -1115,9 +1119,11 @@ def _run_qhas_viewer(log: Path, selected_schematic: Path, output: Path, *, sdk_r assert summary is not None assert operators is not None assert result_path == qhas_output.with_name(f"{qhas_output.stem}_qnn_htp_analysis_summary.json") + assert fallback_reason is None def test_try_qhas_ignores_other_runs_newer_qnn_log(tmp_path, monkeypatch): + from winml.modelkit.session.monitor.qnn.viewer import QHASViewerResult from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor context_model = tmp_path / "model_ctx.onnx" @@ -1144,16 +1150,20 @@ def _run_qhas_viewer(qnn_log: Path, _schematic: Path, output: Path, *, sdk_root: lambda: tmp_path, ) monkeypatch.setattr( - "winml.modelkit.session.monitor.qnn_monitor.run_qhas_viewer", - _run_qhas_viewer, + "winml.modelkit.session.monitor.qnn_monitor.run_qhas_viewer_result", + lambda *args, **kwargs: QHASViewerResult( + path=_run_qhas_viewer(*args, **kwargs), + failure_reason=None, + ), ) - summary, operators, result_path = monitor._try_qhas({}) + summary, operators, result_path, fallback_reason = monitor._try_qhas({}) assert seen_logs == [] assert summary is None assert operators is None assert result_path is None + assert fallback_reason == "qnn_log_missing" def test_find_schematic_returns_none_when_csv_metadata_cannot_be_read(tmp_path, monkeypatch): @@ -1304,6 +1314,7 @@ def test_detail_mode_falls_back_to_basic_when_qhas_unavailable(tmp_path): assert monitor.result is not None assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == "qnn_log_missing" # CSV-only data must still be populated — basic_fallback is degraded # *success*, not failure: operators and summary are non-empty. assert monitor.result.operators, "expected CSV-derived operators in basic_fallback result" @@ -1318,6 +1329,194 @@ def test_detail_mode_falls_back_to_basic_when_qhas_unavailable(tmp_path): assert "csv" in monitor.result.artifacts +def test_detail_mode_reports_schematic_missing(tmp_path): + """A fresh QNN log without its partition sidecar has a distinct reason.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + monitor = QNNMonitor(level="detail", output_dir=tmp_path) + context_model = tmp_path / "model_ctx.onnx" + _write_epcontext_model(context_model, [("missing_schematic_partition", 1)]) + monitor.set_running_model_path(context_model) + fixture = Path(__file__).parent / "qnn" / "fixtures" / "optrace_resnet50.csv" + monitor.__enter__() + monitor._csv_path.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + _qnn_log_for_csv(monitor._csv_path).write_text("qnn log", encoding="utf-8") + + monitor.__exit__(None, None, None) + + assert monitor.result is not None + assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == "schematic_missing" + + +def test_detail_mode_reports_sdk_missing(tmp_path, monkeypatch): + """A complete trace input without QHAS tools reports sdk_missing.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + monitor = QNNMonitor(level="detail", output_dir=tmp_path) + context_model = tmp_path / "model_ctx.onnx" + partition_name = "sdk_missing_partition" + _write_epcontext_model(context_model, [(partition_name, 1)]) + monitor.set_running_model_path(context_model) + fixture = Path(__file__).parent / "qnn" / "fixtures" / "optrace_resnet50.csv" + monitor.__enter__() + monitor._csv_path.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + _qnn_log_for_csv(monitor._csv_path).write_text("qnn log", encoding="utf-8") + _schematic_for_partition(tmp_path, partition_name).write_bytes(b"schematic") + monkeypatch.setattr( + "winml.modelkit.session.monitor.qnn_monitor.find_qnn_sdk", + lambda: None, + ) + + monitor.__exit__(None, None, None) + + assert monitor.result is not None + assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == "sdk_missing" + + +def test_detail_mode_contains_qnn_log_metadata_error(tmp_path, monkeypatch): + """An inaccessible QNN log degrades to basic CSV instead of parse_failed.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + monitor = QNNMonitor(level="detail", output_dir=tmp_path) + fixture = Path(__file__).parent / "qnn" / "fixtures" / "optrace_resnet50.csv" + qnn_log = _qnn_log_for_csv(monitor._csv_path) + qnn_log.write_text("pre-existing QNN log", encoding="utf-8") + original_is_file = Path.is_file + + def _inaccessible_log(self: Path) -> bool: + if self == qnn_log: + raise PermissionError("cannot inspect QNN log") + return original_is_file(self) + + monkeypatch.setattr(Path, "is_file", _inaccessible_log) + + monitor.__enter__() + monitor._csv_path.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + + monitor.__exit__(None, None, None) + + assert monitor.result is not None + assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == "qnn_log_missing" + assert monitor.result.operators + + +def test_detail_mode_contains_sdk_discovery_error(tmp_path, monkeypatch): + """SDK filesystem discovery errors remain sdk_missing basic fallbacks.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + monitor = QNNMonitor(level="detail", output_dir=tmp_path) + context_model = tmp_path / "model_ctx.onnx" + partition_name = "sdk_discovery_error_partition" + _write_epcontext_model(context_model, [(partition_name, 1)]) + monitor.set_running_model_path(context_model) + fixture = Path(__file__).parent / "qnn" / "fixtures" / "optrace_resnet50.csv" + monitor.__enter__() + monitor._csv_path.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + _qnn_log_for_csv(monitor._csv_path).write_text("qnn log", encoding="utf-8") + _schematic_for_partition(tmp_path, partition_name).write_bytes(b"schematic") + + def _fail_sdk_discovery(): + raise PermissionError("cannot enumerate SDK root") + + monkeypatch.setattr( + "winml.modelkit.session.monitor.qnn_monitor.find_qnn_sdk", + _fail_sdk_discovery, + ) + + monitor.__exit__(None, None, None) + + assert monitor.result is not None + assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == "sdk_missing" + assert monitor.result.operators + + +@pytest.mark.parametrize("failure_reason", ["viewer_failed", "qhas_output_missing"]) +def test_detail_mode_reports_viewer_failure_reason(tmp_path, monkeypatch, failure_reason): + """Detailed viewer failures retain their precise machine-readable reason.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn.viewer import QHASViewerResult + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + monitor = QNNMonitor(level="detail", output_dir=tmp_path) + context_model = tmp_path / "model_ctx.onnx" + partition_name = "viewer_failure_partition" + _write_epcontext_model(context_model, [(partition_name, 1)]) + monitor.set_running_model_path(context_model) + fixture = Path(__file__).parent / "qnn" / "fixtures" / "optrace_resnet50.csv" + monitor.__enter__() + monitor._csv_path.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + _qnn_log_for_csv(monitor._csv_path).write_text("qnn log", encoding="utf-8") + _schematic_for_partition(tmp_path, partition_name).write_bytes(b"schematic") + monkeypatch.setattr( + "winml.modelkit.session.monitor.qnn_monitor.find_qnn_sdk", + lambda: tmp_path / "sdk", + ) + monkeypatch.setattr( + "winml.modelkit.session.monitor.qnn_monitor.run_qhas_viewer_result", + lambda *_args, **_kwargs: QHASViewerResult( + path=None, + failure_reason=failure_reason, + ), + ) + + monitor.__exit__(None, None, None) + + assert monitor.result is not None + assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == failure_reason + + +def test_detail_mode_reports_qhas_parse_failure(tmp_path, monkeypatch): + """Malformed QHAS output is distinct from viewer execution failure.""" + from pathlib import Path + + from winml.modelkit.session.monitor.qnn.viewer import QHASViewerResult + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + monitor = QNNMonitor(level="detail", output_dir=tmp_path) + context_model = tmp_path / "model_ctx.onnx" + partition_name = "qhas_parse_partition" + _write_epcontext_model(context_model, [(partition_name, 1)]) + monitor.set_running_model_path(context_model) + fixture = Path(__file__).parent / "qnn" / "fixtures" / "optrace_resnet50.csv" + monitor.__enter__() + monitor._csv_path.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + _qnn_log_for_csv(monitor._csv_path).write_text("qnn log", encoding="utf-8") + _schematic_for_partition(tmp_path, partition_name).write_bytes(b"schematic") + qhas_output = tmp_path / "invalid_qhas.json" + qhas_output.write_text("not json", encoding="utf-8") + monkeypatch.setattr( + "winml.modelkit.session.monitor.qnn_monitor.find_qnn_sdk", + lambda: tmp_path / "sdk", + ) + monkeypatch.setattr( + "winml.modelkit.session.monitor.qnn_monitor.run_qhas_viewer_result", + lambda *_args, **_kwargs: QHASViewerResult( + path=qhas_output, + failure_reason=None, + ), + ) + + monitor.__exit__(None, None, None) + + assert monitor.result is not None + assert monitor.result.status == "basic_fallback" + assert monitor.result.fallback_reason == "qhas_parse_failed" + + def test_detail_mode_uses_basic_fallback_when_schematic_stat_fails(tmp_path, monkeypatch): from pathlib import Path diff --git a/tests/unit/session/monitor/test_qnn_monitor_parse_existing.py b/tests/unit/session/monitor/test_qnn_monitor_parse_existing.py index 489a5c7db..bf5740734 100644 --- a/tests/unit/session/monitor/test_qnn_monitor_parse_existing.py +++ b/tests/unit/session/monitor/test_qnn_monitor_parse_existing.py @@ -145,6 +145,35 @@ def test_parse_existing_artifacts_detail_qhas_override(tmp_path): ) +def test_parse_existing_artifacts_inaccessible_qhas_is_basic_fallback(tmp_path, monkeypatch): + """QHAS metadata I/O failures retain usable CSV data and a stable reason.""" + csv_path = tmp_path / "profiling_output.csv" + csv_path.write_text( + (FIXTURE_DIR / "optrace_resnet50.csv").read_text(encoding="utf-8"), + encoding="utf-8", + ) + qhas_path = tmp_path / "inaccessible_qhas.json" + original_is_file = Path.is_file + + def _inaccessible_qhas(self: Path) -> bool: + if self == qhas_path: + raise PermissionError("cannot inspect QHAS output") + return original_is_file(self) + + monkeypatch.setattr(Path, "is_file", _inaccessible_qhas) + + from winml.modelkit.session.monitor.qnn_monitor import QNNMonitor + + result = QNNMonitor.parse_existing_artifacts( + level="detail", + artifacts={"csv": csv_path, "qhas": qhas_path}, + ) + + assert result.status == "basic_fallback" + assert result.fallback_reason == "qhas_output_missing" + assert result.operators + + def test_parse_existing_artifacts_returns_failed_result_on_corrupt_csv(tmp_path): """When artifacts cannot be parsed, parse_existing_artifacts returns OpTraceResult(status='parse_failed', error=...) rather than raising.