Spectroscopy and calibrations integration V3

qiskit_experiments/calibration/backend_calibrations.py

@@ -40,19 +40,33 @@ class BackendCalibrations(Calibrations):
     any schedule.
     """
 
+    __qubit_freq_parameter__ = "qubit_lo_freq"
+    __readout_freq_parameter__ = "meas_lo_freq"
+
     def __init__(self, backend: Backend):
         """Setup an instance to manage the calibrations of a backend."""
-        super().__init__(backend.configuration().control_channels)
+        if hasattr(backend.configuration(), "control_channels"):
+            control_channels = backend.configuration().control_channels
+        else:
+            control_channels = None
+
+        super().__init__(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.qubit_freq = Parameter(self.__qubit_freq_parameter__)
+        self.meas_freq = Parameter(self.__readout_freq_parameter__)
         self._register_parameter(self.qubit_freq, ())
         self._register_parameter(self.meas_freq, ())
 
         self._qubits = set(range(backend.configuration().n_qubits))
         self._backend = backend
 
+        for qubit, freq in enumerate(backend.defaults().qubit_freq_est):
+            self.add_parameter_value(freq, self.qubit_freq, qubit)
+
+        for meas, freq in enumerate(backend.defaults().meas_freq_est):
+            self.add_parameter_value(freq, self.meas_freq, meas)
+
     def _get_frequencies(
         self,
         element: FrequencyElement,
@@ -70,8 +84,9 @@ def _get_frequencies(
 
         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)
+            schedule = None  # A qubit frequency is not attached to a schedule.
+            if ParameterKey(param, (qubit,), schedule) in self._params:
+                freq = self.get_parameter_value(param, (qubit,), schedule, True, group, cutoff_date)
             else:
                 if element == FrequencyElement.READOUT:
                     freq = self._backend.defaults().meas_freq_est[qubit]

qiskit_experiments/calibration/calibration_key_types.py

@@ -0,0 +1,23 @@
+# 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.
+
+"""Types used by the calibration module."""
+
+from typing import Union
+from collections import namedtuple
+
+from qiskit.circuit import ParameterExpression
+
+
+ParameterKey = namedtuple("ParameterKey", ["parameter", "qubits", "schedule"])
+ScheduleKey = namedtuple("ScheduleKey", ["schedule", "qubits"])
+ParameterValueType = Union[ParameterExpression, float, int, complex]

qiskit_experiments/calibration/calibrations.py

@@ -13,7 +13,7 @@
 """Class to store and manage the results of calibration experiments."""
 
 import os
-from collections import namedtuple, defaultdict
+from collections import defaultdict
 from datetime import datetime
 from typing import Any, Dict, Set, Tuple, Union, List, Optional
 import csv
@@ -37,10 +37,12 @@
 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]
+from qiskit_experiments.experiment_data import ExperimentData
+from qiskit_experiments.calibration.calibration_key_types import (
+    ParameterKey,
+    ParameterValueType,
+    ScheduleKey,
+)
 
 
 class Calibrations:
@@ -897,6 +899,74 @@ def parameters_table(
 
         return data
 
+    def update(
+        self,
+        exp_data: ExperimentData,
+        result_index: int = -1,
+        force_update: bool = False,
+        group: str = "default",
+    ):
+        """Update the calibrations form a result in the given experiment data.
+
+        This function allows users to update their calibrations from experiment data. Typically,
+        the value of the parameter to update is directly stored as the result of the fit.
+        Importantly, this method requires the following keys to be present in the analysis result
+        - cal_parameter: The name of the calibration parameter to update.
+        - qubits: The qubits to which the parameter belongs.
+        - cal_schedule: The name of the schedule which is updated, this can be None (e.g. for
+            qubit frequencies).
+        - cal_value: The value of the parameter which will enter the calibrations.
+        The ParameterKey formed by the values under (calibration_parameter, qubits,
+        calibration_schedule) must therefore be valid.
+
+        Args:
+            exp_data: An analysis result which contains either the value to update under the
+                key value or the information required by the calibration_extraction function
+                which will build the values to update.
+            result_index: The index of the result which defaults to 0.
+            force_update: If set to True then the calibrations will be updated even if the
+                quality of the result is "computer_bad".
+            group: The calibration group from which to draw the parameters.
+                If not specified this defaults to the 'default' group.
+
+        Raises:
+            CalibrationError: If the result does not have the required calibration keys. These
+                keys are "cal_parameter", "qubits", "cal_schedule", and "cal_value".
+        """
+        result = exp_data.analysis_result(result_index)
+
+        if "quality" in result and result["quality"] == "computer_bad" and not force_update:
+            return
+
+        required_keys = ["cal_parameter", "cal_schedule", "cal_value"]
+        if not all(key in result for key in required_keys):
+            raise CalibrationError(
+                f"Cannot update calibrations from result. One of {required_keys} is missing."
+            )
+
+        timestamp = datetime.now()
+        all_times = exp_data.completion_times.values()
+        if all_times:
+            timestamp = max(all_times)
+
+        value = ParameterValue(
+            value=result["cal_value"],
+            date_time=timestamp,
+            group=group,
+            exp_id=exp_data.experiment_id,
+        )
+
+        schedule = result["cal_schedule"]
+        param = result["cal_parameter"]
+
+        # TODO update this with experiment metadata PR #67
+        try:
+            qubits = exp_data.data(0)["metadata"]["qubits"]
+        except KeyError as error:
+            raise CalibrationError("Cannot find qubit information in metadata.") from error
+
+        self.add_parameter_value(value, param, qubits, schedule)
+
     def save(self, file_type: str = "csv", folder: str = None, overwrite: bool = False):
         """Save the parameterized schedules and parameter value.
 
@@ -1025,7 +1095,7 @@ def _to_tuple(qubits: Union[str, int, Tuple[int, ...]]) -> Tuple[int, ...]:
             CalibrationError: If the given input does not conform to an int or
                 tuple of ints.
         """
-        if not qubits:
+        if qubits is None:
             return tuple()
 
         if isinstance(qubits, str):

qiskit_experiments/characterization/qubit_spectroscopy.py

@@ -33,6 +33,7 @@
 )
 from qiskit_experiments.base_experiment import BaseExperiment
 from qiskit_experiments.data_processing.processor_library import get_to_signal_processor
+from qiskit_experiments.calibration.backend_calibrations import BackendCalibrations
 
 
 class SpectroscopyAnalysis(CurveAnalysis):
@@ -169,6 +170,11 @@ def _post_processing(self, analysis_result: CurveAnalysisResult) -> CurveAnalysi
         else:
             analysis_result["quality"] = "computer_bad"
 
+        # Add calibration information
+        analysis_result["cal_parameter"] = BackendCalibrations.__qubit_freq_parameter__
+        analysis_result["cal_schedule"] = None
+        analysis_result["cal_value"] = fit_freq
+
         return analysis_result
 
 
@@ -315,7 +321,7 @@ def circuits(self, backend: Optional[Backend] = None):
             assigned_circ = circuit.assign_parameters({freq_param: freq}, inplace=False)
             assigned_circ.metadata = {
                 "experiment_type": self._type,
-                "qubit": self.physical_qubits[0],
+                "qubits": (self.physical_qubits[0],),
                 "xval": freq,
                 "unit": "Hz",
                 "amplitude": self.experiment_options.amp,

qiskit_experiments/experiment_data.py

@@ -17,6 +17,7 @@
 import os
 import uuid
 from collections import OrderedDict
+from datetime import datetime
 
 from qiskit.result import Result
 from qiskit.providers import Backend
@@ -108,6 +109,16 @@ def job_ids(self) -> List[str]:
         """
         return list(self._jobs.keys())
 
+    @property
+    def completion_times(self) -> Dict[str, datetime]:
+        """Returns the completion times of the jobs."""
+        job_times = {}
+        for job_id, job in self._jobs.items():
+            if job is not None and "COMPLETED" in job.time_per_step():
+                job_times[job_id] = job.time_per_step().get("COMPLETED")
+
+        return job_times
+
     @property
     def backend(self) -> Backend:
         """Return backend.

qiskit_experiments/test/mock_iq_backend.py

@@ -13,6 +13,7 @@
 """An mock IQ backend for testing."""
 
 from typing import Dict, List, Tuple
+from datetime import datetime
 import numpy as np
 
 from qiskit.providers.backend import BackendV1 as Backend
@@ -33,6 +34,11 @@ def result(self) -> Result:
         """Return a result."""
         return Result.from_dict(self._result)
 
+    @staticmethod
+    def time_per_step() -> Dict[str, datetime]:
+        """Return the completion time."""
+        return {"COMPLETED": datetime.now()}
+
     def submit(self):
         pass
 

test/test_qubit_spectroscopy.py

@@ -17,10 +17,12 @@
 import numpy as np
 from qiskit.qobj.utils import MeasLevel
 from qiskit.test import QiskitTestCase
+from qiskit.test.mock import FakeAthens
 
 from qiskit_experiments.characterization.qubit_spectroscopy import QubitSpectroscopy
 from qiskit_experiments.test.mock_iq_backend import TestJob, IQTestBackend
 from qiskit_experiments.analysis import get_opt_value
+from qiskit_experiments.calibration.backend_calibrations import BackendCalibrations
 
 
 class SpectroscopyBackend(IQTestBackend):
@@ -94,6 +96,12 @@ def run(
 class TestQubitSpectroscopy(QiskitTestCase):
     """Test spectroscopy experiment."""
 
+    def setUp(self):
+        """Setup the test."""
+        super().setUp()
+
+        self._cals = BackendCalibrations(FakeAthens())
+
     def test_spectroscopy_end2end_classified(self):
         """End to end test of the spectroscopy experiment."""
 
@@ -114,14 +122,20 @@ def test_spectroscopy_end2end_classified(self):
 
         spec = QubitSpectroscopy(3, np.linspace(-10.0, 10.0, 21), unit="MHz")
         spec.set_run_options(meas_level=MeasLevel.CLASSIFIED)
-        result = spec.run(backend).analysis_result(0)
+        exp_data = spec.run(backend)
+        result = exp_data.analysis_result(0)
 
         value = get_opt_value(result, "freq")
 
         self.assertTrue(value < 5.1e6)
         self.assertTrue(value > 4.9e6)
         self.assertEqual(result["quality"], "computer_good")
 
+        # Test the integration with the BackendCalibrations
+        self.assertNotEqual(self._cals.get_qubit_frequencies()[3], result["cal_value"])
+        self._cals.update(exp_data)
+        self.assertEqual(self._cals.get_qubit_frequencies()[3], result["cal_value"])
+
     def test_spectroscopy_end2end_kerneled(self):
         """End to end test of the spectroscopy experiment on IQ data."""