qamomile.qiskit.runtime

Execute Qiskit circuits on IBM Quantum through Runtime V2 primitives.

Overview¶

Functions¶

hamiltonian_to_sparse_pauli_op [source]¶

def hamiltonian_to_sparse_pauli_op(hamiltonian: qm_o.Hamiltonian) -> 'SparsePauliOp'

Convert qamomile.observable.Hamiltonian to Qiskit SparsePauliOp.

Parameters:

Returns:

'SparsePauliOp' — Qiskit SparsePauliOp representation

Example:

import qamomile.observable as qm_o
from qamomile.qiskit.observable import hamiltonian_to_sparse_pauli_op

# Build Hamiltonian
H = qm_o.Z(0) * qm_o.Z(1) + 0.5 * (qm_o.X(0) + qm_o.X(1))

# Convert to Qiskit
sparse_pauli_op = hamiltonian_to_sparse_pauli_op(H)

Classes¶

CircuitInvocation [source]¶

class CircuitInvocation(Generic[CircuitT])

Keep an emitted circuit and runtime parameter values together.

Engines may bind the values into a new circuit or submit them through a native parameter-input API. Keeping both forms available preserves native parameter sweeps and provider-side compilation caches.

Parameters:

Constructor¶

def __init__(
    self,
    circuit: CircuitT,
    bindings: Mapping[str, Any],
    parameter_metadata: ParameterMetadata,
) -> None

Attributes¶

• bindings: Mapping[str, Any]

• circuit: CircuitT

• parameter_metadata: ParameterMetadata

CompletedExecutionHandle [source]¶

class CompletedExecutionHandle(ExecutionHandle[ResultT])

Wrap an already available result for synchronous executors.

Parameters:

Constructor¶

def __init__(self, value: ResultT) -> None

Initialize an immediately completed execution.

Parameters:

Methods¶

result¶

def result(self, timeout: float | None = None) -> ResultT

Return the completed value without waiting.

Parameters:

Returns:

ResultT — Stored execution value.

snapshot¶

def snapshot(self) -> ExecutionSnapshot

Capture the already available raw result without waiting.

Returns:

ExecutionSnapshot — Owned, type-preserving local value.

Raises:

• TypeError — If the value contains unsupported result objects.

• ValueError — If the value is nonfinite or cyclic.

status¶

def status(self) -> JobStatus

Return the completed status.

Returns:

JobStatus — Always :attr:JobStatus.COMPLETED.

EstimateRequest [source]¶

class EstimateRequest(Generic[CircuitT])

Describe one Hamiltonian expectation execution.

Parameters:

Constructor¶

def __init__(
    self,
    invocation: CircuitInvocation[CircuitT],
    hamiltonian: qm_o.Hamiltonian,
    accuracy: EstimationAccuracy | None = None,
) -> None

Attributes¶

• accuracy: EstimationAccuracy | None

• hamiltonian: qm_o.Hamiltonian

• invocation: CircuitInvocation[CircuitT]

ExecutionCapabilities [source]¶

class ExecutionCapabilities

Declare the execution features implemented by one executor.

Parameters:

Raises:

• ValueError — If an unknown estimation policy type is declared or estimation features are declared without estimation support.

Constructor¶

def __init__(
    self,
    supports_async_sampling: bool = False,
    supports_async_estimation: bool = False,
    supports_estimation: bool = False,
    supports_cancellation: bool = False,
    supports_restoration: bool = False,
    supports_native_batch: bool = False,
    supports_native_parameter_inputs: bool = False,
    estimation_accuracy: frozenset[EstimationPolicyType] = frozenset(),
) -> None

Attributes¶

• estimation_accuracy: frozenset[EstimationPolicyType]

• supports_async_estimation: bool

• supports_async_sampling: bool

• supports_cancellation: bool

• supports_estimation: bool

• supports_native_batch: bool

• supports_native_parameter_inputs: bool

• supports_restoration: bool

ExecutionHandle [source]¶

class ExecutionHandle(ABC, Generic[ResultT])

Expose an engine execution without forcing immediate result retrieval.

Attributes¶

• native: object | None Return the wrapped provider-native task when available.

Methods¶

cancel¶

def cancel(self) -> None

Request best-effort cancellation.

Cancellation is intentionally not reported as a boolean because providers may accept a request after execution has already started. Call :meth:status to observe the eventual state.

metadata¶

def metadata(self) -> Mapping[str, Any]

Return optional provider execution metadata.

Returns:

Mapping[str, Any] — Mapping[str, Any]: Provider metadata such as timestamps or usage.

raw_status¶

def raw_status(self) -> object

Return provider-specific status information.

Returns:

object — Provider status value, or the normalized status when no richer value exists.

references¶

def references(self) -> tuple[ExecutionReference, ...]

Return serializable remote execution references.

Returns:

tuple[ExecutionReference, ...] — tuple[ExecutionReference, ...]: Secret-free provider references.

result¶

def result(self, timeout: float | None = None) -> ResultT

Wait for and return the engine-neutral raw result.

Parameters:

Returns:

ResultT — Raw result normalized by the engine executor.

Raises:

• TimeoutError — If the local wait expires before completion.

result_async¶

def result_async(self, timeout: float | None = None) -> ResultT

Wait asynchronously for the engine-neutral raw result.

Parameters:

Returns:

ResultT — Raw result normalized by the engine executor.

Raises:

• TimeoutError — If the local wait expires before completion.

snapshot¶

def snapshot(self) -> ExecutionSnapshot

Capture one remote execution without fetching its result.

Adapters exposing several logical references must override this method with an explicit reconstruction structure. A provider reference may itself contain multiple physical job IDs.

Returns:

ExecutionSnapshot — One opaque provider execution.

Raises:

• ValueError — If references are absent or their grouping is unknown.

status¶

def status(self) -> JobStatus

Return the current provider-independent execution status.

Returns:

JobStatus — Current normalized status.

ExecutionReference [source]¶

class ExecutionReference

Store secret-free identifiers needed to restore remote execution.

Parameters:

Raises:

• ValueError — If the provider name or any job identifier is empty.

• TypeError — If identifiers or context have incompatible types.

Constructor¶

def __init__(
    self,
    provider: str,
    job_ids: tuple[str, ...],
    target: str | None = None,
    group_id: str | None = None,
    context: Mapping[str, str] = dict(),
) -> None

Attributes¶

• context: Mapping[str, str]

• group_id: str | None

• job_ids: tuple[str, ...]

• provider: str

• target: str | None

Methods¶

from_dict¶

@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> ExecutionReference

Reconstruct a provider reference from JSON-compatible data.

Parameters:

Returns:

ExecutionReference — Validated provider execution reference.

Raises:

• KeyError — If a required provider or job identifier field is absent.

• TypeError — If a field has an incompatible type.

• ValueError — If provider or job identifiers are empty.

to_dict¶

def to_dict(self) -> dict[str, Any]

Convert the provider reference to JSON-compatible data.

Returns:

dict[str, Any] — dict[str, Any]: Provider identifiers and decoding context without credentials or SDK objects.

ParameterMetadata [source]¶

class ParameterMetadata

Describe every scalar slot and runtime array in a compiled segment.

Parameters:

Constructor¶

def __init__(
    self,
    parameters: list[ParameterInfo] = list(),
    arrays: dict[str, ParameterArrayInfo] = dict(),
) -> None

Attributes¶

• arrays: dict[str, ParameterArrayInfo]

• parameters: list[ParameterInfo]

Methods¶

get_array_names¶

def get_array_names(self) -> set[str]

Return unique scalar and array root names.

Returns:

set[str] — set[str]: Root name for every emitted parameter.

get_ordered_params¶

def get_ordered_params(self) -> list[Any]

Return engine parameter objects in ABI definition order.

Returns:

list[Any] — list[Any]: Engine-specific parameter objects.

get_param_by_name¶

def get_param_by_name(self, name: str) -> ParameterInfo | None

Find one scalar slot by its full emitted key.

Parameters:

Returns:

ParameterInfo | None — ParameterInfo | None: Matching slot, or None when absent.

merge¶

@classmethod
def merge(cls, metadata: Sequence[ParameterMetadata]) -> ParameterMetadata

Merge parameter manifests from multiple quantum segments.

Parameters:

Returns:

ParameterMetadata — Combined manifest with first-seen scalar slot ordering and array descriptors derived across all segments.

to_binding_dict¶

def to_binding_dict(self, bindings: Mapping[str, Any]) -> dict[Any, Any]

Map indexed user bindings to engine parameter objects.

Parameters:

Returns:

dict[Any, Any] — dict[Any, Any]: Engine parameter objects mapped to bound values.

validate_array_shapes¶

def validate_array_shapes(self, bindings: Mapping[str, Any] | None) -> None

Validate user array rank and every concrete ABI dimension.

Parameters:

Raises:

• ValueError — If a supplied array has the wrong rank or exceeds a dimension whose ABI extent is known exactly.

validate_required_bindings¶

def validate_required_bindings(self, indexed_bindings: Mapping[str, Any]) -> None

Reject missing scalar slots in an indexed binding map.

Parameters:

Raises:

• ValueError — If one or more emitted scalar slots are missing.

QuantumExecutor [source]¶

class QuantumExecutor(ABC, Generic[T])

Abstract base class for quantum backend execution.

To implement a custom executor:

1. execute() [Required] Execute circuit and return bitstring counts as dict[str, int]. Keys are bitstrings in big-endian format (e.g., “011” means q2=0, q1=1, q0=1).

2. bind_parameters() [Optional] Bind parameter values to parametric circuits. Override if your executor supports parametric circuits (e.g., QAOA variational circuits). Use ParameterMetadata.to_binding_dict() for easy conversion.

3. estimate() [Optional] Compute expectation values <psi|H|psi>. Override if your executor supports estimation primitives (e.g., Qiskit Estimator, QURI Parts).

Example (Minimal): class MyExecutor(QuantumExecutor[QuantumCircuit]): def init(self): from qiskit_aer import AerSimulator self.backend = AerSimulator()

    def execute(self, circuit, shots):
        from qiskit import transpile
        if circuit.num_clbits == 0:
            circuit = circuit.copy()
            circuit.measure_all()
        transpiled = transpile(circuit, self.backend)
        return self.backend.run(transpiled, shots=shots).result().get_counts()

Example (With Parameter Binding): def bind_parameters(self, circuit, bindings, metadata): # metadata.to_binding_dict() converts indexed names to engine params return circuit.assign_parameters(metadata.to_binding_dict(bindings))

Attributes¶

• capabilities: ExecutionCapabilities Describe backend execution features available through this adapter.

Methods¶

bind_invocation¶

def bind_invocation(self, invocation: CircuitInvocation[T]) -> T

Bind one invocation for a backend without native input submission.

Parameters:

Returns:

T — Bound circuit, or the original circuit when it has no runtime parameters.

bind_parameters¶

def bind_parameters(
    self,
    circuit: T,
    bindings: dict[str, Any],
    parameter_metadata: ParameterMetadata,
) -> T

Bind parameter values to the circuit.

Default implementation returns the circuit unchanged. Override for backends that support parametric circuits.

Parameters:

Returns:

T — New circuit with parameters bound

estimate¶

def estimate(
    self,
    circuit: T,
    hamiltonian: 'qm_o.Hamiltonian',
    params: Sequence[float] | None = None,
) -> float

Estimate the expectation value of a Hamiltonian.

This method computes <psi|H|psi> where psi is the quantum state prepared by the circuit and H is the Hamiltonian.

Backends can override this method to provide optimized implementations using their native estimator primitives.

Parameters:

Returns:

float — The estimated expectation value

Raises:

• NotImplementedError — If the executor does not support estimation

execute¶

def execute(self, circuit: T, shots: int) -> dict[str, int]

Execute the circuit and return bitstring counts.

Parameters:

Returns:

dict[str, int] — Dictionary mapping bitstrings to counts. dict[str, int] — {“00”: 512, “11”: 512}

restore¶

def restore(self, reference: ExecutionReference) -> ExecutionHandle[Any]

Restore a remote execution from a secret-free reference.

Parameters:

Returns:

ExecutionHandle[Any] — ExecutionHandle[Any]: Restored provider-backed handle.

Raises:

• NotImplementedError — If this executor cannot restore jobs.

submit_estimate¶

def submit_estimate(self, request: EstimateRequest[T]) -> ExecutionHandle[float]

Submit an expectation request through the synchronous path.

Parameters:

Returns:

ExecutionHandle[float] — ExecutionHandle[float]: Completed compatibility handle.

Raises:

• NotImplementedError — If an explicit accuracy policy is requested from an executor that has not implemented request submission.

submit_estimates¶

def submit_estimates(
    self,
    requests: Sequence[EstimateRequest[T]],
) -> ExecutionHandle[tuple[float, ...]]

Submit an ordered collection of expectation requests.

Parameters:

Returns:

ExecutionHandle[tuple[float, ...]] — ExecutionHandle[tuple[float, ...]]: Composite result handle in request order.

submit_sample¶

def submit_sample(self, request: SampleRequest[T]) -> ExecutionHandle[dict[str, int]]

Submit a sampling request through the synchronous compatibility path.

Remote executors should override this method and return immediately with a provider-backed handle. Existing synchronous executors inherit this implementation unchanged.

Parameters:

Returns:

ExecutionHandle[dict[str, int]] — ExecutionHandle[dict[str, int]]: Completed compatibility handle.

submit_samples¶

def submit_samples(
    self,
    requests: Sequence[SampleRequest[T]],
) -> ExecutionHandle[tuple[dict[str, int], ...]]

Submit an ordered collection of sampling requests.

Backends with native batch or parameter-sweep support should override this method. The default preserves ordering with individual requests.

Parameters:

Returns:

ExecutionHandle[tuple[dict[str, int], ...]] — ExecutionHandle[tuple[dict[str, int], ...]]: Composite result handle preserving request order.

RuntimeExecutionHandle [source]¶

class RuntimeExecutionHandle(ExecutionHandle[ResultT], Generic[ResultT])

Expose one Runtime primitive job without blocking its submission.

Result retrieval runs once in a daemon thread so local wait limits also work with injected Qiskit primitive jobs whose result() method has no timeout argument. Expiring a wait neither cancels nor resubmits the job.

Parameters:

Constructor¶

def __init__(
    self,
    job: Any,
    decoder: Callable[[Any], ResultT],
    reference: ExecutionReference | None = None,
) -> None

Initialize lazy retrieval and provider lifecycle delegation.

Parameters:

Attributes¶

• native: object Return the wrapped primitive job.

Methods¶

cancel¶

def cancel(self) -> None

Request cancellation when the primitive job is unfinished.

Raises:

• ExecutionError — If the status query or cancellation request fails.

metadata¶

def metadata(self) -> Mapping[str, Any]

Return native Runtime job metrics when supported.

Returns:

Mapping[str, Any] — Mapping[str, Any]: Provider metrics or an empty mapping when the local job does not expose metrics.

Raises:

• ExecutionError — If native metric retrieval fails.

raw_status¶

def raw_status(self) -> object

Read the native string or Qiskit status enum.

Returns:

object — Unmodified provider status.

Raises:

• ExecutionError — If the native status query fails.

references¶

def references(self) -> tuple[ExecutionReference, ...]

Return the job’s secret-free restoration reference when available.

Returns:

tuple[ExecutionReference, ...] — tuple[ExecutionReference, ...]: One remote reference or an empty tuple when no restoration service was configured.

result¶

def result(self, timeout: float | None = None) -> ResultT

Wait for the decoded primitive result with a local wait limit.

Successful results and execution failures are cached. A provider-side result-wait timeout permits another retrieval attempt, while an expired local wait leaves the existing retrieval running.

Parameters:

Returns:

ResultT — Cached or newly decoded primitive result.

Raises:

• ValueError — If the timeout is negative, boolean, or non-finite.

• TimeoutError — If the local or provider result wait expires.

• ExecutionError — If execution or result decoding fails.

snapshot¶

def snapshot(self) -> ExecutionSnapshot

Capture a Runtime reference or an already decoded result.

Restorable jobs retain their provider reference after result retrieval. Without a reference, result retrieval must have completed successfully before capturing its cached value. This method never starts retrieval, queries the provider, or waits for another result caller.

Returns:

ExecutionSnapshot — One remote reference or a detached local value.

Raises:

• TypeError — If a cached result contains unsupported objects.

• ValueError — If no reference or successful cached result exists, or the cached value is nonfinite, cyclic, or nested too deeply.

status¶

def status(self) -> JobStatus

Return the normalized provider or cached result status.

Returns:

JobStatus — Current provider-independent execution state.

Raises:

• ExecutionError — If the native job status cannot be retrieved.

SampleRequest [source]¶

class SampleRequest(Generic[CircuitT])

Describe one sampling execution.

Parameters:

Raises:

• ValueError — If shots is not positive.

Constructor¶

def __init__(self, invocation: CircuitInvocation[CircuitT], shots: int) -> None

Attributes¶

• invocation: CircuitInvocation[CircuitT]

• shots: int

TargetPrecision [source]¶

class TargetPrecision

Request an expectation value at a provider target precision.

Parameters:

Raises:

• ValueError — If precision is not positive.

Constructor¶

def __init__(self, precision: float) -> None

Attributes¶

• precision: float