diff --git a/qiskit_experiments/calibration/__init__.py b/qiskit_experiments/calibration/__init__.py new file mode 100644 index 0000000000..ecbf6c6a9a --- /dev/null +++ b/qiskit_experiments/calibration/__init__.py @@ -0,0 +1,123 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +r""" +Qiskit Experiments Calibration Root. + +.. warning:: + The calibrations interface is still in active development. It may have + breaking API changes without deprecation warnings in future releases until + otherwise indicated. + +Calibrations are managed by the Calibrations class. This class stores schedules which are +intended to be fully parameterized, including the index of the channels. This class: +- supports having different schedules share parameters +- allows default schedules for qubits that can be overridden for specific qubits. + +The following code illustrates how a user can create a parameterized schedule, add +values to the parameters and query a schedule. + +.. code-block:: python + + dur = Parameter("dur") + amp = Parameter("amp") + sigma = Parameter("σ") + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(dur, amp, sigma), DriveChannel(Parameter("ch0"))) + + cals = Calibrations() + cals.add_schedule(xp) + + # add duration and sigma parameter values for all qubits. + cals.add_parameter_value(160, "dur", schedule="xp") + cals.add_parameter_value(35.5, "σ", schedule="xp") + + # Add an amplitude for qubit 3. + cals.add_parameter_value(0.2+0.05j, "amp", (3, ), "xp") + + # Retrieve an xp pulse with all parameters assigned + cals.get_schedule("xp", (3, )) + + # Retrieve an xp pulse with unassigned amplitude + cals.get_schedule("xp", (3, ), free_params=["amp"]) + +The Calibrations make a couple of assumptions which are discussed below. + +Parametric channel naming convention +========================= + +Parametrized channel indices must be named according to a predefined pattern to properly +identify the channels and control channels when assigning values to the parametric +channel indices. A channel must have a name that starts with `ch` followed by an integer. +For control channels this integer can be followed by a sequence `.integer`. +Optionally, the name can end with `$integer` to specify the index of a control channel +for the case when a set of qubits share multiple control channels. For example, +valid channel names include "ch0", "ch1", "ch0.1", "ch0$", "ch2$3", and "ch1.0.3$2". +The "." delimiter is used to specify the different qubits when looking for control +channels. The optional $ delimiter is used to specify which control channel to use +if several control channels work together on the same qubits. For example, if the +control channel configuration is {(3,2): [ControlChannel(3), ControlChannel(12)]} +then given qubits (2, 3) the name "ch1.0$1" will resolve to ControlChannel(12) while +"ch1.0$0" will resolve to ControlChannel(3). A channel can only have one parameter. + +Parameter naming restriction +=================== + +Each parameter must have a unique name within each schedule. For example, it is +acceptable to have a parameter named 'amp' in the schedule 'xp' and a different +parameter instance named 'amp' in the schedule named 'xm'. It is not acceptable +to have two parameters named 'amp' in the same schedule. The naming restriction +only applies to parameters used in the immediate scope of the schedule. Schedules +called by Call instructions have their own scope for Parameter names. + +The code block below illustrates the creation of a template schedule for a echoed cross- +resonance gate. + +.. code-block:: python + + amp_cr = Parameter("amp") + amp = Parameter("amp") + d0 = DriveChannel(Parameter("ch0")) + c1 = ControlChannel(Parameter("ch0.1")) + sigma = Parameter("σ") + width = Parameter("w") + dur_xp = Parameter("duration") + dur_cr = Parameter("duration") + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(dur_xp, amp, sigma), d0) + + with pulse.build(name="cr") as cr: + with pulse.align_sequential(): + pulse.play(GaussianSquare(dur_cr, amp_cr, sigma, width), c1) + pulse.call(xp) + pulse.play(GaussianSquare(dur_cr, -amp_cr, sigma, width), c1) + pulse.call(xp) + + cals = Calibrations() + cals.add_schedule(xp) + cals.add_schedule(cr) + +Note that a registered template schedule can be retrieve by doing + +.. code-block:: python + + xp = cals.get_template("xp") + +which would return the default xp schedule block template for all qubits. +""" + +from .calibrations import Calibrations +from .backend_calibrations import BackendCalibrations +from .exceptions import CalibrationError +from .parameter_value import ParameterValue diff --git a/qiskit_experiments/calibration/backend_calibrations.py b/qiskit_experiments/calibration/backend_calibrations.py new file mode 100644 index 0000000000..99414a8a19 --- /dev/null +++ b/qiskit_experiments/calibration/backend_calibrations.py @@ -0,0 +1,149 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Store and manage the results of calibration experiments in the context of a backend.""" + +from datetime import datetime +from enum import Enum +from typing import List +import copy + +from qiskit.providers.backend import BackendV1 as Backend +from qiskit.circuit import Parameter +from qiskit_experiments.calibration.calibrations import Calibrations, ParameterKey +from qiskit_experiments.calibration.exceptions import CalibrationError + + +class FrequencyElement(Enum): + """An extendable enum for components that have a frequency.""" + + QUBIT = "Qubit" + READOUT = "Readout" + + +class BackendCalibrations(Calibrations): + """ + A Calibrations class to enable a seamless interplay with backend objects. + This class enables users to export their calibrations into a backend object. + Additionally, it creates frequency parameters for qubits and readout resonators. + The parameters are named `qubit_lo_freq` and `meas_lo_freq` to be consistent + with the naming in backend.defaults(). These two parameters are not attached to + any schedule. + """ + + def __init__(self, backend: Backend): + """Setup an instance to manage the calibrations of a backend.""" + super().__init__(backend.configuration().control_channels) + + # Use the same naming convention as in backend.defaults() + self.qubit_freq = Parameter("qubit_lo_freq") + self.meas_freq = Parameter("meas_lo_freq") + self._register_parameter(self.qubit_freq, ()) + self._register_parameter(self.meas_freq, ()) + + self._qubits = set(range(backend.configuration().n_qubits)) + self._backend = backend + + def _get_frequencies( + self, + element: FrequencyElement, + group: str = "default", + cutoff_date: datetime = None, + ) -> List[float]: + """Internal helper method.""" + + if element == FrequencyElement.READOUT: + param = self.meas_freq.name + elif element == FrequencyElement.QUBIT: + param = self.qubit_freq.name + else: + raise CalibrationError(f"Frequency element {element} is not supported.") + + freqs = [] + for qubit in self._qubits: + if ParameterKey(None, param, (qubit,)) in self._params: + freq = self.get_parameter_value(param, (qubit,), None, True, group, cutoff_date) + else: + if element == FrequencyElement.READOUT: + freq = self._backend.defaults().meas_freq_est[qubit] + elif element == FrequencyElement.QUBIT: + freq = self._backend.defaults().qubit_freq_est[qubit] + else: + raise CalibrationError(f"Frequency element {element} is not supported.") + + freqs.append(freq) + + return freqs + + def get_qubit_frequencies( + self, + group: str = "default", + cutoff_date: datetime = None, + ) -> List[float]: + """ + Get the most recent qubit frequencies. They can be passed to the run-time + options of :class:`BaseExperiment`. If no calibrated frequency value of a + qubit is found then the default value from the backend defaults is used. + Only valid parameter values are returned. + + Args: + group: The calibration group from which to draw the + parameters. If not specified, this defaults to the 'default' group. + cutoff_date: Retrieve the most recent parameter up until the cutoff date. Parameters + generated after the cutoff date will be ignored. If the cutoff_date is None then + all parameters are considered. This allows users to discard more recent values + that may be erroneous. + + Returns: + A List of qubit frequencies for all qubits of the backend. + """ + return self._get_frequencies(FrequencyElement.QUBIT, group, cutoff_date) + + def get_meas_frequencies( + self, + group: str = "default", + cutoff_date: datetime = None, + ) -> List[float]: + """ + Get the most recent measurement frequencies. They can be passed to the run-time + options of :class:`BaseExperiment`. If no calibrated frequency value of a + measurement is found then the default value from the backend defaults is used. + Only valid parameter values are returned. + + Args: + group: The calibration group from which to draw the + parameters. If not specified, this defaults to the 'default' group. + cutoff_date: Retrieve the most recent parameter up until the cutoff date. Parameters + generated after the cutoff date will be ignored. If the cutoff_date is None then + all parameters are considered. This allows users to discard more recent values + that may be erroneous. + + Returns: + A List of measurement frequencies for all qubits of the backend. + """ + return self._get_frequencies(FrequencyElement.READOUT, group, cutoff_date) + + def export_backend(self) -> Backend: + """ + Exports the calibrations to a backend object that can be used. + + Returns: + calibrated backend: A backend with the calibrations in it. + """ + backend = copy.deepcopy(self._backend) + + backend.defaults().qubit_freq_est = self.get_qubit_frequencies() + backend.defaults().meas_freq_est = self.get_meas_frequencies() + + # TODO: build the instruction schedule map using the stored calibrations + + return backend diff --git a/qiskit_experiments/calibration/calibrations.py b/qiskit_experiments/calibration/calibrations.py new file mode 100644 index 0000000000..228ff93808 --- /dev/null +++ b/qiskit_experiments/calibration/calibrations.py @@ -0,0 +1,1047 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Class to store and manage the results of calibration experiments.""" + +import os +from collections import namedtuple, defaultdict +from datetime import datetime +from typing import Any, Dict, Set, Tuple, Union, List, Optional +import csv +import dataclasses +import warnings +import regex as re + +from qiskit.pulse import ( + ScheduleBlock, + DriveChannel, + ControlChannel, + MeasureChannel, + Call, + Instruction, + AcquireChannel, + RegisterSlot, + MemorySlot, + Schedule, +) +from qiskit.pulse.channels import PulseChannel +from qiskit.circuit import Parameter, ParameterExpression +from qiskit_experiments.calibration.exceptions import CalibrationError +from qiskit_experiments.calibration.parameter_value import ParameterValue + +ParameterKey = namedtuple("ParameterKey", ["parameter", "qubits", "schedule"]) +ScheduleKey = namedtuple("ScheduleKey", ["schedule", "qubits"]) +ParameterValueType = Union[ParameterExpression, float, int, complex] + + +class Calibrations: + """ + A class to manage schedules with calibrated parameter values. Schedules are + intended to be fully parameterized, including the index of the channels. See + the module-level documentation for extra details. Note that only instances of + ScheduleBlock are supported. + """ + + # The channel indices need to be parameterized following this regex. + __channel_pattern__ = r"^ch\d[\.\d]*\${0,1}[\d]*$" + + def __init__(self, control_config: Dict[Tuple[int, ...], List[ControlChannel]] = None): + """Initialize the calibrations. + + Args: + control_config: A configuration dictionary of any control channels. The + keys are tuples of qubits and the values are a list of ControlChannels + that correspond to the qubits in the keys. + """ + + # Mapping between qubits and their control channels. + self._controls_config = control_config if control_config else {} + + # Store the reverse mapping between control channels and qubits for ease of look-up. + self._controls_config_r = {} + for qubits, channels in self._controls_config.items(): + for channel in channels: + self._controls_config_r[channel] = qubits + + # Dict of the form: (schedule.name, parameter.name, qubits): Parameter + self._parameter_map = {} + + # Reverse mapping of _parameter_map + self._parameter_map_r = defaultdict(set) + + # Default dict of the form: (schedule.name, parameter.name, qubits): [ParameterValue, ...] + self._params = defaultdict(list) + + self._schedules = {} + + # A variable to store all parameter hashes encountered and present them as ordered + # indices to the user. + self._hash_to_counter_map = {} + self._parameter_counter = 0 + + def add_schedule(self, schedule: ScheduleBlock, qubits: Union[int, Tuple[int, ...]] = None): + """Add a schedule block and register its parameters. + + Schedules that use Call instructions must register the called schedules separately. + + Args: + schedule: The :class:`ScheduleBlock` to add. + qubits: The qubits for which to add the schedules. If None or an empty tuple is + given then this schedule is the default schedule for all qubits. + + Raises: + CalibrationError: + - If schedule is not an instance of :class:`ScheduleBlock`. + - If the parameterized channel index is not formatted properly. + - If several parameters in the same schedule have the same name. + - If a channel is parameterized by more than one parameter. + - If the schedule name starts with the prefix of ScheduleBlock. + - If the schedule calls subroutines that have not been registered. + - If a :class:`Schedule` is Called instead of a :class:`ScheduleBlock`. + """ + qubits = self._to_tuple(qubits) + + if not isinstance(schedule, ScheduleBlock): + raise CalibrationError(f"{schedule.name} is not a ScheduleBlock.") + + # check that channels, if parameterized, have the proper name format. + if schedule.name.startswith(ScheduleBlock.prefix): + raise CalibrationError( + f"{self.__class__.__name__} uses the `name` property of the schedule as part of a " + f"database key. Using the automatically generated name {schedule.name} may have " + f"unintended consequences. Please define a meaningful and unique schedule name." + ) + + param_indices = set() + for ch in schedule.channels: + if isinstance(ch.index, Parameter): + if len(ch.index.parameters) != 1: + raise CalibrationError(f"Channel {ch} can only have one parameter.") + + param_indices.add(ch.index) + if re.compile(self.__channel_pattern__).match(ch.index.name) is None: + raise CalibrationError( + f"Parameterized channel must correspond to {self.__channel_pattern__}" + ) + + # Check that subroutines are present. + for block in schedule.blocks: + if isinstance(block, Call): + if isinstance(block.subroutine, Schedule): + raise CalibrationError( + "Calling a Schedule is forbidden, call ScheduleBlock instead." + ) + + if (block.subroutine.name, qubits) not in self._schedules: + raise CalibrationError( + f"Cannot register schedule block {schedule.name} with unregistered " + f"subroutine {block.subroutine.name}." + ) + + # Clean the parameter to schedule mapping. This is needed if we overwrite a schedule. + self._clean_parameter_map(schedule.name, qubits) + + # Add the schedule. + self._schedules[ScheduleKey(schedule.name, qubits)] = schedule + + # Register parameters that are not indices. + # Do not register parameters that are in call instructions. + params_to_register = set() + for inst in self._exclude_calls(schedule, []): + for param in inst.parameters: + if param not in param_indices: + params_to_register.add(param) + + if len(params_to_register) != len(set(param.name for param in params_to_register)): + raise CalibrationError(f"Parameter names in {schedule.name} must be unique.") + + for param in params_to_register: + self._register_parameter(param, qubits, schedule) + + def _exclude_calls( + self, schedule: ScheduleBlock, instructions: List[Instruction] + ) -> List[Instruction]: + """Return the non-Call instructions. + + Recursive function to get all non-Call instructions. This will flatten all blocks + in a :class:`ScheduleBlock` and return the instructions of the ScheduleBlock leaving + out any Call instructions. + + Args: + schedule: A :class:`ScheduleBlock` from which to extract the instructions. + instructions: The list of instructions that is recursively populated. + + Returns: + The list of instructions to which all non-Call instructions have been added. + """ + for block in schedule.blocks: + if isinstance(block, ScheduleBlock): + instructions = self._exclude_calls(block, instructions) + else: + if not isinstance(block, Call): + instructions.append(block) + + return instructions + + def get_template( + self, schedule_name: str, qubits: Optional[Tuple[int, ...]] = None + ) -> ScheduleBlock: + """Get a template schedule. + + Allows the user to get a template schedule that was previously registered. + A template schedule will typically be fully parametric, i.e. all pulse + parameters and channel indices are represented by :class:`Parameter`. + + Args: + schedule_name: The name of the template schedule. + qubits: The qubits under which the template schedule was registered. + + Returns: + The registered template schedule. + + Raises: + CalibrationError: if no template schedule for the given schedule name and qubits + was registered. + """ + key = ScheduleKey(schedule_name, self._to_tuple(qubits)) + + if key in self._schedules: + return self._schedules[key] + + if ScheduleKey(schedule_name, ()) in self._schedules: + return self._schedules[ScheduleKey(schedule_name, ())] + + if qubits: + msg = f"Could not find schedule {schedule_name} on qubits {qubits}." + else: + msg = f"Could not find schedule {schedule_name}." + + raise CalibrationError(msg) + + def remove_schedule(self, schedule: ScheduleBlock, qubits: Union[int, Tuple[int, ...]] = None): + """Remove a schedule that was previously registered. + + Allows users to remove a schedule from the calibrations. The history of the parameters + will remain in the calibrations. + + Args: + schedule: The schedule to remove. + qubits: The qubits for which to remove the schedules. If None is given then this + schedule is the default schedule for all qubits. + """ + qubits = self._to_tuple(qubits) + + if ScheduleKey(schedule.name, qubits) in self._schedules: + del self._schedules[ScheduleKey(schedule.name, qubits)] + + # Clean the parameter to schedule mapping. + self._clean_parameter_map(schedule.name, qubits) + + def _clean_parameter_map(self, schedule_name: str, qubits: Tuple[int, ...]): + """Clean the parameter to schedule mapping for the given schedule, parameter and qubits. + + Args: + schedule_name: The name of the schedule. + qubits: The qubits to which this schedule applies. + + """ + keys_to_remove = [] # of the form (schedule.name, parameter.name, qubits) + for key in self._parameter_map: + if key.schedule == schedule_name and key.qubits == qubits: + keys_to_remove.append(key) + + for key in keys_to_remove: + del self._parameter_map[key] + + # Key set is a set of tuples (schedule.name, parameter.name, qubits) + for param, key_set in self._parameter_map_r.items(): + if key in key_set: + key_set.remove(key) + + # Remove entries that do not point to at least one (schedule.name, parameter.name, qubits) + keys_to_delete = [] + for param, key_set in self._parameter_map_r.items(): + if not key_set: + keys_to_delete.append(param) + + for key in keys_to_delete: + del self._parameter_map_r[key] + + def _register_parameter( + self, + parameter: Parameter, + qubits: Tuple[int, ...], + schedule: ScheduleBlock = None, + ): + """Registers a parameter for the given schedule. + + This method allows self to determine the parameter instance that corresponds to the given + schedule name, parameter name and qubits. + + Args: + parameter: The parameter to register. + qubits: The qubits for which to register the parameter. + schedule: The schedule to which this parameter belongs. The schedule can + be None which allows the calibration to accommodate, e.g. qubit frequencies. + """ + if parameter not in self._hash_to_counter_map: + self._hash_to_counter_map[parameter] = self._parameter_counter + self._parameter_counter += 1 + + sched_name = schedule.name if schedule else None + key = ParameterKey(parameter.name, qubits, sched_name) + self._parameter_map[key] = parameter + self._parameter_map_r[parameter].add(key) + + @property + def parameters(self) -> Dict[Parameter, Set[ParameterKey]]: + """Return a mapping between parameters and parameter keys. + + Returns a dictionary mapping parameters managed by the calibrations to the schedules and + qubits and parameter names using the parameters. The values of the dict are sets containing + the parameter keys. Parameters that are not attached to a schedule will have None in place + of a schedule name. + """ + return self._parameter_map_r + + def calibration_parameter( + self, + parameter_name: str, + qubits: Union[int, Tuple[int, ...]] = None, + schedule_name: str = None, + ) -> Parameter: + """Return a parameter given its keys. + + Returns a Parameter object given the triplet parameter_name, qubits and schedule_name + which uniquely determine the context of a parameter. + + Args: + parameter_name: Name of the parameter to get. + qubits: The qubits to which this parameter belongs. If qubits is None then + the default scope is assumed and the key will be an empty tuple. + schedule_name: The name of the schedule to which this parameter belongs. A + parameter may not belong to a schedule in which case None is accepted. + + Returns: + calibration parameter: The parameter that corresponds to the given arguments. + + Raises: + CalibrationError: If the desired parameter is not found. + """ + qubits = self._to_tuple(qubits) + + # 1) Check for qubit specific parameters. + if ParameterKey(parameter_name, qubits, schedule_name) in self._parameter_map: + return self._parameter_map[ParameterKey(parameter_name, qubits, schedule_name)] + + # 2) Check for default parameters. + elif ParameterKey(parameter_name, (), schedule_name) in self._parameter_map: + return self._parameter_map[ParameterKey(parameter_name, (), schedule_name)] + else: + raise CalibrationError( + f"No parameter for {parameter_name} and schedule {schedule_name} " + f"and qubits {qubits}. No default value exists." + ) + + def add_parameter_value( + self, + value: Union[int, float, complex, ParameterValue], + param: Union[Parameter, str], + qubits: Union[int, Tuple[int, ...]] = None, + schedule: Union[ScheduleBlock, str] = None, + ): + """Add a parameter value to the stored parameters. + + This parameter value may be applied to several channels, for instance, all + DRAG pulses may have the same standard deviation. + + Args: + value: The value of the parameter to add. If an int, float, or complex is given + then the timestamp of the parameter value will automatically be generated + and set to the current time. + param: The parameter or its name for which to add the measured value. + qubits: The qubits to which this parameter applies. + schedule: The schedule or its name for which to add the measured parameter value. + + Raises: + CalibrationError: If the schedule name is given but no schedule with that name + exists. + """ + qubits = self._to_tuple(qubits) + + if isinstance(value, (int, float, complex)): + value = ParameterValue(value, datetime.now()) + + param_name = param.name if isinstance(param, Parameter) else param + sched_name = schedule.name if isinstance(schedule, ScheduleBlock) else schedule + + registered_schedules = set(key.schedule for key in self._schedules) + + if sched_name and sched_name not in registered_schedules: + raise CalibrationError(f"Schedule named {sched_name} was never registered.") + + self._params[ParameterKey(param_name, qubits, sched_name)].append(value) + + def _get_channel_index(self, qubits: Tuple[int, ...], chan: PulseChannel) -> int: + """Get the index of the parameterized channel. + + The return index is determined from the given qubits and the name of the parameter + in the channel index. The name of this parameter for control channels must be written + as chqubit_index1.qubit_index2... followed by an optional $index. For example, the + following parameter names are valid: 'ch1', 'ch1.0', 'ch30.12', and 'ch1.0$1'. + + Args: + qubits: The qubits for which we want to obtain the channel index. + chan: The channel with a parameterized name. + + Returns: + index: The index of the channel. For example, if qubits=(10, 32) and + chan is a control channel with parameterized index name 'ch1.0' + the method returns the control channel corresponding to + qubits (qubits[1], qubits[0]) which is here the control channel of + qubits (32, 10). + + Raises: + CalibrationError: + - If the number of qubits is incorrect. + - If the number of inferred ControlChannels is not correct. + - If ch is not a DriveChannel, MeasureChannel, or ControlChannel. + """ + if isinstance(chan.index, Parameter): + if isinstance( + chan, (DriveChannel, MeasureChannel, AcquireChannel, RegisterSlot, MemorySlot) + ): + index = int(chan.index.name[2:].split("$")[0]) + + if len(qubits) <= index: + raise CalibrationError(f"Not enough qubits given for channel {chan}.") + + return qubits[index] + + # Control channels name example ch1.0$1 + if isinstance(chan, ControlChannel): + + channel_index_parts = chan.index.name[2:].split("$") + qubit_channels = channel_index_parts[0] + + indices = [int(sub_channel) for sub_channel in qubit_channels.split(".")] + ch_qubits = tuple(qubits[index] for index in indices) + chs_ = self._controls_config[ch_qubits] + + control_index = 0 + if len(channel_index_parts) == 2: + control_index = int(channel_index_parts[1]) + + if len(chs_) <= control_index: + raise CalibrationError( + f"Control channel index {control_index} not found for qubits {qubits}." + ) + + return chs_[control_index].index + + raise CalibrationError( + f"{chan} must be a sub-type of {PulseChannel} or an {AcquireChannel}, " + f"{RegisterSlot}, or a {MemorySlot}." + ) + + return chan.index + + def get_parameter_value( + self, + param: Union[Parameter, str], + qubits: Union[int, Tuple[int, ...]], + schedule: Union[ScheduleBlock, str, None] = None, + valid_only: bool = True, + group: str = "default", + cutoff_date: datetime = None, + ) -> Union[int, float, complex]: + """Retrieves the value of a parameter. + + Parameters may be linked. get_parameter_value does the following steps: + 1) Retrieve the parameter object corresponding to (param, qubits, schedule) + 2) The values of this parameter may be stored under another schedule since + schedules can share parameters. To deal with this, a list of candidate keys + is created internally based on the current configuration. + 3) Look for candidate parameter values under the candidate keys. + 4) Filter the candidate parameter values according to their date (up until the + cutoff_date), validity and calibration group. + 5) Return the most recent parameter. + + Args: + param: The parameter or the name of the parameter for which to get the parameter value. + qubits: The qubits for which to get the value of the parameter. + schedule: The schedule or its name for which to get the parameter value. + valid_only: Use only parameters marked as valid. + group: The calibration group from which to draw the parameters. + If not specified this defaults to the 'default' group. + cutoff_date: Retrieve the most recent parameter up until the cutoff date. Parameters + generated after the cutoff date will be ignored. If the cutoff_date is None then + all parameters are considered. This allows users to discard more recent values that + may be erroneous. + + Returns: + value: The value of the parameter. + + Raises: + CalibrationError: + - If there is no parameter value for the given parameter name and pulse channel. + """ + qubits = self._to_tuple(qubits) + + # 1) Identify the parameter object. + param_name = param.name if isinstance(param, Parameter) else param + sched_name = schedule.name if isinstance(schedule, ScheduleBlock) else schedule + + param = self.calibration_parameter(param_name, qubits, sched_name) + + # 2) Get a list of candidate keys restricted to the qubits of interest. + candidate_keys = [] + for key in self._parameter_map_r[param]: + candidate_keys.append(ParameterKey(key.parameter, qubits, key.schedule)) + + # 3) Loop though the candidate keys to candidate values + candidates = [] + for key in candidate_keys: + if key in self._params: + candidates += self._params[key] + + # If no candidate parameter values were found look for default parameters + # i.e. parameters that do not specify a qubit. + if len(candidates) == 0: + for key in candidate_keys: + if ParameterKey(key.parameter, (), key.schedule) in self._params: + candidates += self._params[ParameterKey(key.parameter, (), key.schedule)] + + # 4) Filter candidate parameter values. + if valid_only: + candidates = [val for val in candidates if val.valid] + + candidates = [val for val in candidates if val.group == group] + + if cutoff_date: + candidates = [val for val in candidates if val.date_time <= cutoff_date] + + if len(candidates) == 0: + msg = f"No candidate parameter values for {param_name} in calibration group {group} " + + if qubits: + msg += f"on qubits {qubits} " + + if sched_name: + msg += f"in schedule {sched_name} " + + if cutoff_date: + msg += f"with cutoff date: {cutoff_date}" + + raise CalibrationError(msg) + + # 5) Return the most recent parameter. + return max(candidates, key=lambda x: x.date_time).value + + def get_schedule( + self, + name: str, + qubits: Union[int, Tuple[int, ...]], + assign_params: Dict[Union[str, ParameterKey], ParameterValueType] = None, + group: Optional[str] = "default", + cutoff_date: datetime = None, + ) -> ScheduleBlock: + """Get the template schedule with parameters assigned to values. + + All the parameters in the template schedule block will be assigned to the values managed + by the calibrations unless they are specified in assign_params. In this case the value in + assign_params will override the value stored by the calibrations. A parameter value in + assign_params may also be a :class:`ParameterExpression`. + + .. code-block:: python + + # Get an xp schedule with a parametric amplitude + sched = cals.get_schedule("xp", 3, assign_params={"amp": Parameter("amp")}) + + # Get an echoed-cross-resonance schedule between qubits (0, 2) where the xp echo gates + # are Called schedules but leave their amplitudes as parameters. + assign_dict = {("amp", (0,), "xp"): Parameter("my_amp")} + sched = cals.get_schedule("cr", (0, 2), assign_params=assign_dict) + + Args: + name: The name of the schedule to get. + qubits: The qubits for which to get the schedule. + assign_params: The parameters to assign manually. Each parameter is specified by a + ParameterKey which is a named tuple of the form (parameter name, qubits, + schedule name). Each entry in assign_params can also be a string corresponding + to the name of the parameter. In this case, the schedule name and qubits of the + corresponding ParameterKey will be the name and qubits given as arguments to + get_schedule. + group: The calibration group from which to draw the parameters. If not specified + this defaults to the 'default' group. + cutoff_date: Retrieve the most recent parameter up until the cutoff date. Parameters + generated after the cutoff date will be ignored. If the cutoff_date is None then + all parameters are considered. This allows users to discard more recent values that + may be erroneous. + + Returns: + schedule: A copy of the template schedule with all parameters assigned + except for those specified by assign_params. + + Raises: + CalibrationError: + - If the name of the schedule is not known. + - If a parameter could not be found. + """ + qubits = self._to_tuple(qubits) + + # Standardize the input in the assignment dictionary + if assign_params: + assign_params_ = dict() + for assign_param, value in assign_params.items(): + if isinstance(assign_param, str): + assign_params_[ParameterKey(assign_param, qubits, name)] = value + else: + assign_params_[ParameterKey(*assign_param)] = value + + assign_params = assign_params_ + else: + assign_params = dict() + + # Get the template schedule + if (name, qubits) in self._schedules: + schedule = self._schedules[ScheduleKey(name, qubits)] + elif (name, ()) in self._schedules: + schedule = self._schedules[ScheduleKey(name, ())] + else: + raise CalibrationError(f"Schedule {name} is not defined for qubits {qubits}.") + + # Retrieve the channel indices based on the qubits and bind them. + binding_dict = {} + for ch in schedule.channels: + if ch.is_parameterized(): + binding_dict[ch.index] = self._get_channel_index(qubits, ch) + + # Binding the channel indices makes it easier to deal with parameters later on + schedule = schedule.assign_parameters(binding_dict, inplace=False) + + # Now assign the other parameters + assigned_schedule = self._assign(schedule, qubits, assign_params, group, cutoff_date) + + free_params = set() + for param in assign_params.values(): + if isinstance(param, ParameterExpression): + free_params.add(param) + + if len(assigned_schedule.parameters) != len(free_params): + raise CalibrationError( + f"The number of free parameters {len(assigned_schedule.parameters)} in " + f"the assigned schedule differs from the requested number of free " + f"parameters {len(free_params)}." + ) + + return assigned_schedule + + def _assign( + self, + schedule: ScheduleBlock, + qubits: Tuple[int, ...], + assign_params: Dict[Union[str, ParameterKey], ParameterValueType], + group: Optional[str] = "default", + cutoff_date: datetime = None, + ) -> ScheduleBlock: + """Recursively assign parameters in a schedule. + + The recursive behaviour is needed to handle Call instructions as the name of + the called instruction defines the scope of the parameter. Each time a Call + is found _assign recurses on the channel-assigned subroutine of the Call + instruction and the qubits that are in said subroutine. This requires a + careful extraction of the qubits from the subroutine and in the appropriate + order. Next, the parameters are identified and assigned. This is needed to + handle situations where the same parameterized schedule is called but on + different channels. For example, + + .. code-block:: python + + ch0 = Parameter("ch0") + ch1 = Parameter("ch1") + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(duration, amp, sigma), DriveChannel(ch0)) + + with pulse.build(name="xt_xp") as xt: + pulse.call(xp) + pulse.call(xp, value_dict={ch0: ch1}) + + Here, we define the xp :class:`ScheduleBlock` for all qubits as a Gaussian. Next, we define + a schedule where both xp schedules are called simultaneously on different channels. We now + explain a subtlety related to manually assigning values in the case above. In the schedule + above, the parameters of the Gaussian pulses are coupled, e.g. the xp pulse on ch0 and ch1 + share the same instance of :class:`ParameterExpression`. Suppose now that both pulses have + a duration and sigma of 160 and 40 samples, respectively, and that the amplitudes are 0.5 + and 0.3 for qubits 0 and 2, respectively. These values are stored in self._params. When + retrieving a schedule without specifying assign_params, i.e. + + .. code-block:: python + + cals.get_schedule("xt_xp", (0, 2)) + + we will obtain the expected schedule with amplitudes 0.5 and 0.3. When specifying the + following :code:`assign_params = {("amp", (0,), "xp"): Parameter("my_new_amp")}` we + will obtain a schedule where the amplitudes of the xp pulse on qubit 0 is set to + :code:`Parameter("my_new_amp")`. The amplitude of the xp pulse on qubit 2 is set to + the value stored by the calibrations, i.e. 0.3. + + .. code-bloc:: python + + cals.get_schedule( + "xt_xp", + (0, 2), + assign_params = {("amp", (0,), "xp"): Parameter("my_new_amp")} + ) + + Args: + schedule: The schedule with assigned channel indices for which we wish to + assign values to non-channel parameters. + qubits: The qubits for which to get the schedule. + assign_params: The parameters to manually assign. See get_schedules for details. + group: The calibration group of the parameters. + cutoff_date: Retrieve the most recent parameter up until the cutoff date. Parameters + generated after the cutoff date will be ignored. If the cutoff_date is None then + all parameters are considered. This allows users to discard more recent values that + may be erroneous. + + Returns: + ret_schedule: The schedule with assigned parameters. + + Raises: + CalibrationError: + - If a channel has not been assigned. + - If there is an ambiguous parameter assignment. + - If there are inconsistencies between a called schedule and the template + schedule registered under the name of the called schedule. + """ + # 1) Restrict the given qubits to those in the given schedule. + qubit_set = set() + for chan in schedule.channels: + if isinstance(chan.index, ParameterExpression): + raise ( + CalibrationError( + f"All parametric channels must be assigned before searching for " + f"non-channel parameters. {chan} is parametric." + ) + ) + if isinstance(chan, (DriveChannel, MeasureChannel)): + qubit_set.add(chan.index) + + if isinstance(chan, ControlChannel): + for qubit in self._controls_config_r[chan]: + qubit_set.add(qubit) + + qubits_ = tuple(qubit for qubit in qubits if qubit in qubit_set) + + # 2) Recursively assign the parameters in the called instructions. + ret_schedule = ScheduleBlock( + alignment_context=schedule.alignment_context, + name=schedule.name, + metadata=schedule.metadata, + ) + + for inst in schedule.blocks: + if isinstance(inst, Call): + # Check that there are no inconsistencies with the called subroutines. + template_subroutine = self.get_template(inst.subroutine.name, qubits_) + if inst.subroutine != template_subroutine: + raise CalibrationError( + f"The subroutine {inst.subroutine.name} called by {inst.name} does not " + f"match the template schedule stored under {template_subroutine.name}." + ) + + inst = inst.assigned_subroutine() + + if isinstance(inst, ScheduleBlock): + inst = self._assign(inst, qubits_, assign_params, group, cutoff_date) + + ret_schedule.append(inst, inplace=True) + + # 3) Get the parameter keys of the remaining instructions. At this point in + # _assign all parameters in Call instructions that are supposed to be + # assigned have been assigned. + keys = set() + + if ret_schedule.name in set(key.schedule for key in self._parameter_map): + for param in ret_schedule.parameters: + keys.add(ParameterKey(param.name, qubits_, ret_schedule.name)) + + # 4) Build the parameter binding dictionary. + binding_dict = {} + assignment_table = {} + for key, value in assign_params.items(): + key_orig = key + if key.qubits == (): + key = ParameterKey(key.parameter, qubits_, key.schedule) + if key in assign_params: + # if (param, (1,), sched) and (param, (), sched) are both + # in assign_params, skip the default value instead of + # possibly triggering an error about conflicting + # parameters. + continue + elif key.qubits != qubits_: + continue + param = self.calibration_parameter(*key) + if param in ret_schedule.parameters: + assign_okay = ( + param not in binding_dict + or key.schedule == ret_schedule.name + and assignment_table[param].schedule != ret_schedule.name + ) + if assign_okay: + binding_dict[param] = value + assignment_table[param] = key_orig + elif ( + key.schedule == ret_schedule.name + or assignment_table[param].schedule != ret_schedule.name + ) and binding_dict[param] != value: + raise CalibrationError( + "Ambiguous assignment: assign_params keys " + f"{key_orig} and {assignment_table[param]} " + "resolve to the same parameter." + ) + + for key in keys: + # Get the parameter object. Since we are dealing with a schedule the name of + # the schedule is always defined. However, the parameter may be a default + # parameter for all qubits, i.e. qubits may be an empty tuple. + param = self.calibration_parameter(*key) + + if param not in binding_dict: + binding_dict[param] = self.get_parameter_value( + key.parameter, + key.qubits, + key.schedule, + group=group, + cutoff_date=cutoff_date, + ) + + return ret_schedule.assign_parameters(binding_dict, inplace=False) + + def schedules(self) -> List[Dict[str, Any]]: + """Return the managed schedules in a list of dictionaries. + + Returns: + data: A list of dictionaries with all the schedules in it. The key-value pairs are + - 'qubits': the qubits to which this schedule applies. This may be () if the + schedule is the default for all qubits. + - 'schedule': The schedule. + - 'parameters': The parameters in the schedule exposed for convenience. + This list of dictionaries can easily be converted to a data frame. + """ + data = [] + for key, sched in self._schedules.items(): + data.append({"qubits": key.qubits, "schedule": sched, "parameters": sched.parameters}) + + return data + + def parameters_table( + self, + parameters: List[str] = None, + qubit_list: List[Tuple[int, ...]] = None, + schedules: List[Union[ScheduleBlock, str]] = None, + ) -> List[Dict[str, Any]]: + """A convenience function to help users visualize the values of their parameter. + + Args: + parameters: The parameter names that should be included in the returned + table. If None is given then all names are included. + qubit_list: The qubits that should be included in the returned table. + If None is given then all channels are returned. + schedules: The schedules to which to restrict the output. + + Returns: + data: A list of dictionaries with parameter values and metadata which can + easily be converted to a data frame. + """ + if qubit_list: + qubit_list = [self._to_tuple(qubits) for qubits in qubit_list] + + data = [] + + # Convert inputs to lists of strings + if schedules is not None: + schedules = {sdl.name if isinstance(sdl, ScheduleBlock) else sdl for sdl in schedules} + + # Look for exact matches. Default values will be ignored. + keys = set() + for key in self._params.keys(): + if parameters and key.parameter not in parameters: + continue + if schedules and key.schedule not in schedules: + continue + if qubit_list and key.qubits not in qubit_list: + continue + + keys.add(key) + + for key in keys: + for value in self._params[key]: + value_dict = dataclasses.asdict(value) + value_dict["qubits"] = key.qubits + value_dict["parameter"] = key.parameter + value_dict["schedule"] = key.schedule + + data.append(value_dict) + + return data + + def save(self, file_type: str = "csv", folder: str = None, overwrite: bool = False): + """Save the parameterized schedules and parameter value. + + The schedules and parameter values can be stored in csv files. This method creates + three files: + - parameter_config.csv: This file stores a table of parameters which indicates + which parameters appear in which schedules. + - parameter_values.csv: This file stores the values of the calibrated parameters. + - schedules.csv: This file stores the parameterized schedules. + + Warning: + Schedule blocks will only be saved in string format and can therefore not be + reloaded and must instead be rebuilt. + + Args: + file_type: The type of file to which to save. By default this is a csv. + Other file types may be supported in the future. + folder: The folder in which to save the calibrations. + overwrite: If the files already exist then they will not be overwritten + unless overwrite is set to True. + + Raises: + CalibrationError: if the files exist and overwrite is not set to True. + """ + warnings.warn("Schedules are only saved in text format. They cannot be re-loaded.") + + cwd = os.getcwd() + if folder: + os.chdir(folder) + + if os.path.isfile("parameter_config.csv") and not overwrite: + raise CalibrationError("parameter_config.csv already exists. Set overwrite to True.") + + if os.path.isfile("parameter_values.csv") and not overwrite: + raise CalibrationError("parameter_values.csv already exists. Set overwrite to True.") + + if os.path.isfile("parameter_values.csv") and not overwrite: + raise CalibrationError("schedules.csv already exists. Set overwrite to True.") + + # Write the parameter configuration. + header_keys = ["parameter.name", "parameter unique id", "schedule", "qubits"] + body = [] + + for parameter, keys in self.parameters.items(): + for key in keys: + body.append( + { + "parameter.name": parameter.name, + "parameter unique id": self._hash_to_counter_map[parameter], + "schedule": key.schedule, + "qubits": key.qubits, + } + ) + + if file_type == "csv": + with open("parameter_config.csv", "w", newline="", encoding="utf-8") as output_file: + dict_writer = csv.DictWriter(output_file, header_keys) + dict_writer.writeheader() + dict_writer.writerows(body) + + # Write the values of the parameters. + values = self.parameters_table() + if len(values) > 0: + header_keys = values[0].keys() + + with open("parameter_values.csv", "w", newline="", encoding="utf-8") as output_file: + dict_writer = csv.DictWriter(output_file, header_keys) + dict_writer.writeheader() + dict_writer.writerows(values) + + # Serialize the schedules. For now we just print them. + schedules = [] + header_keys = ["name", "qubits", "schedule"] + for key, sched in self._schedules.items(): + schedules.append( + {"name": key.schedule, "qubits": key.qubits, "schedule": str(sched)} + ) + + with open("schedules.csv", "w", newline="", encoding="utf-8") as output_file: + dict_writer = csv.DictWriter(output_file, header_keys) + dict_writer.writeheader() + dict_writer.writerows(schedules) + + else: + raise CalibrationError(f"Saving to .{file_type} is not yet supported.") + + os.chdir(cwd) + + def load_parameter_values(self, file_name: str = "parameter_values.csv"): + """ + Load parameter values from a given file into self._params. + + Args: + file_name: The name of the file that stores the parameters. Will default to + parameter_values.csv. + """ + with open(file_name, encoding="utf-8") as fp: + reader = csv.DictReader(fp, delimiter=",", quotechar='"') + + for row in reader: + param_val = ParameterValue( + row["value"], row["date_time"], row["valid"], row["exp_id"], row["group"] + ) + key = ParameterKey(row["parameter"], self._to_tuple(row["qubits"]), row["schedule"]) + self.add_parameter_value(param_val, *key) + + @classmethod + def load(cls, files: List[str]) -> "Calibrations": + """ + Retrieves the parameterized schedules and pulse parameters from the + given location. + """ + raise CalibrationError("Full calibration loading is not implemented yet.") + + @staticmethod + def _to_tuple(qubits: Union[str, int, Tuple[int, ...]]) -> Tuple[int, ...]: + """Ensure that qubits is a tuple of ints. + + Args: + qubits: An int, a tuple of ints, or a string representing a tuple of ints. + + Returns: + qubits: A tuple of ints. + + Raises: + CalibrationError: If the given input does not conform to an int or + tuple of ints. + """ + if not qubits: + return tuple() + + if isinstance(qubits, str): + try: + return tuple(int(qubit) for qubit in qubits.strip("( )").split(",") if qubit != "") + except ValueError: + pass + + if isinstance(qubits, int): + return (qubits,) + + if isinstance(qubits, tuple): + if all(isinstance(n, int) for n in qubits): + return qubits + + raise CalibrationError( + f"{qubits} must be int, tuple of ints, or str that can be parsed" + f"to a tuple if ints. Received {qubits}." + ) diff --git a/qiskit_experiments/calibration/exceptions.py b/qiskit_experiments/calibration/exceptions.py new file mode 100644 index 0000000000..96e00667a8 --- /dev/null +++ b/qiskit_experiments/calibration/exceptions.py @@ -0,0 +1,19 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Exceptions for calibration.""" + +from qiskit.exceptions import QiskitError + + +class CalibrationError(QiskitError): + """Errors raised by the calibration module.""" diff --git a/qiskit_experiments/calibration/parameter_value.py b/qiskit_experiments/calibration/parameter_value.py new file mode 100644 index 0000000000..baa470f606 --- /dev/null +++ b/qiskit_experiments/calibration/parameter_value.py @@ -0,0 +1,103 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Data class for parameter values.""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Union + +from qiskit_experiments.calibration.exceptions import CalibrationError + + +@dataclass +class ParameterValue: + """A data class to store parameter values.""" + + # Value assumed by the parameter + value: Union[int, float, complex] = None + + # Data time when the value of the parameter was generated + date_time: datetime = datetime.fromtimestamp(0) + + # A bool indicating if the parameter is valid + valid: bool = True + + # The experiment from which the value of this parameter was generated. + exp_id: str = None + + # The group of calibrations to which this parameter belongs + group: str = "default" + + def __post_init__(self): + """ + Ensure that the variables in self have the proper types. This allows + us to give strings to self.__init__ as input which is useful when loading + serialized parameter values. + """ + if isinstance(self.valid, str): + if self.valid == "True": + self.valid = True + else: + self.valid = False + + if isinstance(self.value, str): + self.value = self._validated_value(self.value) + + if isinstance(self.date_time, str): + self.date_time = datetime.strptime(self.date_time, "%Y-%m-%d %H:%M:%S") + + if not isinstance(self.value, (int, float, complex)): + raise CalibrationError(f"Values {self.value} must be int, float or complex.") + + if not isinstance(self.date_time, datetime): + raise CalibrationError(f"Datetime {self.date_time} must be a datetime.") + + if not isinstance(self.valid, bool): + raise CalibrationError(f"Valid {self.valid} is not a boolean.") + + if self.exp_id and not isinstance(self.exp_id, str): + raise CalibrationError(f"Experiment id {self.exp_id} is not a string.") + + if not isinstance(self.group, str): + raise CalibrationError(f"Group {self.group} is not a string.") + + @staticmethod + def _validated_value(value: str) -> Union[int, float, complex]: + """ + Convert the string representation of value to the correct type. + + Args: + value: The string to convert to either an int, float, or complex. + + Returns: + value converted to either int, float, or complex. + + Raises: + CalibrationError: If the conversion fails. + """ + try: + return int(value) + except ValueError: + pass + + try: + return float(value) + except ValueError: + pass + + try: + return complex(value) + except ValueError as val_err: + raise CalibrationError( + f"Could not convert {value} to int, float, or complex." + ) from val_err diff --git a/test/calibration/__init__.py b/test/calibration/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/test/calibration/test_backend_calibrations.py b/test/calibration/test_backend_calibrations.py new file mode 100644 index 0000000000..08c820877e --- /dev/null +++ b/test/calibration/test_backend_calibrations.py @@ -0,0 +1,28 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Class to test the backend calibrations.""" + +from qiskit.test import QiskitTestCase +from qiskit.test.mock import FakeArmonk +from qiskit_experiments.calibration.backend_calibrations import BackendCalibrations + + +class TestBackendCalibrations(QiskitTestCase): + """Class to test the functionality of a BackendCalibrations""" + + def test_run_options(self): + """Test that we can get run options.""" + cals = BackendCalibrations(FakeArmonk()) + + self.assertEqual(cals.get_meas_frequencies(), [6993370669.000001]) + self.assertEqual(cals.get_qubit_frequencies(), [4971852852.405576]) diff --git a/test/calibration/test_calibrations.py b/test/calibration/test_calibrations.py new file mode 100644 index 0000000000..da7e085a57 --- /dev/null +++ b/test/calibration/test_calibrations.py @@ -0,0 +1,1307 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 2021. +# +# This code is licensed under the Apache License, Version 2.0. You may +# obtain a copy of this license in the LICENSE.txt file in the root directory +# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. +# +# Any modifications or derivative works of this code must retain this +# copyright notice, and modified files need to carry a notice indicating +# that they have been altered from the originals. + +"""Class to test the calibrations.""" + +import os +from collections import defaultdict +from datetime import datetime +from qiskit.circuit import Parameter +from qiskit.pulse import ( + Drag, + DriveChannel, + ControlChannel, + AcquireChannel, + Gaussian, + GaussianSquare, + MeasureChannel, + RegisterSlot, + Play, +) +from qiskit.pulse.transforms import inline_subroutines, block_to_schedule +import qiskit.pulse as pulse +from qiskit.test import QiskitTestCase +from qiskit_experiments.calibration.calibrations import Calibrations, ParameterKey +from qiskit_experiments.calibration.parameter_value import ParameterValue +from qiskit_experiments.calibration.exceptions import CalibrationError + + +class TestCalibrationsBasic(QiskitTestCase): + """Class to test the management of schedules and parameters for calibrations.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + self.cals = Calibrations() + + self.sigma = Parameter("σ") + self.amp_xp = Parameter("amp") + self.amp_x90p = Parameter("amp") + self.amp_y90p = Parameter("amp") + self.beta = Parameter("β") + self.chan = Parameter("ch0") + self.drive = DriveChannel(self.chan) + self.duration = Parameter("dur") + + # Define and add template schedules. + with pulse.build(name="xp") as xp: + pulse.play(Drag(self.duration, self.amp_xp, self.sigma, self.beta), self.drive) + + with pulse.build(name="xm") as xm: + pulse.play(Drag(self.duration, -self.amp_xp, self.sigma, self.beta), self.drive) + + with pulse.build(name="x90p") as x90p: + pulse.play(Drag(self.duration, self.amp_x90p, self.sigma, self.beta), self.drive) + + with pulse.build(name="y90p") as y90p: + pulse.play(Drag(self.duration, self.amp_y90p, self.sigma, self.beta), self.drive) + + for sched in [xp, x90p, y90p, xm]: + self.cals.add_schedule(sched) + + self.xm_pulse = xm + + # Add some parameter values. + self.date_time = datetime.strptime("15/09/19 10:21:35", "%d/%m/%y %H:%M:%S") + + self.cals.add_parameter_value(ParameterValue(40, self.date_time), "σ", schedule="xp") + self.cals.add_parameter_value(ParameterValue(160, self.date_time), "dur", schedule="xp") + self.cals.add_parameter_value(ParameterValue(0.2, self.date_time), "amp", 3, "xp") + self.cals.add_parameter_value(ParameterValue(0.1, self.date_time), "amp", (3,), "x90p") + self.cals.add_parameter_value(ParameterValue(0.08, self.date_time), "amp", (3,), "y90p") + self.cals.add_parameter_value(ParameterValue(40, self.date_time), "β", (3,), "xp") + + def test_setup(self): + """Test that the initial setup behaves as expected.""" + expected = {ParameterKey("amp", (), "xp"), ParameterKey("amp", (), "xm")} + self.assertEqual(self.cals.parameters[self.amp_xp], expected) + + expected = {ParameterKey("amp", (), "x90p")} + self.assertEqual(self.cals.parameters[self.amp_x90p], expected) + + expected = {ParameterKey("amp", (), "y90p")} + self.assertEqual(self.cals.parameters[self.amp_y90p], expected) + + expected = { + ParameterKey("β", (), "xp"), + ParameterKey("β", (), "xm"), + ParameterKey("β", (), "x90p"), + ParameterKey("β", (), "y90p"), + } + self.assertEqual(self.cals.parameters[self.beta], expected) + + expected = { + ParameterKey("σ", (), "xp"), + ParameterKey("σ", (), "xm"), + ParameterKey("σ", (), "x90p"), + ParameterKey("σ", (), "y90p"), + } + self.assertEqual(self.cals.parameters[self.sigma], expected) + + self.assertEqual(self.cals.get_parameter_value("amp", (3,), "xp"), 0.2) + self.assertEqual(self.cals.get_parameter_value("amp", (3,), "xm"), 0.2) + self.assertEqual(self.cals.get_parameter_value("amp", 3, "x90p"), 0.1) + self.assertEqual(self.cals.get_parameter_value("amp", 3, "y90p"), 0.08) + + def test_preserve_template(self): + """Test that the template schedule is still fully parametric after we get a schedule.""" + + # First get a schedule + xp = self.cals.get_schedule("xp", (3,)) + self.assertEqual(xp.instructions[0][1].operands[0].amp, 0.2) + + # Find the template schedule for xp and test it. + schedule = pulse.Schedule() + for sched_dict in self.cals.schedules(): + if sched_dict["schedule"].name == "xp": + schedule = sched_dict["schedule"] + + for param in {self.amp_xp, self.sigma, self.beta, self.duration, self.chan}: + self.assertTrue(param in schedule.parameters) + + self.assertEqual(len(schedule.parameters), 5) + self.assertEqual(len(schedule.blocks), 1) + + def test_remove_schedule(self): + """Test that we can easily remove a schedule.""" + + self.assertEqual(len(self.cals.schedules()), 4) + + self.cals.remove_schedule(self.xm_pulse) + + # Removing xm should remove the schedule but not the parameters as they are shared. + self.assertEqual(len(self.cals.schedules()), 3) + for param in [self.sigma, self.amp_xp, self.amp_x90p, self.amp_y90p, self.beta]: + self.assertTrue(param in self.cals.parameters) + + # Add a schedule with a different parameter and then remove it + with pulse.build(name="error") as sched: + pulse.play(Gaussian(160, Parameter("xyz"), 40), DriveChannel(Parameter("ch0"))) + + self.cals.add_schedule(sched) + + self.assertEqual(len(self.cals.schedules()), 4) + self.assertEqual(len(self.cals.parameters), 7) + + self.cals.remove_schedule(sched) + + self.assertEqual(len(self.cals.schedules()), 3) + self.assertEqual(len(self.cals.parameters), 6) + for param in [self.sigma, self.amp_xp, self.amp_x90p, self.amp_y90p, self.beta]: + self.assertTrue(param in self.cals.parameters) + + def test_parameter_dependency(self): + """Check that two schedules that share the same parameter are simultaneously updated.""" + + xp = self.cals.get_schedule("xp", (3,)) + self.assertEqual(xp.instructions[0][1].operands[0].amp, 0.2) + + xm = self.cals.get_schedule("xm", (3,)) + self.assertEqual(xm.instructions[0][1].operands[0].amp, -0.2) + + self.cals.add_parameter_value(ParameterValue(0.25, datetime.now()), "amp", (3,), "xp") + + xp = self.cals.get_schedule("xp", (3,)) + self.assertEqual(xp.instructions[0][1].operands[0].amp, 0.25) + + xm = self.cals.get_schedule("xm", (3,)) + self.assertEqual(xm.instructions[0][1].operands[0].amp, -0.25) + + def test_get_value(self): + """Test the retrieve of parameter values.""" + + self.assertEqual(self.cals.get_parameter_value("amp", (3,), "xp"), 0.2) + self.assertEqual(self.cals.get_parameter_value("amp", (3,), "x90p"), 0.1) + + self.assertEqual(self.cals.get_parameter_value("σ", (3,), "x90p"), 40) + self.assertEqual(self.cals.get_parameter_value("σ", (3,), "xp"), 40) + + self.cals.add_parameter_value(ParameterValue(50, datetime.now()), "σ", (3,), "xp") + self.assertEqual(self.cals.get_parameter_value("σ", (3,), "x90p"), 50) + self.assertEqual(self.cals.get_parameter_value("σ", (3,), "xp"), 50) + + def test_channel_names(self): + """Check the naming of parametric control channels index1.index2.index3...""" + drive_0 = DriveChannel(Parameter("ch0")) + drive_1 = DriveChannel(Parameter("ch1")) + control_bad = ControlChannel(Parameter("u_chan")) + control_good = ControlChannel(Parameter("ch1.0")) + + with pulse.build(name="good_sched") as sched_good: + pulse.play(Drag(160, 0.1, 40, 2), drive_0) + pulse.play(Drag(160, 0.1, 40, 2), drive_1) + pulse.play(Drag(160, 0.1, 40, 2), control_good) + + with pulse.build(name="bad_sched") as sched_bad: + pulse.play(Drag(160, 0.1, 40, 2), drive_0) + pulse.play(Drag(160, 0.1, 40, 2), drive_1) + pulse.play(Drag(160, 0.1, 40, 2), control_bad) + + self.cals.add_schedule(sched_good) + + with self.assertRaises(CalibrationError): + self.cals.add_schedule(sched_bad) + + def test_unique_parameter_names(self): + """Test that we cannot insert schedules in which parameter names are duplicates.""" + with pulse.build() as sched: + pulse.play(Drag(160, Parameter("a"), Parameter("a"), Parameter("a")), DriveChannel(0)) + + with self.assertRaises(CalibrationError): + self.cals.add_schedule(sched) + + def test_parameter_without_schedule(self): + """Test that we can manage parameters that are not bound to a schedule.""" + self.cals._register_parameter(Parameter("a"), ()) + + def test_free_parameters(self): + """Test that we can get a schedule with a free parameter.""" + xp = self.cals.get_schedule("xp", 3, assign_params={"amp": self.amp_xp}) + self.assertEqual(xp.parameters, {self.amp_xp}) + + xp = self.cals.get_schedule("xp", 3, assign_params={"amp": self.amp_xp, "σ": self.sigma}) + self.assertEqual(xp.parameters, {self.amp_xp, self.sigma}) + + def test_qubit_input(self): + """Test the qubit input.""" + + xp = self.cals.get_schedule("xp", 3) + self.assertEqual(xp.instructions[0][1].operands[0].amp, 0.2) + + val = self.cals.get_parameter_value("amp", 3, "xp") + self.assertEqual(val, 0.2) + + val = self.cals.get_parameter_value("amp", (3,), "xp") + self.assertEqual(val, 0.2) + + with self.assertRaises(CalibrationError): + self.cals.get_parameter_value("amp", ("3",), "xp") + + val = self.cals.get_parameter_value("amp", "3", "xp") + self.assertEqual(val, 0.2) + + with self.assertRaises(CalibrationError): + self.cals.get_parameter_value("amp", "(1, a)", "xp") + + with self.assertRaises(CalibrationError): + self.cals.get_parameter_value("amp", [3], "xp") + + +class TestOverrideDefaults(QiskitTestCase): + """ + Test that we can override defaults. For example, this means that all qubits may have a + Gaussian as xp pulse but a specific qubit may have a Drag pulse which overrides the + default Gaussian. + """ + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + self.cals = Calibrations() + + self.sigma = Parameter("σ") + self.amp_xp = Parameter("amp") + self.amp = Parameter("amp") + self.beta = Parameter("β") + self.drive = DriveChannel(Parameter("ch0")) + self.date_time = datetime.strptime("15/09/19 10:21:35", "%d/%m/%y %H:%M:%S") + self.duration = Parameter("dur") + + # Template schedule for qubit 3 + with pulse.build(name="xp") as xp_drag: + pulse.play(Drag(self.duration, self.amp_xp, self.sigma, self.beta), self.drive) + + # Default template schedule for all qubits + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(self.duration, self.amp, self.sigma), self.drive) + + # Add the schedules + self.cals.add_schedule(xp) + self.cals.add_schedule(xp_drag, (3,)) + + def test_parameter_value_adding_and_filtering(self): + """Test that adding parameter values behaves in the expected way.""" + + # Ensure that no parameter values are present when none have been added. + params = self.cals.parameters_table() + self.assertEqual(params, []) + + # Add a default parameter common to all qubits. + self.cals.add_parameter_value(ParameterValue(40, self.date_time), "σ", schedule="xp") + self.assertEqual(len(self.cals.parameters_table()), 1) + + # Check that we can get a default parameter in the parameter table + self.assertEqual(len(self.cals.parameters_table(parameters=["σ"])), 1) + self.assertEqual(len(self.cals.parameters_table(parameters=["σ"], schedules=["xp"])), 1) + self.assertEqual(len(self.cals.parameters_table(parameters=["σ"], schedules=["xm"])), 0) + + # Test behaviour of qubit-specific parameter and without ParameterValue. + self.cals.add_parameter_value(0.25, "amp", (3,), "xp") + self.cals.add_parameter_value(0.15, "amp", (0,), "xp") + + # Check the value for qubit 0 + params = self.cals.parameters_table(parameters=["amp"], qubit_list=[(0,)]) + self.assertEqual(len(params), 1) + self.assertEqual(params[0]["value"], 0.15) + self.assertEqual(params[0]["qubits"], (0,)) + + # Check the value for qubit 3 + params = self.cals.parameters_table(parameters=["amp"], qubit_list=[(3,)]) + self.assertEqual(len(params), 1) + self.assertEqual(params[0]["value"], 0.25) + self.assertEqual(params[0]["qubits"], (3,)) + + def _add_parameters(self): + """Helper function.""" + + # Add the minimum number of parameter values. Sigma is shared across both schedules. + self.cals.add_parameter_value(ParameterValue(40, self.date_time), "σ", schedule="xp") + self.cals.add_parameter_value(ParameterValue(0.25, self.date_time), "amp", (3,), "xp") + self.cals.add_parameter_value(ParameterValue(0.15, self.date_time), "amp", (0,), "xp") + self.cals.add_parameter_value(ParameterValue(10, self.date_time), "β", (3,), "xp") + self.cals.add_parameter_value(160, "dur", schedule="xp") + + def test_default_schedules(self): + """ + In this test we create two xp schedules. A default schedules with a + Gaussian pulse for all qubits and a Drag schedule for qubit three which + should override the default schedule. We also test to see that updating + a common parameter affects both schedules. + """ + self._add_parameters() + + xp0 = self.cals.get_schedule("xp", (0,)) + xp3 = self.cals.get_schedule("xp", (3,)) + + # Check that xp0 is Play(Gaussian(160, 0.15, 40), 0) + self.assertTrue(isinstance(xp0.instructions[0][1].pulse, Gaussian)) + self.assertEqual(xp0.instructions[0][1].channel, DriveChannel(0)) + self.assertEqual(xp0.instructions[0][1].pulse.amp, 0.15) + self.assertEqual(xp0.instructions[0][1].pulse.sigma, 40) + self.assertEqual(xp0.instructions[0][1].pulse.duration, 160) + + # Check that xp3 is Play(Drag(160, 0.25, 40, 10), 3) + self.assertTrue(isinstance(xp3.instructions[0][1].pulse, Drag)) + self.assertEqual(xp3.instructions[0][1].channel, DriveChannel(3)) + self.assertEqual(xp3.instructions[0][1].pulse.amp, 0.25) + self.assertEqual(xp3.instructions[0][1].pulse.sigma, 40) + self.assertEqual(xp3.instructions[0][1].pulse.duration, 160) + self.assertEqual(xp3.instructions[0][1].pulse.beta, 10) + + # Check that updating sigma updates both schedules. + later_date_time = datetime.strptime("16/09/19 10:21:35", "%d/%m/%y %H:%M:%S") + self.cals.add_parameter_value(ParameterValue(50, later_date_time), "σ", schedule="xp") + + xp0 = self.cals.get_schedule("xp", (0,)) + xp3 = self.cals.get_schedule("xp", (3,)) + + self.assertEqual(xp0.instructions[0][1].pulse.sigma, 50) + self.assertEqual(xp3.instructions[0][1].pulse.sigma, 50) + + # Check that we have the expected parameters in the calibrations. + expected = {self.amp_xp, self.amp, self.sigma, self.beta, self.duration} + self.assertEqual(len(set(self.cals.parameters.keys())), len(expected)) + + def test_replace_schedule(self): + """Test that schedule replacement works as expected.""" + + self.cals.add_parameter_value(ParameterValue(0.25, self.date_time), "amp", (3,), "xp") + self.cals.add_parameter_value(ParameterValue(40, self.date_time), "σ", schedule="xp") + self.cals.add_parameter_value(ParameterValue(10, self.date_time), "β", (3,), "xp") + + # Let's replace the schedule for qubit 3 with a double Drag pulse. + with pulse.build(name="xp") as sched: + pulse.play(Drag(160, self.amp_xp / 2, self.sigma, self.beta), self.drive) + pulse.play(Drag(160, self.amp_xp / 2, self.sigma, self.beta), self.drive) + + expected = self.cals.parameters + + # Adding this new schedule should not change the parameter mapping + self.cals.add_schedule(sched, (3,)) + + self.assertEqual(self.cals.parameters, expected) + + # For completeness we check that schedule that comes out. + sched_cal = self.cals.get_schedule("xp", (3,)) + + self.assertTrue(isinstance(sched_cal.instructions[0][1].pulse, Drag)) + self.assertTrue(isinstance(sched_cal.instructions[1][1].pulse, Drag)) + self.assertEqual(sched_cal.instructions[0][1].pulse.amp, 0.125) + self.assertEqual(sched_cal.instructions[1][1].pulse.amp, 0.125) + + # Let's replace the schedule for qubit 3 with a Gaussian pulse. + # This should change the parameter mapping + with pulse.build(name="xp") as sched2: + pulse.play(Gaussian(160, self.amp_xp / 2, self.sigma), self.drive) + + # Check that beta is in the mapping + self.assertEqual( + self.cals.parameters[self.beta], + {ParameterKey("β", (3,), "xp")}, + ) + + self.cals.add_schedule(sched2, (3,)) + + # Check that beta no longer maps to a schedule + self.assertEqual(self.cals.parameters[self.beta], set()) + + def test_parameter_filtering(self): + """Test that we can properly filter parameter values.""" + + self._add_parameters() + + # Check that these values are split between the qubits. + amp_values = self.cals.parameters_table(parameters=["amp"], qubit_list=[(0,)]) + self.assertEqual(len(amp_values), 1) + + # Check that we have one value for sigma. + sigma_values = self.cals.parameters_table(parameters=["σ"]) + self.assertEqual(len(sigma_values), 1) + + # Check that we have two values for amp. + amp_values = self.cals.parameters_table(parameters=["amp"]) + self.assertEqual(len(amp_values), 2) + + amp_values = self.cals.parameters_table(parameters=["amp"], qubit_list=[(3,)]) + self.assertEqual(len(amp_values), 1) + + # Check to see if we get back the two qubits when explicitly specifying them. + amp_values = self.cals.parameters_table(parameters=["amp"], qubit_list=[(3,), (0,)]) + self.assertEqual(len(amp_values), 2) + + +class TestMeasurements(QiskitTestCase): + """Test that schedules on measure channels are handled properly.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + self.amp = Parameter("amp") + self.amp_xp = Parameter("amp") + self.sigma = Parameter("σ") + self.sigma_xp = Parameter("σ") + self.width = Parameter("w") + self.duration = Parameter("dur") + self.duration_xp = Parameter("dur") + ch0 = Parameter("ch0") + ch1 = Parameter("ch1") + self.m0_ = MeasureChannel(ch0) + self.d0_ = DriveChannel(ch0) + self.delay = Parameter("delay") + + with pulse.build(name="meas") as meas: + pulse.play(GaussianSquare(self.duration, self.amp, self.sigma, self.width), self.m0_) + + with pulse.build(name="meas_acquire") as meas_acq: + pulse.play(GaussianSquare(self.duration, self.amp, self.sigma, self.width), self.m0_) + pulse.delay(self.delay, pulse.AcquireChannel(ch0)) + pulse.acquire(self.duration, pulse.AcquireChannel(ch0), pulse.RegisterSlot(ch0)) + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(self.duration_xp, self.amp_xp, self.sigma_xp), self.d0_) + + with pulse.build(name="xp_meas") as xp_meas: + pulse.call(xp) + pulse.call(meas) + + with pulse.build(name="xt_meas") as xt_meas: + with pulse.align_sequential(): + pulse.call(xp) + pulse.call(meas) + with pulse.align_sequential(): + pulse.call(xp, value_dict={ch0: ch1}) + pulse.call(meas, value_dict={ch0: ch1}) + + self.cals = Calibrations() + self.cals.add_schedule(meas) + self.cals.add_schedule(xp) + self.cals.add_schedule(xp_meas) + self.cals.add_schedule(xt_meas) + self.cals.add_schedule(meas_acq) + + # self.cals.add_parameter_value(8000, self.duration, schedule="meas") + self.cals.add_parameter_value(0.5, self.amp, (0,), "meas") + self.cals.add_parameter_value(0.56, self.amp, (123,), "meas") + self.cals.add_parameter_value(0.3, self.amp, (2,), "meas") + self.cals.add_parameter_value(160, self.sigma, schedule="meas") + self.cals.add_parameter_value(7000, self.width, schedule="meas") + self.cals.add_parameter_value(8000, self.duration, schedule="meas") + self.cals.add_parameter_value(100, self.delay, schedule="meas_acquire") + + self.cals.add_parameter_value(0.9, self.amp_xp, (0,), "xp") + self.cals.add_parameter_value(0.7, self.amp_xp, (2,), "xp") + self.cals.add_parameter_value(40, self.sigma_xp, schedule="xp") + self.cals.add_parameter_value(160, self.duration_xp, schedule="xp") + + def test_meas_schedule(self): + """Test that we get a properly assigned measure schedule without drive channels.""" + sched = self.cals.get_schedule("meas", (0,)) + meas = Play(GaussianSquare(8000, 0.5, 160, 7000), MeasureChannel(0)) + self.assertTrue(sched.instructions[0][1], meas) + + sched = self.cals.get_schedule("meas", (2,)) + meas = Play(GaussianSquare(8000, 0.3, 160, 7000), MeasureChannel(0)) + self.assertTrue(sched.instructions[0][1], meas) + + def test_call_meas(self): + """Test that we can call a measurement pulse.""" + sched = self.cals.get_schedule("xp_meas", (0,)) + xp = Play(Gaussian(160, 0.9, 40), DriveChannel(0)) + meas = Play(GaussianSquare(8000, 0.5, 160, 7000), MeasureChannel(0)) + + self.assertTrue(sched.instructions[0][1], xp) + self.assertTrue(sched.instructions[1][1], meas) + + def test_xt_meas(self): + """Test that creating multi-qubit schedules out of calls works.""" + + sched = self.cals.get_schedule("xt_meas", (0, 2)) + + xp0 = Play(Gaussian(160, 0.9, 40), DriveChannel(0)) + xp2 = Play(Gaussian(160, 0.7, 40), DriveChannel(2)) + + meas0 = Play(GaussianSquare(8000, 0.5, 160, 7000), MeasureChannel(0)) + meas2 = Play(GaussianSquare(8000, 0.3, 160, 7000), MeasureChannel(2)) + + self.assertEqual(sched.instructions[0][1], xp0) + self.assertEqual(sched.instructions[1][1], xp2) + self.assertEqual(sched.instructions[2][1], meas0) + self.assertEqual(sched.instructions[3][1], meas2) + + def test_free_parameters(self): + """Test that we can get a schedule with free parameters.""" + + # Test coupling breaking + my_amp = Parameter("my_amp") + schedule = self.cals.get_schedule( + "xt_meas", + (0, 2), + assign_params={("amp", (0,), "xp"): my_amp}, + ) + + schedule = block_to_schedule(schedule) + + with pulse.build(name="xt_meas") as expected: + with pulse.align_sequential(): + pulse.play(Gaussian(160, my_amp, 40), DriveChannel(0)) + pulse.play(GaussianSquare(8000, 0.5, 160, 7000), MeasureChannel(0)) + with pulse.align_sequential(): + pulse.play(Gaussian(160, 0.7, 40), DriveChannel(2)) + pulse.play(GaussianSquare(8000, 0.3, 160, 7000), MeasureChannel(2)) + + expected = block_to_schedule(expected) + + self.assertEqual(schedule.parameters, {my_amp}) + self.assertEqual(schedule, expected) + + def test_free_parameters_check(self): + """ + Test that get_schedule raises an error if the number of parameters does not match. + This test ensures that we forbid ambiguity in free parameters in schedules with + calls that share parameters. + """ + + amp1 = Parameter("amp1") + amp2 = Parameter("amp2") + assign_dict = {("amp", (0,), "xp"): amp1, ("amp", (2,), "xp"): amp2} + + sched = self.cals.get_schedule("xt_meas", (0, 2), assign_params=assign_dict) + + self.assertEqual(sched.parameters, {amp1, amp2}) + + sched = block_to_schedule(sched) + + self.assertEqual(sched.instructions[0][1].parameters, {amp1}) + self.assertEqual(sched.instructions[1][1].parameters, {amp2}) + + def test_measure_and_acquire(self): + """Test that we can get a measurement schedule with an acquire instruction.""" + + sched = self.cals.get_schedule("meas_acquire", (123,)) + + with pulse.build(name="meas_acquire") as expected: + pulse.play(GaussianSquare(8000, 0.56, 160, 7000), MeasureChannel(123)) + pulse.delay(100, AcquireChannel(123)) + pulse.acquire(8000, AcquireChannel(123), RegisterSlot(123)) + + self.assertEqual(sched, expected) + + +class TestInstructions(QiskitTestCase): + """Class to test that instructions like Shift and Set Phase/Frequency are properly managed.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + self.phase = Parameter("φ") + self.freq = Parameter("ν") + self.d0_ = DriveChannel(Parameter("ch0")) + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, 0.5, 40), self.d0_) + + with pulse.build(name="xp12") as xp12: + pulse.shift_phase(self.phase, self.d0_) + pulse.set_frequency(self.freq, self.d0_) + pulse.play(Gaussian(160, 0.5, 40), self.d0_) + + # To make things more interesting we will use a call. + with pulse.build(name="xp02") as xp02: + pulse.call(xp) + pulse.call(xp12) + + self.cals = Calibrations() + self.cals.add_schedule(xp) + self.cals.add_schedule(xp12) + self.cals.add_schedule(xp02) + + self.date_time = datetime.strptime("15/09/19 10:21:35", "%d/%m/%y %H:%M:%S") + + self.cals.add_parameter_value(ParameterValue(1.57, self.date_time), "φ", (3,), "xp12") + self.cals.add_parameter_value(ParameterValue(200, self.date_time), "ν", (3,), "xp12") + + def test_call_registration(self): + """Check that by registering the call we registered three schedules.""" + + self.assertEqual(len(self.cals.schedules()), 3) + + def test_instructions(self): + """Check that we get a properly assigned schedule.""" + + sched = self.cals.get_schedule("xp02", (3,)) + + self.assertEqual(sched.parameters, set()) + + sched = inline_subroutines(sched) # inline makes the check more transparent. + + self.assertTrue(isinstance(sched.instructions[0][1], pulse.Play)) + self.assertEqual(sched.instructions[1][1].phase, 1.57) + self.assertEqual(sched.instructions[2][1].frequency, 200) + + +class TestRegistering(QiskitTestCase): + """Class to test registering of subroutines with calls.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + self.cals = Calibrations() + self.d0_ = DriveChannel(Parameter("ch0")) + + def test_call_registering(self): + """Test registering of schedules with call.""" + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, 0.5, 40), self.d0_) + + with pulse.build(name="call_xp") as call_xp: + pulse.call(xp) + + with self.assertRaises(CalibrationError): + self.cals.add_schedule(call_xp) + + self.cals.add_schedule(xp) + self.cals.add_schedule(call_xp) + + self.assertTrue(isinstance(self.cals.get_schedule("call_xp", 2), pulse.ScheduleBlock)) + + def test_get_template(self): + """Test that we can get a registered template and use it.""" + amp = Parameter("amp") + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, amp, 40), self.d0_) + + self.cals.add_schedule(xp) + + registered_xp = self.cals.get_template("xp") + + self.assertEqual(registered_xp, xp) + + with pulse.build(name="dxp") as dxp: + pulse.call(registered_xp) + pulse.play(Gaussian(160, amp, 40), self.d0_) + + self.cals.add_schedule(dxp) + self.cals.add_parameter_value(0.5, "amp", 3, "xp") + + sched = block_to_schedule(self.cals.get_schedule("dxp", 3)) + + self.assertEqual(sched.instructions[0][1], Play(Gaussian(160, 0.5, 40), DriveChannel(3))) + self.assertEqual(sched.instructions[1][1], Play(Gaussian(160, 0.5, 40), DriveChannel(3))) + + with self.assertRaises(CalibrationError): + self.cals.get_template("not registered") + + self.cals.get_template("xp", (3,)) + + def test_register_schedule(self): + """Test that we cannot register a schedule in a call.""" + + xp = pulse.Schedule(name="xp") + xp.insert(0, pulse.Play(pulse.Gaussian(160, 0.5, 40), pulse.DriveChannel(0)), inplace=True) + + with pulse.build(name="call_xp") as call_xp: + pulse.call(xp) + + try: + self.cals.add_schedule(call_xp) + except CalibrationError as error: + self.assertEqual( + error.message, "Calling a Schedule is forbidden, call ScheduleBlock instead." + ) + + +class CrossResonanceTest(QiskitTestCase): + """Setup class for an echoed cross-resonance calibration.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + controls = { + (3, 2): [ControlChannel(10), ControlChannel(123)], + (2, 3): [ControlChannel(15), ControlChannel(23)], + } + self.cals = Calibrations(control_config=controls) + + self.amp_cr = Parameter("amp") + self.amp_rot = Parameter("amp_rot") + self.amp = Parameter("amp") + self.amp_tcp = Parameter("amp") + self.d0_ = DriveChannel(Parameter("ch0")) + self.d1_ = DriveChannel(Parameter("ch1")) + self.c1_ = ControlChannel(Parameter("ch0.1")) + self.sigma = Parameter("σ") + self.width = Parameter("w") + self.date_time = datetime.strptime("15/09/19 10:21:35", "%d/%m/%y %H:%M:%S") + + cr_tone_p = GaussianSquare(640, self.amp_cr, self.sigma, self.width) + rotary_p = GaussianSquare(640, self.amp_rot, self.sigma, self.width) + + cr_tone_m = GaussianSquare(640, -self.amp_cr, self.sigma, self.width) + rotary_m = GaussianSquare(640, -self.amp_rot, self.sigma, self.width) + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, self.amp, self.sigma), self.d0_) + + with pulse.build(name="cr") as cr: + with pulse.align_sequential(): + with pulse.align_left(): + pulse.play(rotary_p, self.d1_) # Rotary tone + pulse.play(cr_tone_p, self.c1_) # CR tone. + pulse.call(xp) + with pulse.align_left(): + pulse.play(rotary_m, self.d1_) + pulse.play(cr_tone_m, self.c1_) + pulse.call(xp) + + # Mimic a tunable coupler pulse that is just a pulse on a control channel. + with pulse.build(name="tcp") as tcp: + pulse.play(GaussianSquare(640, self.amp_tcp, self.sigma, self.width), self.c1_) + + self.cals.add_schedule(xp) + self.cals.add_schedule(cr) + self.cals.add_schedule(tcp) + + self.cals.add_parameter_value(ParameterValue(40, self.date_time), "σ", schedule="xp") + self.cals.add_parameter_value( + ParameterValue(0.1 + 0.01j, self.date_time), "amp", (3,), "xp" + ) + self.cals.add_parameter_value(ParameterValue(0.3, self.date_time), "amp", (3, 2), "cr") + self.cals.add_parameter_value(ParameterValue(0.2, self.date_time), "amp_rot", (3, 2), "cr") + self.cals.add_parameter_value(ParameterValue(0.8, self.date_time), "amp", (3, 2), "tcp") + self.cals.add_parameter_value(ParameterValue(20, self.date_time), "w", (3, 2), "cr") + + # Reverse gate parameters + self.cals.add_parameter_value(ParameterValue(0.15, self.date_time), "amp", (2,), "xp") + self.cals.add_parameter_value(ParameterValue(0.5, self.date_time), "amp", (2, 3), "cr") + self.cals.add_parameter_value(ParameterValue(0.4, self.date_time), "amp_rot", (2, 3), "cr") + self.cals.add_parameter_value(ParameterValue(30, self.date_time), "w", (2, 3), "cr") + + +class TestControlChannels(CrossResonanceTest): + """ + Test the echoed cross-resonance schedule which is more complex than single-qubit + schedules. The example also shows that a schedule with call instructions can + support parameters with the same names. + """ + + def test_get_schedule(self): + """Check that we can get a CR schedule with a built in Call.""" + + with pulse.build(name="cr") as cr_32: + with pulse.align_sequential(): + with pulse.align_left(): + pulse.play(GaussianSquare(640, 0.2, 40, 20), DriveChannel(2)) # Rotary tone + pulse.play(GaussianSquare(640, 0.3, 40, 20), ControlChannel(10)) # CR tone. + pulse.play(Gaussian(160, 0.1 + 0.01j, 40), DriveChannel(3)) + with pulse.align_left(): + pulse.play(GaussianSquare(640, -0.2, 40, 20), DriveChannel(2)) # Rotary tone + pulse.play(GaussianSquare(640, -0.3, 40, 20), ControlChannel(10)) # CR tone. + pulse.play(Gaussian(160, 0.1 + 0.01j, 40), DriveChannel(3)) + + # We inline to make the schedules comparable with the construction directly above. + schedule = self.cals.get_schedule("cr", (3, 2)) + inline_schedule = inline_subroutines(schedule) + for idx, inst in enumerate(inline_schedule.instructions): + self.assertTrue(inst == cr_32.instructions[idx]) + + self.assertEqual(schedule.parameters, set()) + + # Do the CR in the other direction + with pulse.build(name="cr") as cr_23: + with pulse.align_sequential(): + with pulse.align_left(): + pulse.play(GaussianSquare(640, 0.4, 40, 30), DriveChannel(3)) # Rotary tone + pulse.play(GaussianSquare(640, 0.5, 40, 30), ControlChannel(15)) # CR tone. + pulse.play(Gaussian(160, 0.15, 40), DriveChannel(2)) + with pulse.align_left(): + pulse.play(GaussianSquare(640, -0.4, 40, 30), DriveChannel(3)) # Rotary tone + pulse.play(GaussianSquare(640, -0.5, 40, 30), ControlChannel(15)) # CR tone. + pulse.play(Gaussian(160, 0.15, 40), DriveChannel(2)) + + schedule = self.cals.get_schedule("cr", (2, 3)) + inline_schedule = inline_subroutines(schedule) + for idx, inst in enumerate(inline_schedule.instructions): + self.assertTrue(inst == cr_23.instructions[idx]) + + self.assertEqual(schedule.parameters, set()) + + def test_free_parameters(self): + """Test that we can get a schedule with free parameters.""" + + schedule = self.cals.get_schedule("cr", (3, 2), assign_params={"amp": self.amp_cr}) + + self.assertEqual(schedule.parameters, {self.amp_cr}) + + def test_single_control_channel(self): + """Test that getting a correct pulse on a control channel only works.""" + + with pulse.build(name="tcp") as expected: + pulse.play(GaussianSquare(640, 0.8, 40, 20), ControlChannel(10)) + + self.assertEqual(self.cals.get_schedule("tcp", (3, 2)), expected) + + +class TestAssignment(QiskitTestCase): + """Test simple assignment""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + controls = {(3, 2): [ControlChannel(10)]} + + self.cals = Calibrations(control_config=controls) + + self.amp_xp = Parameter("amp") + self.ch0 = Parameter("ch0") + self.d0_ = DriveChannel(self.ch0) + self.ch1 = Parameter("ch1") + self.d1_ = DriveChannel(self.ch1) + self.sigma = Parameter("σ") + self.width = Parameter("w") + self.dur = Parameter("duration") + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, self.amp_xp, self.sigma), self.d0_) + + with pulse.build(name="xpxp") as xpxp: + with pulse.align_left(): + pulse.call(xp) + pulse.call(xp, value_dict={self.ch0: self.ch1}) + + self.xp_ = xp + self.cals.add_schedule(xp) + self.cals.add_schedule(xpxp) + + self.cals.add_parameter_value(0.2, "amp", (2,), "xp") + self.cals.add_parameter_value(0.3, "amp", (3,), "xp") + self.cals.add_parameter_value(40, "σ", (), "xp") + + def test_short_key(self): + """Test simple value assignment""" + sched = self.cals.get_schedule("xp", (2,), assign_params={"amp": 0.1}) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, 0.1, 40), DriveChannel(2)) + + self.assertEqual(sched, expected) + + def test_assign_to_parameter(self): + """Test assigning to a Parameter instance""" + my_amp = Parameter("my_amp") + sched = self.cals.get_schedule("xp", (2,), assign_params={"amp": my_amp}) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, my_amp, 40), DriveChannel(2)) + + self.assertEqual(sched, expected) + + def test_assign_to_parameter_in_call(self): + """Test assigning to a Parameter instance in a call""" + with pulse.build(name="call_xp") as call_xp: + pulse.call(self.xp_) + self.cals.add_schedule(call_xp) + + my_amp = Parameter("my_amp") + sched = self.cals.get_schedule("call_xp", (2,), assign_params={("amp", (2,), "xp"): my_amp}) + sched = block_to_schedule(sched) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, my_amp, 40), DriveChannel(2)) + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + def test_assign_to_parameter_in_call_and_to_value_in_caller(self): + """Test assigning to a Parameter instances in a call and caller""" + with pulse.build(name="call_xp_xp") as call_xp_xp: + pulse.call(self.xp_) + pulse.play(Gaussian(160, self.amp_xp, self.sigma), self.d0_) + self.cals.add_schedule(call_xp_xp) + + my_amp = Parameter("amp") + sched = self.cals.get_schedule( + "call_xp_xp", + (2,), + assign_params={ + ("amp", (2,), "xp"): my_amp, + ("amp", (2,), "call_xp_xp"): 0.2, + }, + ) + sched = block_to_schedule(sched) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, my_amp, 40), DriveChannel(2)) + pulse.play(Gaussian(160, 0.2, 40), DriveChannel(2)) + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + def test_assign_to_same_parameter_in_call_and_caller(self): + """ + Test assigning to a Parameter in a call and reassigning in caller raises + + Check that it is not allowed to leave a parameter in a subschedule free + by assigning it to a Parameter that is also used in the calling + schedule as that will re-bind the Parameter in the subschedule as well. + """ + with pulse.build(name="call_xp_xp") as call_xp_xp: + pulse.call(self.xp_) + pulse.play(Gaussian(160, self.amp_xp, self.sigma), self.d0_) + self.cals.add_schedule(call_xp_xp) + + my_amp = Parameter("amp") + with self.assertRaises(CalibrationError): + self.cals.get_schedule( + "call_xp_xp", + (2,), + assign_params={ + ("amp", (2,), "xp"): self.amp_xp, + ("amp", (2,), "call_xp_xp"): my_amp, + }, + ) + + def test_full_key(self): + """Test value assignment with full key""" + sched = self.cals.get_schedule("xp", (2,), assign_params={("amp", (2,), "xp"): 0.1}) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, 0.1, 40), DriveChannel(2)) + + self.assertEqual(sched, expected) + + def test_default_qubit(self): + """Test value assignment with default qubit""" + sched = self.cals.get_schedule("xp", (2,), assign_params={("amp", (), "xp"): 0.1}) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, 0.1, 40), DriveChannel(2)) + + self.assertEqual(sched, expected) + + def test_default_across_qubits(self): + """Test assigning to multiple schedules through default parameter""" + sched = self.cals.get_schedule("xpxp", (2, 3), assign_params={("amp", (), "xp"): 0.4}) + sched = block_to_schedule(sched) + + with pulse.build(name="xpxp") as expected: + with pulse.align_left(): + pulse.play(Gaussian(160, 0.4, 40), DriveChannel(2)) + pulse.play(Gaussian(160, 0.4, 40), DriveChannel(3)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + +class TestReplaceScheduleAndCall(QiskitTestCase): + """A test to ensure that inconsistencies are picked up when a schedule is reassigned.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + self.cals = Calibrations() + + self.amp = Parameter("amp") + self.dur = Parameter("duration") + self.sigma = Parameter("σ") + self.beta = Parameter("β") + self.ch0 = Parameter("ch0") + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(self.dur, self.amp, self.sigma), DriveChannel(self.ch0)) + + with pulse.build(name="call_xp") as call_xp: + pulse.call(xp) + + self.cals.add_schedule(xp) + self.cals.add_schedule(call_xp) + + self.cals.add_parameter_value(0.2, "amp", (4,), "xp") + self.cals.add_parameter_value(160, "duration", (4,), "xp") + self.cals.add_parameter_value(40, "σ", (), "xp") + + def test_call_replaced(self): + """Test that we get an error when there is an inconsistency in subroutines.""" + + sched = self.cals.get_schedule("call_xp", (4,)) + sched = block_to_schedule(sched) + + with pulse.build(name="xp") as expected: + pulse.play(Gaussian(160, 0.2, 40), DriveChannel(4)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + # Now update the xp pulse without updating the call_xp schedule and ensure that + # an error is raised. + with pulse.build(name="xp") as drag: + pulse.play(Drag(self.dur, self.amp, self.sigma, self.beta), DriveChannel(self.ch0)) + + self.cals.add_schedule(drag) + self.cals.add_parameter_value(10.0, "β", (4,), "xp") + + with self.assertRaises(CalibrationError): + self.cals.get_schedule("call_xp", (4,)) + + +class TestCoupledAssigning(QiskitTestCase): + """Test that assigning parameters works when they are coupled in calls.""" + + def setUp(self): + """Create the setting to test.""" + super().setUp() + + controls = {(3, 2): [ControlChannel(10)]} + + self.cals = Calibrations(control_config=controls) + + self.amp_cr = Parameter("amp") + self.amp_xp = Parameter("amp") + self.ch0 = Parameter("ch0") + self.d0_ = DriveChannel(self.ch0) + self.ch1 = Parameter("ch1") + self.d1_ = DriveChannel(self.ch1) + self.c1_ = ControlChannel(Parameter("ch0.1")) + self.sigma = Parameter("σ") + self.width = Parameter("w") + self.dur = Parameter("duration") + + with pulse.build(name="cr_p") as cr_p: + pulse.play(GaussianSquare(self.dur, self.amp_cr, self.sigma, self.width), self.c1_) + + with pulse.build(name="cr_m") as cr_m: + pulse.play(GaussianSquare(self.dur, -self.amp_cr, self.sigma, self.width), self.c1_) + + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, self.amp_xp, self.sigma), self.d0_) + + with pulse.build(name="ecr") as ecr: + with pulse.align_sequential(): + pulse.call(cr_p) + pulse.call(xp) + pulse.call(cr_m) + + with pulse.build(name="cr_echo_both") as cr_echo_both: + with pulse.align_sequential(): + pulse.call(cr_p) + with pulse.align_left(): + pulse.call(xp) + pulse.call(xp, value_dict={self.ch0: self.ch1}) + pulse.call(cr_m) + + self.cals.add_schedule(cr_p) + self.cals.add_schedule(cr_m) + self.cals.add_schedule(xp) + self.cals.add_schedule(ecr) + self.cals.add_schedule(cr_echo_both) + + self.cals.add_parameter_value(0.3, "amp", (3, 2), "cr_p") + self.cals.add_parameter_value(0.2, "amp", (3,), "xp") + self.cals.add_parameter_value(0.4, "amp", (2,), "xp") + self.cals.add_parameter_value(40, "σ", (), "xp") + self.cals.add_parameter_value(640, "w", (3, 2), "cr_p") + self.cals.add_parameter_value(800, "duration", (3, 2), "cr_p") + + def test_assign_coupled_explicitly(self): + """Test that we get the proper schedules when they are coupled.""" + + # Test that we can preserve the coupling + my_amp = Parameter("my_amp") + assign_params = {("amp", (3, 2), "cr_p"): my_amp, ("amp", (3, 2), "cr_m"): my_amp} + sched = self.cals.get_schedule("ecr", (3, 2), assign_params=assign_params) + sched = block_to_schedule(sched) + + with pulse.build(name="ecr") as expected: + with pulse.align_sequential(): + pulse.play(GaussianSquare(800, my_amp, 40, 640), ControlChannel(10)) + pulse.play(Gaussian(160, 0.2, 40), DriveChannel(3)) + pulse.play(GaussianSquare(800, -my_amp, 40, 640), ControlChannel(10)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + def test_assign_coupled_implicitly_float(self): + """Test that we get the proper schedules when they are coupled.""" + assign_params = {("amp", (3, 2), "cr_m"): 0.8} + sched = self.cals.get_schedule("ecr", (3, 2), assign_params=assign_params) + sched = block_to_schedule(sched) + + with pulse.build(name="ecr") as expected: + with pulse.align_sequential(): + pulse.play(GaussianSquare(800, 0.8, 40, 640), ControlChannel(10)) + pulse.play(Gaussian(160, 0.2, 40), DriveChannel(3)) + pulse.play(GaussianSquare(800, -0.8, 40, 640), ControlChannel(10)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + def test_assign_coupled_implicitly(self): + """Test that we get the proper schedules when they are coupled.""" + my_amp = Parameter("my_amp") + assign_params = {("amp", (3, 2), "cr_p"): my_amp} + sched = self.cals.get_schedule("ecr", (3, 2), assign_params=assign_params) + sched = block_to_schedule(sched) + + with pulse.build(name="ecr") as expected: + with pulse.align_sequential(): + pulse.play(GaussianSquare(800, my_amp, 40, 640), ControlChannel(10)) + pulse.play(Gaussian(160, 0.2, 40), DriveChannel(3)) + pulse.play(GaussianSquare(800, -my_amp, 40, 640), ControlChannel(10)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + def test_break_coupled(self): + """Test that we get the proper schedules when they are coupled.""" + my_amp = Parameter("my_amp") + my_amp2 = Parameter("my_amp2") + assign_params = {("amp", (3, 2), "cr_p"): my_amp, ("amp", (3, 2), "cr_m"): my_amp2} + sched = self.cals.get_schedule("ecr", (3, 2), assign_params=assign_params) + sched = block_to_schedule(sched) + + with pulse.build(name="ecr") as expected: + with pulse.align_sequential(): + pulse.play(GaussianSquare(800, my_amp, 40, 640), ControlChannel(10)) + pulse.play(Gaussian(160, 0.2, 40), DriveChannel(3)) + pulse.play(GaussianSquare(800, -my_amp2, 40, 640), ControlChannel(10)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + def test_assign_coupled_explicitly_two_channel(self): + """Test that we get the proper schedules when they are coupled.""" + + # Test that we can preserve the coupling + my_amp = Parameter("my_amp") + my_amp2 = Parameter("my_amp2") + assign_params = {("amp", (3,), "xp"): my_amp, ("amp", (2,), "xp"): my_amp2} + sched = self.cals.get_schedule("cr_echo_both", (3, 2), assign_params=assign_params) + sched = block_to_schedule(sched) + + with pulse.build(name="cr_echo_both") as expected: + with pulse.align_sequential(): + pulse.play(GaussianSquare(800, 0.3, 40, 640), ControlChannel(10)) + with pulse.align_left(): + pulse.play(Gaussian(160, my_amp, 40), DriveChannel(3)) + pulse.play(Gaussian(160, my_amp2, 40), DriveChannel(2)) + pulse.play(GaussianSquare(800, -0.3, 40, 640), ControlChannel(10)) + + expected = block_to_schedule(expected) + + self.assertEqual(sched, expected) + + +class TestFiltering(QiskitTestCase): + """Test that the filtering works as expected.""" + + def setUp(self): + """Setup a calibration.""" + super().setUp() + + self.cals = Calibrations() + + self.sigma = Parameter("σ") + self.amp = Parameter("amp") + self.drive = DriveChannel(Parameter("ch0")) + + # Define and add template schedules. + with pulse.build(name="xp") as xp: + pulse.play(Gaussian(160, self.amp, self.sigma), self.drive) + + self.cals.add_schedule(xp) + + self.date_time1 = datetime.strptime("15/09/19 10:21:35", "%d/%m/%y %H:%M:%S") + self.date_time2 = datetime.strptime("15/09/19 11:21:35", "%d/%m/%y %H:%M:%S") + + self.cals.add_parameter_value(ParameterValue(40, self.date_time1), "σ", schedule="xp") + self.cals.add_parameter_value( + ParameterValue(45, self.date_time2, False), "σ", schedule="xp" + ) + self.cals.add_parameter_value(ParameterValue(0.1, self.date_time1), "amp", (0,), "xp") + self.cals.add_parameter_value(ParameterValue(0.2, self.date_time2), "amp", (0,), "xp") + self.cals.add_parameter_value( + ParameterValue(0.4, self.date_time2, group="super_cal"), "amp", (0,), "xp" + ) + + def test_get_parameter_value(self): + """Test that getting parameter values funcions properly.""" + + amp = self.cals.get_parameter_value(self.amp, (0,), "xp") + self.assertEqual(amp, 0.2) + + amp = self.cals.get_parameter_value(self.amp, (0,), "xp", group="super_cal") + self.assertEqual(amp, 0.4) + + cutoff_date = datetime.strptime("15/09/19 11:21:34", "%d/%m/%y %H:%M:%S") + amp = self.cals.get_parameter_value(self.amp, (0,), "xp", cutoff_date=cutoff_date) + self.assertEqual(amp, 0.1) + + sigma = self.cals.get_parameter_value(self.sigma, (0,), "xp") + self.assertEqual(sigma, 40) + + sigma = self.cals.get_parameter_value(self.sigma, (0,), "xp", valid_only=False) + self.assertEqual(sigma, 45) + + +class TestSavingAndLoading(CrossResonanceTest): + """Test that calibrations can be saved and loaded to and from files.""" + + def test_save_load_parameter_values(self): + """Test that we can save and load parameter values.""" + + self.cals.save("csv", overwrite=True) + self.assertEqual(self.cals.get_parameter_value("amp", (3,), "xp"), 0.1 + 0.01j) + + self.cals._params = defaultdict(list) + + with self.assertRaises(CalibrationError): + self.cals.get_parameter_value("amp", (3,), "xp") + + # Load the parameters, check value and type. + self.cals.load_parameter_values("parameter_values.csv") + + val = self.cals.get_parameter_value("amp", (3,), "xp") + self.assertEqual(val, 0.1 + 0.01j) + self.assertTrue(isinstance(val, complex)) + + val = self.cals.get_parameter_value("σ", (3,), "xp") + self.assertEqual(val, 40) + self.assertTrue(isinstance(val, int)) + + val = self.cals.get_parameter_value("amp", (3, 2), "cr") + self.assertEqual(val, 0.3) + self.assertTrue(isinstance(val, float)) + + # Check that we cannot rewrite files as they already exist. + with self.assertRaises(CalibrationError): + self.cals.save("csv") + + self.cals.save("csv", overwrite=True) + + # Clean-up + os.remove("parameter_values.csv") + os.remove("parameter_config.csv") + os.remove("schedules.csv") diff --git a/tox.ini b/tox.ini index 61d3da7976..aa69e4a514 100644 --- a/tox.ini +++ b/tox.ini @@ -9,7 +9,9 @@ install_command = pip install -U {opts} {packages} setenv = VIRTUAL_ENV={envdir} QISKIT_SUPPRESS_PACKAGING_WARNINGS=Y -deps = -r{toxinidir}/requirements-dev.txt +deps = + -r{toxinidir}/requirements-dev.txt + git+https://github.com/Qiskit/qiskit-terra commands = stestr run {posargs} [testenv:lint]