QPI Driver API Reference

qpi_driver

BlueforsGen1Driver

Bases: QpiDriver

Polls Bluefors Gen. 1 Control API channels and emits readings on a timer.

Attributes:

Name	Type	Description
bluefors_base_url		Base URL of the Bluefors Control API, e.g. http://localhost:49099.
channels		Maps a value-tree channel path (e.g. "mapper.bf.tmc") to a display unit (e.g. "K"), which the Bluefors API's basic read response does not itself report. An empty unit is fine.
api_key		Optional Bluefors API access key, sent as the key query parameter (Bluefors reference §3.5.1).
poll_interval		Seconds between polls.
timeout		HTTP timeout per channel read, in seconds.
read_channel	ChannelReader	What reads one channel; see :data:ChannelReader. Defaults to the Gen. 1 Control API read, which is what bluefors_base_url, api_key and timeout configure. A monitor for other control software supplies its own and leaves those alone.

Source code in qpi-driver/py/qpi_driver/builtins/bluefors_gen1.py

class BlueforsGen1Driver(QpiDriver):
    """Polls Bluefors Gen. 1 Control API channels and emits readings on a timer.

    Attributes:
        bluefors_base_url: Base URL of the Bluefors Control API, e.g.
            ``http://localhost:49099``.
        channels: Maps a value-tree channel path (e.g. ``"mapper.bf.tmc"``) to
            a display unit (e.g. ``"K"``), which the Bluefors API's basic read
            response does not itself report. An empty unit is fine.
        api_key: Optional Bluefors API access key, sent as the ``key`` query
            parameter (Bluefors reference §3.5.1).
        poll_interval: Seconds between polls.
        timeout: HTTP timeout per channel read, in seconds.
        read_channel: What reads one channel; see :data:`ChannelReader`. Defaults to
            the Gen. 1 Control API read, which is what ``bluefors_base_url``,
            ``api_key`` and ``timeout`` configure. A monitor for other control
            software supplies its own and leaves those alone.
    """

    def __init__(
        self,
        qpi_addr: str = "http://127.0.0.1:8090",
        token: str = "",
        bluefors_base_url: str = "http://127.0.0.1:49099",
        channels: dict[str, str] | list[str] | None = None,
        api_key: str = "",
        poll_interval: float = DEFAULT_POLL_INTERVAL,
        timeout: float = DEFAULT_TIMEOUT,
        ca_fingerprint: str = "",
        ca_file_path: str = "./bin/qpi.ca.pem",
        recv_timeout_ms: int = DEFAULT_RECV_TIMEOUT_MS,
        read_channel: ChannelReader | None = None,
    ) -> None:
        super().__init__(
            qpi_addr=_normalize_qpi_addr(qpi_addr),
            token=token,
            ca_fingerprint=ca_fingerprint,
            ca_file_path=ca_file_path,
            recv_timeout_ms=recv_timeout_ms,
        )
        self.bluefors_base_url = bluefors_base_url.rstrip("/")
        self.channels = normalize_channels(channels)
        self.api_key = api_key
        self.poll_interval = poll_interval
        self.timeout = timeout
        self.read_channel: ChannelReader = read_channel or self._read_gen1_channel

        self.every(self.poll_interval, self._poll)

    def handle_event(self, event: Event) -> None:
        """Ignore every inbound event — the monitor only reports upward.

        It never handles ``JobDispatch``; it is a separate driver from the
        QPU, not part of it (RFC 0001 §4).
        """
        log.warning(
            "dropping event %s: bluefors_gen1 driver does not handle %s",
            event.id,
            event.type.value,
        )

    def _poll(self) -> None:
        """Read every configured channel and emit whatever succeeded.

        A channel that fails to read (timeout, HTTP error, unexpected shape)
        is recorded with a ``None`` value and an ``ERROR`` status rather than
        raising, so one bad channel does not lose the rest of the tick. If
        every channel fails, nothing is emitted for this tick.
        """
        readings: dict[str, dict[str, Any]] = {}
        for channel, unit in self.channels.items():
            readings[channel] = self.read_channel(channel, unit)

        if not any(r["status"] != "ERROR" for r in readings.values()):
            log.warning(
                "all %d channel(s) failed this tick; skipping emit", len(readings)
            )
            return

        self.emit(
            Event(
                type=EventType.CRYOSTAT_READING,
                driver=self.name,
                payload={"readings": readings},
            )
        )

    def _read_gen1_channel(self, channel: str, unit: str) -> dict[str, Any]:
        """Read a single value-tree channel from the Bluefors Gen. 1 Control API.

        Mirrors the "values" endpoint example in the Bluefors reference: GET
        the endpoint path (channel with dots replaced by slashes) and read
        ``data.content.latest_valid_value``, falling back to
        ``latest_value`` if there is no recent valid sample.

        The default :data:`ChannelReader`, and the only Gen. 1-specific code here.
        """
        url = f"{self.bluefors_base_url}/values/{channel.replace('.', '/')}"
        params = {"key": self.api_key} if self.api_key else {}

        try:
            resp = requests.get(url, params=params, timeout=self.timeout)
            resp.raise_for_status()
            content = resp.json()["data"]["content"]
            sample = (
                content.get("latest_valid_value") or content.get("latest_value") or {}
            )
            raw_value = sample.get("value")
            value = float(raw_value) if raw_value not in (None, "") else None
            status = sample.get("status", "UNKNOWN")
        except Exception:
            log.exception("failed to read channel %s", channel)
            return {"value": None, "unit": unit, "status": "ERROR"}

        return {"value": value, "unit": unit, "status": status}

handle_event(event)

Ignore every inbound event — the monitor only reports upward.

It never handles JobDispatch; it is a separate driver from the QPU, not part of it (RFC 0001 §4).

Source code in qpi-driver/py/qpi_driver/builtins/bluefors_gen1.py

def handle_event(self, event: Event) -> None:
    """Ignore every inbound event — the monitor only reports upward.

    It never handles ``JobDispatch``; it is a separate driver from the
    QPU, not part of it (RFC 0001 §4).
    """
    log.warning(
        "dropping event %s: bluefors_gen1 driver does not handle %s",
        event.id,
        event.type.value,
    )

CalibrateDriver

Bases: QpiDriver

A QPI driver that calibrates a chip through a tuner backend.

Source code in qpi-driver/py/qpi_driver/builtins/calibrate.py

class CalibrateDriver(QpiDriver):
    """A QPI driver that calibrates a chip through a tuner backend."""

    def __init__(
        self,
        *,
        tuner: str | type[Tuner] | Tuner,
        calibration_config: Path,
        qpi_addr: str = "http://127.0.0.1:8090",
        token: str = "",
        drift_check_interval: float = 0,
        fidelity_threshold: float = 0.999,
        fidelity_2q_threshold: float = 0.99,
        ca_fingerprint: str = "",
        ca_file_path: Path = Path("./bin/qpi.ca.pem"),
        recv_timeout_ms: int = DEFAULT_RECV_TIMEOUT_MS,
        **tuner_options: Any,
    ) -> None:
        from qpi_driver.builtins.qpu import _normalize_qpi_addr

        super().__init__(
            qpi_addr=_normalize_qpi_addr(qpi_addr),
            token=token,
            ca_fingerprint=ca_fingerprint,
            ca_file_path=Path(ca_file_path).as_posix(),
            recv_timeout_ms=recv_timeout_ms,
        )
        self.tuner = tuner
        self.calibration_config_path = Path(calibration_config)
        self.drift_check_interval = drift_check_interval
        self.fidelity_threshold = fidelity_threshold
        self.fidelity_2q_threshold = fidelity_2q_threshold
        self.tuner_options = tuner_options

        self._job_queue: multiprocessing.Queue | None = None
        self._result_queue: multiprocessing.Queue | None = None
        self._worker: multiprocessing.Process | None = None
        self._result_pump: threading.Thread | None = None
        self._busy = threading.Event()
        #: What QPI-UI last said this QPU's state is. Assumed online until told
        #: otherwise, so a driver whose server predates the event still calibrates.
        self._qpu_state = "online"

        # Registered here, started by `run` after `_on_start` has made the queue
        # — the order QpiDriver.run guarantees and BlueforsGen1Driver relies on.
        if self.drift_check_interval > 0:
            self.every(self.drift_check_interval, self._check_fidelity)

    def handle_event(self, event: Event) -> None:
        """Queue a dispatched calibration; ignore everything else."""
        if event.type is EventType.QPU_STATE:
            self._qpu_state = str(event.payload.get("state") or "online")
            log.info("QPU is %s", self._qpu_state)
            return

        if event.type is not EventType.CALIBRATE_DISPATCH:
            log.warning(
                "dropping event %s: calibrate driver does not handle %s",
                event.id,
                event.type.value,
            )
            return

        payload = dict(event.payload)
        job_id = payload.get("job_id", "unknown")
        if self._qpu_state == "disabled":
            log.warning("rejecting calibration %s: the QPU is switched off", job_id)
            self._emit_result(job_id, {"error": "this QPU is switched off"})
            return
        if self._busy.is_set():
            # A calibration takes hours and the worker runs one at a time.
            # Queueing a second is not patience, it is a report nobody will
            # connect to a request, so it is refused and said so.
            log.warning("rejecting calibration %s: one is already running", job_id)
            self._emit_result(
                job_id, {"error": "a calibration is already running on this driver"}
            )
            return

        log.info("Received calibration %s (mode=%s)", job_id, payload.get("mode"))
        self._busy.set()
        self._job_queue.put(payload)

    def _check_fidelity(self) -> None:
        """Queue a periodic drift check, unless something says not to.

        The one path no server-side gate reaches: it runs on this driver's clock and
        never asks. So the state event is what stops it.
        """
        if self._qpu_state != "online":
            log.info("skipping drift check: the QPU is %s", self._qpu_state)
            return
        if self._busy.is_set():
            log.info("skipping drift check: a calibration is already running")
            return
        log.info("Queuing periodic drift check")
        self._busy.set()
        self._emit_queued(DRIFT_CHECK_JOB_ID, "fidelity_check", [], "the drift timer")
        self._job_queue.put(
            {
                "mode": "fidelity_check",
                "job_id": DRIFT_CHECK_JOB_ID,
                "fidelity_threshold": self.fidelity_threshold,
                "fidelity_2q_threshold": self.fidelity_2q_threshold,
            }
        )

    def _on_start(self) -> None:
        self._job_queue = multiprocessing.Queue()
        self._result_queue = multiprocessing.Queue()

        self._worker = multiprocessing.Process(
            target=calibrate_worker,
            kwargs={
                "job_queue": self._job_queue,
                "result_queue": self._result_queue,
                "tuner": self.tuner,
                "calibration_config_path": self.calibration_config_path,
                **self.tuner_options,
            },
            name="QPI-CalibrateWorker",
            daemon=True,
        )
        self._worker.start()

        self._result_pump = threading.Thread(
            target=self._pump_results, name="QPI-CalibrateResultPump", daemon=True
        )
        self._result_pump.start()

    def _pump_results(self) -> None:
        """Drain the worker's reports and emit each as a CalibrationResult."""
        while True:
            item = self._result_queue.get()
            if item is None:
                log.info("Result pump received shutdown signal")
                return

            # Both before the clear below: neither is an outcome, and treating one
            # as such would free the driver to accept another calibration.
            if "plan" in item:
                self._emit_queued(
                    item["job_id"],
                    item["mode"],
                    item.get("target_qubits") or [],
                    "the walk it is about to make",
                    plan=item["plan"],
                )
                continue
            if "progress" in item:
                self._emit_progress(item["job_id"], item["progress"])
                continue

            self._busy.clear()
            job_id = item.get("job_id", "unknown")
            if "error" in item:
                self._emit_result(job_id, {"error": item["error"]})
                continue

            report = item["report"]
            log.info("Emitting calibration result for %s: %s", job_id, report["status"])
            self._emit_result(job_id, report)

            for follow_up in item.get("follow_up", []):
                log.info("Drift detected; queuing recalibration of %s", follow_up)
                self._busy.set()
                self._emit_queued(
                    follow_up["job_id"],
                    follow_up["mode"],
                    follow_up.get("target_qubits") or [],
                    f"drift measured by {job_id}",
                )
                self._job_queue.put(follow_up)

    def _emit_queued(
        self,
        job_id: str,
        mode: str,
        target_qubits: list[str],
        reason: str,
        plan: dict[str, Any] | None = None,
    ) -> None:
        """Say that this driver has queued a calibration nobody dispatched.

        A drift check and the recalibration it triggers run on this driver's clock,
        so QPI-UI has no queued row for either and its Calibration tab showed hours
        of nothing. This is what gives them one — announced before the job is queued,
        so the row exists before any progress reports against it.

        Emitted a second time, with *plan*, once the walk has resolved the graph it
        will actually take (RFC 0006 §5.1). The two facts are known in different
        processes and at different moments, and the server's handler is already
        idempotent, so sending one small event twice is cheaper than a second event
        type — and a dispatched calibration, which makes no announcement of its own,
        gets its plan by the same route.

        Best-effort: the calibration is the point and the announcement is not, so a
        socket that will not carry it costs the dashboard a row, not the chip a
        recalibration.
        """
        payload: dict[str, Any] = {
            "job_id": job_id,
            "mode": mode,
            "target_qubits": list(target_qubits),
            "reason": reason,
        }
        if plan is not None:
            payload["plan"] = plan
        try:
            self.emit(
                Event(
                    type=EventType.CALIBRATION_QUEUED,
                    driver=self.name,
                    payload=payload,
                )
            )
        except Exception:  # noqa: BLE001 - see above
            log.warning("could not announce calibration %s", job_id, exc_info=True)

    def _emit_progress(self, job_id: str, update: dict[str, Any]) -> None:
        """Emit one CalibrationProgress, so the dashboard can show a walk in flight.

        Flat beside ``job_id`` for the same reason the result is: QPI-UI unmarshals
        the payload straight into its own struct.
        """
        self.emit(
            Event(
                type=EventType.CALIBRATION_PROGRESS,
                driver=self.name,
                payload={"job_id": job_id, **update},
            )
        )

    def _emit_result(self, job_id: str, report: dict[str, Any]) -> None:
        """Emit one CalibrationResult.

        The report's fields are the payload's own — QPI-UI unmarshals this
        straight into ``CalibrationResultPayload``, so nesting them under a
        ``results`` key would leave every field at its zero value.
        """
        self.emit(
            Event(
                type=EventType.CALIBRATION_RESULT,
                driver=self.name,
                payload={"job_id": job_id, **report},
            )
        )

    def _on_stop(self) -> None:
        from qpi_driver.builtins.qpu import _safe_put

        if self._job_queue is not None:
            _safe_put(self._job_queue, None)
        if self._result_queue is not None:
            _safe_put(self._result_queue, None)

        if self._worker is not None:
            # A full DAG walk will not finish inside this window, so the join is
            # a courtesy to a worker that is between routines; the terminate is
            # the expected path mid-calibration.
            self._worker.join(timeout=5)
            if self._worker.is_alive():
                log.warning("Terminating calibration worker mid-run...")
                self._worker.terminate()
                self._worker.join()

handle_event(event)

Queue a dispatched calibration; ignore everything else.

Source code in qpi-driver/py/qpi_driver/builtins/calibrate.py

def handle_event(self, event: Event) -> None:
    """Queue a dispatched calibration; ignore everything else."""
    if event.type is EventType.QPU_STATE:
        self._qpu_state = str(event.payload.get("state") or "online")
        log.info("QPU is %s", self._qpu_state)
        return

    if event.type is not EventType.CALIBRATE_DISPATCH:
        log.warning(
            "dropping event %s: calibrate driver does not handle %s",
            event.id,
            event.type.value,
        )
        return

    payload = dict(event.payload)
    job_id = payload.get("job_id", "unknown")
    if self._qpu_state == "disabled":
        log.warning("rejecting calibration %s: the QPU is switched off", job_id)
        self._emit_result(job_id, {"error": "this QPU is switched off"})
        return
    if self._busy.is_set():
        # A calibration takes hours and the worker runs one at a time.
        # Queueing a second is not patience, it is a report nobody will
        # connect to a request, so it is refused and said so.
        log.warning("rejecting calibration %s: one is already running", job_id)
        self._emit_result(
            job_id, {"error": "a calibration is already running on this driver"}
        )
        return

    log.info("Received calibration %s (mode=%s)", job_id, payload.get("mode"))
    self._busy.set()
    self._job_queue.put(payload)

CircuitPayload dataclass

A single circuit within a batch job.

Source code in qpi-driver/py/qpi_driver/executors/base/dtos.py

@dataclass
class CircuitPayload:
    """A single circuit within a batch job."""

    circuit: str  # QASM string
    parameter_values: list[list[float]] | None = None  # Optional parameter bindings
    shots: int | None = None  # Per-circuit override

DeviceSpec dataclass

One device: what --device names, and what to call when it does.

Attributes:

Name	Type	Description
name	str	How --device names it, e.g. qblox. A plain identifier — values containing . or : are import paths, not names.
operation	Operation	Which operation this device implements.
build	DeviceBuilder	Returns an unstarted driver; see :data:DeviceBuilder. It is called with an :class:~qpi_driver.options.Options and the transport arguments, and reads whichever options it understands.

Source code in qpi-driver/py/qpi_driver/builtins/registry.py

@dataclass(frozen=True)
class DeviceSpec:
    """One device: what ``--device`` names, and what to call when it does.

    Attributes:
        name: How ``--device`` names it, e.g. ``qblox``. A plain identifier —
            values containing ``.`` or ``:`` are import paths, not names.
        operation: Which operation this device implements.
        build: Returns an unstarted driver; see :data:`DeviceBuilder`. It is called
            with an :class:`~qpi_driver.options.Options` and the transport
            arguments, and reads whichever options it understands.
    """

    name: str
    operation: Operation
    build: DeviceBuilder

Event dataclass

A single typed message exchanged with QPI-UI in either direction.

Attributes:

Name	Type	Description
type	EventType	The event type, which determines the payload shape.
payload	dict[str, Any]	Type-specific body, validated by whoever handles the event.
driver	str	Identifier of the driver this event belongs to.
id	str	Unique identifier of this envelope.
ts	str	Creation time as an ISO-8601 UTC timestamp with millisecond precision.

Source code in qpi-driver/py/qpi_driver/events.py

@dataclass
class Event:
    """A single typed message exchanged with QPI-UI in either direction.

    Attributes:
        type: The event type, which determines the payload shape.
        payload: Type-specific body, validated by whoever handles the event.
        driver: Identifier of the driver this event belongs to.
        id: Unique identifier of this envelope.
        ts: Creation time as an ISO-8601 UTC timestamp with millisecond precision.
    """

    type: EventType
    payload: dict[str, Any] = field(default_factory=dict)
    driver: str = ""
    id: str = ""
    ts: str = ""

    def __post_init__(self) -> None:
        self.type = EventType(self.type)
        if not self.id:
            self.id = _new_event_id()
        if not self.ts:
            self.ts = _now_timestamp()

    def to_dict(self) -> dict[str, Any]:
        """Return the envelope as a plain dict matching the wire shape."""
        return {
            "id": self.id,
            "driver": self.driver,
            "type": self.type.value,
            "ts": self.ts,
            "payload": self.payload,
        }

    def to_json(self) -> str:
        """Serialise the envelope to a JSON string for sending over NNG."""
        return json.dumps(self.to_dict())

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "Event":
        """Build an event from a decoded envelope dict."""
        return cls(
            type=EventType(data["type"]),
            payload=data.get("payload") or {},
            driver=data.get("driver", ""),
            id=data.get("id", ""),
            ts=data.get("ts", ""),
        )

    @classmethod
    def from_json(cls, raw: str | bytes) -> "Event":
        """Build an event from a JSON string or bytes received over NNG."""
        if isinstance(raw, bytes):
            raw = raw.decode()
        return cls.from_dict(json.loads(raw))

from_dict(data) classmethod

Build an event from a decoded envelope dict.

Source code in qpi-driver/py/qpi_driver/events.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Event":
    """Build an event from a decoded envelope dict."""
    return cls(
        type=EventType(data["type"]),
        payload=data.get("payload") or {},
        driver=data.get("driver", ""),
        id=data.get("id", ""),
        ts=data.get("ts", ""),
    )

from_json(raw) classmethod

Build an event from a JSON string or bytes received over NNG.

Source code in qpi-driver/py/qpi_driver/events.py

@classmethod
def from_json(cls, raw: str | bytes) -> "Event":
    """Build an event from a JSON string or bytes received over NNG."""
    if isinstance(raw, bytes):
        raw = raw.decode()
    return cls.from_dict(json.loads(raw))

to_dict()

Return the envelope as a plain dict matching the wire shape.

Source code in qpi-driver/py/qpi_driver/events.py

def to_dict(self) -> dict[str, Any]:
    """Return the envelope as a plain dict matching the wire shape."""
    return {
        "id": self.id,
        "driver": self.driver,
        "type": self.type.value,
        "ts": self.ts,
        "payload": self.payload,
    }

to_json()

Serialise the envelope to a JSON string for sending over NNG.

Source code in qpi-driver/py/qpi_driver/events.py

def to_json(self) -> str:
    """Serialise the envelope to a JSON string for sending over NNG."""
    return json.dumps(self.to_dict())

EventType

Bases: str, Enum

The fixed set of event types a QPI-UI version understands.

Maintainers grow the framework by adding new types over releases.

Source code in qpi-driver/py/qpi_driver/events.py

class EventType(str, Enum):
    """The fixed set of event types a QPI-UI version understands.

    Maintainers grow the framework by adding new types over releases.
    """

    JOB_DISPATCH = "JobDispatch"
    JOB_RESULT = "JobResult"
    CRYOSTAT_READING = "CryostatReading"
    CALIBRATE_DISPATCH = "CalibrateDispatch"
    CALIBRATION_RESULT = "CalibrationResult"
    CALIBRATION_PROGRESS = "CalibrationProgress"
    CALIBRATION_QUEUED = "CalibrationQueued"
    QPU_STATE = "QPUState"

Executor

Bases: ABC

Source code in qpi-driver/py/qpi_driver/executors/base/__init__.py

class Executor(ABC):
    def __init__(self, name: str = "executor", **kwargs: Any) -> None:
        self.name = name

    @abstractmethod
    def execute(self, payload: JobPayload) -> xr.Dataset:
        """Execute the quantum circuit/instructions payload.

        Args:
            payload: JobPayload object containing circuit QASM, qubit count, shots, etc.

        Returns:
            xr.Dataset: Dataset mimicking the raw measurement counts and frequencies.
        """
        ...

    @abstractmethod
    def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
        """Convert a raw xr.Dataset from execute() into a Qiskit-compatible result dict.

        Each executor knows its own data format and how to:
        - Perform state discrimination (if meas_level=2)
        - Return IQ memory (if meas_level=1)
        - Return raw traces (if meas_level=0)
        - Handle meas_return averaging

        Args:
            dataset: The xr.Dataset returned by execute().
            job_id: The unique ID of the quantum job.

        Returns:
            dict: Qiskit-compatible result dict with keys like 'counts', 'memory',
                  'shots', 'backend', 'success', etc.
        """
        ...

    def close(self) -> None:
        """Release resources."""
        pass

close()

Release resources.

Source code in qpi-driver/py/qpi_driver/executors/base/__init__.py

def close(self) -> None:
    """Release resources."""
    pass

execute(payload) abstractmethod

Execute the quantum circuit/instructions payload.

Parameters:

Name	Type	Description	Default
payload	JobPayload	JobPayload object containing circuit QASM, qubit count, shots, etc.	required

Returns:

Type	Description
Dataset	xr.Dataset: Dataset mimicking the raw measurement counts and frequencies.

Source code in qpi-driver/py/qpi_driver/executors/base/__init__.py

@abstractmethod
def execute(self, payload: JobPayload) -> xr.Dataset:
    """Execute the quantum circuit/instructions payload.

    Args:
        payload: JobPayload object containing circuit QASM, qubit count, shots, etc.

    Returns:
        xr.Dataset: Dataset mimicking the raw measurement counts and frequencies.
    """
    ...

process_result(dataset, job_id) abstractmethod

Convert a raw xr.Dataset from execute() into a Qiskit-compatible result dict.

Each executor knows its own data format and how to: - Perform state discrimination (if meas_level=2) - Return IQ memory (if meas_level=1) - Return raw traces (if meas_level=0) - Handle meas_return averaging

Parameters:

Name	Type	Description	Default
dataset	Dataset	The xr.Dataset returned by execute().	required
job_id	str	The unique ID of the quantum job.	required

Returns:

Name	Type	Description
dict	dict	Qiskit-compatible result dict with keys like 'counts', 'memory', 'shots', 'backend', 'success', etc.

Source code in qpi-driver/py/qpi_driver/executors/base/__init__.py

@abstractmethod
def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
    """Convert a raw xr.Dataset from execute() into a Qiskit-compatible result dict.

    Each executor knows its own data format and how to:
    - Perform state discrimination (if meas_level=2)
    - Return IQ memory (if meas_level=1)
    - Return raw traces (if meas_level=0)
    - Handle meas_return averaging

    Args:
        dataset: The xr.Dataset returned by execute().
        job_id: The unique ID of the quantum job.

    Returns:
        dict: Qiskit-compatible result dict with keys like 'counts', 'memory',
              'shots', 'backend', 'success', etc.
    """
    ...

JobPayload dataclass

Source code in qpi-driver/py/qpi_driver/executors/base/dtos.py

@dataclass
class JobPayload:
    circuits: list[CircuitPayload]
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    shots: int = 1024
    meas_level: int = 2  # 2=counts, 1=kerneled IQ, 0=raw IQ
    meas_return: str = "single"  # "single" or "avg"
    acq_rotation: float | None = None
    acq_threshold: float | None = None

    @property
    def qasm(self) -> str:
        """Backward-compatible accessor: returns the first circuit's QASM string."""
        return self.circuits[0].circuit

    def __post_init__(self):
        if not self.id.strip():
            raise ValueError("id cannot be empty or just whitespace.")

        if not self.circuits:
            raise ValueError("circuits list cannot be empty.")

        for i, cp in enumerate(self.circuits):
            if not cp.circuit.strip():
                raise ValueError(f"Circuit at index {i} has an empty circuit string.")

        if self.meas_level not in (0, 1, 2):
            raise ValueError(f"meas_level must be 0, 1, or 2, got {self.meas_level}")

        if self.meas_return not in ("single", "avg"):
            raise ValueError(
                f"meas_return must be 'single' or 'avg', got '{self.meas_return}'"
            )

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "JobPayload":
        shots = data.get("shots") or 1024
        identifier = data.get("id") or str(uuid.uuid4())
        meas_level = data.get("meas_level", 2)
        meas_return = data.get("meas_return", "single")
        acq_rotation = data.get("acq_rotation")
        if acq_rotation is not None:
            acq_rotation = float(acq_rotation)
        acq_threshold = data.get("acq_threshold")
        if acq_threshold is not None:
            acq_threshold = float(acq_threshold)

        # New-style: list of circuit dicts under "circuits"
        raw_circuits = data.get("circuits")
        if raw_circuits and isinstance(raw_circuits, list):
            circuits = []
            for entry in raw_circuits:
                if isinstance(entry, dict):
                    circ_str = (
                        entry.get("circuit")
                        or entry.get("qasm")
                        or entry.get("circuit_qasm")
                        or ""
                    )
                    param_vals = entry.get("parameter_values")
                    per_circuit_shots = entry.get("shots")
                    circuits.append(
                        CircuitPayload(
                            circuit=circ_str,
                            parameter_values=param_vals,
                            shots=per_circuit_shots,
                        )
                    )
                elif isinstance(entry, str):
                    # Plain QASM string in list
                    circuits.append(CircuitPayload(circuit=entry))
                else:
                    raise ValueError(
                        f"Invalid circuit entry type: {type(entry)}. "
                        "Expected dict or string."
                    )
            if not circuits:
                raise ValueError("No circuits provided in payload")
        else:
            # Old-style: single QASM string
            qasm = (
                data.get("qasm")
                or data.get("circuit_qasm")
                or data.get("circuit")
                or ""
            )
            if not qasm:
                raise ValueError("No QASM string/circuit provided in payload")
            circuits = [CircuitPayload(circuit=qasm)]

        return cls(
            circuits=circuits,
            shots=shots,
            id=identifier,
            meas_level=meas_level,
            meas_return=meas_return,
            acq_rotation=acq_rotation,
            acq_threshold=acq_threshold,
        )

qasm property

Backward-compatible accessor: returns the first circuit's QASM string.

MockExecutor

Bases: Executor

Source code in qpi-driver/py/qpi_driver/executors/mock/__init__.py

class MockExecutor(Executor):
    def __init__(self, name: str = "mock", **kwargs: Any):
        super().__init__(name, **kwargs)
        self._simulator = BasicSimulator()

    def execute(self, payload: JobPayload) -> xr.Dataset:
        """Execute quantum circuit simulation using Qiskit BasicSimulator.

        Every circuit is run honouring its per-circuit ``shots`` override.  A
        single-circuit payload returns that circuit's flat dataset; multi-circuit
        payloads are bundled so circuits with different classical-bit widths or
        shot counts stay independent (see ``combine_circuit_datasets``).

        Args:
            payload: JobPayload specifying shots and circuits.

        Returns:
            xr.Dataset: Dataset containing measured state outcomes.
        """
        sub_datasets: list[xr.Dataset] = []

        for circ in payload.circuits:
            circ_shots = circ.shots if circ.shots is not None else payload.shots
            qasm_str = circ.circuit
            circuit = load_qasm(qasm_str)

            param_sets = circ.parameter_values or [None]
            for param_vals in param_sets:
                bound_circuit = circuit
                if param_vals is not None and circuit.parameters:
                    bound_circuit = circuit.assign_parameters(param_vals)

                t_qc = transpile(bound_circuit, self._simulator)
                result = self._simulator.run(
                    t_qc, shots=circ_shots, memory=True
                ).result()
                memory = result.get_memory(t_qc)

                ds = memory_to_dataset(
                    memory, circuit.num_clbits, circ_shots, payload.meas_level
                )
                ds.attrs.update(
                    {
                        "shots": circ_shots,
                        "n_qubits": circuit.num_qubits,
                        "backend": self.name,
                        "meas_level": payload.meas_level,
                        "meas_return": payload.meas_return,
                    }
                )
                sub_datasets.append(ds)

        return combine_circuit_datasets(sub_datasets)

    def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
        """Convert the simulator's xr.Dataset into a Qiskit-compatible result dict.

        Supports meas_level 1 (IQ memory) and 2 (classified counts).
        meas_level 0 is not supported for simulators.

        Args:
            dataset: xr.Dataset from execute().
            job_id: Unique job ID.

        Returns:
            dict: Qiskit-compatible result dict.
        """
        meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
        meas_return = str(dataset.attrs.get("meas_return", "single"))
        backend = dataset.attrs.get("backend", self.name)

        circuit_results = [
            simulator_dataset_to_result(sub_ds, meas_level, meas_return)
            for sub_ds in iter_circuit_datasets(dataset)
        ]
        return build_qiskit_result(circuit_results, job_id, backend)

execute(payload)

Execute quantum circuit simulation using Qiskit BasicSimulator.

Every circuit is run honouring its per-circuit shots override. A single-circuit payload returns that circuit's flat dataset; multi-circuit payloads are bundled so circuits with different classical-bit widths or shot counts stay independent (see combine_circuit_datasets).

Parameters:

Name	Type	Description	Default
payload	JobPayload	JobPayload specifying shots and circuits.	required

Returns:

Type	Description
Dataset	xr.Dataset: Dataset containing measured state outcomes.

Source code in qpi-driver/py/qpi_driver/executors/mock/__init__.py

def execute(self, payload: JobPayload) -> xr.Dataset:
    """Execute quantum circuit simulation using Qiskit BasicSimulator.

    Every circuit is run honouring its per-circuit ``shots`` override.  A
    single-circuit payload returns that circuit's flat dataset; multi-circuit
    payloads are bundled so circuits with different classical-bit widths or
    shot counts stay independent (see ``combine_circuit_datasets``).

    Args:
        payload: JobPayload specifying shots and circuits.

    Returns:
        xr.Dataset: Dataset containing measured state outcomes.
    """
    sub_datasets: list[xr.Dataset] = []

    for circ in payload.circuits:
        circ_shots = circ.shots if circ.shots is not None else payload.shots
        qasm_str = circ.circuit
        circuit = load_qasm(qasm_str)

        param_sets = circ.parameter_values or [None]
        for param_vals in param_sets:
            bound_circuit = circuit
            if param_vals is not None and circuit.parameters:
                bound_circuit = circuit.assign_parameters(param_vals)

            t_qc = transpile(bound_circuit, self._simulator)
            result = self._simulator.run(
                t_qc, shots=circ_shots, memory=True
            ).result()
            memory = result.get_memory(t_qc)

            ds = memory_to_dataset(
                memory, circuit.num_clbits, circ_shots, payload.meas_level
            )
            ds.attrs.update(
                {
                    "shots": circ_shots,
                    "n_qubits": circuit.num_qubits,
                    "backend": self.name,
                    "meas_level": payload.meas_level,
                    "meas_return": payload.meas_return,
                }
            )
            sub_datasets.append(ds)

    return combine_circuit_datasets(sub_datasets)

process_result(dataset, job_id)

Convert the simulator's xr.Dataset into a Qiskit-compatible result dict.

Supports meas_level 1 (IQ memory) and 2 (classified counts). meas_level 0 is not supported for simulators.

Parameters:

Name	Type	Description	Default
dataset	Dataset	xr.Dataset from execute().	required
job_id	str	Unique job ID.	required

Returns:

Name	Type	Description
dict	dict	Qiskit-compatible result dict.

Source code in qpi-driver/py/qpi_driver/executors/mock/__init__.py

def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
    """Convert the simulator's xr.Dataset into a Qiskit-compatible result dict.

    Supports meas_level 1 (IQ memory) and 2 (classified counts).
    meas_level 0 is not supported for simulators.

    Args:
        dataset: xr.Dataset from execute().
        job_id: Unique job ID.

    Returns:
        dict: Qiskit-compatible result dict.
    """
    meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
    meas_return = str(dataset.attrs.get("meas_return", "single"))
    backend = dataset.attrs.get("backend", self.name)

    circuit_results = [
        simulator_dataset_to_result(sub_ds, meas_level, meas_return)
        for sub_ds in iter_circuit_datasets(dataset)
    ]
    return build_qiskit_result(circuit_results, job_id, backend)

Operation

Bases: str, Enum

What a driver does, and the contract QPI-UI implements for it.

Closed by design: every value needs a handler on the server side, so a new operation is a coordinated change across the SDKs and QPI-UI, never a third-party extension (RFC 0003 §13.1). Devices are the open half.

Source code in qpi-driver/py/qpi_driver/builtins/registry.py

class Operation(str, Enum):
    """What a driver does, and the contract QPI-UI implements for it.

    Closed by design: every value needs a handler on the server side, so a new
    operation is a coordinated change across the SDKs and QPI-UI, never a
    third-party extension (RFC 0003 §13.1). Devices are the open half.
    """

    PROCESS = "process"
    MONITOR = "monitor"
    CALIBRATE = "calibrate"

Options

Raw -o values, read by the device that understands them.

Every accessor takes the fallback used when the key is absent, written as the value itself rather than as a string to parse: the default belongs to the one piece of code that acts on it.

A value the accessor cannot read raises :class:ValueError naming the option, which the CLI turns into a single line.

Source code in qpi-driver/py/qpi_driver/options.py

class Options:
    """Raw ``-o`` values, read by the device that understands them.

    Every accessor takes the fallback used when the key is absent, written as the
    value itself rather than as a string to parse: the default belongs to the one
    piece of code that acts on it.

    A value the accessor cannot read raises :class:`ValueError` naming the option,
    which the CLI turns into a single line.
    """

    def __init__(self, values: Mapping[str, str] | None = None) -> None:
        self._values = {key: str(value) for key, value in (values or {}).items()}
        self._read: set[str] = set()

    def __repr__(self) -> str:
        return f"Options({self._values!r})"

    def get_str(self, key: str, default: str = "") -> str:
        """The value of *key* as typed, or *default*."""
        self._read.add(key)
        return self._values.get(key, default)

    def require(self, key: str, example: str = "") -> str:
        """The value of *key*, or raise because the device cannot run without it.

        Raises:
            ValueError: if *key* was not given. *example* is appended as a
                ready-to-paste ``-o`` pair when there is a sensible one.
        """
        self._read.add(key)
        if key not in self._values:
            hint = f", e.g. -o {key}={example}" if example else ""
            raise ValueError(f"missing required option {key!r}{hint}")
        return self._values[key]

    def get_int(self, key: str, default: int) -> int:
        """The value of *key* as an integer, or *default*."""
        return self._parsed(key, default, int)

    def get_float(self, key: str, default: float) -> float:
        """The value of *key* as a float, or *default*."""
        return self._parsed(key, default, float)

    def get_bool(self, key: str, default: bool = False) -> bool:
        """The value of *key* as a boolean, or *default*.

        ``1``, ``true``, ``yes`` and ``on`` are true in any case; every other
        value, the empty string included, is false. Nothing here can fail, so a
        misspelled ``-o is_dummy=ture`` is false rather than an error — the
        alternative is refusing to start over a flag.
        """
        self._read.add(key)
        if key not in self._values:
            return default
        return self._values[key].strip().lower() in _TRUTHY

    def get_path(self, key: str, default: Path | str) -> Path:
        """The value of *key* as a path, or *default*. Not checked for safety."""
        return self._parsed(key, Path(default), Path)

    def get_dir(self, key: str, default: Path | str, *, default_name: str = "") -> Path:
        """The value of *key* as a directory a driver may write to, or *default*.

        The safe-location check of :func:`~qpi_driver.paths.as_safe_dir` applies to
        the given value and to *default* alike, so a device cannot default itself
        somewhere it may not write.

        *default_name* is what to call *default* in that error, for a fallback the
        operator set themselves — ``--data-dir``, rather than ``-o <key> (default)``.
        """
        return self._parsed(
            key,
            None,
            as_safe_dir,
            fallback_raw=str(default),
            fallback_name=default_name,
        )

    def remaining(self) -> dict[str, str]:
        """Every option not read yet, as typed, and marks them read.

        For a device that passes options on to something this SDK has never seen —
        an executor named by import path, whose constructor is the only thing that
        knows its keys. A device that reads its own options should not need this.
        """
        rest = {
            key: value for key, value in self._values.items() if key not in self._read
        }
        self._read.update(rest)
        return rest

    def unread(self) -> tuple[str, ...]:
        """The keys given that nothing read, sorted.

        The caller reports them: the device has finished building by then, so a key
        left over is one it does not understand.
        """
        return tuple(sorted(set(self._values) - self._read))

    def _parsed(
        self,
        key,
        default,
        parse,
        *,
        fallback_raw: str | None = None,
        fallback_name: str = "",
    ):
        """Read *key* through *parse*, falling back to *default* or *fallback_raw*.

        A fallback given as a raw string goes through *parse* like any other value,
        which is how a default directory is checked for safety too. An error names
        whichever of the two the operator set: being told about an ``-o`` they never
        wrote sends them looking in the wrong place.
        """
        self._read.add(key)
        raw = self._values.get(key)
        label = f"-o {key}"
        if raw is None:
            if fallback_raw is None:
                return default
            raw = fallback_raw
            label = fallback_name or f"-o {key} (default)"

        try:
            return parse(raw)
        except ValueError as exc:
            raise ValueError(f"bad value for {label}: {exc}") from exc

get_bool(key, default=False)

The value of key as a boolean, or default.

1, true, yes and on are true in any case; every other value, the empty string included, is false. Nothing here can fail, so a misspelled -o is_dummy=ture is false rather than an error — the alternative is refusing to start over a flag.

Source code in qpi-driver/py/qpi_driver/options.py

def get_bool(self, key: str, default: bool = False) -> bool:
    """The value of *key* as a boolean, or *default*.

    ``1``, ``true``, ``yes`` and ``on`` are true in any case; every other
    value, the empty string included, is false. Nothing here can fail, so a
    misspelled ``-o is_dummy=ture`` is false rather than an error — the
    alternative is refusing to start over a flag.
    """
    self._read.add(key)
    if key not in self._values:
        return default
    return self._values[key].strip().lower() in _TRUTHY

get_dir(key, default, *, default_name='')

The value of key as a directory a driver may write to, or default.

The safe-location check of :func:~qpi_driver.paths.as_safe_dir applies to the given value and to default alike, so a device cannot default itself somewhere it may not write.

default_name is what to call default in that error, for a fallback the operator set themselves — --data-dir, rather than -o <key> (default).

Source code in qpi-driver/py/qpi_driver/options.py

def get_dir(self, key: str, default: Path | str, *, default_name: str = "") -> Path:
    """The value of *key* as a directory a driver may write to, or *default*.

    The safe-location check of :func:`~qpi_driver.paths.as_safe_dir` applies to
    the given value and to *default* alike, so a device cannot default itself
    somewhere it may not write.

    *default_name* is what to call *default* in that error, for a fallback the
    operator set themselves — ``--data-dir``, rather than ``-o <key> (default)``.
    """
    return self._parsed(
        key,
        None,
        as_safe_dir,
        fallback_raw=str(default),
        fallback_name=default_name,
    )

get_float(key, default)

The value of key as a float, or default.

Source code in qpi-driver/py/qpi_driver/options.py

def get_float(self, key: str, default: float) -> float:
    """The value of *key* as a float, or *default*."""
    return self._parsed(key, default, float)

get_int(key, default)

The value of key as an integer, or default.

Source code in qpi-driver/py/qpi_driver/options.py

def get_int(self, key: str, default: int) -> int:
    """The value of *key* as an integer, or *default*."""
    return self._parsed(key, default, int)

get_path(key, default)

The value of key as a path, or default. Not checked for safety.

Source code in qpi-driver/py/qpi_driver/options.py

def get_path(self, key: str, default: Path | str) -> Path:
    """The value of *key* as a path, or *default*. Not checked for safety."""
    return self._parsed(key, Path(default), Path)

get_str(key, default='')

The value of key as typed, or default.

Source code in qpi-driver/py/qpi_driver/options.py

def get_str(self, key: str, default: str = "") -> str:
    """The value of *key* as typed, or *default*."""
    self._read.add(key)
    return self._values.get(key, default)

remaining()

Every option not read yet, as typed, and marks them read.

For a device that passes options on to something this SDK has never seen — an executor named by import path, whose constructor is the only thing that knows its keys. A device that reads its own options should not need this.

Source code in qpi-driver/py/qpi_driver/options.py

def remaining(self) -> dict[str, str]:
    """Every option not read yet, as typed, and marks them read.

    For a device that passes options on to something this SDK has never seen —
    an executor named by import path, whose constructor is the only thing that
    knows its keys. A device that reads its own options should not need this.
    """
    rest = {
        key: value for key, value in self._values.items() if key not in self._read
    }
    self._read.update(rest)
    return rest

require(key, example='')

The value of key, or raise because the device cannot run without it.

Raises:

Type	Description
ValueError	if key was not given. example is appended as a ready-to-paste -o pair when there is a sensible one.

Source code in qpi-driver/py/qpi_driver/options.py

def require(self, key: str, example: str = "") -> str:
    """The value of *key*, or raise because the device cannot run without it.

    Raises:
        ValueError: if *key* was not given. *example* is appended as a
            ready-to-paste ``-o`` pair when there is a sensible one.
    """
    self._read.add(key)
    if key not in self._values:
        hint = f", e.g. -o {key}={example}" if example else ""
        raise ValueError(f"missing required option {key!r}{hint}")
    return self._values[key]

unread()

The keys given that nothing read, sorted.

The caller reports them: the device has finished building by then, so a key left over is one it does not understand.

Source code in qpi-driver/py/qpi_driver/options.py

def unread(self) -> tuple[str, ...]:
    """The keys given that nothing read, sorted.

    The caller reports them: the device has finished building by then, so a key
    left over is one it does not understand.
    """
    return tuple(sorted(set(self._values) - self._read))

PrestoExecutor

Bases: Executor

Executor subclass for interacting with Presto RF signal generators.

Source code in qpi-driver/py/qpi_driver/executors/presto/__init__.py

class PrestoExecutor(Executor):
    """Executor subclass for interacting with Presto RF signal generators."""

    def execute(self, payload: dict) -> xr.Dataset:
        """Execute quantum instructions on Presto hardware.

        Args:
            payload: Dictionary containing circuit execution options.

        Returns:
            xr.Dataset: The control acquisition dataset.

        Raises:
            NotImplementedError: Always raised since the executor is a placeholder.
        """
        raise NotImplementedError("PrestoExecutor is not implemented yet.")

    def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
        """Convert raw dataset to Qiskit-compatible result dict.

        Args:
            dataset: The xr.Dataset returned by execute().
            job_id: The unique ID of the quantum job.

        Returns:
            dict: Qiskit-compatible result dict.

        Raises:
            NotImplementedError: Always raised since the executor is a placeholder.
        """
        raise NotImplementedError("PrestoExecutor is not implemented yet.")

execute(payload)

Execute quantum instructions on Presto hardware.

Parameters:

Name	Type	Description	Default
payload	dict	Dictionary containing circuit execution options.	required

Returns:

Type	Description
Dataset	xr.Dataset: The control acquisition dataset.

Raises:

Type	Description
NotImplementedError	Always raised since the executor is a placeholder.

Source code in qpi-driver/py/qpi_driver/executors/presto/__init__.py

def execute(self, payload: dict) -> xr.Dataset:
    """Execute quantum instructions on Presto hardware.

    Args:
        payload: Dictionary containing circuit execution options.

    Returns:
        xr.Dataset: The control acquisition dataset.

    Raises:
        NotImplementedError: Always raised since the executor is a placeholder.
    """
    raise NotImplementedError("PrestoExecutor is not implemented yet.")

process_result(dataset, job_id)

Convert raw dataset to Qiskit-compatible result dict.

Parameters:

Name	Type	Description	Default
dataset	Dataset	The xr.Dataset returned by execute().	required
job_id	str	The unique ID of the quantum job.	required

Returns:

Name	Type	Description
dict	dict	Qiskit-compatible result dict.

Raises:

Type	Description
NotImplementedError	Always raised since the executor is a placeholder.

Source code in qpi-driver/py/qpi_driver/executors/presto/__init__.py

def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
    """Convert raw dataset to Qiskit-compatible result dict.

    Args:
        dataset: The xr.Dataset returned by execute().
        job_id: The unique ID of the quantum job.

    Returns:
        dict: Qiskit-compatible result dict.

    Raises:
        NotImplementedError: Always raised since the executor is a placeholder.
    """
    raise NotImplementedError("PrestoExecutor is not implemented yet.")

QbloxExecutor

Bases: Executor

Executor subclass for interacting with Qblox instruments and modules via qblox-scheduler.

Source code in qpi-driver/py/qpi_driver/executors/qblox/__init__.py

class QbloxExecutor(Executor):
    """Executor subclass for interacting with Qblox instruments and modules via qblox-scheduler."""

    def __new__(cls, *args, **kwargs):
        if not IS_QBLOX_SCHEDULER_INSTALLED:
            raise ImportError(
                "qblox-scheduler is not installed. Install the [qblox] extra to use QbloxExecutor."
            )
        return super().__new__(cls)

    def __init__(
        self,
        name: str = "qblox",
        quantify_hardware_config: QbloxHardwareCompilationConfig | Path | dict = Path(
            "quantify.hardware.json"
        ),
        quantify_device_config: Path | dict = Path("quantify.device.json"),
        is_dummy: bool = False,
        data_dir: Path = Path("data"),
        acquisition_timeout: int = 10,
        save_raw_data: bool = False,
        **kwargs: Any,
    ) -> None:
        """Initialize the QbloxExecutor.

        Args:
            name: the name of the executor
            quantify_hardware_config: Hardware-layer configuration dictionary, file path, or config as dict.
            quantify_device_config: Device-layer configuration dictionary, file path or config as dict
            is_dummy: If True, uses a dummy Cluster instrument.
            data_dir: Directory to where data is temporarily stored.
            acquisition_timeout: Timeout in seconds to wait for acquisition.
            save_raw_data: If True, the agent writes a dataset and an instrument
                snapshot under *data_dir* for every job. Off by default: nothing
                here reads them and nothing prunes them, so a QPU serving jobs
                would fill its data directory over a run of months.
            **kwargs: Arbitrary keyword arguments passed to the base class.
        """
        super().__init__(name, **kwargs)
        # Clean up any previously registered instruments to avoid name collision errors in QCoDeS
        with suppress(Exception):
            Instrument.close_all()

        self._data_dir = data_dir
        self._should_save_raw_data = bool(save_raw_data)
        is_simulated = bool(kwargs.pop("is_simulated", False))
        if is_dummy and is_simulated:
            raise ValueError(
                "is_dummy and is_simulated both replace the cluster; pick one"
            )
        self._is_dummy = is_dummy
        self._is_simulated = is_simulated
        self._acquisition_timeout = acquisition_timeout
        self._hardware_config = load_quantify_hardware_config(quantify_hardware_config)
        self._watched_hardware_config = (
            ConfigFile(quantify_hardware_config)
            if isinstance(quantify_hardware_config, Path)
            else None
        )
        self._watched_device_config = (
            ConfigFile(quantify_device_config)
            if isinstance(quantify_device_config, Path)
            else None
        )
        self._device = load_quantum_device(name=name, config=quantify_device_config)
        if is_simulated:
            # A real agent for compilation, the simulator for execution — the
            # agent's `run` is the only part a chip is needed for. See
            # `qpi_driver.simulation.agent`.
            from qpi_driver.executors.utils.coupler_bias import (
                declared_parking_currents,
                declared_sideband_gaps,
            )
            from qpi_driver.simulation.agent import simulated_agent

            self._agent = simulated_agent(
                hardware_configuration=self._hardware_config,
                quantum_device_configuration=self._device,
                output_dir=data_dir,
                simulator=kwargs.get("simulator"),
                sideband_gaps=declared_sideband_gaps(self._device),
                parking_currents=declared_parking_currents(self._device),
            )
        else:
            self._agent = HardwareAgent(
                hardware_configuration=self._hardware_config,
                quantum_device_configuration=self._device,
                create_dummy_connections=is_dummy,
                output_dir=data_dir,
            )
        self._bias_source = self._park_couplers(kwargs.get("spi_rack_address"))

    def _reload_device_config(self) -> None:
        """Apply the device config file, which has changed on disk, to the device.

        A calibration on this node rewrites that file (RFC 0004 §8), and so does
        anyone restoring parameters by hand. Neither can reach into this process,
        so the file is the whole channel.

        A file that will not parse leaves the device as it was and costs the next
        job its new parameters, not the driver its device. The tuner's own writes
        are atomic, so only a hand-dropped file can be seen part-written.
        """
        path = self._watched_device_config.path
        try:
            unknown = apply_device_config(self._device, path)
        except Exception:
            log.exception("could not reload %s; keeping the current parameters", path)
            return

        if unknown:
            log.warning(
                "%s names %s, which this device does not have; a new element needs a "
                "restart",
                path,
                ", ".join(unknown),
            )
        log.info("reloaded device parameters from %s", path)

        if self._is_simulated:
            from qpi_driver.executors.utils.coupler_bias import (
                declared_parking_currents,
                declared_sideband_gaps,
            )

            coordinator = getattr(self._agent, "instrument_coordinator", None)
            if coordinator is not None:
                coordinator.sideband_gaps = declared_sideband_gaps(self._device)
                coordinator.parking_currents = declared_parking_currents(self._device)

        # The device object is the same one, so the agent still points at it. The
        # bias does not follow: it is a current sitting in a rack, and only
        # re-applying it moves the coupler.
        self._reapply_coupler_bias()

    def _warn_if_hardware_config_moved(self) -> None:
        """Say so when the hardware config changes, and keep running on the old one.

        Deliberately not reloaded. It builds the instrument coordinator and the
        Cluster behind it, so applying a new one means closing a live connection to
        the rack and dialling it again — and a reconnect that fails leaves this
        driver with no coordinator and no way back, the old one being already gone.
        The device config has somewhere to fall back to; this does not.

        A restart is the honest answer: rewiring a rack is not a runtime event.
        Warned once per change so a stale hardware config is at least not a silent
        one.
        """
        if not self._watched_hardware_config:
            return
        if not self._watched_hardware_config.changed():
            return
        self._watched_hardware_config.mark_read()
        log.warning(
            "%s has changed; this driver is still running on the hardware config it "
            "started with. Restart it to pick the new one up.",
            self._watched_hardware_config.path,
        )

    def _reapply_coupler_bias(self) -> None:
        """Hold the couplers at the reloaded currents, on the rack already open.

        Resolving a second source would open a second connection to the same
        rack, or clash on its qcodes name.
        """
        from qpi_driver.executors.utils.coupler_bias import apply_coupler_bias

        try:
            self._parked = apply_coupler_bias(self._device, self._bias_source)
        except Exception:
            log.exception("could not park the couplers; two-qubit gates will be wrong")
            self._parked = {}

    @property
    def hardware_config(self) -> QbloxHardwareCompilationConfig:
        return self._hardware_config

    def execute(self, payload: JobPayload) -> xr.Dataset:
        """Execute quantum instructions using the Qblox scheduler.

        Every circuit in ``payload.circuits`` is executed, honouring each
        circuit's ``shots`` override and ``parameter_values`` bindings.  A
        single-circuit payload returns that circuit's flat dataset; multi-circuit
        payloads are bundled so circuits with different qubit widths or shot
        counts stay independent (see ``combine_circuit_datasets``).

        Args:
            payload: JobPayload specifying shots, circuits, meas_level, etc.

        Returns:
            xr.Dataset: Raw acquisition dataset.
        """
        # Between jobs, never during one: a reload part-way through a compilation
        # would be worse than a stale parameter.
        if self._watched_device_config and self._watched_device_config.changed():
            self._reload_device_config()
            self._watched_device_config.mark_read()
        self._warn_if_hardware_config_moved()

        acq_protocol, acq_kwargs, acq_overrides = self._resolve_acq_protocol(payload)
        sub_datasets: list[xr.Dataset] = []

        for circ in payload.circuits:
            circ_shots = circ.shots if circ.shots is not None else payload.shots
            circuit = load_qasm(circ.circuit)

            for param_values in circ.parameter_values or [None]:
                bound_circuit = circuit
                if param_values is not None and circuit.parameters:
                    bound_circuit = circuit.assign_parameters(param_values)
                sub_datasets.append(
                    self._acquire_circuit(
                        payload,
                        bound_circuit,
                        circ_shots,
                        acq_protocol,
                        acq_kwargs,
                        acq_overrides,
                    )
                )

        return combine_circuit_datasets(sub_datasets)

    def _park_couplers(self, spi_rack_address: str | None):
        """Hold every tunable coupler at its calibrated DC bias.

        The same thing the quantify executor does, for the same reason: the bias
        is a seconds-scale DC current that no schedule can express, so if the
        driver does not set it then nothing does and every CZ runs against a
        coupler parked wherever it was left.
        """
        from qpi_driver.executors.utils.coupler_bias import (
            RecordingBias,
            apply_coupler_bias,
            resolve_bias_source,
        )

        try:
            source = (
                RecordingBias()
                if (self._is_simulated or self._is_dummy)
                else resolve_bias_source(
                    self._device,
                    cluster=self._cluster(),
                    spi_address=spi_rack_address,
                )
            )
            self._parked = apply_coupler_bias(self._device, source)
        except Exception:
            # A coupler that cannot be parked is a broken two-qubit gate, not a
            # broken node.
            log.exception("could not park the couplers; two-qubit gates will be wrong")
            self._parked = {}
            return RecordingBias()
        return source

    def _cluster(self):
        """The first cluster the agent is connected to, if any."""
        clusters = getattr(self._agent, "get_clusters", lambda: [])()
        return clusters[0] if clusters else None

    def _acquire_circuit(
        self,
        payload: JobPayload,
        circuit: QuantumCircuit,
        shots: int,
        acq_protocol: str,
        acq_kwargs: dict,
        acq_overrides: dict[int, dict[str, float]],
    ) -> xr.Dataset:
        """One circuit's acquisition, in as many runs as the hardware requires.

        Almost always one. The exception is a raw trace over more than one
        qubit: a Qblox module can put a single sequencer into scope mode, so
        asking two qubits for a trace at once does not compile. The circuit is
        played once per measured qubit instead, capturing one trace each time,
        at an honest cost of N runs for N qubits.

        The same constraint and the same remedy as the quantify executor's —
        it is a property of the module, not of the scheduler driving it.
        """
        measured = measured_qubits(circuit)
        if acq_protocol != "Trace" or len(measured) < 2:
            return self._run_circuit(
                payload, circuit, shots, acq_protocol, acq_kwargs, acq_overrides
            )

        log.info(
            "raw trace over %d qubits: taking %d runs, one scope-mode acquisition each",
            len(measured),
            len(measured),
        )
        passes = [
            self._run_circuit(
                payload,
                circuit,
                shots,
                acq_protocol,
                acq_kwargs,
                acq_overrides,
                only_qubit=qubit,
            )
            for qubit in measured
        ]
        return xr.merge(passes, combine_attrs="override")

    def _run_circuit(
        self,
        payload: JobPayload,
        circuit: QuantumCircuit,
        shots: int,
        acq_protocol: str,
        acq_kwargs: dict,
        acq_overrides: dict[int, dict[str, float]],
        only_qubit: int | None = None,
    ) -> xr.Dataset:
        """Run a single (parameter-bound) circuit and return its acquisition dataset."""
        schedule, clbit_map, num_clbits = generate_schedule(
            name=payload.id,
            circuit=circuit,
            shots=shots,
            acq_protocol=acq_protocol,
            acq_kwargs=acq_kwargs,
            acq_overrides=acq_overrides,
            readout_points=readout_points_by_qubit(self._device),
            only_qubit=only_qubit,
        )

        dataset = self._agent.run(
            schedule,
            timeout=self._acquisition_timeout,
            save_to_experiment=self._should_save_raw_data,
            save_snapshot=self._should_save_raw_data,
        )
        dataset.attrs.update(
            {
                "shots": shots,
                "n_qubits": circuit.num_qubits,
                "backend": self.name,
                "meas_level": payload.meas_level,
                "meas_return": payload.meas_return,
                "acq_protocol": acq_protocol,
                "clbit_map": [list(entry) for entry in clbit_map],
                "num_clbits": num_clbits,
            }
        )
        # Per qubit, because the discriminator is. `process_result` reads this back
        # when it has to threshold in software, and a single pair here would send it
        # down the same collapse the schedule no longer has.
        if acq_overrides:
            dataset.attrs["acq_discriminators"] = {
                str(index): dict(values) for index, values in acq_overrides.items()
            }
        return dataset

    def _resolve_acq_protocol(
        self, payload: JobPayload
    ) -> tuple[str, dict, dict[int, dict[str, float]]]:
        """Determine the qblox-scheduler acquisition protocol for the given meas_level.

        Args:
            payload: JobPayload specifying meas_level, acq_threshold, acq_rotation, etc.

        Returns:
            Tuple of (protocol_name, kwargs_for_every_Measure, kwargs_per_qubit).
        """
        bin_mode = self._resolve_bin_mode(payload.meas_level, payload.meas_return)

        if payload.meas_level == 0:
            return "Trace", {"bin_mode": bin_mode}, {}

        if payload.meas_level == 1:
            return "SSBIntegrationComplex", {"bin_mode": bin_mode}, {}

        # meas_level == 2: threshold on the instrument when every qubit has a line to
        # threshold against, in software otherwise. Per qubit either way — see
        # `qpi_driver.executors.utils.discriminator`.
        per_qubit, complete = resolve_discriminators(
            self._device, payload.acq_rotation, payload.acq_threshold
        )
        # And the power those shots are taken at, where a `CalibratedTransmon` says so.
        # The matching frequency cannot ride on `Measure` — it is a clock, not a pulse
        # parameter — so `_readout_overrides` sets it on the schedule instead.
        for index, point in readout_points_by_qubit(self._device).items():
            if index in per_qubit:
                per_qubit[index]["pulse_amp"] = point["pulse_amp"]
        protocol = "ThresholdedAcquisition" if complete else "SSBIntegrationComplex"
        return protocol, {"bin_mode": bin_mode}, per_qubit

    def _resolve_bin_mode(self, meas_level: int, meas_return: str) -> BinMode:
        """Choose the acquisition bin mode implied by the requested measurement mode.

        Raw traces (level 0) and averaged integrated results (level 1 with
        ``meas_return="avg"``) collapse every repetition into a single bin.
        Counts (level 2) and single-shot integrated results keep each shot as a
        separate bin so results can be processed per shot.
        """
        if meas_level == 0:
            return BinMode.AVERAGE
        if meas_level == 1 and meas_return == "avg":
            return BinMode.AVERAGE
        return BinMode.APPEND

    def _get_threshold_params(self) -> dict[int, dict[str, float]]:
        """Every qubit's discriminator, by qubit index."""
        return discriminators_by_qubit(self._device)

    def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
        """Convert a qblox-scheduler acquisition dataset into a Qiskit-compatible result dict.

        Handles all meas_levels:
        - meas_level=0 (Trace): Returns raw complex waveform data as [[real, imag], ...] per time sample.
        - meas_level=1 (SSBIntegrationComplex): Returns IQ values as [[real, imag]] per shot per qubit.
        - meas_level=2 with ThresholdedAcquisition: Aggregates 0/1 values into counts dict.
        - meas_level=2 with SSBIntegrationComplex: Performs software discrimination using
          acq_threshold and acq_rotation from the device config.

        Args:
            dataset: xr.Dataset from execute().
            job_id: Unique job ID.

        Returns:
            dict: Qiskit-compatible result dict.
        """
        from qpi_driver.executors.utils.result import build_qiskit_result

        meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
        meas_return = str(dataset.attrs.get("meas_return", "single"))
        acq_protocol = str(dataset.attrs.get("acq_protocol", "SSBIntegrationComplex"))
        backend = dataset.attrs.get("backend", self.name)

        circuit_results = [
            self._single_dataset_to_result(
                sub_ds, meas_level, meas_return, acq_protocol
            )
            for sub_ds in iter_circuit_datasets(dataset)
        ]
        return build_qiskit_result(circuit_results, job_id, backend)

    def _single_dataset_to_result(
        self, dataset: xr.Dataset, meas_level: int, meas_return: str, acq_protocol: str
    ) -> dict:
        """Extract result data from a single-circuit quantify dataset."""
        qubit_vars, q0_key, shots = self._extract_qubit_vars(dataset)
        if not qubit_vars:
            return {"raw": str(dataset), "shots": 0}

        if meas_level == 0:
            return self._process_meas_level_0(dataset, qubit_vars, shots)
        if meas_level == 1:
            return self._process_meas_level_1(dataset, qubit_vars, shots, meas_return)
        return self._process_meas_level_2(dataset, qubit_vars, shots, acq_protocol)

    def _extract_qubit_vars(self, dataset: xr.Dataset) -> tuple[list[int], str, int]:
        """Identify qubit variables (integer-named data vars) and shots."""
        qubit_vars = []
        for var in dataset.data_vars:
            try:
                qubit_vars.append(int(var))
            except ValueError:
                pass
        if not qubit_vars:
            return [], "", 0
        qubit_vars.sort()
        q0_key = qubit_key(dataset, qubit_vars[0])
        shots = cast_to(int, dataset.attrs.get("shots"), len(dataset[q0_key]))
        return qubit_vars, q0_key, shots

    def _process_meas_level_0(
        self, dataset: xr.Dataset, qubit_vars: list[int], shots: int
    ) -> dict:
        """Extract raw complex trace data (meas_level=0)."""
        memory: list[list[list[float]]] = []
        for q_idx in qubit_vars:
            var_key = q_idx if q_idx in dataset else str(q_idx)
            trace = dataset[var_key].values.flatten()
            qubit_trace = [[float(v.real), float(v.imag)] for v in trace]
            memory.append(qubit_trace)
        return {"memory": memory, "shots": shots}

    def _process_meas_level_1(
        self, dataset: xr.Dataset, qubit_vars: list[int], shots: int, meas_return: str
    ) -> dict:
        """Extract integrated IQ memory (meas_level=1)."""
        from qpi_driver.executors.utils.result import iq_memory_avg

        per_shot = {
            q_idx: per_shot_values(dataset[qubit_key(dataset, q_idx)])
            for q_idx in qubit_vars
        }
        num_samples = len(per_shot[qubit_vars[0]])
        memory = []
        for s in range(num_samples):
            shot_iq = []
            for q_idx in qubit_vars:
                val = per_shot[q_idx][s]
                r = float(val.real) if not np.isnan(val.real) else 0.0
                i = float(val.imag) if not np.isnan(val.imag) else 0.0
                shot_iq.append([r, i])
            memory.append(shot_iq)

        if meas_return == "avg" and memory:
            memory = iq_memory_avg(memory, len(qubit_vars))
        return {"memory": memory, "shots": shots}

    def _process_meas_level_2(
        self, dataset: xr.Dataset, qubit_vars: list[int], shots: int, acq_protocol: str
    ) -> dict:
        """Extract classified counts (meas_level=2) performing software discrimination if needed.

        Counts are keyed by classical register, one bit per measured clbit,
        positioned by clbit index (see ``build_acquisition_counts``).
        """
        discriminate = build_discriminator(
            dataset, acq_protocol, self._get_threshold_params
        )
        counts_dict = build_acquisition_counts(dataset, qubit_vars, discriminate)
        return {"counts": counts_dict, "shots": shots}

    def close(self) -> None:
        """Release resources."""
        with suppress(Exception):
            self._agent.instrument_coordinator.close()

__init__(name='qblox', quantify_hardware_config=Path('quantify.hardware.json'), quantify_device_config=Path('quantify.device.json'), is_dummy=False, data_dir=Path('data'), acquisition_timeout=10, save_raw_data=False, **kwargs)

Initialize the QbloxExecutor.

Parameters:

Name	Type	Description	Default
name	str	the name of the executor	'qblox'
quantify_hardware_config	QbloxHardwareCompilationConfig | Path | dict	Hardware-layer configuration dictionary, file path, or config as dict.	Path('quantify.hardware.json')
quantify_device_config	Path | dict	Device-layer configuration dictionary, file path or config as dict	Path('quantify.device.json')
is_dummy	bool	If True, uses a dummy Cluster instrument.	False
data_dir	Path	Directory to where data is temporarily stored.	Path('data')
acquisition_timeout	int	Timeout in seconds to wait for acquisition.	10
save_raw_data	bool	If True, the agent writes a dataset and an instrument snapshot under data_dir for every job. Off by default: nothing here reads them and nothing prunes them, so a QPU serving jobs would fill its data directory over a run of months.	False
**kwargs	Any	Arbitrary keyword arguments passed to the base class.	{}

Source code in qpi-driver/py/qpi_driver/executors/qblox/__init__.py

def __init__(
    self,
    name: str = "qblox",
    quantify_hardware_config: QbloxHardwareCompilationConfig | Path | dict = Path(
        "quantify.hardware.json"
    ),
    quantify_device_config: Path | dict = Path("quantify.device.json"),
    is_dummy: bool = False,
    data_dir: Path = Path("data"),
    acquisition_timeout: int = 10,
    save_raw_data: bool = False,
    **kwargs: Any,
) -> None:
    """Initialize the QbloxExecutor.

    Args:
        name: the name of the executor
        quantify_hardware_config: Hardware-layer configuration dictionary, file path, or config as dict.
        quantify_device_config: Device-layer configuration dictionary, file path or config as dict
        is_dummy: If True, uses a dummy Cluster instrument.
        data_dir: Directory to where data is temporarily stored.
        acquisition_timeout: Timeout in seconds to wait for acquisition.
        save_raw_data: If True, the agent writes a dataset and an instrument
            snapshot under *data_dir* for every job. Off by default: nothing
            here reads them and nothing prunes them, so a QPU serving jobs
            would fill its data directory over a run of months.
        **kwargs: Arbitrary keyword arguments passed to the base class.
    """
    super().__init__(name, **kwargs)
    # Clean up any previously registered instruments to avoid name collision errors in QCoDeS
    with suppress(Exception):
        Instrument.close_all()

    self._data_dir = data_dir
    self._should_save_raw_data = bool(save_raw_data)
    is_simulated = bool(kwargs.pop("is_simulated", False))
    if is_dummy and is_simulated:
        raise ValueError(
            "is_dummy and is_simulated both replace the cluster; pick one"
        )
    self._is_dummy = is_dummy
    self._is_simulated = is_simulated
    self._acquisition_timeout = acquisition_timeout
    self._hardware_config = load_quantify_hardware_config(quantify_hardware_config)
    self._watched_hardware_config = (
        ConfigFile(quantify_hardware_config)
        if isinstance(quantify_hardware_config, Path)
        else None
    )
    self._watched_device_config = (
        ConfigFile(quantify_device_config)
        if isinstance(quantify_device_config, Path)
        else None
    )
    self._device = load_quantum_device(name=name, config=quantify_device_config)
    if is_simulated:
        # A real agent for compilation, the simulator for execution — the
        # agent's `run` is the only part a chip is needed for. See
        # `qpi_driver.simulation.agent`.
        from qpi_driver.executors.utils.coupler_bias import (
            declared_parking_currents,
            declared_sideband_gaps,
        )
        from qpi_driver.simulation.agent import simulated_agent

        self._agent = simulated_agent(
            hardware_configuration=self._hardware_config,
            quantum_device_configuration=self._device,
            output_dir=data_dir,
            simulator=kwargs.get("simulator"),
            sideband_gaps=declared_sideband_gaps(self._device),
            parking_currents=declared_parking_currents(self._device),
        )
    else:
        self._agent = HardwareAgent(
            hardware_configuration=self._hardware_config,
            quantum_device_configuration=self._device,
            create_dummy_connections=is_dummy,
            output_dir=data_dir,
        )
    self._bias_source = self._park_couplers(kwargs.get("spi_rack_address"))

close()

Release resources.

Source code in qpi-driver/py/qpi_driver/executors/qblox/__init__.py

def close(self) -> None:
    """Release resources."""
    with suppress(Exception):
        self._agent.instrument_coordinator.close()

execute(payload)

Execute quantum instructions using the Qblox scheduler.

Every circuit in payload.circuits is executed, honouring each circuit's shots override and parameter_values bindings. A single-circuit payload returns that circuit's flat dataset; multi-circuit payloads are bundled so circuits with different qubit widths or shot counts stay independent (see combine_circuit_datasets).

Parameters:

Name	Type	Description	Default
payload	JobPayload	JobPayload specifying shots, circuits, meas_level, etc.	required

Returns:

Type	Description
Dataset	xr.Dataset: Raw acquisition dataset.

Source code in qpi-driver/py/qpi_driver/executors/qblox/__init__.py

def execute(self, payload: JobPayload) -> xr.Dataset:
    """Execute quantum instructions using the Qblox scheduler.

    Every circuit in ``payload.circuits`` is executed, honouring each
    circuit's ``shots`` override and ``parameter_values`` bindings.  A
    single-circuit payload returns that circuit's flat dataset; multi-circuit
    payloads are bundled so circuits with different qubit widths or shot
    counts stay independent (see ``combine_circuit_datasets``).

    Args:
        payload: JobPayload specifying shots, circuits, meas_level, etc.

    Returns:
        xr.Dataset: Raw acquisition dataset.
    """
    # Between jobs, never during one: a reload part-way through a compilation
    # would be worse than a stale parameter.
    if self._watched_device_config and self._watched_device_config.changed():
        self._reload_device_config()
        self._watched_device_config.mark_read()
    self._warn_if_hardware_config_moved()

    acq_protocol, acq_kwargs, acq_overrides = self._resolve_acq_protocol(payload)
    sub_datasets: list[xr.Dataset] = []

    for circ in payload.circuits:
        circ_shots = circ.shots if circ.shots is not None else payload.shots
        circuit = load_qasm(circ.circuit)

        for param_values in circ.parameter_values or [None]:
            bound_circuit = circuit
            if param_values is not None and circuit.parameters:
                bound_circuit = circuit.assign_parameters(param_values)
            sub_datasets.append(
                self._acquire_circuit(
                    payload,
                    bound_circuit,
                    circ_shots,
                    acq_protocol,
                    acq_kwargs,
                    acq_overrides,
                )
            )

    return combine_circuit_datasets(sub_datasets)

process_result(dataset, job_id)

Convert a qblox-scheduler acquisition dataset into a Qiskit-compatible result dict.

Handles all meas_levels: - meas_level=0 (Trace): Returns raw complex waveform data as [[real, imag], ...] per time sample. - meas_level=1 (SSBIntegrationComplex): Returns IQ values as [[real, imag]] per shot per qubit. - meas_level=2 with ThresholdedAcquisition: Aggregates 0/1 values into counts dict. - meas_level=2 with SSBIntegrationComplex: Performs software discrimination using acq_threshold and acq_rotation from the device config.

Parameters:

Name	Type	Description	Default
dataset	Dataset	xr.Dataset from execute().	required
job_id	str	Unique job ID.	required

Returns:

Name	Type	Description
dict	dict	Qiskit-compatible result dict.

Source code in qpi-driver/py/qpi_driver/executors/qblox/__init__.py

def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
    """Convert a qblox-scheduler acquisition dataset into a Qiskit-compatible result dict.

    Handles all meas_levels:
    - meas_level=0 (Trace): Returns raw complex waveform data as [[real, imag], ...] per time sample.
    - meas_level=1 (SSBIntegrationComplex): Returns IQ values as [[real, imag]] per shot per qubit.
    - meas_level=2 with ThresholdedAcquisition: Aggregates 0/1 values into counts dict.
    - meas_level=2 with SSBIntegrationComplex: Performs software discrimination using
      acq_threshold and acq_rotation from the device config.

    Args:
        dataset: xr.Dataset from execute().
        job_id: Unique job ID.

    Returns:
        dict: Qiskit-compatible result dict.
    """
    from qpi_driver.executors.utils.result import build_qiskit_result

    meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
    meas_return = str(dataset.attrs.get("meas_return", "single"))
    acq_protocol = str(dataset.attrs.get("acq_protocol", "SSBIntegrationComplex"))
    backend = dataset.attrs.get("backend", self.name)

    circuit_results = [
        self._single_dataset_to_result(
            sub_ds, meas_level, meas_return, acq_protocol
        )
        for sub_ds in iter_circuit_datasets(dataset)
    ]
    return build_qiskit_result(circuit_results, job_id, backend)

QiskitAerExecutor

Bases: Executor

Source code in qpi-driver/py/qpi_driver/executors/qiskit_aer/__init__.py

class QiskitAerExecutor(Executor):
    def __init__(self, name: str = "qiskit_aer", **kwargs: Any):
        if not IS_AER_INSTALLED:
            raise ImportError(
                "qiskit-aer is not installed. Install the [aer] extra to use QiskitAerExecutor."
            )

        super().__init__(name, **kwargs)
        self._simulator = AerSimulator()

    def execute(self, payload: JobPayload) -> xr.Dataset:
        """Run quantum circuit simulation using Qiskit Aer backend.

        Every circuit is run honouring its per-circuit ``shots`` override.  A
        single-circuit payload returns that circuit's flat dataset; multi-circuit
        payloads are bundled so circuits with different classical-bit widths or
        shot counts stay independent (see ``combine_circuit_datasets``).

        Args:
            payload: JobPayload specifying shots and circuits.

        Returns:
            xr.Dataset: Dataset containing measured state outcomes, counts, and frequencies.

        Raises:
            ImportError: If qiskit-aer is not installed.
            ValueError: If the provided QASM circuit cannot be loaded.
        """
        sub_datasets: list[xr.Dataset] = []

        for circ in payload.circuits:
            circ_shots = circ.shots if circ.shots is not None else payload.shots
            qasm_str = circ.circuit
            circuit = load_qasm(qasm_str)

            param_sets = circ.parameter_values or [None]
            for param_vals in param_sets:
                bound_circuit = circuit
                if param_vals is not None and circuit.parameters:
                    bound_circuit = circuit.assign_parameters(param_vals)

                t_qc = transpile(bound_circuit, self._simulator)
                result = self._simulator.run(
                    t_qc, shots=circ_shots, memory=True
                ).result()
                memory = result.get_memory(t_qc)

                ds = memory_to_dataset(
                    memory, circuit.num_clbits, circ_shots, payload.meas_level
                )
                ds.attrs.update(
                    {
                        "shots": circ_shots,
                        "n_qubits": circuit.num_qubits,
                        "backend": self.name,
                        "meas_level": payload.meas_level,
                        "meas_return": payload.meas_return,
                    }
                )
                sub_datasets.append(ds)

        return combine_circuit_datasets(sub_datasets)

    def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
        """Convert the simulator's xr.Dataset into a Qiskit-compatible result dict.

        Supports meas_level 1 (IQ memory) and 2 (classified counts).
        meas_level 0 is not supported for simulators.

        Args:
            dataset: xr.Dataset from execute().
            job_id: Unique job ID.

        Returns:
            dict: Qiskit-compatible result dict.
        """
        meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
        meas_return = str(dataset.attrs.get("meas_return", "single"))
        backend = dataset.attrs.get("backend", self.name)

        circuit_results = [
            simulator_dataset_to_result(sub_ds, meas_level, meas_return)
            for sub_ds in iter_circuit_datasets(dataset)
        ]
        return build_qiskit_result(circuit_results, job_id, backend)

execute(payload)

Run quantum circuit simulation using Qiskit Aer backend.

Every circuit is run honouring its per-circuit shots override. A single-circuit payload returns that circuit's flat dataset; multi-circuit payloads are bundled so circuits with different classical-bit widths or shot counts stay independent (see combine_circuit_datasets).

Parameters:

Name	Type	Description	Default
payload	JobPayload	JobPayload specifying shots and circuits.	required

Returns:

Type	Description
Dataset	xr.Dataset: Dataset containing measured state outcomes, counts, and frequencies.

Raises:

Type	Description
ImportError	If qiskit-aer is not installed.
ValueError	If the provided QASM circuit cannot be loaded.

Source code in qpi-driver/py/qpi_driver/executors/qiskit_aer/__init__.py

def execute(self, payload: JobPayload) -> xr.Dataset:
    """Run quantum circuit simulation using Qiskit Aer backend.

    Every circuit is run honouring its per-circuit ``shots`` override.  A
    single-circuit payload returns that circuit's flat dataset; multi-circuit
    payloads are bundled so circuits with different classical-bit widths or
    shot counts stay independent (see ``combine_circuit_datasets``).

    Args:
        payload: JobPayload specifying shots and circuits.

    Returns:
        xr.Dataset: Dataset containing measured state outcomes, counts, and frequencies.

    Raises:
        ImportError: If qiskit-aer is not installed.
        ValueError: If the provided QASM circuit cannot be loaded.
    """
    sub_datasets: list[xr.Dataset] = []

    for circ in payload.circuits:
        circ_shots = circ.shots if circ.shots is not None else payload.shots
        qasm_str = circ.circuit
        circuit = load_qasm(qasm_str)

        param_sets = circ.parameter_values or [None]
        for param_vals in param_sets:
            bound_circuit = circuit
            if param_vals is not None and circuit.parameters:
                bound_circuit = circuit.assign_parameters(param_vals)

            t_qc = transpile(bound_circuit, self._simulator)
            result = self._simulator.run(
                t_qc, shots=circ_shots, memory=True
            ).result()
            memory = result.get_memory(t_qc)

            ds = memory_to_dataset(
                memory, circuit.num_clbits, circ_shots, payload.meas_level
            )
            ds.attrs.update(
                {
                    "shots": circ_shots,
                    "n_qubits": circuit.num_qubits,
                    "backend": self.name,
                    "meas_level": payload.meas_level,
                    "meas_return": payload.meas_return,
                }
            )
            sub_datasets.append(ds)

    return combine_circuit_datasets(sub_datasets)

process_result(dataset, job_id)

Convert the simulator's xr.Dataset into a Qiskit-compatible result dict.

Supports meas_level 1 (IQ memory) and 2 (classified counts). meas_level 0 is not supported for simulators.

Parameters:

Name	Type	Description	Default
dataset	Dataset	xr.Dataset from execute().	required
job_id	str	Unique job ID.	required

Returns:

Name	Type	Description
dict	dict	Qiskit-compatible result dict.

Source code in qpi-driver/py/qpi_driver/executors/qiskit_aer/__init__.py

def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
    """Convert the simulator's xr.Dataset into a Qiskit-compatible result dict.

    Supports meas_level 1 (IQ memory) and 2 (classified counts).
    meas_level 0 is not supported for simulators.

    Args:
        dataset: xr.Dataset from execute().
        job_id: Unique job ID.

    Returns:
        dict: Qiskit-compatible result dict.
    """
    meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
    meas_return = str(dataset.attrs.get("meas_return", "single"))
    backend = dataset.attrs.get("backend", self.name)

    circuit_results = [
        simulator_dataset_to_result(sub_ds, meas_level, meas_return)
        for sub_ds in iter_circuit_datasets(dataset)
    ]
    return build_qiskit_result(circuit_results, job_id, backend)

QpiDriver

Bases: ABC

Base class for a QPI driver: handles inbound events, emits its own.

Attributes:

Name	Type	Description
qpi_addr		Full URL of the QPI-UI server.
token		The driver's access token; identifies it (and its QPU) to QPI-UI.
name		The display label QPI-UI has this driver registered under, used to tag emitted events. It comes from the drivers/connect response, so it is empty until :meth:run has connected.
ca_fingerprint		Expected SHA-256 of the server root CA, pinned over TLS.
ca_file_path		Where the downloaded root CA certificate is written.
recv_timeout_ms		How long the inbound receive loop blocks per attempt before checking for a shutdown signal, in milliseconds.

Source code in qpi-driver/py/qpi_driver/sdk.py

class QpiDriver(ABC):
    """Base class for a QPI driver: handles inbound events, emits its own.

    Attributes:
        qpi_addr: Full URL of the QPI-UI server.
        token: The driver's access token; identifies it (and its QPU) to QPI-UI.
        name: The display label QPI-UI has this driver registered under, used to
            tag emitted events. It comes from the ``drivers/connect`` response,
            so it is empty until :meth:`run` has connected.
        ca_fingerprint: Expected SHA-256 of the server root CA, pinned over TLS.
        ca_file_path: Where the downloaded root CA certificate is written.
        recv_timeout_ms: How long the inbound receive loop blocks per attempt
            before checking for a shutdown signal, in milliseconds.
    """

    def __init__(
        self,
        qpi_addr: str,
        token: str,
        ca_fingerprint: str = "",
        ca_file_path: str = "./bin/qpi.ca.pem",
        recv_timeout_ms: int = DEFAULT_RECV_TIMEOUT_MS,
    ) -> None:
        self.qpi_addr = qpi_addr
        self.token = token
        self.name = ""
        self.ca_fingerprint = ca_fingerprint
        self.ca_file_path = ca_file_path
        self.recv_timeout_ms = recv_timeout_ms

        self._out_sock: pynng.Push0 | None = None
        self._emit_lock = threading.Lock()
        self._stop = threading.Event()
        self._periodic: list[tuple[float, Callable[[], None]]] = []
        self._threads: list[threading.Thread] = []

    def emit(self, event: Event) -> None:
        """Send an event upward to QPI-UI over the outbound NNG channel.

        Delivery is best-effort, as today: if nothing is listening the event is
        dropped rather than buffered (RFC 0001 §5).

        Raises:
            RuntimeError: If called before the driver has connected.
        """
        if self._out_sock is None:
            raise RuntimeError("cannot emit before the driver is running")

        payload = self._encode_outbound(event)
        with self._emit_lock:
            self._out_sock.send(payload)

    def every(self, interval: float, fn: Callable[[], None]) -> None:
        """Register a callback to run every *interval* seconds while the driver runs.

        Used by drivers that report on their own schedule — e.g. a monitor that
        emits a reading on a timer — independently of any inbound event.
        """
        self._periodic.append((interval, fn))

    def run(self) -> None:
        """Connect to QPI-UI and process events until interrupted.

        Performs the handshake, opens the outbound channel, starts any periodic
        callbacks, then blocks on the inbound receive loop.
        """
        conn = self._connect()
        # The server owns the label, so the driver only knows it from here on.
        self.name = conn.name

        tls_config = TLSConfig(
            TLSConfig.MODE_CLIENT,
            server_name=conn.host,
            ca_files=conn.ca_file,
        )

        self._out_sock = pynng.Push0(tls_config=tls_config)
        out_addr = f"tls+tcp://{conn.host}:{conn.out_port}"
        self._out_sock.dial(out_addr, block=True)
        log.info("NNG PUSH connected to %s", out_addr)

        self._on_start()
        self._start_periodic()

        try:
            self._recv_loop(conn, tls_config)
        except KeyboardInterrupt:
            log.info("Shutdown signal received")
        finally:
            self._shutdown()

    def _recv_loop(self, conn: Connection, tls_config: TLSConfig) -> None:
        """Pull inbound events and dispatch each to its handler until stopped."""
        in_addr = f"tls+tcp://{conn.host}:{conn.in_port}"
        with pynng.Pull0(
            tls_config=tls_config, recv_timeout=self.recv_timeout_ms
        ) as sock:
            sock.dial(in_addr, block=True)
            log.info("NNG PULL connected to %s", in_addr)

            while not self._stop.is_set():
                try:
                    raw = sock.recv()
                except pynng.Timeout:
                    continue
                except pynng.Closed:
                    return

                event = self._decode_inbound(raw)
                if event is not None:
                    self._deliver(event)

    @abstractmethod
    def handle_event(self, event: Event) -> None:
        """Act on a single inbound event, dispatching on ``event.type``.

        Implemented per driver. An event a driver does not care about is simply
        ignored; raising signals a rejected event, which is logged and dropped.
        There is no application-level ACK/NACK (RFC 0001 §4).
        """

    def _deliver(self, event: Event) -> None:
        """Pass an event to :meth:`handle_event`, logging and dropping on failure."""
        try:
            self.handle_event(event)
        except Exception:
            log.exception(
                "dropping event %s of type %s: handler failed",
                event.id,
                event.type.value,
            )

    def _start_periodic(self) -> None:
        for interval, fn in self._periodic:
            thread = threading.Thread(
                target=self._run_periodic, args=(interval, fn), daemon=True
            )
            thread.start()
            self._threads.append(thread)

    def _run_periodic(self, interval: float, fn: Callable[[], None]) -> None:
        while not self._stop.wait(interval):
            try:
                fn()
            except Exception:
                log.exception("periodic callback failed")

    def _shutdown(self) -> None:
        log.info("Shutting down driver...")
        self._stop.set()
        self._on_stop()
        if self._out_sock is not None:
            self._out_sock.close()
            self._out_sock = None
        log.info("Shutdown complete.")

    def _connect(self) -> Connection:
        """Handshake with QPI-UI over the shared driver connect endpoint.

        Every driver connects the same way: the token identifies the driver
        (and, transitively, its QPU), and QPI-UI returns the NNG ports and host.
        What differs between drivers is only which events they handle and emit,
        not how they connect (RFC 0001 §3, §8).

        The token is the whole of the identity asserted here. The driver's display
        label comes back in the response rather than going out in the request: it
        belongs to the admin who typed it into the dashboard, and a driver sending
        one meant every restart silently overwrote what they chose.
        """
        resp = requests.post(
            f"{self.qpi_addr}/api/op/drivers/connect",
            json={"token": self.token},
            timeout=10,
        )
        resp.raise_for_status()
        data = resp.json()

        ca_file = _download_root_ca_cert(
            self.qpi_addr, self.ca_fingerprint, Path(self.ca_file_path)
        )
        return Connection(
            name=data.get("name", ""),
            host=data["nng_host"],
            in_port=int(data["nng_in_port"]),
            out_port=int(data["nng_out_port"]),
            ca_file=ca_file,
        )

    def _on_start(self) -> None:
        """Hook run after the outbound channel opens, before the receive loop.

        Subclasses that need background work (e.g. an executor subprocess) start
        it here.
        """

    def _on_stop(self) -> None:
        """Hook run once the receive loop exits, for releasing resources."""

    def _decode_inbound(self, raw: bytes) -> Event | None:
        """Turn a received wire message into an :class:`Event`, or ``None`` to drop it.

        The default parses the shared envelope (RFC 0001 §6); subclasses speaking
        a legacy wire shape override this.
        """
        try:
            return Event.from_json(raw)
        except Exception:
            log.exception("dropping malformed inbound message")
            return None

    def _encode_outbound(self, event: Event) -> bytes:
        """Serialise an outbound event to wire bytes.

        The default emits the shared envelope; subclasses speaking a legacy wire
        shape override this.
        """
        return event.to_json().encode()

emit(event)

Send an event upward to QPI-UI over the outbound NNG channel.

Delivery is best-effort, as today: if nothing is listening the event is dropped rather than buffered (RFC 0001 §5).

Raises:

Type	Description
RuntimeError	If called before the driver has connected.

Source code in qpi-driver/py/qpi_driver/sdk.py

def emit(self, event: Event) -> None:
    """Send an event upward to QPI-UI over the outbound NNG channel.

    Delivery is best-effort, as today: if nothing is listening the event is
    dropped rather than buffered (RFC 0001 §5).

    Raises:
        RuntimeError: If called before the driver has connected.
    """
    if self._out_sock is None:
        raise RuntimeError("cannot emit before the driver is running")

    payload = self._encode_outbound(event)
    with self._emit_lock:
        self._out_sock.send(payload)

every(interval, fn)

Register a callback to run every interval seconds while the driver runs.

Used by drivers that report on their own schedule — e.g. a monitor that emits a reading on a timer — independently of any inbound event.

Source code in qpi-driver/py/qpi_driver/sdk.py

def every(self, interval: float, fn: Callable[[], None]) -> None:
    """Register a callback to run every *interval* seconds while the driver runs.

    Used by drivers that report on their own schedule — e.g. a monitor that
    emits a reading on a timer — independently of any inbound event.
    """
    self._periodic.append((interval, fn))

handle_event(event) abstractmethod

Act on a single inbound event, dispatching on event.type.

Implemented per driver. An event a driver does not care about is simply ignored; raising signals a rejected event, which is logged and dropped. There is no application-level ACK/NACK (RFC 0001 §4).

Source code in qpi-driver/py/qpi_driver/sdk.py

@abstractmethod
def handle_event(self, event: Event) -> None:
    """Act on a single inbound event, dispatching on ``event.type``.

    Implemented per driver. An event a driver does not care about is simply
    ignored; raising signals a rejected event, which is logged and dropped.
    There is no application-level ACK/NACK (RFC 0001 §4).
    """

run()

Connect to QPI-UI and process events until interrupted.

Performs the handshake, opens the outbound channel, starts any periodic callbacks, then blocks on the inbound receive loop.

Source code in qpi-driver/py/qpi_driver/sdk.py

def run(self) -> None:
    """Connect to QPI-UI and process events until interrupted.

    Performs the handshake, opens the outbound channel, starts any periodic
    callbacks, then blocks on the inbound receive loop.
    """
    conn = self._connect()
    # The server owns the label, so the driver only knows it from here on.
    self.name = conn.name

    tls_config = TLSConfig(
        TLSConfig.MODE_CLIENT,
        server_name=conn.host,
        ca_files=conn.ca_file,
    )

    self._out_sock = pynng.Push0(tls_config=tls_config)
    out_addr = f"tls+tcp://{conn.host}:{conn.out_port}"
    self._out_sock.dial(out_addr, block=True)
    log.info("NNG PUSH connected to %s", out_addr)

    self._on_start()
    self._start_periodic()

    try:
        self._recv_loop(conn, tls_config)
    except KeyboardInterrupt:
        log.info("Shutdown signal received")
    finally:
        self._shutdown()

QpuDriver

Bases: QpiDriver

A QPI driver that runs quantum jobs on an executor.

Source code in qpi-driver/py/qpi_driver/builtins/qpu.py

class QpuDriver(QpiDriver):
    """A QPI driver that runs quantum jobs on an executor."""

    def __init__(
        self,
        qpi_addr: str = "http://127.0.0.1:8090",
        token: str = "",
        executor: str | type[Executor] | Executor = "mock",
        data_dir: Path = Path("bin/data"),
        ca_fingerprint: str = "",
        ca_file_path: Path = Path("./bin/qpi.ca.pem"),
        recv_timeout_ms: int = DEFAULT_RECV_TIMEOUT_MS,
        **executor_options: Any,
    ) -> None:
        super().__init__(
            qpi_addr=_normalize_qpi_addr(qpi_addr),
            token=token,
            ca_fingerprint=ca_fingerprint,
            ca_file_path=Path(ca_file_path).as_posix(),
            recv_timeout_ms=recv_timeout_ms,
        )
        self.executor = executor
        self.data_dir = data_dir
        self.executor_options = executor_options

        self._job_queue: multiprocessing.Queue | None = None
        self._result_queue: multiprocessing.Queue | None = None
        self._worker: multiprocessing.Process | None = None
        self._result_pump: threading.Thread | None = None

    def handle_event(self, event: Event) -> None:
        """Run dispatched jobs; ignore everything else (RFC 0001 §8).

        A JobDispatch payload is the job envelope QPI-UI's dispatcher builds —
        ``{job_id, payload}`` — which is exactly what the worker consumes.
        """
        if event.type is EventType.JOB_DISPATCH:
            log.info("Received job %s", event.payload.get("job_id"))
            self._job_queue.put(event.payload)
        else:
            log.warning(
                "dropping event %s: QPU driver does not handle %s",
                event.id,
                event.type.value,
            )

    def _on_start(self) -> None:
        self._job_queue = multiprocessing.Queue()
        self._result_queue = multiprocessing.Queue()

        # The executor keeps its own name, which is what a dataset's "backend"
        # attribute is for. This used to be overridden with the driver's display
        # label — the reason a _sanitize_name existed at all — so a cryostat called
        # "lab-1" produced datasets claiming a backend of "lab_1".
        self._worker = multiprocessing.Process(
            target=job_worker,
            kwargs={
                "job_queue": self._job_queue,
                "result_queue": self._result_queue,
                "executor": self.executor,
                "data_dir": self.data_dir,
                **self.executor_options,
            },
            name="QPI-Worker",
            daemon=True,
        )
        self._worker.start()

        self._result_pump = threading.Thread(
            target=self._pump_results, name="QPI-ResultPump", daemon=True
        )
        self._result_pump.start()

    def _pump_results(self) -> None:
        """Drain executor results and emit each as a JobResult event."""
        while True:
            item = self._result_queue.get()
            if item is None:
                log.info("Result pump received shutdown signal")
                return

            job_id = item["job_id"]
            results = {"error": item["error"]} if "error" in item else item["results"]
            log.info("Emitting result for job %s", job_id)
            self.emit(
                Event(
                    type=EventType.JOB_RESULT,
                    driver=self.name,
                    payload={"job_id": job_id, "results": results},
                )
            )

    def _on_stop(self) -> None:
        if self._job_queue is not None:
            _safe_put(self._job_queue, None)
        if self._result_queue is not None:
            _safe_put(self._result_queue, None)

        if self._worker is not None:
            self._worker.join(timeout=2)
            if self._worker.is_alive():
                log.warning("Terminating worker process...")
                self._worker.terminate()
                self._worker.join()

handle_event(event)

Run dispatched jobs; ignore everything else (RFC 0001 §8).

A JobDispatch payload is the job envelope QPI-UI's dispatcher builds — {job_id, payload} — which is exactly what the worker consumes.

Source code in qpi-driver/py/qpi_driver/builtins/qpu.py

def handle_event(self, event: Event) -> None:
    """Run dispatched jobs; ignore everything else (RFC 0001 §8).

    A JobDispatch payload is the job envelope QPI-UI's dispatcher builds —
    ``{job_id, payload}`` — which is exactly what the worker consumes.
    """
    if event.type is EventType.JOB_DISPATCH:
        log.info("Received job %s", event.payload.get("job_id"))
        self._job_queue.put(event.payload)
    else:
        log.warning(
            "dropping event %s: QPU driver does not handle %s",
            event.id,
            event.type.value,
        )

QuantifyExecutor

Bases: Executor

Executor subclass for interacting with Quantify-scheduler acquisition backends.

Source code in qpi-driver/py/qpi_driver/executors/quantify/__init__.py

class QuantifyExecutor(Executor):
    """Executor subclass for interacting with Quantify-scheduler acquisition backends."""

    def __new__(cls, *args, **kwargs):
        if not IS_QUANTIFY_INSTALLED:
            raise ImportError(
                "quantify-scheduler is not installed. Install the [quantify] extra to use QuantifyExecutor."
            )
        return super().__new__(cls)

    def __init__(
        self,
        name: str = "quantify",
        quantify_hardware_config: QbloxHardwareCompilationConfig | Path | dict = Path(
            "quantify.hardware.json"
        ),
        quantify_device_config: Path | dict = Path("quantify.device.json"),
        is_dummy: bool = False,
        is_simulated: bool = False,
        data_dir: Path = Path("data"),
        acquisition_timeout: int = 10,
        **kwargs: Any,
    ) -> None:
        """Initialize the QuantifyExecutor.

        Args:
            name: the name of the executor
            quantify_hardware_config: Hardware-layer configuration dictionary, file path, or config as dict.
            quantify_device_config: Device-layer configuration dictionary, file path or config as dict
            is_dummy: If True, uses a dummy Cluster instrument. It compiles and
                runs, but every acquisition comes back `nan`, so results carry
                no information about the circuit.
            is_simulated: If True, plays the compiled schedule through
                `qpi_driver.simulation.SimulatedCoordinator` instead — a
                transmon model driven by the schedule's own pulses, so counts
                reflect both the circuit and the device calibration. Needs the
                `sim` extra. Mutually exclusive with `is_dummy`.
            data_dir: Directory to where data is temporarily stored.
            acquisition_timeout: Timeout in seconds to wait for acquisition.
            **kwargs: Arbitrary keyword arguments passed to the base class.
        """
        super().__init__(name, **kwargs)
        set_datadir(data_dir)
        # Clean up any previously registered instruments to avoid name collision errors in QCoDeS
        with suppress(Exception):
            Instrument.close_all()

        if is_dummy and is_simulated:
            raise ValueError(
                "is_dummy and is_simulated both replace the cluster; pick one"
            )
        self._is_dummy = is_dummy
        self._is_simulated = is_simulated
        self._acquisition_timeout = acquisition_timeout
        hardware_config = load_quantify_hardware_config(quantify_hardware_config)
        self._hardware_config = hardware_config
        self._watched_hardware_config = (
            ConfigFile(quantify_hardware_config)
            if isinstance(quantify_hardware_config, Path)
            else None
        )
        self._watched_device_config = (
            ConfigFile(quantify_device_config)
            if isinstance(quantify_device_config, Path)
            else None
        )
        self._device = load_quantum_device(name=name, config=quantify_device_config)
        if is_simulated:
            from qpi_driver.executors.utils.coupler_bias import (
                declared_parking_currents,
                declared_sideband_gaps,
            )
            from qpi_driver.simulation import SimulatedCoordinator

            self._instrument_coordinator = SimulatedCoordinator(
                kwargs.get("simulator"),
                sideband_gaps=declared_sideband_gaps(self._device),
                parking_currents=declared_parking_currents(self._device),
            )
        else:
            self._instrument_coordinator = load_instrument_coordinator(
                f"{name}_ic", hardware_config=hardware_config, is_dummy=is_dummy
            )
        self._device.hardware_config(hardware_config)
        self._bias_source = self._park_couplers(kwargs.get("spi_rack_address"))
        self._compiler = SerialCompiler(
            name=f"{name}_compiler", quantum_device=self._device
        )

    def _reload_device_config(self) -> None:
        """Apply the device config file, which has changed on disk, to the device.

        A calibration on this node rewrites that file (RFC 0004 §8), and so does
        anyone restoring parameters by hand. Neither can reach into this process,
        so the file is the whole channel.

        A file that will not parse leaves the device as it was and costs the next
        job its new parameters, not the driver its device. The tuner's own writes
        are atomic, so only a hand-dropped file can be seen part-written.
        """
        path = self._watched_device_config.path
        try:
            unknown = apply_device_config(self._device, path)
        except Exception:
            log.exception("could not reload %s; keeping the current parameters", path)
            return

        if unknown:
            log.warning(
                "%s names %s, which this device does not have; a new element needs a "
                "restart",
                path,
                ", ".join(unknown),
            )
        log.info("reloaded device parameters from %s", path)

        if self._is_simulated:
            from qpi_driver.executors.utils.coupler_bias import (
                declared_parking_currents,
                declared_sideband_gaps,
            )

            self._instrument_coordinator.sideband_gaps = declared_sideband_gaps(
                self._device
            )
            self._instrument_coordinator.parking_currents = declared_parking_currents(
                self._device
            )

        # The device object is the same one, so the compiler and the hardware
        # config still point at it. The bias does not follow: it is a current
        # sitting in a rack, and only re-applying it moves the coupler.
        self._reapply_coupler_bias()

    def _warn_if_hardware_config_moved(self) -> None:
        """Say so when the hardware config changes, and keep running on the old one.

        Deliberately not reloaded. It builds the instrument coordinator and the
        Cluster behind it, so applying a new one means closing a live connection to
        the rack and dialling it again — and a reconnect that fails leaves this
        driver with no coordinator and no way back, the old one being already gone.
        The device config has somewhere to fall back to; this does not.

        A restart is the honest answer: rewiring a rack is not a runtime event.
        Warned once per change so a stale hardware config is at least not a silent
        one.
        """
        if not self._watched_hardware_config:
            return
        if not self._watched_hardware_config.changed():
            return
        self._watched_hardware_config.mark_read()
        log.warning(
            "%s has changed; this driver is still running on the hardware config it "
            "started with. Restart it to pick the new one up.",
            self._watched_hardware_config.path,
        )

    def _reapply_coupler_bias(self) -> None:
        """Hold the couplers at the reloaded currents, on the rack already open.

        Resolving a second source would open a second connection to the same
        rack, or clash on its qcodes name.
        """
        from qpi_driver.executors.utils.coupler_bias import apply_coupler_bias

        try:
            self._parked = apply_coupler_bias(self._device, self._bias_source)
        except Exception:
            log.exception("could not park the couplers; two-qubit gates will be wrong")
            self._parked = {}

    @property
    def hardware_config(self) -> QbloxHardwareCompilationConfig:
        return self._hardware_config

    def execute(self, payload: JobPayload) -> xr.Dataset:
        """Execute quantum instructions using the Quantify scheduler.

        The acquisition protocol is selected based on ``payload.meas_level``:

        * ``meas_level=0`` → ``Trace`` (raw waveform)
        * ``meas_level=1`` → ``SSBIntegrationComplex`` (kerneled IQ)
        * ``meas_level=2`` → ``ThresholdedAcquisition`` if threshold params
          are configured on the device elements, else ``SSBIntegrationComplex``
          (software discrimination deferred to ``process_result()``).

        Every circuit in ``payload.circuits`` is executed, honouring each
        circuit's ``shots`` override and ``parameter_values`` bindings.  A
        single-circuit payload returns that circuit's flat dataset; multi-circuit
        payloads are bundled so circuits with different qubit widths or shot
        counts stay independent (see ``combine_circuit_datasets``).

        Args:
            payload: JobPayload specifying shots, circuits, meas_level, etc.

        Returns:
            xr.Dataset: Raw acquisition dataset.
        """
        # Between jobs, never during one: a reload part-way through a compilation
        # would be worse than a stale parameter.
        if self._watched_device_config and self._watched_device_config.changed():
            self._reload_device_config()
            self._watched_device_config.mark_read()
        self._warn_if_hardware_config_moved()

        acq_protocol, acq_kwargs, acq_overrides = self._resolve_acq_protocol(payload)
        sub_datasets: list[xr.Dataset] = []

        for circ in payload.circuits:
            circ_shots = circ.shots if circ.shots is not None else payload.shots
            circuit = load_qasm(circ.circuit)

            for param_values in circ.parameter_values or [None]:
                bound_circuit = circuit
                if param_values is not None and circuit.parameters:
                    bound_circuit = circuit.assign_parameters(param_values)
                sub_datasets.append(
                    self._acquire_circuit(
                        payload,
                        bound_circuit,
                        circ_shots,
                        acq_protocol,
                        acq_kwargs,
                        acq_overrides,
                    )
                )

        return combine_circuit_datasets(sub_datasets)

    def _acquire_circuit(
        self,
        payload: JobPayload,
        circuit: QuantumCircuit,
        shots: int,
        acq_protocol: str,
        acq_kwargs: dict,
        acq_overrides: dict[int, dict[str, float]],
    ) -> xr.Dataset:
        """One circuit's acquisition, in as many runs as the hardware requires.

        Almost always one. The exception is a raw trace over more than one
        qubit: a Qblox module can put a single sequencer into scope mode, so
        asking two qubits for a trace at once does not compile —
        *"Only one sequencer per device can trigger raw trace capture"*.

        Rather than refuse a `meas_level=0` job on a multi-qubit circuit, run
        the circuit once per measured qubit and capture one trace each time.
        That is what the instrument allows and what a lab does by hand; the
        cost is honest and unavoidable — N runs' worth of time for N qubits,
        because the shots are repeated per qubit rather than shared.
        """
        measured = measured_qubits(circuit)
        if acq_protocol != "Trace" or len(measured) < 2:
            return self._run_circuit(
                payload, circuit, shots, acq_protocol, acq_kwargs, acq_overrides
            )

        log.info(
            "raw trace over %d qubits: taking %d runs, one scope-mode acquisition each",
            len(measured),
            len(measured),
        )
        passes = [
            self._run_circuit(
                payload,
                circuit,
                shots,
                acq_protocol,
                acq_kwargs,
                acq_overrides,
                only_qubit=qubit,
            )
            for qubit in measured
        ]
        # Each pass carries exactly one qubit's variable, on its own
        # channel-suffixed dimensions, so there is nothing to collide.
        return xr.merge(passes, combine_attrs="override")

    def _run_circuit(
        self,
        payload: JobPayload,
        circuit: QuantumCircuit,
        shots: int,
        acq_protocol: str,
        acq_kwargs: dict,
        acq_overrides: dict[int, dict[str, float]],
        only_qubit: int | None = None,
    ) -> xr.Dataset:
        """Run a single (parameter-bound) circuit and return its acquisition dataset."""
        schedule = Schedule(name=payload.id, repetitions=shots)
        _open_readout_clocks(schedule, self._device, acq_protocol, only_qubit)
        acq_indices: dict[int, int] = {}
        clbit_map: list[tuple[int, int, int]] = []

        for instruction in circuit.data:
            parsed_ops = to_quantify_gates(
                circuit=circuit,
                instruction=instruction,
                acq_indices=acq_indices,
                acq_protocol=acq_protocol,
                acq_kwargs=acq_kwargs,
                acq_overrides=acq_overrides,
                clbit_map=clbit_map,
                only_qubit=only_qubit,
            )
            is_parallel_op = isinstance(
                instruction.operation,
                (qiskit_library.Measure, qiskit.circuit.Delay, qiskit_library.Barrier),
            )

            if is_parallel_op and parsed_ops:
                first_op = schedule.add(parsed_ops[0])
                for op in parsed_ops[1:]:
                    schedule.add(op, ref_op=first_op, ref_pt="start")
            else:
                for op in parsed_ops:
                    schedule.add(op)

        compiled_sched = self._compiler.compile(schedule=schedule)

        self._instrument_coordinator.prepare(compiled_sched)
        self._instrument_coordinator.start()
        self._instrument_coordinator.wait_done(timeout_sec=self._acquisition_timeout)
        dataset = self._instrument_coordinator.retrieve_acquisition()
        dataset.attrs.update(
            {
                "shots": shots,
                "n_qubits": circuit.num_qubits,
                "backend": self.name,
                "meas_level": payload.meas_level,
                "meas_return": payload.meas_return,
                "acq_protocol": acq_protocol,
                "clbit_map": [list(entry) for entry in clbit_map],
                "num_clbits": circuit.num_clbits,
            }
        )
        # Per qubit, because the discriminator is. `process_result` reads this back
        # when it has to threshold in software, and a single pair here would send it
        # down the same collapse the schedule no longer has.
        if acq_overrides:
            dataset.attrs["acq_discriminators"] = {
                str(index): dict(values) for index, values in acq_overrides.items()
            }
        return dataset

    def _park_couplers(self, spi_rack_address: str | None):
        """Hold every tunable coupler at its calibrated DC bias.

        Done once at startup rather than per job, because that is what the bias
        physically is: a current that sits there while the fridge is cold. It is
        not part of any schedule — quantify has no way to express an SPI rack —
        so if this does not happen, nothing else will do it, and every CZ runs
        against a coupler parked wherever it was left.

        Replacing the cluster replaces the rack too: with ``is_simulated`` or
        ``is_dummy`` there is no instrument to talk to, so the intended currents
        are recorded and not applied.
        """
        from qpi_driver.executors.utils.coupler_bias import (
            RecordingBias,
            apply_coupler_bias,
            resolve_bias_source,
        )

        try:
            source = (
                RecordingBias()
                if (self._is_simulated or self._is_dummy)
                else resolve_bias_source(
                    self._device,
                    cluster=self._cluster(),
                    spi_address=spi_rack_address,
                )
            )
            self._parked = apply_coupler_bias(self._device, source)
        except Exception:
            # A coupler that cannot be parked is a broken two-qubit gate, not a
            # broken node: single-qubit work is unaffected, and refusing to
            # start would take the whole QPU out for it.
            log.exception("could not park the couplers; two-qubit gates will be wrong")
            self._parked = {}
            return RecordingBias()
        return source

    def _cluster(self):
        """The Cluster behind the instrument coordinator, if there is one."""
        for component in getattr(
            self._instrument_coordinator, "components", lambda: []
        )():
            instrument = getattr(component, "instrument", None)
            if instrument is not None:
                return instrument
        return None

    def _resolve_acq_protocol(
        self, payload: JobPayload
    ) -> tuple[str, dict, dict[int, dict[str, float]]]:
        """Determine the quantify-scheduler acquisition protocol for the given meas_level.

        Args:
            payload: JobPayload specifying meas_level, acq_threshold, acq_rotation, etc.

        Returns:
            Tuple of (protocol_name, kwargs_for_every_Measure, kwargs_per_qubit).
        """
        bin_mode = self._resolve_bin_mode(payload.meas_level, payload.meas_return)

        if payload.meas_level == 0:
            return "Trace", {"bin_mode": bin_mode}, {}

        if payload.meas_level == 1:
            return "SSBIntegrationComplex", {"bin_mode": bin_mode}, {}

        # meas_level == 2: threshold on the instrument when every qubit has a line to
        # threshold against, in software otherwise. Per qubit either way — see
        # `qpi_driver.executors.utils.discriminator`.
        per_qubit, complete = resolve_discriminators(
            self._device, payload.acq_rotation, payload.acq_threshold
        )
        # And the power those shots are taken at, where a `CalibratedTransmon` says so.
        # The matching frequency cannot ride on `Measure` — it is a clock, not a pulse
        # parameter — so `_readout_overrides` sets it on the schedule instead.
        for index, point in readout_points_by_qubit(self._device).items():
            if index in per_qubit:
                per_qubit[index]["pulse_amp"] = point["pulse_amp"]
        protocol = "ThresholdedAcquisition" if complete else "SSBIntegrationComplex"
        return protocol, {"bin_mode": bin_mode}, per_qubit

    def _resolve_bin_mode(self, meas_level: int, meas_return: str) -> BinMode:
        """Choose the acquisition bin mode implied by the requested measurement mode.

        Raw traces (level 0) and averaged integrated results (level 1 with
        ``meas_return="avg"``) collapse every repetition into a single bin.
        Counts (level 2) and single-shot integrated results keep each shot as a
        separate bin so results can be processed per shot.
        """
        if meas_level == 0:
            return BinMode.AVERAGE
        if meas_level == 1 and meas_return == "avg":
            return BinMode.AVERAGE
        return BinMode.APPEND

    def _get_threshold_params(self) -> dict[int, dict[str, float]]:
        """Every qubit's discriminator, by qubit index."""
        return discriminators_by_qubit(self._device)

    def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
        """Convert a quantify-scheduler acquisition dataset into a Qiskit-compatible result dict.

        Handles all meas_levels:
        - meas_level=0 (Trace): Returns raw complex waveform data as [[real, imag], ...] per time sample.
        - meas_level=1 (SSBIntegrationComplex): Returns IQ values as [[real, imag]] per shot per qubit.
        - meas_level=2 with ThresholdedAcquisition: Aggregates 0/1 values into counts dict.
        - meas_level=2 with SSBIntegrationComplex: Performs software discrimination using
          acq_threshold and acq_rotation from the device config.

        Args:
            dataset: xr.Dataset from execute().
            job_id: Unique job ID.

        Returns:
            dict: Qiskit-compatible result dict.
        """
        from qpi_driver.executors.utils.result import build_qiskit_result

        meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
        meas_return = str(dataset.attrs.get("meas_return", "single"))
        acq_protocol = str(dataset.attrs.get("acq_protocol", "SSBIntegrationComplex"))
        backend = dataset.attrs.get("backend", self.name)

        circuit_results = [
            self._single_dataset_to_result(
                sub_ds, meas_level, meas_return, acq_protocol
            )
            for sub_ds in iter_circuit_datasets(dataset)
        ]
        return build_qiskit_result(circuit_results, job_id, backend)

    def _single_dataset_to_result(
        self, dataset: xr.Dataset, meas_level: int, meas_return: str, acq_protocol: str
    ) -> dict:
        """Extract result data from a single-circuit quantify dataset."""
        qubit_vars, q0_key, shots = self._extract_qubit_vars(dataset)
        if not qubit_vars:
            return {"raw": str(dataset), "shots": 0}

        if meas_level == 0:
            return self._process_meas_level_0(dataset, qubit_vars, shots)
        if meas_level == 1:
            return self._process_meas_level_1(dataset, qubit_vars, shots, meas_return)
        return self._process_meas_level_2(dataset, qubit_vars, shots, acq_protocol)

    def _extract_qubit_vars(self, dataset: xr.Dataset) -> tuple[list[int], str, int]:
        """Identify qubit variables (integer-named data vars) and shots."""
        qubit_vars = []
        for var in dataset.data_vars:
            try:
                qubit_vars.append(int(var))
            except ValueError:
                pass
        if not qubit_vars:
            return [], "", 0
        qubit_vars.sort()
        q0_key = qubit_key(dataset, qubit_vars[0])
        shots = cast_to(int, dataset.attrs.get("shots"), len(dataset[q0_key]))
        return qubit_vars, q0_key, shots

    def _process_meas_level_0(
        self, dataset: xr.Dataset, qubit_vars: list[int], shots: int
    ) -> dict:
        """Extract raw complex trace data (meas_level=0)."""
        memory: list[list[list[float]]] = []
        for q_idx in qubit_vars:
            var_key = q_idx if q_idx in dataset else str(q_idx)
            trace = dataset[var_key].values.flatten()
            qubit_trace = [[float(v.real), float(v.imag)] for v in trace]
            memory.append(qubit_trace)
        return {"memory": memory, "shots": shots}

    def _process_meas_level_1(
        self, dataset: xr.Dataset, qubit_vars: list[int], shots: int, meas_return: str
    ) -> dict:
        """Extract integrated IQ memory (meas_level=1)."""
        from qpi_driver.executors.utils.result import iq_memory_avg

        per_shot = {
            q_idx: per_shot_values(dataset[qubit_key(dataset, q_idx)])
            for q_idx in qubit_vars
        }
        num_samples = len(per_shot[qubit_vars[0]])
        memory = []
        for s in range(num_samples):
            shot_iq = []
            for q_idx in qubit_vars:
                val = per_shot[q_idx][s]
                r = float(val.real) if not np.isnan(val.real) else 0.0
                i = float(val.imag) if not np.isnan(val.imag) else 0.0
                shot_iq.append([r, i])
            memory.append(shot_iq)

        if meas_return == "avg" and memory:
            memory = iq_memory_avg(memory, len(qubit_vars))
        return {"memory": memory, "shots": shots}

    def _process_meas_level_2(
        self, dataset: xr.Dataset, qubit_vars: list[int], shots: int, acq_protocol: str
    ) -> dict:
        """Extract classified counts (meas_level=2) performing software discrimination if needed.

        Counts are keyed by classical register, one bit per measured clbit,
        positioned by clbit index (see ``build_acquisition_counts``).
        """
        discriminate = build_discriminator(
            dataset, acq_protocol, self._get_threshold_params
        )
        counts_dict = build_acquisition_counts(dataset, qubit_vars, discriminate)
        return {"counts": counts_dict, "shots": shots}

    def close(self) -> None:
        """Detach the coordinator's components, then release it.

        ``InstrumentCoordinator.components`` is a qcodes ``ManualParameter``
        holding component *names*, so it has to be called — iterating it
        directly raises, and a shutdown that raises leaves the cluster held
        against the next driver that wants it.
        """
        components: list = []
        with suppress(Exception):
            components = list(self._instrument_coordinator.components())

        for name in components:
            with suppress(Exception):
                self._instrument_coordinator.remove_component(name)

        with suppress(Exception):
            self._instrument_coordinator.close()

__init__(name='quantify', quantify_hardware_config=Path('quantify.hardware.json'), quantify_device_config=Path('quantify.device.json'), is_dummy=False, is_simulated=False, data_dir=Path('data'), acquisition_timeout=10, **kwargs)

Initialize the QuantifyExecutor.

Parameters:

Name	Type	Description	Default
name	str	the name of the executor	'quantify'
quantify_hardware_config	QbloxHardwareCompilationConfig | Path | dict	Hardware-layer configuration dictionary, file path, or config as dict.	Path('quantify.hardware.json')
quantify_device_config	Path | dict	Device-layer configuration dictionary, file path or config as dict	Path('quantify.device.json')
is_dummy	bool	If True, uses a dummy Cluster instrument. It compiles and runs, but every acquisition comes back nan, so results carry no information about the circuit.	False
is_simulated	bool	If True, plays the compiled schedule through qpi_driver.simulation.SimulatedCoordinator instead — a transmon model driven by the schedule's own pulses, so counts reflect both the circuit and the device calibration. Needs the sim extra. Mutually exclusive with is_dummy.	False
data_dir	Path	Directory to where data is temporarily stored.	Path('data')
acquisition_timeout	int	Timeout in seconds to wait for acquisition.	10
**kwargs	Any	Arbitrary keyword arguments passed to the base class.	{}

Source code in qpi-driver/py/qpi_driver/executors/quantify/__init__.py

def __init__(
    self,
    name: str = "quantify",
    quantify_hardware_config: QbloxHardwareCompilationConfig | Path | dict = Path(
        "quantify.hardware.json"
    ),
    quantify_device_config: Path | dict = Path("quantify.device.json"),
    is_dummy: bool = False,
    is_simulated: bool = False,
    data_dir: Path = Path("data"),
    acquisition_timeout: int = 10,
    **kwargs: Any,
) -> None:
    """Initialize the QuantifyExecutor.

    Args:
        name: the name of the executor
        quantify_hardware_config: Hardware-layer configuration dictionary, file path, or config as dict.
        quantify_device_config: Device-layer configuration dictionary, file path or config as dict
        is_dummy: If True, uses a dummy Cluster instrument. It compiles and
            runs, but every acquisition comes back `nan`, so results carry
            no information about the circuit.
        is_simulated: If True, plays the compiled schedule through
            `qpi_driver.simulation.SimulatedCoordinator` instead — a
            transmon model driven by the schedule's own pulses, so counts
            reflect both the circuit and the device calibration. Needs the
            `sim` extra. Mutually exclusive with `is_dummy`.
        data_dir: Directory to where data is temporarily stored.
        acquisition_timeout: Timeout in seconds to wait for acquisition.
        **kwargs: Arbitrary keyword arguments passed to the base class.
    """
    super().__init__(name, **kwargs)
    set_datadir(data_dir)
    # Clean up any previously registered instruments to avoid name collision errors in QCoDeS
    with suppress(Exception):
        Instrument.close_all()

    if is_dummy and is_simulated:
        raise ValueError(
            "is_dummy and is_simulated both replace the cluster; pick one"
        )
    self._is_dummy = is_dummy
    self._is_simulated = is_simulated
    self._acquisition_timeout = acquisition_timeout
    hardware_config = load_quantify_hardware_config(quantify_hardware_config)
    self._hardware_config = hardware_config
    self._watched_hardware_config = (
        ConfigFile(quantify_hardware_config)
        if isinstance(quantify_hardware_config, Path)
        else None
    )
    self._watched_device_config = (
        ConfigFile(quantify_device_config)
        if isinstance(quantify_device_config, Path)
        else None
    )
    self._device = load_quantum_device(name=name, config=quantify_device_config)
    if is_simulated:
        from qpi_driver.executors.utils.coupler_bias import (
            declared_parking_currents,
            declared_sideband_gaps,
        )
        from qpi_driver.simulation import SimulatedCoordinator

        self._instrument_coordinator = SimulatedCoordinator(
            kwargs.get("simulator"),
            sideband_gaps=declared_sideband_gaps(self._device),
            parking_currents=declared_parking_currents(self._device),
        )
    else:
        self._instrument_coordinator = load_instrument_coordinator(
            f"{name}_ic", hardware_config=hardware_config, is_dummy=is_dummy
        )
    self._device.hardware_config(hardware_config)
    self._bias_source = self._park_couplers(kwargs.get("spi_rack_address"))
    self._compiler = SerialCompiler(
        name=f"{name}_compiler", quantum_device=self._device
    )

close()

Detach the coordinator's components, then release it.

InstrumentCoordinator.components is a qcodes ManualParameter holding component names, so it has to be called — iterating it directly raises, and a shutdown that raises leaves the cluster held against the next driver that wants it.

Source code in qpi-driver/py/qpi_driver/executors/quantify/__init__.py

def close(self) -> None:
    """Detach the coordinator's components, then release it.

    ``InstrumentCoordinator.components`` is a qcodes ``ManualParameter``
    holding component *names*, so it has to be called — iterating it
    directly raises, and a shutdown that raises leaves the cluster held
    against the next driver that wants it.
    """
    components: list = []
    with suppress(Exception):
        components = list(self._instrument_coordinator.components())

    for name in components:
        with suppress(Exception):
            self._instrument_coordinator.remove_component(name)

    with suppress(Exception):
        self._instrument_coordinator.close()

execute(payload)

Execute quantum instructions using the Quantify scheduler.

The acquisition protocol is selected based on payload.meas_level:

• meas_level=0 → Trace (raw waveform)
• meas_level=1 → SSBIntegrationComplex (kerneled IQ)
• meas_level=2 → ThresholdedAcquisition if threshold params are configured on the device elements, else SSBIntegrationComplex (software discrimination deferred to process_result()).

Every circuit in payload.circuits is executed, honouring each circuit's shots override and parameter_values bindings. A single-circuit payload returns that circuit's flat dataset; multi-circuit payloads are bundled so circuits with different qubit widths or shot counts stay independent (see combine_circuit_datasets).

Parameters:

Name	Type	Description	Default
payload	JobPayload	JobPayload specifying shots, circuits, meas_level, etc.	required

Returns:

Type	Description
Dataset	xr.Dataset: Raw acquisition dataset.

Source code in qpi-driver/py/qpi_driver/executors/quantify/__init__.py

def execute(self, payload: JobPayload) -> xr.Dataset:
    """Execute quantum instructions using the Quantify scheduler.

    The acquisition protocol is selected based on ``payload.meas_level``:

    * ``meas_level=0`` → ``Trace`` (raw waveform)
    * ``meas_level=1`` → ``SSBIntegrationComplex`` (kerneled IQ)
    * ``meas_level=2`` → ``ThresholdedAcquisition`` if threshold params
      are configured on the device elements, else ``SSBIntegrationComplex``
      (software discrimination deferred to ``process_result()``).

    Every circuit in ``payload.circuits`` is executed, honouring each
    circuit's ``shots`` override and ``parameter_values`` bindings.  A
    single-circuit payload returns that circuit's flat dataset; multi-circuit
    payloads are bundled so circuits with different qubit widths or shot
    counts stay independent (see ``combine_circuit_datasets``).

    Args:
        payload: JobPayload specifying shots, circuits, meas_level, etc.

    Returns:
        xr.Dataset: Raw acquisition dataset.
    """
    # Between jobs, never during one: a reload part-way through a compilation
    # would be worse than a stale parameter.
    if self._watched_device_config and self._watched_device_config.changed():
        self._reload_device_config()
        self._watched_device_config.mark_read()
    self._warn_if_hardware_config_moved()

    acq_protocol, acq_kwargs, acq_overrides = self._resolve_acq_protocol(payload)
    sub_datasets: list[xr.Dataset] = []

    for circ in payload.circuits:
        circ_shots = circ.shots if circ.shots is not None else payload.shots
        circuit = load_qasm(circ.circuit)

        for param_values in circ.parameter_values or [None]:
            bound_circuit = circuit
            if param_values is not None and circuit.parameters:
                bound_circuit = circuit.assign_parameters(param_values)
            sub_datasets.append(
                self._acquire_circuit(
                    payload,
                    bound_circuit,
                    circ_shots,
                    acq_protocol,
                    acq_kwargs,
                    acq_overrides,
                )
            )

    return combine_circuit_datasets(sub_datasets)

process_result(dataset, job_id)

Convert a quantify-scheduler acquisition dataset into a Qiskit-compatible result dict.

Handles all meas_levels: - meas_level=0 (Trace): Returns raw complex waveform data as [[real, imag], ...] per time sample. - meas_level=1 (SSBIntegrationComplex): Returns IQ values as [[real, imag]] per shot per qubit. - meas_level=2 with ThresholdedAcquisition: Aggregates 0/1 values into counts dict. - meas_level=2 with SSBIntegrationComplex: Performs software discrimination using acq_threshold and acq_rotation from the device config.

Parameters:

Name	Type	Description	Default
dataset	Dataset	xr.Dataset from execute().	required
job_id	str	Unique job ID.	required

Returns:

Name	Type	Description
dict	dict	Qiskit-compatible result dict.

Source code in qpi-driver/py/qpi_driver/executors/quantify/__init__.py

def process_result(self, dataset: xr.Dataset, job_id: str) -> dict:
    """Convert a quantify-scheduler acquisition dataset into a Qiskit-compatible result dict.

    Handles all meas_levels:
    - meas_level=0 (Trace): Returns raw complex waveform data as [[real, imag], ...] per time sample.
    - meas_level=1 (SSBIntegrationComplex): Returns IQ values as [[real, imag]] per shot per qubit.
    - meas_level=2 with ThresholdedAcquisition: Aggregates 0/1 values into counts dict.
    - meas_level=2 with SSBIntegrationComplex: Performs software discrimination using
      acq_threshold and acq_rotation from the device config.

    Args:
        dataset: xr.Dataset from execute().
        job_id: Unique job ID.

    Returns:
        dict: Qiskit-compatible result dict.
    """
    from qpi_driver.executors.utils.result import build_qiskit_result

    meas_level = cast_to(int, dataset.attrs.get("meas_level"), 2)
    meas_return = str(dataset.attrs.get("meas_return", "single"))
    acq_protocol = str(dataset.attrs.get("acq_protocol", "SSBIntegrationComplex"))
    backend = dataset.attrs.get("backend", self.name)

    circuit_results = [
        self._single_dataset_to_result(
            sub_ds, meas_level, meas_return, acq_protocol
        )
        for sub_ds in iter_circuit_datasets(dataset)
    ]
    return build_qiskit_result(circuit_results, job_id, backend)

Tuner

Bases: ABC

Runs calibration routines against a quantum device and updates it.

Subclasses supply :meth:backend and :attr:device; the three entry points below — full calibration, partial recalibration and the drift check — are the same DAG walk over different subsets, so they are implemented once here.

Source code in qpi-driver/py/qpi_driver/tuners/base/__init__.py

class Tuner(ABC):
    """Runs calibration routines against a quantum device and updates it.

    Subclasses supply :meth:`backend` and :attr:`device`; the three entry points
    below — full calibration, partial recalibration and the drift check — are
    the same DAG walk over different subsets, so they are implemented once here.
    """

    def __init__(self, name: str, **kwargs: Any) -> None:
        self.name = name
        self._watched_device_config: ConfigFile | None = None
        self._device_config_path = None
        #: Where to report progress, set per calibration by the worker that owns the
        #: queue it reports through. An attribute rather than an argument to the
        #: three entry points below, so a tuner that overrides one of them keeps
        #: working and simply reports nothing.
        self.on_progress: ProgressSink | None = None

    # A property so that setting the path arms the watcher with it. Every tuner
    # assigns the path in its constructor; none of them should have to remember
    # a second line to make the reload work.
    @property
    def _device_config_path(self) -> Path | None:
        return self.__path

    @_device_config_path.setter
    def _device_config_path(self, path: Path | None) -> None:
        self.__path = path
        self._watched_device_config = ConfigFile(path) if path is not None else None

    @property
    @abstractmethod
    def backend(self) -> SchedulerBackend:
        """The scheduler this tuner composes and runs schedules through."""

    @property
    @abstractmethod
    def device(self) -> Any:
        """The in-memory ``QuantumDevice`` being calibrated."""

    @property
    def bias(self) -> Any:
        """Something that can park a coupler at a DC current, or ``None``.

        Only `coupler_anticrossing` asks for it, and only because the bias is not a
        pulse: it is held over qcodes for as long as the fridge is cold, so a routine
        sweeping it has to set instrument state between acquisitions. A tuner with no
        rack to talk to returns ``None`` and that routine declines.
        """
        return None

    def routines(self) -> list[CalibrationRoutine]:
        """The routines this tuner can run. Every tuner runs the same set."""
        return all_routines()

    def calibrate(self, config: CalibrationConfig) -> CalibrationReport:
        """Walk the whole enabled DAG, then persist what it calibrated."""
        self._refresh_device_config()
        config.validate_against(routine_names())
        config.validate_targets()
        dag = CalibrationDAG(self.routines(), config, bias=self.bias)
        report = dag.run(
            self.device, self.backend, config, mode="full", on_progress=self.on_progress
        )
        self._persist(report)
        return report

    def recalibrate(
        self, qubits: list[str], config: CalibrationConfig
    ) -> CalibrationReport:
        """Recalibrate *qubits* only, running whichever routines the checks blame.

        Two narrowings, and both matter: this is what a drift check triggers, and
        it has to be meaningfully cheaper than a full run to be worth having.

        The targets narrow to *qubits* plus any edge touching one of them. The
        routines narrow by :meth:`CalibrationDAG.diagnose` — each routine is asked
        whether its parameters still hold, and the shallowest node with evidence
        against it is recalibrated along with everything downstream of it.
        Downstream because recalibrating a frequency invalidates the gates tuned
        against it, so re-running the blamed node alone would leave the chip in a
        worse state than not running at all.

        Where RFC 0004 assumed the boundary, this measures it. A run in which every
        check passes recalibrates nothing and says so, which is the cheapest
        possible outcome and was not previously reachable.
        """
        self._refresh_device_config()
        config.validate_against(routine_names())
        config.validate_targets()
        narrowed = self._narrow_to(qubits, config)
        dag = CalibrationDAG(self.routines(), narrowed, bias=self.bias)
        order, notes = dag.diagnose(
            list(RECALIBRATION_SEEDS), self.device, self.backend, narrowed
        )
        for note in notes:
            log.info("diagnose: %s", note)

        if not order:
            report = CalibrationReport(
                timestamp=utc_timestamp(),
                duration_s=0.0,
                mode="partial",
                backend=self.backend.name,
            )
            report.notes.extend(notes)
            return report

        report = dag.run(
            self.device,
            self.backend,
            narrowed,
            mode="partial",
            only=order,
            on_progress=self.on_progress,
        )
        report.notes.extend(notes)
        self._persist(report)
        return report

    def check_fidelity(self, config: CalibrationConfig) -> CalibrationReport:
        """Run only the benchmarks, calibrating nothing.

        Returns a report rather than a bare mapping so the caller emits the same
        payload shape it does for every other mode; :meth:`CalibrationReport.fidelities`
        is what reduces it to the per-target numbers a threshold is compared with.
        """
        self._refresh_device_config()
        config.validate_against(routine_names())
        config.validate_targets()
        routines = self.routines()
        benchmarks = [r.name for r in routines if r.is_benchmark]
        dag = CalibrationDAG(routines, config, bias=self.bias)
        order = [name for name in dag.execution_order() if name in benchmarks]
        return dag.run(
            self.device,
            self.backend,
            config,
            mode="fidelity_check",
            only=order,
            on_progress=self.on_progress,
        )

    def _narrow_to(
        self, qubits: list[str], config: CalibrationConfig
    ) -> CalibrationConfig:
        """*config* restricted to *qubits*, the edges touching them, and their partners.

        The partners are not an afterthought. An edge is calibrated *through* its two
        qubits — a chevron prepares ``|11>`` with a pi pulse on each — so narrowing to
        the drifted qubit alone would carry the edge in and leave the other end
        wherever it was. `CalibrationConfig.validate_targets` refuses that outright,
        and it is right to: the failure it prevents is a confident, wrong gate rather
        than an error. So a partial recalibration that keeps an edge keeps both its
        ends, which is the smallest honest unit of work.
        """
        wanted = [q for q in qubits if q in config.target_qubits] or list(qubits)
        edges = [
            edge
            for edge in config.target_edges
            if any(part in wanted for part in edge.split("_"))
        ]
        for edge in edges:
            for part in edge.split("_"):
                if part not in wanted:
                    wanted.append(part)
        return CalibrationConfig(
            target_qubits=wanted,
            target_edges=edges,
            routines=config.routines,
            monitoring=config.monitoring,
            routine_timeout_s=config.routine_timeout_s,
        )

    def _refresh_device_config(self) -> None:
        """Re-read the device config if it changed since this tuner last looked.

        Something else may have moved the chip — a hand edit, a restored backup —
        and this tuner would otherwise calibrate from startup values and write them
        back over it.

        At DAG start, never between routines: a device moving mid-walk leaves a fit
        and the parameters it was measured against disagreeing.
        """
        if self._watched_device_config is None:
            return
        if not self._watched_device_config.changed():
            return

        path = self._watched_device_config.path
        try:
            unknown = apply_device_config(self.device, path)
        except Exception:
            log.exception(
                "could not reload %s; calibrating from what is in memory", path
            )
            return

        self._watched_device_config.mark_read()
        if unknown:
            log.warning(
                "%s names %s, which this device does not have; a new element needs a "
                "restart",
                path,
                ", ".join(unknown),
            )
        log.info("reloaded device parameters from %s before calibrating", path)

    def _persist(self, report: CalibrationReport) -> None:
        """Write the calibrated device back, unless there is nothing to write.

        A run that calibrated nothing must not touch the file. A failed run is
        the case that matters: the device may hold half-applied parameters, and
        overwriting a good config with them is how a calibration takes a QPU
        down (RFC 0004 §10).
        """
        if self._device_config_path is None:
            return
        if report.status == "failed":
            log.warning(
                "not writing back device config: calibration failed (%d error(s))",
                len(report.errors),
            )
            return
        if not any(r for r in report.routine_results):
            log.info("not writing back device config: no routine produced a result")
            return

        try:
            save_device_config(self.device, self._device_config_path)
        except Exception as exc:
            log.exception("failed to persist calibrated device config")
            report.errors.append(f"write-back failed: {exc}")
            report.status = "partial_failure"

    def close(self) -> None:
        """Release instruments. Safe to call more than once."""

backend abstractmethod property

The scheduler this tuner composes and runs schedules through.

bias property

Something that can park a coupler at a DC current, or None.

Only coupler_anticrossing asks for it, and only because the bias is not a pulse: it is held over qcodes for as long as the fridge is cold, so a routine sweeping it has to set instrument state between acquisitions. A tuner with no rack to talk to returns None and that routine declines.

device abstractmethod property

The in-memory QuantumDevice being calibrated.

calibrate(config)

Walk the whole enabled DAG, then persist what it calibrated.

Source code in qpi-driver/py/qpi_driver/tuners/base/__init__.py

def calibrate(self, config: CalibrationConfig) -> CalibrationReport:
    """Walk the whole enabled DAG, then persist what it calibrated."""
    self._refresh_device_config()
    config.validate_against(routine_names())
    config.validate_targets()
    dag = CalibrationDAG(self.routines(), config, bias=self.bias)
    report = dag.run(
        self.device, self.backend, config, mode="full", on_progress=self.on_progress
    )
    self._persist(report)
    return report

check_fidelity(config)

Run only the benchmarks, calibrating nothing.

Returns a report rather than a bare mapping so the caller emits the same payload shape it does for every other mode; :meth:CalibrationReport.fidelities is what reduces it to the per-target numbers a threshold is compared with.

Source code in qpi-driver/py/qpi_driver/tuners/base/__init__.py

def check_fidelity(self, config: CalibrationConfig) -> CalibrationReport:
    """Run only the benchmarks, calibrating nothing.

    Returns a report rather than a bare mapping so the caller emits the same
    payload shape it does for every other mode; :meth:`CalibrationReport.fidelities`
    is what reduces it to the per-target numbers a threshold is compared with.
    """
    self._refresh_device_config()
    config.validate_against(routine_names())
    config.validate_targets()
    routines = self.routines()
    benchmarks = [r.name for r in routines if r.is_benchmark]
    dag = CalibrationDAG(routines, config, bias=self.bias)
    order = [name for name in dag.execution_order() if name in benchmarks]
    return dag.run(
        self.device,
        self.backend,
        config,
        mode="fidelity_check",
        only=order,
        on_progress=self.on_progress,
    )

close()

Release instruments. Safe to call more than once.

Source code in qpi-driver/py/qpi_driver/tuners/base/__init__.py

def close(self) -> None:
    """Release instruments. Safe to call more than once."""

recalibrate(qubits, config)

Recalibrate qubits only, running whichever routines the checks blame.

Two narrowings, and both matter: this is what a drift check triggers, and it has to be meaningfully cheaper than a full run to be worth having.

The targets narrow to qubits plus any edge touching one of them. The routines narrow by :meth:CalibrationDAG.diagnose — each routine is asked whether its parameters still hold, and the shallowest node with evidence against it is recalibrated along with everything downstream of it. Downstream because recalibrating a frequency invalidates the gates tuned against it, so re-running the blamed node alone would leave the chip in a worse state than not running at all.

Where RFC 0004 assumed the boundary, this measures it. A run in which every check passes recalibrates nothing and says so, which is the cheapest possible outcome and was not previously reachable.

Source code in qpi-driver/py/qpi_driver/tuners/base/__init__.py

def recalibrate(
    self, qubits: list[str], config: CalibrationConfig
) -> CalibrationReport:
    """Recalibrate *qubits* only, running whichever routines the checks blame.

    Two narrowings, and both matter: this is what a drift check triggers, and
    it has to be meaningfully cheaper than a full run to be worth having.

    The targets narrow to *qubits* plus any edge touching one of them. The
    routines narrow by :meth:`CalibrationDAG.diagnose` — each routine is asked
    whether its parameters still hold, and the shallowest node with evidence
    against it is recalibrated along with everything downstream of it.
    Downstream because recalibrating a frequency invalidates the gates tuned
    against it, so re-running the blamed node alone would leave the chip in a
    worse state than not running at all.

    Where RFC 0004 assumed the boundary, this measures it. A run in which every
    check passes recalibrates nothing and says so, which is the cheapest
    possible outcome and was not previously reachable.
    """
    self._refresh_device_config()
    config.validate_against(routine_names())
    config.validate_targets()
    narrowed = self._narrow_to(qubits, config)
    dag = CalibrationDAG(self.routines(), narrowed, bias=self.bias)
    order, notes = dag.diagnose(
        list(RECALIBRATION_SEEDS), self.device, self.backend, narrowed
    )
    for note in notes:
        log.info("diagnose: %s", note)

    if not order:
        report = CalibrationReport(
            timestamp=utc_timestamp(),
            duration_s=0.0,
            mode="partial",
            backend=self.backend.name,
        )
        report.notes.extend(notes)
        return report

    report = dag.run(
        self.device,
        self.backend,
        narrowed,
        mode="partial",
        only=order,
        on_progress=self.on_progress,
    )
    report.notes.extend(notes)
    self._persist(report)
    return report

routines()

The routines this tuner can run. Every tuner runs the same set.

Source code in qpi-driver/py/qpi_driver/tuners/base/__init__.py

def routines(self) -> list[CalibrationRoutine]:
    """The routines this tuner can run. Every tuner runs the same set."""
    return all_routines()

load_installed_devices()

Register every device advertised under :data:ENTRY_POINT_GROUP.

An entry point must resolve to a :class:DeviceSpec or an iterable of them. One that does not — because it will not import, resolves to something else, or names a device already registered — is logged and skipped: a third party's mistake must not stop the CLI from starting or --help from printing.

Returns the names of the devices registered, for the caller to log or test.

Source code in qpi-driver/py/qpi_driver/builtins/discovery.py

def load_installed_devices() -> tuple[str, ...]:
    """Register every device advertised under :data:`ENTRY_POINT_GROUP`.

    An entry point must resolve to a :class:`DeviceSpec` or an iterable of them.
    One that does not — because it will not import, resolves to something else, or
    names a device already registered — is logged and skipped: a third party's
    mistake must not stop the CLI from starting or ``--help`` from printing.

    Returns the names of the devices registered, for the caller to log or test.
    """
    registered: list[str] = []
    for entry_point in importlib.metadata.entry_points(group=ENTRY_POINT_GROUP):
        try:
            for spec in _specs_from(entry_point.load()):
                register(spec)
                registered.append(spec.name)
        except Exception as exc:
            log.warning(
                "skipping device entry point %r: %s: %s",
                entry_point.name,
                type(exc).__name__,
                exc,
            )
    return tuple(registered)

register(spec)

Add spec to the registry, or raise if its name is already taken.

Names are unique per operation, not globally: nothing stops a process and a monitor device from sharing a name, since --device is always read in the context of an operation.

Raises:

Type	Description
ValueError	if a device of the same name is already registered for the same operation. Silently replacing it would make the winner depend on import order.

Source code in qpi-driver/py/qpi_driver/builtins/registry.py

def register(spec: DeviceSpec) -> None:
    """Add *spec* to the registry, or raise if its name is already taken.

    Names are unique per operation, not globally: nothing stops a ``process``
    and a ``monitor`` device from sharing a name, since ``--device`` is always
    read in the context of an operation.

    Raises:
        ValueError: if a device of the same name is already registered for the
            same operation. Silently replacing it would make the winner depend
            on import order.
    """
    existing = _DEVICES[spec.operation].get(spec.name)
    if existing is not None:
        raise ValueError(
            f"{spec.operation.value} device {spec.name!r} is already registered"
        )
    _DEVICES[spec.operation][spec.name] = spec