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
22 changes: 20 additions & 2 deletions src/winml/modelkit/commands/perf.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down Expand Up @@ -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:
Expand Down
17 changes: 17 additions & 0 deletions src/winml/modelkit/session/monitor/op_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.

Expand All @@ -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,
}

Expand Down
124 changes: 78 additions & 46 deletions src/winml/modelkit/session/monitor/qnn/viewer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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")
79 changes: 58 additions & 21 deletions src/winml/modelkit/session/monitor/qnn_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand All @@ -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)

Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading