Pipelines

The hardware API is just WebSocket JSON — anything the IDE can do, a script can do. That makes Nuclei’s kernel usable as a batch driver for parameter sweeps, calibration curves, and simulator-vs-hardware comparisons.

Prerequisites: the client examples (connection + message-draining patterns) and the replay-tested simulator session (the exact submit → status → results message shapes this page builds on).

The scripted loop

Per job, the protocol sequence is always:

1. hardware_submit with the source code (not a circuit object) → hardware_job_submitted with a JobHandle.
2. Poll hardware_results until data carries measurements (or error). hardware_status only refreshes queue position — it will never tell you a real-hardware job finished.
3. Handle both failure shapes: an error envelope (no job exists) and a handle with status: "failed" (the convention).

Worked example: parameter sweep

sweep_driver.py drives an RY(θ) rotation sweep — θ over a linspace, one hardware_submit per point against the always-available simulator provider, polled to completion, printed as a θ → P(1) table. The expected curve is P(1) = sin²(θ/2).

This exact file is executed by the kernel test suite against a live server on every CI run (kernel/tests/test_docs_fixtures.py), so it speaks the current protocol.

sweep_driver.py

#!/usr/bin/env python3
"""Batch parameter sweep over the Nuclei hardware API: an RY(theta) curve.

Submits one single-qubit circuit per sweep point through `hardware_submit`
to the local simulator provider, polls each job to completion, and prints a
theta -> P(1) table.

Usage: python sweep_driver.py [ws://localhost:9742] [shots] [points]
Requires: pip install websockets

This file is replay-tested: kernel/tests/test_docs_fixtures.py runs it as a
subprocess against a live kernel server (3 points, 64 shots) and asserts
the table output.
"""
import asyncio
import json
import math
import sys

import websockets

PROVIDER = "simulator"
BACKEND = "sim_qasm"

CIRCUIT_TEMPLATE = """from qiskit import QuantumCircuit

qc = QuantumCircuit(1, 1)
qc.ry({theta}, 0)
qc.measure(0, 0)
"""


async def request(ws, payload: dict, reply_type: str) -> dict:
    """Send one hardware request and wait for its single reply.

    Every hardware request gets exactly one reply: the named response type
    or an `error` envelope — which this raises so the sweep fails loudly.
    """
    await ws.send(json.dumps(payload))
    while True:
        message = json.loads(await ws.recv())
        if message["type"] == "error":
            raise RuntimeError(message["message"])
        if message["type"] == reply_type:
            return message


async def run_point(ws, theta: float, shots: int) -> dict:
    """Submit one circuit and poll until its counts arrive."""
    submitted = await request(ws, {
        "type": "hardware_submit",
        "provider": PROVIDER,
        "backend": BACKEND,
        "shots": shots,
        "code": CIRCUIT_TEMPLATE.format(theta=repr(theta)),
        "language": "python",
    }, "hardware_job_submitted")
    job_id = submitted["job"]["id"]

    # Completion is discovered via hardware_results, NOT hardware_status
    # (status polling only refreshes queue position). The simulator
    # completes synchronously; real providers return {"status": "running"}
    # here until done — hence the loop.
    while True:
        result = await request(ws, {
            "type": "hardware_results", "job_id": job_id,
        }, "hardware_result")
        data = result["data"]
        if "error" in data:
            raise RuntimeError(f"job {job_id} failed: {data['error']}")
        if "measurements" in data:
            return data["measurements"]
        await asyncio.sleep(1.0)


async def main(url: str, shots: int, points: int) -> int:
    step = math.pi / (points - 1) if points > 1 else 0.0
    async with websockets.connect(url) as ws:
        print(f"sweep: RY(theta) on {PROVIDER}/{BACKEND}, {shots} shots/point")
        for i in range(points):
            theta = i * step
            counts = await run_point(ws, theta, shots)
            p1 = counts.get("1", 0) / shots
            print(f"theta={theta:.4f}  P(1)={p1:.4f}  counts={counts}")
    return 0


if __name__ == "__main__":
    url = sys.argv[1] if len(sys.argv) > 1 else "ws://localhost:9742"
    shots = int(sys.argv[2]) if len(sys.argv) > 2 else 256
    points = int(sys.argv[3]) if len(sys.argv) > 3 else 9
    sys.exit(asyncio.run(main(url, shots, points)))

cd kernel && python server.py          # terminal 1
python sweep_driver.py                 # terminal 2 — 9 points, 256 shots
python sweep_driver.py ws://localhost:9742 1024 17

Sample output, abridged (counts vary run to run):

sweep: RY(theta) on simulator/sim_qasm, 256 shots/point
theta=0.0000  P(1)=0.0000  counts={'0': 256}
theta=0.7854  P(1)=0.1484  counts={'0': 218, '1': 38}
theta=1.5708  P(1)=0.5039  counts={'1': 129, '0': 127}
...
theta=3.1416  P(1)=1.0000  counts={'1': 256}

Into analysis

Collect the table into a DataFrame and plot it (illustrative — adapt freely, not replay-tested):

import pandas as pd
import matplotlib.pyplot as plt
import numpy as np

# rows collected from run_point(): [(theta, counts), ...]
df = pd.DataFrame(
    {"theta": theta, "p1": counts.get("1", 0) / shots}
    for theta, counts in rows
)

ax = df.plot(x="theta", y="p1", marker="o", label="measured")
thetas = np.linspace(0, np.pi, 200)
ax.plot(thetas, np.sin(thetas / 2) ** 2, "--", label="sin²(θ/2)")
ax.set_xlabel("θ (rad)"); ax.set_ylabel("P(1)"); ax.legend()
plt.savefig("ry_sweep.png", dpi=150)

Remember the per-framework bitstring-key conventions before aggregating counts across frameworks — see schemas.

Simulator vs real hardware

The driver is provider-agnostic by design — switch PROVIDER/BACKEND (after a hardware_connect with credentials) and the same loop runs on real devices. Adjust expectations:

	Simulator	Real hardware
Submit	Completes synchronously — the handle arrives already complete.	Returns queued; minutes-to-hours in queue is normal.
Polling	First hardware_results has the data.	{"status": "running"} until done — poll with backoff (seconds, not the driver’s 1 s, for long queues).
Results	Full SimulationResult.	Counts-style dict only — no statevector, no Bloch coordinates.
Failure modes	data.error carries a KernelError.	SDK error strings; plus queue cancellations and per-job credit limits.
Restart safety	Cheap to re-run.	A kernel restart mid-sweep turns in-flight jobs stale — results unrecoverable through Nuclei. Keep sweeps short or checkpoint your own (point → counts) pairs as you go.

For sweeps on Azure Quantum specifically, start with the free targets listed in testing tips.