mandatory_workload_surface_validation.py

Go to the documentation of this file.

#!/usr/bin/env python3
"""Mandatory workload planner surface validation.

Builds the required 2 to 4 qubit microcases and the structured noisy
`U3` / `CNOT` workload families, then verifies that all of them enter the same
canonical Phase 3 planner contract.

Run with:
    python benchmarks/density_matrix/planner_surface/mandatory_workload_surface_validation.py
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[3]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from benchmarks.density_matrix.planner_surface.common import build_software_metadata
from benchmarks.density_matrix.planner_surface.workloads import (
    MANDATORY_NOISE_PATTERNS,
    STRUCTURED_FAMILY_NAMES,
    STRUCTURED_QUBITS,
    iter_microcase_surfaces,
    iter_structured_surfaces,
)
from squander.partitioning.noisy_planner import (
    PLANNER_OP_KIND_GATE,
    PLANNER_OP_KIND_NOISE,
)

SUITE_NAME = "phase3_planner_surface_workload_surface"
ARTIFACT_FILENAME = "mandatory_workload_surface_bundle.json"
DEFAULT_OUTPUT_DIR = (
    REPO_ROOT / "benchmarks" / "density_matrix" / "artifacts" / "planner_surface"
)
ARTIFACT_CORE_FIELDS = (
    "suite_name",
    "status",
    "requested_mode",
    "software",
    "summary",
    "cases",
)


def _surface_case(case_kind: str, metadata: dict, surface) -> dict:
    payload = surface.to_dict()
    return {
        "case_kind": case_kind,
        "case_name": payload["workload_id"],
        "status": "pass",
        "requested_mode": payload["requested_mode"],
        "source_type": payload["source_type"],
        "workload_id": payload["workload_id"],
        "qbit_num": payload["qbit_num"],
        "operation_count": payload["operation_count"],
        "gate_count": payload["gate_count"],
        "noise_count": payload["noise_count"],
        "gate_sequence": [
            op["name"]
            for op in payload["operations"]
            if op["kind"] == PLANNER_OP_KIND_GATE
        ],
        "noise_sequence": [
            op["name"]
            for op in payload["operations"]
            if op["kind"] == PLANNER_OP_KIND_NOISE
        ],
        "planner_operations": payload["operations"],
        "metadata": metadata,
    }


def run_validation(verbose: bool = True) -> list[dict]:
    cases: list[dict] = []
    for metadata, surface in iter_microcase_surfaces():
        cases.append(_surface_case("microcase", metadata, surface))
    for metadata, surface in iter_structured_surfaces():
        cases.append(_surface_case("structured_family", metadata, surface))

    if verbose:
        print("Mandatory workload planner surfaces:")
        for case in cases:
            print(
                "  {case_name}: q={qbit_num}, ops={operation_count}, gates={gate_count}, noise={noise_count}".format(
                    **case
                )
            )
    return cases


def build_artifact_bundle(cases: list[dict]) -> dict:
    payload_key_sets = {frozenset(case.keys()) for case in cases}
    operation_key_sets = {
        frozenset(case["planner_operations"][0].keys()) for case in cases if case["planner_operations"]
    }
    passed_cases = sum(case["status"] == "pass" for case in cases)
    bundle = {
        "suite_name": SUITE_NAME,
        "status": "pass"
        if passed_cases == len(cases) and len(payload_key_sets) == 1 and len(operation_key_sets) == 1
        else "fail",
        "requested_mode": "partitioned_density",
        "software": build_software_metadata(),
        "summary": {
            "total_cases": len(cases),
            "passed_cases": passed_cases,
            "microcase_count": sum(case["case_kind"] == "microcase" for case in cases),
            "structured_family_count": sum(
                case["case_kind"] == "structured_family" for case in cases
            ),
            "required_structured_families": list(STRUCTURED_FAMILY_NAMES),
            "required_structured_qubits": list(STRUCTURED_QUBITS),
            "required_noise_patterns": list(MANDATORY_NOISE_PATTERNS),
            "payload_schema_count": len(payload_key_sets),
            "operation_schema_count": len(operation_key_sets),
        },
        "cases": cases,
    }
    missing = [field for field in ARTIFACT_CORE_FIELDS if field not in bundle]
    if missing:
        raise ValueError(
            "Mandatory workload surface bundle missing required fields: {}".format(
                ", ".join(missing)
            )
        )
    return bundle


def write_artifact_bundle(bundle: dict, output_dir: Path = DEFAULT_OUTPUT_DIR) -> Path:
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / ARTIFACT_FILENAME
    output_path.write_text(json.dumps(bundle, indent=2, sort_keys=True) + "\n")
    return output_path


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=DEFAULT_OUTPUT_DIR,
        help="Directory to write the mandatory workload surface bundle into.",
    )
    parser.add_argument(
        "--quiet",
        action="store_true",
        help="Suppress per-case console output.",
    )
    args = parser.parse_args(argv)

    results = run_validation(verbose=not args.quiet)
    bundle = build_artifact_bundle(results)
    output_path = write_artifact_bundle(bundle, output_dir=args.output_dir)

    if not args.quiet:
        print("Wrote {}".format(output_path))

    return 0 if bundle["status"] == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main())