# Copyright 2026 Arthur Strauss
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""QMBackend: Qiskit BackendV2 for Quantum Machines (circuit-to-QUA, run, primitives).
Author: Arthur Strauss
Date: 2026-02-08
"""
from __future__ import annotations
import warnings
from typing import (
Iterable,
List,
Dict,
Optional,
Callable,
Sequence,
Union,
Tuple,
TYPE_CHECKING,
)
from inspect import Signature, Parameter as sigParam
from qiskit.circuit import (
QuantumCircuit,
Parameter as QiskitParameter,
Instruction,
)
from qiskit.circuit.controlflow import CONTROL_FLOW_OP_NAMES
from qiskit.circuit.classical.expr import Var
from qiskit.providers import BackendV2 as Backend, QubitProperties, Options
from qiskit.result.models import MeasLevel, MeasReturnType
from qiskit.transpiler import Target, InstructionProperties, CouplingMap
from qiskit.qasm3 import Exporter
# QUA and Quam imports
from qm import QuantumMachinesManager, DictQuaConfig, QuantumMachine
from quam.components import Channel as QuAMChannel, QubitPair, Qubit
from quam.core import QuamRoot
# Helper modules
from ..parameter_table import ParameterTable, InputType, Parameter
from .backend_utils import (
validate_machine,
look_for_standard_op,
get_extended_gate_name_mapping,
control_flow_name_mapping,
qasm3_keyword_instructions,
)
from .qm_instruction_properties import QMInstructionProperties
if TYPE_CHECKING:
from iqcc_cloud_client.qmm_cloud import (
CloudQuantumMachine,
CloudQuantumMachinesManager,
)
from qm_qasm import QubitsMapping, Compiler
from quam.utils.qua_types import Scalar
from ..job.qm_job import QMJob
from .qua_circuit_compilation import QuaCircuitCompilation
__all__ = ["QMBackend", "QISKIT_PULSE_AVAILABLE"]
try: # Importing Qiskit Pulse components
from qiskit.pulse import (
DriveChannel,
MeasureChannel,
AcquireChannel,
ControlChannel,
Schedule,
ScheduleBlock,
Play,
Waveform,
SymbolicPulse,
)
from qiskit.pulse.channels import Channel as QiskitChannel
QISKIT_PULSE_AVAILABLE = True
except ImportError:
warnings.warn(
"Qiskit Pulse is not available, some features of the QM backend will not be available",
ImportWarning,
)
QISKIT_PULSE_AVAILABLE = False
QiskitChannel = DriveChannel = MeasureChannel = AcquireChannel = ControlChannel = Schedule = ScheduleBlock = (
Play
) = Waveform = SymbolicPulse = None
def requires_qiskit_pulse(func):
"""
Decorator to check if Qiskit Pulse is available before executing a function.
"""
def wrapper(*args, **kwargs):
if not QISKIT_PULSE_AVAILABLE:
raise ImportError("Current Qiskit version does not have Qiskit Pulse, lower it to 1.x to use this feature.")
return func(*args, **kwargs)
return wrapper
[docs]
class QMBackend(Backend):
"""Qiskit backend for Quantum Machines (QuAM).
Represents at the Qiskit level all available operations and qubit properties
of a Quantum Abstract Machine (QuAM) instance. The QuAM instance must expose
``active_qubits``, ``active_qubit_pairs``, and per-qubit/pair ``macros``.
"""
def __init__(
self,
machine: QuamRoot,
channel_mapping: Dict[QiskitChannel, QuAMChannel] | None = None,
init_macro: Optional[Callable] = None,
qmm: QuantumMachinesManager | CloudQuantumMachinesManager | None = None,
name: Optional[str] = None,
**fields,
):
"""Initialize the QM backend.
Args:
machine: The QuAM instance to wrap.
channel_mapping: Optional mapping from Qiskit Pulse channels
(``DriveChannel``, ``ControlChannel``, ``MeasureChannel``, …) to
QuAM channels. Required for converting Pulse schedules into
parametric QUA macros (Qiskit < 2.0; schedules must have fixed
durations).
init_macro: Optional QUA macro invoked at the start of each program to
initialize the QPU.
qmm: Optional ``QuantumMachinesManager`` or
``CloudQuantumMachinesManager``. Inferred from the machine when
omitted.
name: Optional backend name. Defaults to the network backend name or
``"QMBackend"``.
fields: Optional keyword overrides for default run options:
``shots`` (1024), ``compiler_options``, ``simulate``, ``memory``
(False), ``skip_reset`` (False), ``meas_level``
(``MeasLevel.CLASSIFIED``), ``meas_return``
(``MeasReturnType.AVERAGE``), ``timeout`` (60 seconds), and
``max_circuits`` (30).
"""
if name is None:
if "quantum_computer_backend" in machine.network:
name = machine.network["quantum_computer_backend"]
else:
name = "QMBackend"
Backend.__init__(self, name=name, **fields)
self._custom_instructions = {}
self.machine = validate_machine(machine)
self._qmm: Optional[QuantumMachinesManager | CloudQuantumMachinesManager] = qmm
self._qm: Optional[QuantumMachine] = None
self.channel_mapping: Dict[QiskitChannel, QuAMChannel] = channel_mapping or {}
self.reverse_channel_mapping: Dict[QuAMChannel, QiskitChannel] = {v: k for k, v in self.channel_mapping.items()}
self._qubit_dict = {qubit.name: i for i, qubit in enumerate(machine.active_qubits)}
self._qubit_pair_dict = {
qubit_pair.name: (
self._qubit_dict.get(qubit_pair.qubit_control.name, None),
self._qubit_dict.get(qubit_pair.qubit_target.name, None),
)
for qubit_pair in machine.active_qubit_pairs
}
self._target = Target(
f"Qiskit Backend for Quantum Abstract Machine (Quam) of {self.name}",
dt=1e-9,
granularity=4,
num_qubits=len(machine.active_qubits),
min_length=16,
qubit_properties=[
QubitProperties(t1=qubit.T1, t2=qubit.T2echo, frequency=qubit.f_01) for qubit in machine.active_qubits
],
)
# Base mapping: operations from machine macros and target updates
self._operation_mapping_QUA = {}
self._populate_target()
# Calibration mapping: working copy for circuit-specific pulse calibrations
self._calibration_operation_mapping_QUA = self._operation_mapping_QUA.copy()
self._qasm3_custom_gates = []
self._init_macro = init_macro if init_macro is not None else lambda: None
def __deepcopy__(self, memo):
"""
Custom deepcopy implementation to avoid copying non-picklable resources.
In particular, the underlying QuAM `machine` and live connection objects
(`_qmm`, `_qm`) may contain threading primitives such as RLocks, which
are not picklable and cause `copy.deepcopy` (used by Qiskit Runtime's
local service) to fail.
For the purposes of backend options copying, it is sufficient – and
typically preferable – to share these resources rather than duplicate
them, so we shallow-copy those attributes and deep-copy the rest.
"""
return self
@classmethod
def _default_options(cls) -> Options:
"""Return the default backend run options.
Defaults: ``shots=1024``, ``compiler_options=None``, ``simulate=None``,
``memory=False``, ``skip_reset=False``, ``meas_level=MeasLevel.CLASSIFIED``,
``meas_return=MeasReturnType.AVERAGE``, ``timeout=60``.
"""
return Options(
shots=1024,
compiler_options=None,
simulate=None,
memory=False,
skip_reset=False,
meas_level=MeasLevel.CLASSIFIED,
meas_return=MeasReturnType.AVERAGE,
timeout=60,
max_circuits=30,
)
@property
def target(self) -> Target:
return self._target
@property
def custom_instructions(self) -> Dict[str, Instruction]:
"""
Get the custom instructions for the backend (those that are part of the target but not in the
standard Qiskit gate set, inferred from the available macros)
"""
return self._custom_instructions
@property
def qubit_dict(self) -> Dict[str, int]:
"""
Get the qubit dictionary for the backend
"""
return self._qubit_dict
@property
def qubit_pair_dict(self) -> Dict[str, Tuple[int, int]]:
"""
Get the qubit pair dictionary for the backend
"""
return self._qubit_pair_dict
[docs]
def get_qubit_index(self, qubit: str | Qubit) -> int:
"""
Get the index of the given qubit or qubit name in Quam.
Args:
qubit: The qubit or qubit name
Returns:
The index of the given qubit or qubit name in Quam
"""
if isinstance(qubit, str):
return self.qubit_dict[qubit]
elif isinstance(qubit, Qubit):
return self.qubit_dict[qubit.name]
else:
raise ValueError("Qubit should be a string name or a Qubit object")
[docs]
def get_qubit_pair_indices(self, qubit_pair: str | QubitPair) -> Tuple[int, int]:
"""
Get the indices of the given qubit pair or qubit pair name in Quam.
Args:
qubit_pair: The qubit pair or qubit pair name
Returns:
The indices of the given qubit pair or qubit pair name in Quam
"""
if isinstance(qubit_pair, str):
return self.qubit_pair_dict[qubit_pair]
elif isinstance(qubit_pair, QubitPair):
return self.qubit_pair_dict[qubit_pair.name]
else:
raise ValueError("Qubit pair should be a string name or a QubitPair object")
[docs]
def get_qubit(self, qubit: int | str) -> Qubit:
"""Get the Qubit object for a qubit index or name.
Args:
qubit: Qubit index or name.
Returns:
The corresponding :class:`~quam.components.Qubit`.
"""
if isinstance(qubit, int):
return self.machine.active_qubits[qubit]
elif isinstance(qubit, str):
return self.machine.active_qubits[self.qubit_dict[qubit]]
else:
raise ValueError("Qubit should be an integer index or a string name")
[docs]
def get_qubit_pair(self, qubits: Tuple[int | str | Qubit, int | str | Qubit]) -> QubitPair:
"""Get the QubitPair for two qubit indices or names.
Args:
qubits: Pair of qubit indices, names, or :class:`~quam.components.Qubit`
objects.
Returns:
The corresponding :class:`~quam.components.QubitPair`.
"""
if isinstance(qubits, tuple) and len(qubits) == 2:
qubit1, qubit2 = qubits
if isinstance(qubit1, int):
q1 = self.machine.active_qubits[qubit1]
elif isinstance(qubit1, str):
q1 = self.machine.active_qubits[self.qubit_dict[qubit1]]
elif isinstance(qubit1, Qubit):
q1 = qubit1
else:
raise ValueError("First qubit should be an integer index, a string name or a Qubit object")
if isinstance(qubit2, int):
q2 = self.machine.active_qubits[qubit2]
elif isinstance(qubit2, str):
q2 = self.machine.active_qubits[self.qubit_dict[qubit2]]
elif isinstance(qubit2, Qubit):
q2 = qubit2
else:
raise ValueError("Second qubit should be an integer index, a string name or a Qubit object")
try:
qubit_pair = q1 @ q2 # Using the @ operator to get the QubitPair
if qubit_pair.name not in self._qubit_pair_dict:
raise ValueError(f"Qubit pair {qubits} not found in the machine's active qubit pairs")
return qubit_pair # Using the @ operator to get the QubitPair
except TypeError:
raise ValueError(f"Qubit pair {qubits} not found in the machine's active qubit pairs")
else:
raise ValueError("Qubit pair should be a tuple of two qubits")
@property
def qubit_mapping(self) -> QubitsMapping:
"""
Build the qubit to quantum elements mapping for the backend.
Should be of the form {qubit_index: (quantum_element1, quantum_element2, ...)}
"""
return {i: tuple(channel for channel in qubit.channels) for i, qubit in enumerate(self.machine.active_qubits)}
@property
def qubit_index_dict(self):
"""
Returns a dictionary mapping qubit indices (Qiskit numbering) to corresponding Qubit objects (based on
the active_qubits attribute of QuAM instance)
"""
return {i: qubit for i, qubit in enumerate(self.machine.active_qubits)}
@property
def qmm(self) -> QuantumMachinesManager | CloudQuantumMachinesManager:
"""
Returns the QuantumMachinesManager instance. Gets a new QuantumMachinesManager instance if none is already set.
"""
if self._qmm is None:
self._qmm = self.machine.connect()
return self._qmm
@qmm.setter
def qmm(self, qmm: QuantumMachinesManager | CloudQuantumMachinesManager):
"""
Set the QuantumMachinesManager instance.
"""
self._qmm = qmm
@property
def qm(self) -> QuantumMachine | CloudQuantumMachine:
"""
Returns the QuantumMachine instance. Gets a new QuantumMachine instance if none is already set, using the current QM config.
"""
if self._qm is None:
self._qm = self.qmm.open_qm(self.qm_config, close_other_machines=True)
from ..parameter_table.parameter_pool import ParameterPool
if ParameterPool.has_quarc_module():
ParameterPool.quarc_module().bind_connection(self.qmm)
return self._qm
[docs]
def close_all_qms(self):
"""
Close all QuantumMachines managed by the QuantumMachinesManager
"""
if isinstance(self.qmm, QuantumMachinesManager):
self.qmm.close_all_qms()
self._qm = None
else:
warnings.warn(
"Closing all QuantumMachines is not supported for the current QuantumMachinesManager type",
UserWarning,
)
@property
def qm_config(self) -> DictQuaConfig:
"""
Returns the QUA configuration for the backend
"""
return self.machine.generate_config()
@property
def max_circuits(self):
"""Maximum number of circuits (or PUBs for Primitives) packed into a single QUA program.
``backend.run``, ``QMSamplerV2``, and ``QMEstimatorV2`` all split larger batches into
several queued QUA programs whose results are stitched back transparently.
Must be a positive integer (>= 1). Defaults to 30.
Can be updated at any time via ``backend.set_options(max_circuits=N)``.
"""
return self.options.max_circuits
def _populate_target(self) -> None:
"""
Populate the target instructions with the QOP configuration from machine macros.
Updates both Target and operation_mapping_QUA incrementally (does not clear existing entries).
"""
from qm_qasm import OperationIdentifier
gate_map = get_extended_gate_name_mapping()
operations_dict = {}
operations_qua_dict = self._operation_mapping_QUA
name_to_op_dict = {}
coupling_map = []
# Add single qubit instructions
for q, qubit in enumerate(self.machine.active_qubits):
for op, func in qubit.macros.items():
op_ = look_for_standard_op(op)
prop = QMInstructionProperties(qua_pulse_macro=func)
if op_ in gate_map:
gate_op = gate_map[op_]
num_params = len(gate_op.params)
operations_dict.setdefault(op_, {})[(q,)] = prop
operations_qua_dict[OperationIdentifier(op_, num_params, (q,))] = func.apply
name_to_op_dict[op_] = gate_op
else:
# Create custom gate
signature = Signature.from_callable(func.apply)
params = signature.parameters.values()
positional_params = [
param
for param in params
if param.kind in (sigParam.POSITIONAL_OR_KEYWORD, sigParam.POSITIONAL_ONLY)
]
params = [QiskitParameter(param.name) for param in positional_params]
return_type = signature.return_annotation
if return_type is not None and return_type is not Signature.empty:
raise ValueError(f"Return type {return_type} not yet supported for custom gate {op_}")
gate_op = Instruction(op_, 1, 0, params)
operations_dict.setdefault(op_, {})[(q,)] = prop
operations_qua_dict[OperationIdentifier(op_, len(params), (q,))] = func.apply
name_to_op_dict[op_] = gate_op
self._custom_instructions[op_] = gate_op
for qubit_pair in self.machine.active_qubit_pairs:
q_ctrl = self.qubit_dict.get(qubit_pair.qubit_control.name, None)
q_tgt = self.qubit_dict.get(qubit_pair.qubit_target.name, None)
if q_ctrl is None or q_tgt is None:
warnings.warn(
f"Qubit pair {qubit_pair.name} contains a qubit ({qubit_pair.qubit_control.name if q_ctrl is None else qubit_pair.qubit_target.name}) that is not part of the active qubits."
)
continue
coupling_map.append([q_ctrl, q_tgt])
for op, func in qubit_pair.macros.items():
op_ = look_for_standard_op(op)
prop = QMInstructionProperties(qua_pulse_macro=func)
if op_ in gate_map:
gate_op = gate_map[op_]
num_params = len(gate_op.params)
operations_dict.setdefault(op_, {})[(q_ctrl, q_tgt)] = prop
operations_qua_dict[OperationIdentifier(op_, num_params, (q_ctrl, q_tgt))] = func.apply
name_to_op_dict[op_] = gate_op
else:
# Create custom gate
signature = Signature.from_callable(func.apply)
params = signature.parameters.values()
positional_params = [
param
for param in params
if param.kind in (sigParam.POSITIONAL_OR_KEYWORD, sigParam.POSITIONAL_ONLY)
]
params = [QiskitParameter(param.name) for param in positional_params]
return_type = signature.return_annotation
if return_type is not None and return_type is not Signature.empty:
raise ValueError(f"Return type {return_type} not yet supported for custom gate {op_}")
gate_op = Instruction(op_, 2, 0, params)
operations_dict.setdefault(op_, {})[(q_ctrl, q_tgt)] = prop
operations_qua_dict[OperationIdentifier(op_, len(params), (q_ctrl, q_tgt))] = func.apply
name_to_op_dict[op_] = gate_op
self._custom_instructions[op_] = gate_op
# Update Target object incrementally
for op, properties in operations_dict.items():
if self._target.instruction_supported(op):
for qargs, prop in properties.items():
# Check if this qargs combination already exists for this instruction
if self._target.instruction_supported(op, qargs):
self._target.update_instruction_properties(op, qargs, prop)
else:
raise ValueError(f"Instruction {op} with qargs {qargs} is not supported by the target")
else:
# Add new instruction to target
self._target.add_instruction(name_to_op_dict[op], properties=properties)
for flow_op_name, control_flow_op in control_flow_name_mapping.items():
if flow_op_name not in self._target.operation_names:
self._target.add_instruction(control_flow_op, name=flow_op_name)
self._coupling_map = CouplingMap(coupling_map)
@requires_qiskit_pulse
def get_quam_channel(self, channel: QiskitChannel):
"""
Convert a Qiskit Pulse channel to a QuAM channel
Args:
channel: The Qiskit Pulse Channel to convert
Returns:
The corresponding QuAM channel
"""
try:
return self.channel_mapping[channel]
except KeyError:
raise ValueError(f"Channel {channel} not in the channel mapping")
@requires_qiskit_pulse
def get_pulse_channel(self, channel: QuAMChannel):
"""
Convert a QuAM channel to a Qiskit Pulse channel
Args:
channel: The QuAM channel to convert
Returns:
The corresponding pulse channel
"""
return self.reverse_channel_mapping[channel]
@property
def meas_map(self) -> List[List[int]]:
"""
Retrieve the measurement map for the backend.
"""
return self._target.concurrent_measurements
@requires_qiskit_pulse
def drive_channel(self, qubit: int):
"""
Get the drive channel for a given qubit (should be mapped to a quantum element in configuration)
"""
return DriveChannel(qubit)
@requires_qiskit_pulse
def measure_channel(self, qubit: int):
return MeasureChannel(qubit)
@requires_qiskit_pulse
def acquire_channel(self, qubit: int):
return AcquireChannel(qubit)
@requires_qiskit_pulse
def control_channel(self, qubits: Iterable[int]):
"""Return the secondary drive channel for the given qubit
This is typically used for controlling multiqubit interactions.
This channel is derived from other channels.
This is required to be implemented if the backend supports Pulse
scheduling.
Args:
qubits: Tuple or list of qubits of the form
``(control_qubit, target_qubit)``.
Returns:
List[ControlChannel]: The multi qubit control line.
Raises:
NotImplementedError: if the backend doesn't support querying the
measurement mapping
"""
channels = []
qubits = list(qubits)
if len(qubits) != 2:
raise ValueError("Control channel should be defined for a qubit pair")
if self.channel_mapping is None:
raise ValueError("Channel mapping not defined")
for channel, element in self.channel_mapping.items():
if isinstance(channel, ControlChannel):
qubit_pair: QubitPair = element.parent
qubit_control = qubit_pair.qubit_control
qubit_target = qubit_pair.qubit_target
q_ctrl_idx = self.qubit_dict.get(qubit_control.name, None)
q_tgt_idx = self.qubit_dict.get(qubit_target.name, None)
if q_ctrl_idx is None or q_tgt_idx is None:
continue
if (q_ctrl_idx, q_tgt_idx) == tuple(qubits):
channels.append(channel)
if len(channels) == 0:
raise ValueError(f"Control channel not found for qubit pair {qubits} in the channel mapping")
return channels
[docs]
def run(self, run_input: QuantumCircuit | List[QuantumCircuit], **options) -> QMJob:
"""
Run one or more ``QuantumCircuit`` objects on the QM backend.
This method now delegates the heavy lifting (circuit validation,
compilation to QUA, job submission and result assembly) to the
:class:`QMJob` interface, keeping the public behaviour identical
while centralising execution logic in the job layer.
Args:
run_input: The ``QuantumCircuit`` (or list thereof) to run on the backend.
options: Backend run options (shots, simulate, compiler options, etc.).
Returns:
A :class:`QMJob` (or :class:`IQCCJob` for cloud backends) instance.
"""
from ..job.qm_job import QMJob
return QMJob.from_circuits(self, run_input, **options)
@requires_qiskit_pulse
def schedule_to_qua_macro(
self,
sched: Schedule,
param_table: Optional[ParameterTable] = None,
input_type: Optional[InputType] = None,
gate_param_names: Optional[Sequence[str]] = None,
) -> Callable:
"""
Convert a Qiskit Pulse Schedule to a QUA macro
Args:
sched: The Qiskit Pulse Schedule to convert
param_table: The parameter table to use for the conversion of parameterized pulses to QUA variables
input_type: The input type to use for the conversion of parameterized pulses to QUA variables.
Should be specified only if the schedule is parameterized and the parameter table is not provided.
gate_param_names: Optional sequence of parameter names for gate-level parameters when the schedule
is not parameterized; used so the macro signature matches the gate arity expected by the QM compiler.
Returns:
The QUA macro corresponding to the Qiskit Pulse Schedule
"""
from ..pulse import schedule_to_qua_macro
return schedule_to_qua_macro(self, sched, param_table, input_type, gate_param_names=gate_param_names)
@requires_qiskit_pulse
def add_pulse_operations(
self,
pulse_input: Union[Schedule, ScheduleBlock],
name: Optional[str] = None,
):
"""
Add pulse operations created in Qiskit to QuAM operations mapping
Args:
pulse_input: Pulse schedule or schedule block to register in the QuAM
operations mapping.
name: Optional base name for the added operations. For a
:class:`~qiskit.pulse.Schedule` or
:class:`~qiskit.pulse.ScheduleBlock`, each ``Play`` instruction
is stored as ``"{name}_{i}"``; for a single pulse, ``"{name}"``.
"""
from ..pulse import validate_schedule, QuAMQiskitPulse
pulse_input = validate_schedule(pulse_input)
# Update QuAM with additional custom pulses
for idx, (time, instruction) in enumerate(pulse_input.filter(instruction_types=[Play]).instructions):
instruction: Play
pulse, channel = instruction.pulse, instruction.channel
if not isinstance(pulse, (SymbolicPulse, Waveform)):
raise ValueError("Only SymbolicPulse and Waveform pulses are supported")
pulse_name = pulse.name
if not channel.is_parameterized() and pulse_name in self.get_quam_channel(channel).operations:
pulse_name += str(pulse.id)
pulse.name = pulse_name
# Check if pulse fits QOP constraints
if pulse.duration < 16:
raise ValueError("Pulse duration must be at least 16 ns")
elif pulse.duration % 4 != 0:
raise ValueError("Pulse duration must be a multiple of 4 ns")
if pulse.name is None:
if name is not None:
pulse.name = f"{name}_{idx}"
else:
pulse.name = f"qiskit_pulse_{id(pulse)}"
quam_pulse = QuAMQiskitPulse(pulse)
if quam_pulse.is_compile_time_parameterized():
raise ValueError("Pulse contains unassigned parameters that cannot be adjusted in real-time")
if channel.is_parameterized(): # Add pulse to each channel of same type
for ch in filter(
lambda x: isinstance(x, type(channel)),
self.channel_mapping.keys(),
):
self.get_quam_channel(ch).operations[pulse.name] = QuAMQiskitPulse(pulse)
else:
self.get_quam_channel(channel).operations[pulse.name] = QuAMQiskitPulse(pulse)
[docs]
def update_target(self, input_type: Optional[InputType] = None):
"""Synchronize Target object with ``_operation_mapping_QUA``.
This method performs a one-way sync from Target to ``_operation_mapping_QUA``:
1. Updates ``_operation_mapping_QUA`` from machine macros (incrementally).
2. Syncs operations from Target to ``_operation_mapping_QUA``, overwriting
entries for the same ``OperationIdentifier``.
3. Updates the calibration mapping.
The sync is additive (never removes operations) and Target entries take
precedence over machine macros for the same identifier.
Args:
input_type: Input type for converting parameterized instructions to
QUA variables. Required when the Target contains parameterized
pulse schedules.
"""
from qm_qasm import OperationIdentifier
# Step 1: Update from machine macros (incremental - doesn't clear, additive only)
self._populate_target()
# Step 2: Sync operations from Target to _operation_mapping_QUA (overwrites existing entries)
for op_name, op_properties in self.target.items():
# Skip control flow operations (they're handled separately)
if op_name in CONTROL_FLOW_OP_NAMES:
continue
for qubits, properties in op_properties.items():
if properties is None:
raise ValueError(
f"Operation {op_name} with qargs {qubits} has no properties defined in the target,"
f"hence cannot be added to the QUA operations mapping"
)
# Determine the OperationIdentifier and QUA macro/schedule
if isinstance(properties, QMInstructionProperties):
if properties.qua_pulse_macro is None:
raise ValueError(
f"Operation {op_name} with qargs {qubits} has no QUA macro defined in the target,"
f"hence cannot be added to the QUA operations mapping"
)
sched = properties.qua_pulse_macro
sig = Signature.from_callable(sched)
positional_params = [
param
for param in sig.parameters.values()
if param.kind in (sigParam.POSITIONAL_OR_KEYWORD, sigParam.POSITIONAL_ONLY)
]
num_params = len(positional_params)
op_id = OperationIdentifier(op_name, num_params, qubits)
# Overwrite existing entry if present (Target takes precedence over machine macros)
self._operation_mapping_QUA[op_id] = sched
elif isinstance(properties, InstructionProperties) and hasattr(properties, "calibration"):
from ..pulse.pulse_support_utils import validate_schedule
sched = validate_schedule(properties.calibration)
num_params = len(sched.parameters)
gate_param_names = None
if num_params == 0:
gate_map = get_extended_gate_name_mapping()
gate = gate_map.get(op_name)
if gate is not None and getattr(gate, "params", None):
num_params = len(gate.params)
gate_param_names = [getattr(p, "name", f"param_{i}") for i, p in enumerate(gate.params)]
op_id = OperationIdentifier(op_name, num_params, qubits)
if num_params > 0 and sched.is_parameterized():
param_table = ParameterTable.from_qiskit(
sched,
input_type=input_type,
name=sched.name + "_param_table",
)
else:
param_table = None
# Overwrite existing entry if present (Target takes precedence over machine macros)
self._operation_mapping_QUA[op_id] = self.schedule_to_qua_macro(
sched, param_table, gate_param_names=gate_param_names
)
# Step 3: Update calibration mapping
self._calibration_operation_mapping_QUA = self._operation_mapping_QUA.copy()
@requires_qiskit_pulse
def update_calibrations(self, qc: QuantumCircuit, input_type: Optional[InputType] = None):
"""Update the QUA operations mapping from circuit calibrations.
Requires Qiskit < 2.0 (Qiskit Pulse).
Args:
qc: Circuit whose ``calibrations`` attribute defines custom gates.
input_type: Input type for converting parameterized instructions to
QUA variables when the circuit or its calibrations are
parameterized.
"""
from qm_qasm import OperationIdentifier
if hasattr(qc, "calibrations") and qc.calibrations: # Check for custom calibrations
from ..pulse.pulse_support_utils import (
validate_schedule,
handle_parameterized_channel,
)
if qc.parameters or qc.iter_vars():
param_table = qc.metadata.get(
"qua",
ParameterTable.from_qiskit(qc, input_type=input_type, name=qc.name + "_param_table"),
)
if isinstance(param_table, Dict):
if len(param_table) == 1:
param_table = list(param_table.values())[0]
else:
param_table = ParameterTable.from_other_tables(list(param_table.values()))
else:
param_table = None
for gate_name, cal_info in qc.calibrations.items():
if gate_name not in self._qasm3_custom_gates: # Make it a basis gate for OQ compiler
self._qasm3_custom_gates.append(gate_name)
for (qubits, parameters), schedule in cal_info.items():
schedule = validate_schedule(schedule) # Check that schedule has fixed duration
# Convert type of parameters to int if required (for switch case over channels)
if param_table is not None:
param_table = handle_parameterized_channel(schedule, param_table)
gate_param_names = [getattr(p, "name", f"param_{i}") for i, p in enumerate(parameters)]
self._calibration_operation_mapping_QUA[
OperationIdentifier(
gate_name,
len(parameters),
qubits,
)
] = self.schedule_to_qua_macro(schedule, param_table, gate_param_names=gate_param_names)
self.add_pulse_operations(schedule, name=schedule.name)
[docs]
def quantum_circuit_to_qua(
self,
qc: QuantumCircuit,
param_table: Optional[
ParameterTable | Sequence[ParameterTable | Parameter] | Dict[str | QiskitParameter | Var, Scalar]
] = None,
) -> QuaCircuitCompilation:
"""Convert a :class:`~qiskit.circuit.QuantumCircuit` to a QUA program fragment.
Can be called inside an existing ``with program():`` block or standalone.
When called standalone, access the generated program via
``result.qua_program`` or ``result.result_program.dsl_program``.
Args:
qc: The circuit to compile.
param_table: Parameter mapping for real-time QUA variables. Required
when the circuit contains symbolic parameters or classical inputs
that must be streamed during execution.
Returns:
:class:`~qiskit_qm_provider.backend.qua_circuit_compilation.QuaCircuitCompilation`
wrapping the compilation result and
wired measurement outputs.
"""
from .qua_circuit_compilation import QuaCircuitCompilation
basis_gates = self.qm_qasm_basis_gates
# Check if all custom calibrations are in the qasm3 basis gates
if hasattr(qc, "calibrations") and qc.calibrations:
for gate_name in qc.calibrations.keys():
if gate_name not in basis_gates:
raise ValueError(
f"Custom calibration {gate_name} not in basis gates {basis_gates}",
f"Run update_calibrations() before compiling the circuit",
)
exporter = Exporter(includes=(), basis_gates=basis_gates, disable_constants=True)
open_qasm_code = exporter.dumps(qc)
open_qasm_code = "\n".join(
line for line in open_qasm_code.splitlines() if not line.strip().startswith(("barrier",))
)
inputs = None
if param_table is not None:
inputs = {}
if isinstance(param_table, (ParameterTable, Parameter)):
param_table = [param_table]
if isinstance(param_table, Sequence):
for table in param_table:
if not table.is_declared:
# Unified API: declare() works for both Parameter and ParameterTable.
table.declare(pause_program=False)
variables = table.variables_dict if isinstance(table, ParameterTable) else {table.name: table.var}
inputs.update(variables)
elif isinstance(param_table, Dict):
for key, value in param_table.items():
if isinstance(key, (QiskitParameter, Var)):
inputs[key.name] = value
else:
inputs[key] = value
result = self.compiler.compile(
open_qasm_code,
compilation_name=f"{qc.name}_qua",
inputs=inputs,
)
return QuaCircuitCompilation(result, qc)
@property
def compiler(self) -> Compiler:
"""
The OpenQASM to QUA compiler.
"""
from qm_qasm import Compiler, HardwareConfig
return Compiler(
hardware_config=HardwareConfig(
quantum_operations_db=self._calibration_operation_mapping_QUA,
physical_qubits=self.qubit_mapping,
)
)
[docs]
def connect(self) -> QuantumMachinesManager:
"""
Connect to the Quantum Machines Manager
"""
return self.machine.connect()
[docs]
def generate_config(self) -> DictQuaConfig:
"""
Generate the configuration for the Quantum Machine
"""
return self.machine.generate_config()
@property
def init_macro(self) -> Callable:
"""
The macro to be called at the beginning of the QUA program
"""
return self._init_macro
@init_macro.setter
def init_macro(self, macro: Callable):
"""
Set the macro to be called at the beginning of the QUA program
"""
if not callable(macro):
raise ValueError("Init macro must be a callable")
self._init_macro = macro
@property
def qubits(self) -> List[Qubit]:
"""
Retrieve the list of active qubits of the machine
"""
return self.machine.active_qubits
@property
def qubit_pairs(self) -> List[QubitPair]:
"""
Retrieve the list of active qubit pairs of the machine
"""
return self.machine.active_qubit_pairs
@property
def qm_qasm_basis_gates(self) -> List[str]:
"""
Retrieve the list of OpenQASM 3 basis gates supported by the backend
"""
basis_gates = list(
set(self._qasm3_custom_gates + list(self.target.operation_names)) - set(qasm3_keyword_instructions)
)
return basis_gates
@property
def qasm3_exporter(self) -> Exporter:
"""
Retrieve the OpenQASM 3 exporter for the backend
"""
return Exporter(
includes=(),
basis_gates=self.qm_qasm_basis_gates,
disable_constants=True,
)