Profile Spark attention across Qualcomm edge devices

benchmarks/edge_harness/experiments/audit_spark_qualcomm_attention.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Audit a real Spark attention-layer AI Hub result without claiming model E2E."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+from collections import Counter
+from pathlib import Path
+
+import numpy as np
+
+
+OUTPUT_NAMES = ("hidden", "k_new_0", "v_new_0")
+SOURCE_ONNX_SHA256 = "4e2b1c8d791b1ee0e9b75f3322a20a743c0bf8c40f9b98edd963f2944d6300bb"
+QUANTIZED_ZIP_SHA256 = "703821ca8ace5672228919ea2add4901def4887748c381b12ffc8b259665c553"
+FIXTURE_SHA256 = "5a439e6c6d148f1720938770daa110039a39edde014f1c1592c823c7d7dfc8ed"
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for block in iter(lambda: source.read(8 << 20), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def read_outputs(path: Path, *, hub: bool) -> dict[str, np.ndarray]:
+    with np.load(path, allow_pickle=False) as data:
+        actual_names = tuple(data.files)
+        expected_names = tuple(f"output_{i}__0" for i in range(3)) if hub else OUTPUT_NAMES
+        if actual_names != expected_names:
+            raise ValueError(f"{path}: output names/order differ: {actual_names}")
+        values = {name: np.asarray(data[key]) for name, key in zip(OUTPUT_NAMES, expected_names)}
+    shapes = ((1, 1, 2048), (1, 2, 1, 256), (1, 2, 1, 256))
+    for name, shape in zip(OUTPUT_NAMES, shapes):
+        value = values[name]
+        if value.shape != shape or value.dtype != np.float32 or not np.isfinite(value).all():
+            raise ValueError(f"{path}: {name} violates FP32 finite layer-output contract")
+    return values
+
+
+def compare(reference: dict[str, np.ndarray], actual: dict[str, np.ndarray]) -> dict:
+    result = {}
+    for name in OUTPUT_NAMES:
+        left = reference[name].astype(np.float64).ravel()
+        right = actual[name].astype(np.float64).ravel()
+        if not np.linalg.norm(left):
+            raise ValueError(f"{name}: zero reference cannot define relative L2")
+        result[name] = {
+            "relative_l2": float(np.linalg.norm(left - right) / np.linalg.norm(left)),
+            "max_abs": float(np.max(np.abs(left - right))),
+            "cosine": float(np.dot(left, right) / (np.linalg.norm(left) * np.linalg.norm(right))),
+        }
+    return result
+
+
+def nearest_rank(values: list[int], fraction: float) -> int:
+    ordered = sorted(values)
+    return ordered[math.ceil(fraction * len(ordered)) - 1]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--device-name", required=True)
+    for name in (
+        "source-onnx", "quantized-zip", "fixture", "source-cpu-output",
+        "quantized-cpu-output", "s25-output", "s25-inference-report", "device-output", "compile-report",
+        "inference-report", "profile-report", "output-report",
+    ):
+        parser.add_argument(f"--{name}", type=Path, required=True)
+    args = parser.parse_args()
+    compile_report = json.loads(args.compile_report.read_text(encoding="utf-8"))
+    inference = json.loads(args.inference_report.read_text(encoding="utf-8"))
+    s25_inference = json.loads(args.s25_inference_report.read_text(encoding="utf-8"))
+    profile = json.loads(args.profile_report.read_text(encoding="utf-8"))
+    if compile_report.get("status") != "SUCCESS" or inference.get("status") != "SUCCESS" or profile.get("status") != "SUCCESS":
+        raise RuntimeError("QNN DLC compile, device inference and profile must all succeed")
+    if (compile_report.get("source_quantized_model_id") != "mmd5e3v3q"
+            or compile_report.get("options") != "--target_runtime qnn_dlc --quantize_io"):
+        raise RuntimeError("compile did not use the pinned quantized Spark layer and QNN DLC layout")
+    target = compile_report["target_model_id"]
+    if inference["model_id"] != target or profile["model_id"] != target:
+        raise RuntimeError("device inference/profile used a different compiled artifact")
+    if inference["input_dataset_id"] != "d7m83gjy2":
+        raise RuntimeError("device inference did not use the pinned cross-device fixture")
+    if (s25_inference.get("status") != "SUCCESS" or s25_inference.get("model_id") != "mnowl6xpq"
+            or s25_inference.get("input_dataset_id") != inference["input_dataset_id"]
+            or s25_inference.get("device", {}).get("name") != "Samsung Galaxy S25"):
+        raise RuntimeError("historical S25 output lacks matching model/device/dataset provenance")
+    if inference["device"]["name"] != args.device_name or profile["device"]["name"] != args.device_name:
+        raise RuntimeError("jobs did not execute on the exact requested device")
+    if inference["options"] != "--compute_unit npu" or profile["options"] != "--compute_unit npu":
+        raise RuntimeError("jobs did not request NPU placement")
+
+    execution = profile["profile"]
+    times_us = execution["execution_summary"]["all_inference_times"]
+    if not times_us or any(not isinstance(value, int) or value <= 0 for value in times_us):
+        raise RuntimeError("missing or invalid device samples")
+    units = Counter(row.get("compute_unit") for row in execution["execution_detail"])
+    if units["NPU"] <= 0:
+        raise RuntimeError("profile has no attributed NPU execution rows")
+
+    source = read_outputs(args.source_cpu_output, hub=False)
+    quantized = read_outputs(args.quantized_cpu_output, hub=False)
+    s25 = read_outputs(args.s25_output, hub=True)
+    device_output = read_outputs(args.device_output, hub=True)
+    files = {
+        name: {"bytes": path.stat().st_size, "sha256": sha256(path)}
+        for name, path in {
+            "source_onnx": args.source_onnx,
+            "quantized_onnx_zip": args.quantized_zip,
+            "fixture": args.fixture,
+            "source_cpu_output": args.source_cpu_output,
+            "quantized_cpu_output": args.quantized_cpu_output,
+            "s25_output": args.s25_output,
+            "device_output": args.device_output,
+        }.items()
+    }
+    for name, expected in {
+        "source_onnx": SOURCE_ONNX_SHA256,
+        "quantized_onnx_zip": QUANTIZED_ZIP_SHA256,
+        "fixture": FIXTURE_SHA256,
+    }.items():
+        if files[name]["sha256"] != expected:
+            raise RuntimeError(f"{name} hash differs from pinned layer artifact or fixture")
+    report = {
+        "scope": "one real Spark-X2.5 full-attention decoder layer at one static 1024-token cache shape; no generation E2E",
+        "status": "component_executed_on_npu_quality_unqualified",
+        "device": profile["device"],
+        "compile_job_id": compile_report["job_id"],
+        "inference_job_id": inference["job_id"],
+        "profile_job_id": profile["job_id"],
+        "target_model_id": target,
+        "input_dataset_id": inference["input_dataset_id"],
+        "s25_prior_inference_job_id": s25_inference["job_id"],
+        "files": files,
+        "source_vs_quantized_cpu": compare(source, quantized),
+        "quantized_cpu_vs_device_npu": compare(quantized, device_output),
+        "source_cpu_vs_device_npu": compare(source, device_output),
+        "s25_npu_vs_device_npu": compare(s25, device_output),
+        "profile": {
+            "sample_count": len(times_us),
+            "sample_min_us": min(times_us),
+            "nearest_rank_p50_us": nearest_rank(times_us, 0.50),
+            "nearest_rank_p95_us": nearest_rank(times_us, 0.95),
+            "reported_estimated_inference_time_us": execution["execution_summary"]["estimated_inference_time"],
+            "reported_peak_memory_bytes": execution["execution_summary"]["estimated_inference_peak_memory"],
+            "compute_unit_row_counts": dict(units),
+        },
+        "limits": [
+            "No task-level numerical tolerance or next-token quality gate is established.",
+            "Hub component jobs do not prove device-local prefill, KV/ring update, sampling, continuous generation or Omni deployment.",
+            "Profile sample latency is one component, not complete-request or sustained thermal performance.",
+        ],
+    }
+    args.output_report.parent.mkdir(parents=True, exist_ok=True)
+    args.output_report.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps({"status": report["status"], "profile": report["profile"]}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

benchmarks/edge_harness/experiments/audit_spark_qualcomm_fallback.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""Audit an RB3 Spark attention-layer fallback against the FP32 source."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import Counter
+from pathlib import Path
+
+from audit_spark_qualcomm_attention import (
+    FIXTURE_SHA256,
+    SOURCE_ONNX_SHA256,
+    compare,
+    nearest_rank,
+    read_outputs,
+    sha256,
+)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--device-name", default="Dragonwing RB3 Gen 2 Vision Kit")
+    parser.add_argument("--compute-unit", choices=("cpu", "npu"), required=True)
+    parser.add_argument("--compile-options", required=True)
+    for name in (
+        "source-onnx", "fixture", "source-cpu-output", "compile-report",
+        "inference-report", "profile-report", "device-output", "output-report",
+    ):
+        parser.add_argument(f"--{name}", type=Path, required=True)
+    args = parser.parse_args()
+    compile_report = json.loads(args.compile_report.read_text(encoding="utf-8"))
+    inference = json.loads(args.inference_report.read_text(encoding="utf-8"))
+    profile = json.loads(args.profile_report.read_text(encoding="utf-8"))
+    if any(report.get("status") != "SUCCESS" for report in (compile_report, inference, profile)):
+        raise RuntimeError("compile, inference and profile must all succeed")
+    if compile_report.get("source_model_id") != "mn14dw6zn" or compile_report.get("options") != args.compile_options:
+        raise RuntimeError("compile did not use the pinned FP32 Spark source and requested route")
+    target_id = compile_report["target_model_id"]
+    if inference.get("model_id") != target_id or profile.get("model_id") != target_id:
+        raise RuntimeError("inference/profile did not use the compiled artifact")
+    if inference.get("input_dataset_id") != "d7m83gjy2":
+        raise RuntimeError("inference did not use the pinned cross-device fixture")
+    for report in (compile_report, inference, profile):
+        if report.get("device", {}).get("name") != args.device_name:
+            raise RuntimeError("job did not use the exact requested device")
+    requested = f"--compute_unit {args.compute_unit}"
+    if inference.get("options") != requested or profile.get("options") != requested:
+        raise RuntimeError("inference/profile requested a different compute unit")
+    files = {
+        name: {"bytes": path.stat().st_size, "sha256": sha256(path)}
+        for name, path in {
+            "source_onnx": args.source_onnx,
+            "fixture": args.fixture,
+            "source_cpu_output": args.source_cpu_output,
+            "device_output": args.device_output,
+        }.items()
+    }
+    if files["source_onnx"]["sha256"] != SOURCE_ONNX_SHA256 or files["fixture"]["sha256"] != FIXTURE_SHA256:
+        raise RuntimeError("source or fixture differs from the pinned experiment")
+    output = read_outputs(args.device_output, hub=True)
+    source = read_outputs(args.source_cpu_output, hub=False)
+    execution = profile["profile"]
+    times_us = execution["execution_summary"]["all_inference_times"]
+    if not times_us or any(not isinstance(value, int) or value <= 0 for value in times_us):
+        raise RuntimeError("missing or invalid device samples")
+    units = Counter(row.get("compute_unit") for row in execution["execution_detail"])
+    if units[args.compute_unit.upper()] <= 0:
+        raise RuntimeError("profile has no rows attributed to the requested compute unit")
+    report = {
+        "scope": "one real Spark-X2.5 attention layer at one static 1024-token cache shape; no generation E2E",
+        "status": f"component_executed_on_{args.compute_unit}_quality_unqualified",
+        "device": profile["device"],
+        "compile_job_id": compile_report["job_id"],
+        "inference_job_id": inference["job_id"],
+        "profile_job_id": profile["job_id"],
+        "target_model_id": target_id,
+        "input_dataset_id": inference["input_dataset_id"],
+        "compile_options": args.compile_options,
+        "files": files,
+        "source_cpu_vs_device": compare(source, output),
+        "profile": {
+            "sample_count": len(times_us),
+            "sample_min_us": min(times_us),
+            "nearest_rank_p50_us": nearest_rank(times_us, 0.50),
+            "nearest_rank_p95_us": nearest_rank(times_us, 0.95),
+            "reported_estimated_inference_time_us": execution["execution_summary"]["estimated_inference_time"],
+            "reported_peak_memory_bytes": execution["execution_summary"]["estimated_inference_peak_memory"],
+            "compute_unit_row_counts": dict(units),
+        },
+        "limits": [
+            "No next-token or task-level numerical tolerance is established.",
+            "No complete prefill, KV/ring, sampling, continuous generation or Omni deployment is tested.",
+            "Hosted component latency is not a complete request or sustained thermal profile.",
+        ],
+    }
+    args.output_report.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps({"status": report["status"], "profile": report["profile"]}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

benchmarks/edge_harness/experiments/collect_spark_aihub_component.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Preserve raw Workbench Spark-layer jobs submitted in an evidence directory.
+
+Requires qai-hub authentication through the local client configuration. The
+submission JSON files identify existing jobs; this script never resubmits them.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+import numpy as np
+import qai_hub as hub
+
+
+def device_record(device: object) -> dict:
+    return {
+        "name": device.name,
+        "os": device.os,
+        "attributes": list(device.attributes),
+    }
+
+
+def collect(evidence_dir: Path, kind: str, client: hub.Client) -> str:
+    submission = json.loads((evidence_dir / f"{kind}_submission.json").read_text(encoding="utf-8"))
+    report_path = evidence_dir / f"{kind}_report.json"
+    if report_path.exists():
+        existing = json.loads(report_path.read_text(encoding="utf-8"))
+        if existing.get("job_id") != submission["job_id"]:
+            raise RuntimeError(f"{report_path}: job ID differs from its submission")
+        return f"{kind}: {submission['job_id']} {existing['status']} (already collected)"
+    job = client.get_job(submission["job_id"])
+    status = job.get_status()
+    if status.code not in {"SUCCESS", "FAILED", "CANCELED"}:
+        return f"{kind}: {submission['job_id']} {status.code} (pending)"
+    report = {
+        "timestamp_utc": datetime.now(timezone.utc).isoformat(),
+        "qai_hub_version": str(hub.__version__),
+        "facility": "Qualcomm AI Hub Workbench",
+        "job_id": submission["job_id"],
+        "url": submission["url"],
+        "status": status.code,
+        "status_message": status.message or "",
+        "options": job.options,
+        "hub_version": str(job.hub_version),
+        "device": device_record(job.device),
+        "model_id": job.model.model_id,
+    }
+    if kind == "inference":
+        report["input_dataset_id"] = job.inputs.dataset_id
+        if status.code == "SUCCESS":
+            output_path = evidence_dir / "device_output.npz"
+            outputs = job.download_output_data()
+            arrays = {f"{name}__{i}": value for name, values in outputs.items() for i, value in enumerate(values)}
+            np.savez_compressed(output_path, **arrays)
+            report["outputs"] = output_path.name
+    elif kind == "profile":
+        report["shapes"] = job.shapes
+        if status.code == "SUCCESS":
+            report["profile"] = job.download_profile()
+    else:
+        raise ValueError(f"unexpected job kind {kind}")
+    report_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    return f"{kind}: {submission['job_id']} {status.code}"
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("evidence_dirs", nargs="+", type=Path)
+    args = parser.parse_args()
+    client = hub.Client()
+    for directory in args.evidence_dirs:
+        for kind in ("inference", "profile"):
+            print(f"{directory.name}: {collect(directory, kind, client)}")
+
+
+if __name__ == "__main__":
+    main()