Calibrations fixes

qiskit_experiments/calibration_management/calibration_utils.py

@@ -0,0 +1,62 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2019-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.
+
+"""Calibration helper functions"""
+
+from typing import List, Set
+from qiskit.pulse import ScheduleBlock, Call
+
+
+def used_in_calls(schedule_name: str, schedules: List[ScheduleBlock]) -> Set[str]:
+    """Find the schedules in the given list that call a given schedule by name.
+
+    Args:
+        schedule_name: The name of the callee to identify.
+        schedules: A list of potential caller schedules to search.
+
+    Returns:
+        A set of schedule names that call the given schedule.
+    """
+    caller_names = set()
+
+    for schedule in schedules:
+        if _used_in_calls(schedule_name, schedule):
+            caller_names.add(schedule.name)
+
+    return caller_names
+
+
+def _used_in_calls(schedule_name: str, schedule: ScheduleBlock) -> bool:
+    """Recursively find if the schedule calls a schedule with name ``schedule_name``.
+
+    Args:
+        schedule_name: The name of the callee to identify.
+        schedule: The schedule to parse.
+
+    Returns:
+        True if ``schedule``calls a ``ScheduleBlock`` with name ``schedule_name``.
+    """
+    blocks_have_schedule = False
+
+    for block in schedule.blocks:
+        if isinstance(block, Call):
+            if block.subroutine.name == schedule_name:
+                return True
+            else:
+                blocks_have_schedule = blocks_have_schedule or _used_in_calls(
+                    schedule_name, block.subroutine
+                )
+
+        if isinstance(block, ScheduleBlock):
+            blocks_have_schedule = blocks_have_schedule or _used_in_calls(schedule_name, block)
+
+    return blocks_have_schedule

qiskit_experiments/calibration_management/calibrations.py

@@ -43,6 +43,7 @@
 from qiskit_experiments.calibration_management.basis_gate_library import BasisGateLibrary
 from qiskit_experiments.calibration_management.parameter_value import ParameterValue
 from qiskit_experiments.calibration_management.control_channel_map import ControlChannelMap
+from qiskit_experiments.calibration_management.calibration_utils import used_in_calls
 from qiskit_experiments.calibration_management.calibration_key_types import (
     ParameterKey,
     ParameterValueType,
@@ -96,7 +97,8 @@ def __init__(
                 qubits is :code:`[[0, 1], [1, 0], [1, 2], [2, 1], [2, 0], [0, 2]]`.
             control_channel_map: 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.
+                that correspond to the qubits in the keys. If a control_channel_map is given
+                then the qubits must be in the coupling_map.
             library: A library instance from which to get template schedules to register as well
                 as default parameter values.
             add_parameter_defaults: A boolean to indicate weather the default parameter values of
@@ -174,15 +176,35 @@ def __init__(
         self._register_parameter(self.meas_freq, ())
 
         # Backends with a single qubit may not have a coupling map.
-        num_qubits = max(max(coupling_map)) + 1 if coupling_map is not None else 1
+        self._coupling_map = coupling_map if coupling_map is not None else []
 
-        self._qubits = list(range(num_qubits))
-        self._coupling_map = coupling_map
+        # A dict extension of the coupling map where the key is the number of qubits and
+        # the values are a list of qubits coupled.
         self._operated_qubits = self._get_operated_qubits()
+        self._check_consistency()
 
         # Push the schedules to the instruction schedule map.
         self.update_inst_map()
 
+    def _check_consistency(self):
+        """Check that the attributes defined in self are consistent.
+
+        Raises:
+            CalibrationError: If there is a control channel map but no coupling map.
+            CalibrationError: If a qubit in the control channel map is not in the
+                coupling map.
+        """
+        if not self._coupling_map and self._control_channel_map:
+            raise CalibrationError("No coupling map but a control channel map was found.")
+
+        if self._coupling_map and self._control_channel_map:
+            cmap_qubits = set(qubit for pair in self._coupling_map for qubit in pair)
+            for qubits in self._control_channel_map:
+                if not set(qubits).issubset(cmap_qubits):
+                    raise CalibrationError(
+                        f"Qubits {qubits} of control_channel_map are not in the coupling map."
+                    )
+
     @property
     def backend_name(self) -> str:
         """Return the name of the backend."""
@@ -221,7 +243,7 @@ def from_backend(
             backend_name = None
 
         cals = Calibrations(
-            getattr(backend.configuration(), "coupling_map", None),
+            getattr(backend.configuration(), "coupling_map", []),
             getattr(backend.configuration(), "control_channels", None),
             library,
             add_parameter_defaults,
@@ -268,13 +290,16 @@ def _get_operated_qubits(self) -> Dict[int, List[int]]:
         operated_qubits = defaultdict(list)
 
         # Single qubits
-        for qubit in self._qubits:
-            operated_qubits[1].append([qubit])
+        if self._coupling_map:
+            for qubit in set(qubit for coupled in self._coupling_map for qubit in coupled):
+                operated_qubits[1].append([qubit])
+        else:
+            # Edge case for single-qubit device.
+            operated_qubits[1].append([0])
 
         # Multi-qubit couplings
-        if self._coupling_map is not None:
-            for coupling in self._coupling_map:
-                operated_qubits[len(coupling)].append(coupling)
+        for coupling in self._coupling_map:
+            operated_qubits[len(coupling)].append(coupling)
 
         return operated_qubits
 
@@ -353,7 +378,7 @@ def update_inst_map(
             if schedules is not None and sched_name not in schedules:
                 continue
 
-            if qubits is not None:
+            if qubits:
                 self._robust_inst_map_add(inst_map, sched_name, qubits, group, cutoff_date)
             else:
                 for qubits_ in self._operated_qubits[self._schedules_qubits[key]]:
@@ -371,23 +396,65 @@ def _robust_inst_map_add(
 
         get_schedule may raise an error if not all parameters have values or
         default values. In this case we ignore and continue updating inst_map.
+        Note that ``qubits`` may only be a sub-set of the qubits of the schedule that
+        we want to update. This may arise in cases such as an ECR gate schedule that calls
+        an X-gate schedule. When updating the X-gate schedule we need to also update the
+        corresponding ECR schedules which operate on a larger number of qubits.
 
         Args:
             sched_name: The name of the schedule.
-            qubits: The qubit to which the schedule applies.
+            qubits: The qubit to which the schedule applies. Note, these may be only a
+                subset of the qubits in the schedule. For example, if the name of the
+                schedule is `"cr"` we may have `qubits` be `(3, )` and this function
+                will update the CR schedules on all schedules which involve qubit 3.
             group: The calibration group.
             cutoff: The cutoff date.
         """
-        try:
-            inst_map.add(
-                instruction=sched_name,
-                qubits=qubits,
-                schedule=self.get_schedule(sched_name, qubits, group=group, cutoff_date=cutoff),
-            )
-        except CalibrationError:
-            # get_schedule may raise an error if not all parameters have values or
-            # default values. In this case we ignore and continue updating inst_map.
-            pass
+        for update_qubits in self._get_full_qubits_of_schedule(sched_name, qubits):
+            try:
+                schedule = self.get_schedule(
+                    sched_name, update_qubits, group=group, cutoff_date=cutoff
+                )
+                inst_map.add(instruction=sched_name, qubits=update_qubits, schedule=schedule)
+            except CalibrationError:
+                # get_schedule may raise an error if not all parameters have values or
+                # default values. In this case we ignore and continue updating inst_map.
+                pass
+
+    def _get_full_qubits_of_schedule(
+        self, schedule_name: str, partial_qubits: Tuple[int, ...]
+    ) -> List[Tuple[int, ...]]:
+        """Find all qubits for which there is a schedule ``schedule_name`` on ``partial_qubits``.
+
+        This method uses the map between the schedules and the number of qubits that they
+        operate on as well as the extension of the coupling map ``_operated_qubits`` to find
+        which qubits are involved in the schedule named ``schedule_name`` involving the
+        ``partial_qubits``.
+
+        Args:
+            schedule_name: The name of the schedule as registered in ``self``.
+            partial_qubits: A sub-set of qubits on which the schedule applies.
+
+        Returns:
+            A list of tuples. Each tuple is the set of qubits for which there is a schedule
+            named ``schedule_name`` and ``partial_qubits`` is a sub-set of said qubits.
+        """
+        for key, circuit_inst_num_qubits in self._schedules_qubits.items():
+            if key.schedule == schedule_name:
+
+                if len(partial_qubits) == circuit_inst_num_qubits:
+                    return [partial_qubits]
+
+                else:
+                    candidates = self._operated_qubits[circuit_inst_num_qubits]
+                    qubits_for_update = []
+                    for candidate_qubits in candidates:
+                        if set(partial_qubits).issubset(set(candidate_qubits)):
+                            qubits_for_update.append(tuple(candidate_qubits))
+
+                    return qubits_for_update
+
+        return []
 
     def inst_map_add(
         self,
@@ -770,7 +837,11 @@ def add_parameter_value(
         if update_inst_map and schedule is not None:
             param_obj = self.calibration_parameter(param_name, qubits, sched_name)
             schedules = set(key.schedule for key in self._parameter_map_r[param_obj])
-            self.update_inst_map(schedules)
+
+            # Find schedules that may call the schedule we want to update.
+            schedules.update(used_in_calls(sched_name, list(self._schedules.values())))
+
+            self.update_inst_map(schedules, qubits=qubits)
 
     def _get_channel_index(self, qubits: Tuple[int, ...], chan: PulseChannel) -> int:
         """Get the index of the parameterized channel.
@@ -816,7 +887,7 @@ def _get_channel_index(self, qubits: Tuple[int, ...], chan: PulseChannel) -> int
 
                 indices = [int(sub_channel) for sub_channel in qubit_channels.split(".")]
                 ch_qubits = tuple(qubits[index] for index in indices)
-                chs_ = self._control_channel_map[ch_qubits]
+                chs_ = self._control_channel_map.get(ch_qubits, [])
 
                 control_index = 0
                 if len(channel_index_parts) == 2:
@@ -910,6 +981,7 @@ def get_parameter_value(
         candidates = [val for val in candidates if val.group == group]
 
         if cutoff_date:
+            cutoff_date = cutoff_date.astimezone()
             candidates = [val for val in candidates if val.date_time <= cutoff_date]
 
         if len(candidates) == 0:

qiskit_experiments/calibration_management/parameter_value.py

@@ -69,6 +69,8 @@ def __post_init__(self):
                     f"Cannot parse {self.date_time} in either of {formats} formats."
                 )
 
+        self.date_time = self.date_time.astimezone()
+
         if not isinstance(self.value, (int, float, complex)):
             raise CalibrationError(f"Values {self.value} must be int, float or complex.")
 

test/calibration/test_base_calibration_experiment.py

@@ -32,7 +32,7 @@ class CorrectOrder(BaseCalibrationExperiment, QubitSpectroscopy):
 
             def __init__(self):
                 """A dummy class for parent order testing."""
-                super().__init__(Calibrations(), 0, [0, 1, 2])
+                super().__init__(Calibrations(coupling_map=[]), 0, [0, 1, 2])
 
         CorrectOrder()
 
@@ -44,4 +44,4 @@ class WrongOrder(QubitSpectroscopy, BaseCalibrationExperiment):
 
                 def __init__(self):
                     """A dummy class for parent order testing."""
-                    super().__init__(Calibrations(), 0, [0, 1, 2])
+                    super().__init__(Calibrations(coupling_map=[]), 0, [0, 1, 2])

test/calibration/test_calibration_utils.py

@@ -0,0 +1,65 @@
+# 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 utility functions for calibrations."""
+
+from test.base import QiskitExperimentsTestCase
+import qiskit.pulse as pulse
+from qiskit_experiments.calibration_management.calibration_utils import used_in_calls
+
+
+class TestCalibrationUtils(QiskitExperimentsTestCase):
+    """Test the function in CalUtils."""
+
+    def test_used_in_calls(self):
+        """Test that we can identify schedules by name when calls are present."""
+
+        with pulse.build(name="xp") as xp:
+            pulse.play(pulse.Gaussian(160, 0.5, 40), pulse.DriveChannel(1))
+
+        with pulse.build(name="xp2") as xp2:
+            pulse.play(pulse.Gaussian(160, 0.5, 40), pulse.DriveChannel(1))
+
+        with pulse.build(name="call_xp") as xp_call:
+            pulse.call(xp)
+
+        with pulse.build(name="call_call_xp") as xp_call_call:
+            pulse.play(pulse.Drag(160, 0.5, 40, 0.2), pulse.DriveChannel(1))
+            pulse.call(xp_call)
+
+        self.assertSetEqual(used_in_calls("xp", [xp_call]), {"call_xp"})
+        self.assertSetEqual(used_in_calls("xp", [xp2]), set())
+        self.assertSetEqual(
+            used_in_calls("xp", [xp_call, xp_call_call]), {"call_xp", "call_call_xp"}
+        )
+
+        with pulse.build(name="xp") as xp:
+            pulse.play(pulse.Gaussian(160, 0.5, 40), pulse.DriveChannel(2))
+
+        cr_tone_p = pulse.GaussianSquare(640, 0.2, 64, 500)
+        rotary_p = pulse.GaussianSquare(640, 0.1, 64, 500)
+
+        cr_tone_m = pulse.GaussianSquare(640, -0.2, 64, 500)
+        rotary_m = pulse.GaussianSquare(640, -0.1, 64, 500)
+
+        with pulse.build(name="cr") as cr:
+            with pulse.align_sequential():
+                with pulse.align_left():
+                    pulse.play(rotary_p, pulse.DriveChannel(3))  # Rotary tone
+                    pulse.play(cr_tone_p, pulse.ControlChannel(2))  # CR tone.
+                pulse.call(xp)
+                with pulse.align_left():
+                    pulse.play(rotary_m, pulse.DriveChannel(3))
+                    pulse.play(cr_tone_m, pulse.ControlChannel(2))
+                pulse.call(xp)
+
+        self.assertSetEqual(used_in_calls("xp", [cr]), {"cr"})