# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the main branch
"""Compilation backend for quantum-circuit to quantum-device layer."""
from __future__ import annotations
import warnings
from copy import deepcopy
from itertools import permutations
from typing import Dict, List, Sequence, overload
import numpy as np
from quantify_scheduler.backends.graph_compilation import (
CompilationConfig,
DeviceCompilationConfig,
OperationCompilationConfig,
)
from quantify_scheduler.operations.control_flow_library import ControlFlowOperation
from quantify_scheduler.operations.operation import Operation
from quantify_scheduler.resources import ClockResource
from quantify_scheduler.schedules.schedule import Schedulable, Schedule, ScheduleBase
[docs]
def compile_circuit_to_device_with_config_validation(
schedule: Schedule,
config: CompilationConfig,
) -> Schedule:
"""
Add pulse information to all gates in the schedule.
Before calling this function, the schedule can contain abstract operations (gates or
measurements). This function adds pulse and acquisition information with respect to
``config`` as they are expected to arrive to device (latency or distortion corrections
are not taken into account).
From a point of view of :ref:`sec-compilation`, this function converts a schedule
defined on a quantum-circuit layer to a schedule defined on a quantum-device layer.
Parameters
----------
schedule
The schedule to be compiled.
config
Compilation config for
:class:`~quantify_scheduler.backends.graph_compilation.QuantifyCompiler`, of
which only the :attr:`.CompilationConfig.device_compilation_config`
is used in this compilation step.
Returns
-------
:
The modified ``schedule`` with pulse information added to all gates, or the unmodified
schedule if circuit to device compilation is not necessary.
"""
device_cfg = DeviceCompilationConfig.model_validate(
config.device_compilation_config
)
return _compile_circuit_to_device(operation=schedule, device_cfg=device_cfg)
# It is important that if the operation is a Schedule type, we always return a Schedule.
# Otherwise, we can return an Operation or a Schedule.
@overload
[docs]
def _compile_circuit_to_device(
operation: Schedule,
device_cfg: DeviceCompilationConfig,
) -> Schedule: ...
@overload
def _compile_circuit_to_device(
operation: Operation | Schedule,
device_cfg: DeviceCompilationConfig,
) -> Operation | Schedule: ...
def _compile_circuit_to_device(
operation,
device_cfg,
):
if isinstance(operation, ScheduleBase):
for inner_op_key in operation.operations:
operation.operations[inner_op_key] = _compile_circuit_to_device(
operation=operation.operations[inner_op_key], device_cfg=device_cfg
)
return operation
elif isinstance(operation, ControlFlowOperation):
operation.body = _compile_circuit_to_device(
operation=operation.body, device_cfg=device_cfg
)
return operation
elif not (operation.valid_pulse or operation.valid_acquisition):
# If operation is a valid pulse or acquisition it will not attempt to
# add pulse/acquisition info in the lines below (if operation.valid_gate
# will not work here for e.g. Measure, which is also a valid
# acquisition)
qubits: Sequence[str] = operation.data["gate_info"]["qubits"]
operation_type: str = operation.data["gate_info"]["operation_type"]
# single qubit operations
if len(qubits) == 1:
return _compile_single_qubit(
operation=operation,
qubit=qubits[0],
operation_type=operation_type,
device_cfg=device_cfg,
)
# it is a two-qubit operation if the operation not in the qubit config
elif len(qubits) == 2 and operation_type not in device_cfg.elements[qubits[0]]:
return _compile_two_qubits(
operation=operation,
qubits=qubits,
operation_type=operation_type,
device_cfg=device_cfg,
)
# we only support 2-qubit operations and single-qubit operations.
# some single-qubit operations (reset, measure) can be expressed as acting
# on multiple qubits simultaneously. That is covered through this for-loop.
else:
return _compile_multiplexed(
operation=operation,
qubits=qubits,
operation_type=operation_type,
device_cfg=device_cfg,
)
else:
return operation
[docs]
def set_pulse_and_acquisition_clock(
schedule: Schedule,
config: CompilationConfig,
) -> Schedule:
"""
Ensures that each pulse/acquisition-level clock resource is added to the schedule,
and validates the given configuration.
If a pulse/acquisition-level clock resource has not been added
to the schedule and is present in device_cfg, it is added to the schedule.
A warning is given when a clock resource has conflicting frequency
definitions, and an error is raised if the clock resource is unknown.
Parameters
----------
schedule
The schedule to be compiled.
config
Compilation config for
:class:`~quantify_scheduler.backends.graph_compilation.QuantifyCompiler`, of
which only the :attr:`.CompilationConfig.device_compilation_config`
is used in this compilation step.
Returns
-------
:
The modified ``schedule`` with all clock resources added, or the unmodified
schedule if circuit to device compilation is not necessary.
Warns
-----
RuntimeWarning
When clock has conflicting frequency definitions.
Raises
------
RuntimeError
When operation is not at pulse/acquisition-level.
ValueError
When clock frequency is unknown.
ValueError
When clock frequency is NaN.
"""
device_cfg = DeviceCompilationConfig.model_validate(
config.device_compilation_config
)
all_clock_freqs: Dict[str, float] = {}
_extract_clock_freqs(schedule, all_clock_freqs)
for clock, freq in device_cfg.clocks.items():
if clock in all_clock_freqs:
_clocks_compatible(clock, device_cfg, all_clock_freqs)
else:
all_clock_freqs[clock] = freq
return _set_pulse_and_acquisition_clock(
schedule=schedule,
operation=schedule,
all_clock_freqs=all_clock_freqs,
verified_clocks=[],
)
# It is important that if the operation is a Schedule type, we always return a Schedule.
# Otherwise, we can return an Operation or a Schedule.
@overload
[docs]
def _set_pulse_and_acquisition_clock(
schedule: Schedule,
operation: Schedule,
all_clock_freqs: Dict[str, float],
verified_clocks: List,
) -> Schedule: ...
@overload
def _set_pulse_and_acquisition_clock(
schedule: Schedule,
operation: Operation | Schedule,
all_clock_freqs: Dict[str, float],
verified_clocks: List,
) -> Operation | Schedule: ...
def _set_pulse_and_acquisition_clock(
schedule,
operation,
all_clock_freqs,
verified_clocks,
):
"""
Ensures that each pulse/acquisition-level clock resource is added to the schedule.
Parameters
----------
schedule
The resources from ``operation`` are added to ``schedule``
if ``operation`` is not a ``Schedule``.
operation
The ``operation`` to collect resources from.
all_clock_freqs
All clock frequencies.
verified_clocks
Already verified clocks.
Returns
-------
:
The modified ``operation`` with all clock resources added.
"""
if isinstance(operation, ScheduleBase):
# verify that required clocks are present; print warning if they are inconsistent
verified_clocks = []
for inner_op_key in operation.operations:
# Only if we have a valid device-level operation, we can assign clocks
operation.operations[inner_op_key] = _set_pulse_and_acquisition_clock(
schedule=operation,
all_clock_freqs=all_clock_freqs,
operation=operation.operations[inner_op_key],
verified_clocks=verified_clocks,
)
elif isinstance(operation, ControlFlowOperation):
operation.body = _set_pulse_and_acquisition_clock(
schedule=schedule,
operation=operation.body,
all_clock_freqs=all_clock_freqs,
verified_clocks=verified_clocks,
)
else:
_assert_operation_valid_device_level(operation)
operation_info = operation["pulse_info"] + operation["acquisition_info"]
clocks_used = set([info["clock"] for info in operation_info])
for clock in clocks_used:
if clock in verified_clocks:
continue
# raises ValueError if no clock found;
# enters if condition if clock only in device config
if not _valid_clock_in_schedule(
clock, all_clock_freqs, schedule, operation
):
clock_resource = ClockResource(name=clock, freq=all_clock_freqs[clock])
schedule.add_resource(clock_resource)
verified_clocks.append(clock)
return operation
[docs]
def _valid_clock_in_schedule(
clock: str,
all_clock_freqs: Dict[str, float],
schedule: Schedule,
operation: Operation,
) -> bool:
"""
Asserts that valid clock is present. Returns whether clock is already in schedule.
Parameters
----------
clock
Name of the clock
all_clock_freqs
All clock frequencies
schedule
Schedule that potentially has the clock in its resources
operation
Quantify operation, to which the clock belongs. Only used for error message.
Raises
------
ValueError
Returns ValueError if (i) the device config is the only defined clock and
contains nan values or (ii) no clock is defined.
"""
if clock in schedule.resources:
return True
else:
if clock in all_clock_freqs:
# Clock only in device config
if np.isnan(all_clock_freqs[clock]).any():
raise ValueError(
f"Operation '{operation}' contains clock '{clock}' with an "
f"undefined (initial) frequency; ensure this resource has been "
f"added to the schedule or to the device config."
)
return False
# Clock neither in device config nor schedule.
raise ValueError(
f"Operation '{operation}' contains an unknown clock '{clock}'; "
f"ensure this resource has been added to the schedule "
f"or to the device config."
)
[docs]
def _clocks_compatible(
clock,
device_cfg: DeviceCompilationConfig,
schedule_clock_resources: Dict[str, float],
) -> bool:
"""
Compare device config and schedule resources for compatibility of their clocks.
Clocks can be defined in the device_cfg and in the schedule. They are consistent if
- they have the same value
- if the clock in the device config is nan (not the other way around)
These conditions are also generalized to numpy arrays. Arrays of different length
are only equal if all frequencies in the device config are nan.
If the clocks are inconsistent, a warning message is emitted.
Parameters
----------
clock
Name of the clock found in the device config and schedule
device_cfg
Device config containing the ``clock``
schedule_clock_resources
All clock resources in the schedule
Returns
-------
True if the clock frequencies are consistent.
"""
clock_freq_device_cfg = np.asarray(device_cfg.clocks[clock])
clock_freq_schedule = np.asarray(schedule_clock_resources[clock])
is_nan = np.isnan(clock_freq_device_cfg)
if is_nan.all():
return True
try:
is_equal = clock_freq_device_cfg == clock_freq_schedule
except ValueError:
return False
if (np.logical_or(is_nan, is_equal)).all():
return True
warnings.warn(
f"Clock '{clock}' has conflicting frequency definitions: "
f"{clock_freq_schedule} Hz in the schedule and "
f"{clock_freq_device_cfg} Hz in the device config. "
f"The clock is set to '{clock_freq_schedule}'. "
f"Ensure the schedule clock resource matches the "
f"device config clock frequency or set the "
f"clock frequency in the device config to np.NaN "
f"to omit this warning.",
RuntimeWarning,
)
return False
[docs]
def _assert_operation_valid_device_level(operation: Operation) -> None:
"""
Verifies that the operation has been compiled to device level.
Parameters
----------
operation
Quantify operation
"""
if not (
operation.valid_pulse
or operation.valid_acquisition
or operation.has_voltage_offset
):
raise RuntimeError(
f"Operation '{operation}' is a gate-level operation and must be "
f"compiled from circuit to device; ensure compilation "
f"is made in the correct order."
)
[docs]
def _compile_multiplexed(
operation: Operation,
qubits: Sequence[str],
operation_type: str,
device_cfg: DeviceCompilationConfig,
) -> Operation | Schedule:
"""
Compiles gate with multiple qubits.
Note: it updates the `operation`, if it can directly add pulse representation.
"""
inner_subschedules: list = []
operation_has_device_representation: bool = False
for mux_idx, qubit in enumerate(qubits):
if qubit not in device_cfg.elements:
raise ConfigKeyError(
kind="element",
missing=qubit,
allowed=list(device_cfg.elements.keys()),
)
element_cfg = device_cfg.elements[qubit]
if operation_type not in element_cfg:
raise ConfigKeyError(
kind="operation",
missing=operation_type,
allowed=list(element_cfg.keys()),
)
device_op: Operation | Schedule = _get_device_repr_from_cfg_multiplexed(
operation, element_cfg[operation_type], mux_idx=mux_idx
)
if isinstance(device_op, ScheduleBase):
inner_subschedules.append(
_compile_circuit_to_device(operation=device_op, device_cfg=device_cfg)
)
else:
operation.add_device_representation(device_op)
operation_has_device_representation = True
if len(inner_subschedules) != 0:
inner_schedule: Schedule = Schedule(f"Inner schedule for {str(operation)}")
# All operations in the inner schedule
# should happen at the same time;
# this reference time is the start time
# of the `ref_schedulable` operation / schedule
ref_schedulable: Schedulable | None = None
if operation_has_device_representation:
ref_schedulable = inner_schedule.add(operation)
for inner_subschedule in inner_subschedules:
if ref_schedulable is not None:
inner_schedule.add(
operation=_compile_circuit_to_device(
operation=inner_subschedule, device_cfg=device_cfg
),
rel_time=0,
ref_op=ref_schedulable,
ref_pt="start",
)
else:
ref_schedulable = inner_schedule.add(operation=inner_subschedule)
return inner_schedule
else:
return operation
[docs]
def _compile_single_qubit(
operation: Operation,
qubit: str,
operation_type: str,
device_cfg: DeviceCompilationConfig,
) -> Operation | Schedule:
"""
Compiles gate with multiple qubits.
Note: it updates the `operation`, if it can directly add pulse representation.
"""
if qubit not in device_cfg.elements:
raise ConfigKeyError(
kind="element",
missing=qubit,
allowed=list(device_cfg.elements.keys()),
)
element_cfg = device_cfg.elements[qubit]
if operation_type not in element_cfg:
raise ConfigKeyError(
kind="operation",
missing=operation_type,
allowed=list(element_cfg.keys()),
)
device_op: Operation | Schedule = _get_device_repr_from_cfg(
operation=operation,
operation_cfg=element_cfg[operation_type],
)
if isinstance(device_op, ScheduleBase):
return _compile_circuit_to_device(operation=device_op, device_cfg=device_cfg)
else:
operation.add_device_representation(device_op)
return operation
[docs]
def _compile_two_qubits(
operation: Operation,
qubits: Sequence[str],
operation_type: str,
device_cfg: DeviceCompilationConfig,
) -> Operation | Schedule:
"""
Compiles gate with multiple qubits.
Note: it updates the `operation`, if it can directly add pulse representation.
"""
parent_qubit, child_qubit = qubits
edge = f"{parent_qubit}_{child_qubit}"
symmetric_operation = operation.get("gate_info", {}).get("symmetric", False)
if symmetric_operation:
possible_permutations = permutations(qubits, 2)
operable_edges = {
f"{permutation[0]}_{permutation[1]}"
for permutation in possible_permutations
}
valid_edge_list = list(operable_edges.intersection(device_cfg.edges))
if len(valid_edge_list) == 1:
edge = valid_edge_list[0]
elif len(valid_edge_list) < 1:
raise ConfigKeyError(
kind="edge", missing=edge, allowed=list(device_cfg.edges.keys())
)
elif len(valid_edge_list) > 1:
raise MultipleKeysError(operation=operation_type, matches=valid_edge_list)
if edge not in device_cfg.edges:
raise ConfigKeyError(
kind="edge", missing=edge, allowed=list(device_cfg.edges.keys())
)
edge_config = device_cfg.edges[edge]
if operation_type not in edge_config:
# only raise exception if it is also not a single-qubit operation
raise ConfigKeyError(
kind="operation",
missing=operation_type,
allowed=list(edge_config.keys()),
)
device_op: Operation | Schedule = _get_device_repr_from_cfg(
operation, edge_config[operation_type]
)
if isinstance(device_op, ScheduleBase):
return _compile_circuit_to_device(operation=device_op, device_cfg=device_cfg)
else:
operation.add_device_representation(device_op)
return operation
[docs]
def _get_device_repr_from_cfg(
operation: Operation, operation_cfg: OperationCompilationConfig
) -> Operation | Schedule:
# deepcopy because operation_type can occur multiple times
# (e.g., parametrized operations).
operation_cfg = deepcopy(operation_cfg)
factory_func = operation_cfg.factory_func
factory_kwargs: Dict = operation_cfg.factory_kwargs
# retrieve keyword args for parametrized operations from the gate info
if operation_cfg.gate_info_factory_kwargs is not None:
for key in operation_cfg.gate_info_factory_kwargs:
factory_kwargs[key] = operation.data["gate_info"][key]
return factory_func(**factory_kwargs)
[docs]
def _get_device_repr_from_cfg_multiplexed(
operation: Operation, operation_cfg: OperationCompilationConfig, mux_idx: int
) -> Operation | Schedule:
operation_cfg = deepcopy(operation_cfg)
factory_func = operation_cfg.factory_func
factory_kwargs: Dict = operation_cfg.factory_kwargs
# retrieve keyword args for parametrized operations from the gate info
if operation_cfg.gate_info_factory_kwargs is not None:
for key in operation_cfg.gate_info_factory_kwargs:
gate_info = operation.data["gate_info"][key]
# Hack alert: not all parameters in multiplexed operation are
# necessary passed for each element separately. We assume that if they do
# (say, acquisition index and channel for measurement), they are passed as
# a list or tuple. If they don't (say, it is hard to imagine different
# acquisition protocols for qubits during multiplexed readout), they are
# assumed to NOT be a list or tuple. If this spoils the correct behaviour of
# your program in future: sorry :(
if isinstance(gate_info, (tuple, list)):
factory_kwargs[key] = gate_info[mux_idx]
else:
factory_kwargs[key] = gate_info
return factory_func(**factory_kwargs)
[docs]
class ConfigKeyError(KeyError):
"""Custom exception for when a key is missing in a configuration file."""
def __init__(self, kind, missing, allowed):
self.value = (
f'{kind} "{missing}" is not present in the configuration file;'
+ f" {kind} must be one of the following: {allowed}"
)
def __str__(self):
return repr(self.value)
[docs]
class MultipleKeysError(KeyError):
"""Custom exception for when symmetric keys are found in a configuration file."""
def __init__(self, operation, matches):
self.value = (
f"Symmetric Operation {operation} matches the following edges {matches}"
f" in the QuantumDevice. You can only specify a single edge for a symmetric"
" operation."
)
def __str__(self):
return repr(self.value)
