diff --git a/qiskit_experiments/calibration_management/calibrations.py b/qiskit_experiments/calibration_management/calibrations.py index 5e68af388e..7ace1fba3b 100644 --- a/qiskit_experiments/calibration_management/calibrations.py +++ b/qiskit_experiments/calibration_management/calibrations.py @@ -12,14 +12,14 @@ """Class to store and manage the results of calibration experiments.""" +import warnings import os -from collections import defaultdict +from collections import defaultdict, Counter from datetime import datetime, timezone from typing import Any, Dict, Set, Tuple, Union, List, Optional import csv import dataclasses import json -import warnings import rustworkx as rx from qiskit.pulse import ( @@ -35,6 +35,7 @@ from qiskit.pulse.channels import PulseChannel from qiskit.circuit import Parameter, ParameterExpression from qiskit.providers.backend import Backend +from qiskit.utils.deprecation import deprecate_func, deprecate_arg from qiskit_experiments.exceptions import CalibrationError from qiskit_experiments.calibration_management.basis_gate_library import BasisGateLibrary @@ -51,7 +52,7 @@ ParameterValueType, ScheduleKey, ) -from qiskit_experiments.framework import BackendData +from qiskit_experiments.framework import BackendData, ExperimentEncoder, ExperimentDecoder class Calibrations: @@ -1352,9 +1353,16 @@ def _append_to_list( value_dict["date_time"] = value_dict["date_time"].strftime("%Y-%m-%d %H:%M:%S.%f%z") data.append(value_dict) + @deprecate_arg( + name="file_type", + since="0.6", + additional_msg="Full calibration saving is now supported in json format. csv is deprecated.", + package_name="qiskit-experiments", + predicate=lambda file_type: file_type == "csv", + ) def save( self, - file_type: str = "csv", + file_type: str = "json", folder: str = None, overwrite: bool = False, file_prefix: str = "", @@ -1362,20 +1370,14 @@ def save( ): """Save the parameterized schedules and parameter value. - The schedules and parameter values can be stored in csv files. This method creates - three files: + .. note:: - * 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. + Full round-trip serialization of a :class:`.Calibrations` instance + is only supported in JSON format. + This may be extended to other file formats in future version. Args: - file_type: The type of file to which to save. By default this is a csv. + file_type: The type of file to which to save. By default, this is a json. 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 @@ -1388,43 +1390,56 @@ def save( 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) - parameter_config_file = file_prefix + "parameter_config.csv" - parameter_value_file = file_prefix + "parameter_values.csv" - schedule_file = file_prefix + "schedules.csv" + if file_type == "json": + from .save_utils import calibrations_to_dict - if os.path.isfile(parameter_config_file) and not overwrite: - raise CalibrationError( - f"{parameter_config_file} already exists. Set overwrite to True." - ) + file_path = file_prefix + ".json" + if os.path.isfile(file_path) and not overwrite: + raise CalibrationError(f"{file_path} already exists. Set overwrite to True.") - if os.path.isfile(parameter_value_file) and not overwrite: - raise CalibrationError(f"{parameter_value_file} already exists. Set overwrite to True.") + canonical_data = calibrations_to_dict(self, most_recent_only=most_recent_only) + with open(file_path, "w", encoding="utf-8") as file: + json.dump(canonical_data, file, cls=ExperimentEncoder) - if os.path.isfile(schedule_file) and not overwrite: - raise CalibrationError(f"{schedule_file} already exists. Set overwrite to True.") + elif file_type == "csv": + warnings.warn("Schedules are only saved in text format. They cannot be re-loaded.") - # Write the parameter configuration. - header_keys = ["parameter.name", "parameter unique id", "schedule", "qubits"] - body = [] + parameter_config_file = file_prefix + "parameter_config.csv" + parameter_value_file = file_prefix + "parameter_values.csv" + schedule_file = file_prefix + "schedules.csv" - 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 os.path.isfile(parameter_config_file) and not overwrite: + raise CalibrationError( + f"{parameter_config_file} already exists. Set overwrite to True." ) - if file_type == "csv": + if os.path.isfile(parameter_value_file) and not overwrite: + raise CalibrationError( + f"{parameter_value_file} already exists. Set overwrite to True." + ) + + if os.path.isfile(schedule_file) and not overwrite: + raise CalibrationError(f"{schedule_file} 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, + } + ) + with open(parameter_config_file, "w", newline="", encoding="utf-8") as output_file: dict_writer = csv.DictWriter(output_file, header_keys) dict_writer.writeheader() @@ -1453,6 +1468,14 @@ def save( os.chdir(cwd) + @deprecate_func( + since="0.6", + additional_msg=( + "Saving calibration in csv format is deprecate " + "as well as functions that support this functionality." + ), + package_name="qiskit-experiments", + ) def schedule_information(self) -> Tuple[List[str], List[Dict]]: """Get the information on the schedules stored in the calibrations. @@ -1471,6 +1494,11 @@ def schedule_information(self) -> Tuple[List[str], List[Dict]]: return ["name", "qubits", "schedule"], schedules + @deprecate_func( + since="0.6", + additional_msg="Loading and saving calibrations in CSV format is deprecated.", + package_name="qiskit-experiments", + ) def load_parameter_values(self, file_name: str = "parameter_values.csv"): """ Load parameter values from a given file into self._params. @@ -1514,6 +1542,8 @@ def _add_parameter_value_from_conf( parameter: The name of the parameter. qubits: The qubits on which the parameter acts. """ + # TODO remove this after load_parameter_values method is removed. + param_val = ParameterValue(value, date_time, valid, exp_id, group) if schedule == "": @@ -1525,12 +1555,32 @@ def _add_parameter_value_from_conf( self.add_parameter_value(param_val, *key, update_inst_map=False) @classmethod - def load(cls, files: List[str]) -> "Calibrations": + @deprecate_arg( + name="files", + new_alias="file_path", + since="0.6", + package_name="qiskit-experiments", + ) + def load(cls, file_path: str) -> "Calibrations": """ Retrieves the parameterized schedules and pulse parameters from the given location. + + Args: + file_path: Path to file location. + + Returns: + Calibration instance restored from the file. """ - raise CalibrationError("Full calibration loading is not implemented yet.") + from .save_utils import calibrations_from_dict + + with open(file_path, "r", encoding="utf-8") as file: + # Do we really need branching for data types? + # Parsing data format and dispatching the loader seems an overkill, + # but save method intend to support multiple formats. + cal_data = json.load(file, cls=ExperimentDecoder) + + return calibrations_from_dict(cal_data) @staticmethod def _to_tuple(qubits: Union[str, int, Tuple[int, ...]]) -> Tuple[int, ...]: @@ -1595,14 +1645,22 @@ def __eq__(self, other: "Calibrations") -> bool: if self._schedules.keys() != other._schedules.keys(): return False - def _hash(data: dict): - return hash(json.dumps(data)) - - sorted_params_a = sorted(self.parameters_table()["data"], key=_hash) - sorted_params_b = sorted(other.parameters_table()["data"], key=_hash) - - return sorted_params_a == sorted_params_b - + def _counting(table): + return Counter(map(lambda d: tuple(d.items()), table["data"])) + + # Use counting sort algorithm to compare unordered sequences + # https://en.wikipedia.org/wiki/Counting_sort + return _counting(self.parameters_table()) == _counting(other.parameters_table()) + + @deprecate_func( + since="0.6", + additional_msg=( + "Configuration data for Calibrations instance is deprecate. " + "Please use ExperimentEncoder and ExperimentDecoder to " + "serialize and deserialize this instance with JSON format." + ), + package_name="qiskit-experiments", + ) def config(self) -> Dict[str, Any]: """Return the settings used to initialize the calibrations. @@ -1635,23 +1693,33 @@ def config(self) -> Dict[str, Any]: } @classmethod + @deprecate_func( + since="0.6", + additional_msg="This method will be removed and no alternative will be provided.", + package_name="qiskit-experiments", + ) def from_config(cls, config: Dict) -> "Calibrations": - """Deserialize the calibrations given the input dictionary""" - - config["kwargs"]["control_channel_map"] = config["kwargs"]["control_channel_map"].chan_map + """Restore Calibration from config data. - calibrations = cls(**config["kwargs"]) + Args: + config: Configuration data. - for param_config in config["parameters"]: - calibrations._add_parameter_value_from_conf(**param_config) + Returns: + Calibration instance restored from configuration data. + """ + from .save_utils import calibrations_from_dict - return calibrations + return calibrations_from_dict(config) def __json_encode__(self): """Convert to format that can be JSON serialized.""" - return self.config() + from .save_utils import calibrations_to_dict + + return calibrations_to_dict(self, most_recent_only=False) @classmethod def __json_decode__(cls, value: Dict[str, Any]) -> "Calibrations": """Load from JSON compatible format.""" - return cls.from_config(value) + from .save_utils import calibrations_from_dict + + return calibrations_from_dict(value) diff --git a/qiskit_experiments/calibration_management/save_utils.py b/qiskit_experiments/calibration_management/save_utils.py new file mode 100644 index 0000000000..716c50119c --- /dev/null +++ b/qiskit_experiments/calibration_management/save_utils.py @@ -0,0 +1,262 @@ +# 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. + +"""A helper module to save calibration data to local storage. + +.. warning:: + + This module is expected to be internal and is not intended as a stable user-facing API. + +.. note:: + + Because a locally saved :class:`.Calibrations` instance may not conform to the + data model of the latest Qiskit Experiments implementation, the calibration loader + must be aware of the data model version. + CalibrationModel classes representing the data model must have + the version suffix, e.g., `CalibrationModelV1` and the `schema_version` field. + This helps the loader to raise user-friendly error rather than being crashed by + incompatible data, and possibly to dispatch the loader function based on the version number. + + When a developer refactors the :class:`.Calibrations` class to a new data model, + the developer must also define a corresponding CalibrationModel class with new version number. + Existing CalibrationModel classes should be preserved for backward compatibility. + + +.. note:: + + We don't guarantee the portability of stored data across different Qiskit Experiments + versions. We allow the calibration loader to raise an error for non-supported + data models. + +""" + +from dataclasses import dataclass, field, asdict +from datetime import datetime +from typing import List, Dict, Any + +from qiskit.pulse import ScheduleBlock + +from .calibrations import Calibrations +from .control_channel_map import ControlChannelMap +from .parameter_value import ParameterValue + + +@dataclass(frozen=True) +class ParameterModelV1: + """A data schema of a single calibrated parameter. + + .. note:: + This is intentionally agnostic to the data structure of + Qiskit Experiments Calibrations for portability. + + """ + + param_name: str + """Name of the parameter.""" + + qubits: List[int] + """List of associated qubits.""" + + schedule: str = "" + """Associated schedule name.""" + + value: float = 0.0 + """Parameter value.""" + + datetime: datetime = None + """A date time at which this value is obtained.""" + + valid: bool = True + """If this parameter is valid.""" + + exp_id: str = "" + """Associated experiment ID which is used to obtain this value.""" + + group: str = "" + """Name of calibration group in which this calibration parameter belongs to.""" + + @classmethod + def apply_schema(cls, data: Dict[str, Any]): + """Consume dictionary and returns canonical data model.""" + return ParameterModelV1(**data) + + +@dataclass +class CalibrationModelV1: + """A data schema to represent instances of Calibrations. + + .. note:: + This is intentionally agnostic to the data structure of + Qiskit Experiments Calibrations for portability. + + """ + + backend_name: str + """Name of the backend.""" + + backend_version: str + """Version of the backend.""" + + device_coupling_graph: List[List[int]] + """Qubit coupling graph of the device.""" + + control_channel_map: ControlChannelMap + """Mapping of ControlChannel to qubit index.""" + + schedules: List[ScheduleBlock] = field(default_factory=list) + """Template schedules. It must contain the metadata for qubits and num_qubits.""" + + parameters: List[ParameterModelV1] = field(default_factory=list) + """List of calibrated pulse parameters.""" + + schema_version: str = "1.0" + """Version of this data model. This must be static.""" + + @classmethod + def apply_schema(cls, data: Dict[str, Any]): + """Consume dictionary and returns canonical data model.""" + in_data = {} + for key, value in data.items(): + if key == "parameters": + value = list(map(ParameterModelV1.apply_schema, value)) + in_data[key] = value + return CalibrationModelV1(**in_data) + + +def calibrations_to_dict( + cals: Calibrations, + most_recent_only: bool = True, +) -> Dict[str, Any]: + """A helper function to convert calibration data into dictionary. + + Args: + cals: A calibration instance to save. + most_recent_only: Set True to save calibration parameters with most recent time stamps. + + Returns: + Canonical calibration data in dictionary format. + """ + schedules = getattr(cals, "_schedules") + num_qubits = getattr(cals, "_schedules_qubits") + parameters = getattr(cals, "_params") + if most_recent_only: + # Get values with most recent time stamps. + parameters = {k: [max(parameters[k], key=lambda x: x.date_time)] for k in parameters} + + data_entries = [] + for param_key, param_values in parameters.items(): + for param_value in param_values: + entry = ParameterModelV1( + param_name=param_key.parameter, + qubits=param_key.qubits, + schedule=param_key.schedule, + value=param_value.value, + datetime=param_value.date_time, + valid=param_value.valid, + exp_id=param_value.exp_id, + group=param_value.group, + ) + data_entries.append(entry) + + sched_entries = [] + for sched_key, sched_obj in schedules.items(): + if "qubits" not in sched_obj.metadata or "num_qubits" not in sched_obj.metadata: + qubit_metadata = { + "qubits": sched_key.qubits, + "num_qubits": num_qubits[sched_key], + } + sched_obj.metadata.update(qubit_metadata) + sched_entries.append(sched_obj) + + model = CalibrationModelV1( + backend_name=cals.backend_name, + backend_version=cals.backend_version, + device_coupling_graph=getattr(cals, "_coupling_map"), + control_channel_map=ControlChannelMap(getattr(cals, "_control_channel_map")), + schedules=sched_entries, + parameters=data_entries, + ) + + return asdict(model) + + +def calibrations_from_dict( + cal_data: Dict[str, Any], +) -> Calibrations: + """A helper function to build calibration instance from canonical dictionary. + + Args: + cal_data: Calibration data dictionary which is formatted according to the + predefined data schema provided by Qiskit Experiments. + This formatting is implicitly performed when the calibration data is + dumped into dictionary with the :func:`calibrations_to_dict` function. + + Returns: + Calibration instance. + + Raises: + ValueError: When input data model version is not supported. + KeyError: When input data model doesn't conform to data schema. + """ + # Apply schema for data field validation + try: + version = cal_data["schema_version"] + if version == "1.0": + model = CalibrationModelV1.apply_schema(cal_data) + else: + raise ValueError( + f"Loading calibration data with schema version {version} is no longer supported. " + "Use the same version of Qiskit Experiments at the time of saving." + ) + except (KeyError, TypeError) as ex: + raise KeyError( + "Loaded data doesn't match with the defined data schema. " + "Check if this object is dumped from the Calibrations instance." + ) from ex + + # This can dispatch loading mechanism depending on schema version + cals = Calibrations( + coupling_map=model.device_coupling_graph, + control_channel_map=model.control_channel_map.chan_map, + backend_name=model.backend_name, + backend_version=model.backend_version, + ) + + # Add schedules + for sched in model.schedules: + qubits = sched.metadata.pop("qubits", tuple()) + num_qubits = sched.metadata.pop("num_qubits", None) + cals.add_schedule( + schedule=sched, + qubits=qubits if qubits and len(qubits) != 0 else None, + num_qubits=num_qubits, + ) + + # Add parameters + for param in model.parameters: + param_value = ParameterValue( + value=param.value, + date_time=param.datetime, + valid=param.valid, + exp_id=param.exp_id, + group=param.group, + ) + cals.add_parameter_value( + value=param_value, + param=param.param_name, + qubits=tuple(param.qubits), + schedule=param.schedule, + update_inst_map=False, + ) + cals.update_inst_map() + + return cals diff --git a/qiskit_experiments/framework/json.py b/qiskit_experiments/framework/json.py index 294cc14710..9308ea2bc5 100644 --- a/qiskit_experiments/framework/json.py +++ b/qiskit_experiments/framework/json.py @@ -23,6 +23,7 @@ import traceback import warnings import zlib +from datetime import datetime from functools import lru_cache from types import FunctionType, MethodType from typing import Any, Dict, Type, Optional, Union, Callable @@ -34,6 +35,7 @@ from qiskit import qpy from qiskit.circuit import ParameterExpression, QuantumCircuit, Instruction from qiskit.circuit.library import BlueprintCircuit +from qiskit.pulse import ScheduleBlock from qiskit.quantum_info import DensityMatrix from qiskit.quantum_info.operators.channel.quantum_channel import QuantumChannel from qiskit.result import LocalReadoutMitigator, CorrelatedReadoutMitigator @@ -460,6 +462,8 @@ def default(self, obj: Any) -> Any: # pylint: disable=arguments-renamed return {"__type__": "spmatrix", "__value__": value} if isinstance(obj, bytes): return _serialize_bytes(obj) + if isinstance(obj, datetime): + return {"__type__": "datetime", "__value__": obj.isoformat()} if isinstance(obj, np.number): return obj.item() if dataclasses.is_dataclass(obj): @@ -511,6 +515,11 @@ def default(self, obj: Any) -> Any: # pylint: disable=arguments-renamed data=obj, serializer=lambda buff, data: qpy.dump(data, buff) ) return {"__type__": "QuantumCircuit", "__value__": value} + if isinstance(obj, ScheduleBlock): + value = _serialize_and_encode( + data=obj, serializer=lambda buff, data: qpy.dump(data, buff) + ) + return {"__type__": "ScheduleBlock", "__value__": value} if isinstance(obj, ParameterExpression): value = _serialize_and_encode( data=obj, @@ -581,6 +590,8 @@ def object_hook(self, obj): return _deserialize_bytes(obj_val) if obj_type == "set": return set(obj_val) + if obj_type == "datetime": + return datetime.fromisoformat(obj_val) if obj_type == "LMFIT.Model": tmp = lmfit.Model(func=None) load_obj = tmp.loads(s=obj_val) @@ -590,6 +601,8 @@ def object_hook(self, obj): return circuit.data[0][0] if obj_type == "QuantumCircuit": return _decode_and_deserialize(obj_val, qpy.load, name=obj_type)[0] + if obj_type == "ScheduleBlock": + return _decode_and_deserialize(obj_val, qpy.load, name=obj_type)[0] if obj_type == "ParameterExpression": return _decode_and_deserialize( obj_val, qpy._read_parameter_expression, name=obj_type diff --git a/releasenotes/notes/feature-support-calibrations-roundtrip-47f09bd9ff803479.yaml b/releasenotes/notes/feature-support-calibrations-roundtrip-47f09bd9ff803479.yaml new file mode 100644 index 0000000000..99a90f22cf --- /dev/null +++ b/releasenotes/notes/feature-support-calibrations-roundtrip-47f09bd9ff803479.yaml @@ -0,0 +1,32 @@ +--- +features: + - | + JSON data format is supoorted for saving :class:`.Calibrations` instance. + This leverages a custom JSON encoder and decoder to serialize + the entire calibration data including user provided schedule templates. + Output JSON data is formatted into the standard data model which is intentionally + agnostic to the calibration data structure, and this allows community + developer to reuse the calibration data in their platform. + See :mod:`qiskit_experiments.calibration_management.save_utils` for data models. +deprecations: + - | + Saving :class:`.Calibrations` instance into CSV file was deprecated. + This only provides serialization for limited set of calibraiton data, + and loading from the local file is not supported. + - | + :meth:`.Calibrations.schedule_information` was deprecated. + This method returns attached calibration templates in the string format, + but this cannot be converted back to the original Qiskit representation. + Now better serialization is provided with :meth:`Calibrations.save` with JSON mode + and it internally dumps these schedule in through QPY format. + - | + :meth:`.Calibrations.load_parameter_values` was deprecated. + Since saving :.Calibrations: instance into the CSV format was deprecated, + the required data file to invoke this method will be no longer generated + in future calibrations instance. Full calibration instance roundtrip + is now supported with the save and load method. + - | + :meth:`.Calibrations.config` and :meth:`.Calibrations.from_config` were deprecated. + Now canonical data representation is generated for calibration by the + newly introduced :mod:`~qiskit_experiments.calibration_management.save_utils` module, + and the legacy configuration dictionary is no longer used for JSON encoding. diff --git a/test/calibration/test_calibrations.py b/test/calibration/test_calibrations.py index d77b2f47a9..af70681421 100644 --- a/test/calibration/test_calibrations.py +++ b/test/calibration/test_calibrations.py @@ -121,6 +121,18 @@ def setUp(self): 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_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_setup(self): """Test that the initial setup behaves as expected.""" expected = {ParameterKey("amp", (), "xp"), ParameterKey("amp", (), "xm")} @@ -312,7 +324,8 @@ def test_from_backend(self): the data is passed correctly""" backend = FakeBelemV2() cals = Calibrations.from_backend(backend) - config_args = cals.config()["kwargs"] + with self.assertWarns(DeprecationWarning): + config_args = cals.config()["kwargs"] control_channel_map_size = len(config_args["control_channel_map"].chan_map) coupling_map_size = len(config_args["coupling_map"]) self.assertEqual(control_channel_map_size, 8) @@ -350,6 +363,65 @@ def test_from_minimal_backend(self, num_qubits, gate_name, pass_properties): backend.target.add_instruction(gate, properties=properties) Calibrations.from_backend(backend) + def test_equality(self): + """Test the equal method on calibrations.""" + backend = FakeBelemV2() + library = FixedFrequencyTransmon(basis_gates=["sx", "x"]) + + cals1 = Calibrations.from_backend( + backend, libraries=[library], add_parameter_defaults=False + ) + cals2 = Calibrations.from_backend( + backend, libraries=[library], add_parameter_defaults=False + ) + self.assertTrue(cals1 == cals2) + + date_time = datetime.now(timezone.utc).astimezone() + param_val = ParameterValue(0.12345, date_time=date_time) + cals1.add_parameter_value(param_val, "amp", 3, "x") + + # The two objects are different due to missing parameter value + self.assertFalse(cals1 == cals2) + + # The two objects are different due to time stamps + param_val2 = ParameterValue(0.12345, date_time=date_time - timedelta(seconds=1)) + cals2.add_parameter_value(param_val2, "amp", 3, "x") + self.assertFalse(cals1 == cals2) + + # The two objects are different due to missing parameter value + cals3 = Calibrations.from_backend( + backend, libraries=[library], add_parameter_defaults=False + ) + self.assertFalse(cals1 == cals3) + + # The two objects are identical due to time stamps + cals2.add_parameter_value(param_val, "amp", 3, "x") + self.assertFalse(cals1 == cals3) + + # The schedules contained in the cals are different. + library2 = FixedFrequencyTransmon(basis_gates=["sx", "x", "y"]) + cals1 = Calibrations.from_backend(backend, libraries=[library]) + cals2 = Calibrations.from_backend(backend, libraries=[library2]) + self.assertFalse(cals1 == cals2) + + # Ensure that the equality is not sensitive to parameter adding order. + cals1 = Calibrations.from_backend( + backend, libraries=[library], add_parameter_defaults=False + ) + cals2 = Calibrations.from_backend( + backend, libraries=[library], add_parameter_defaults=False + ) + param_val1 = ParameterValue(0.54321, date_time=date_time) + param_val2 = ParameterValue(0.12345, date_time=date_time - timedelta(seconds=1)) + + cals1.add_parameter_value(param_val2, "amp", 3, "x") + cals1.add_parameter_value(param_val1, "amp", 3, "x") + + cals2.add_parameter_value(param_val1, "amp", 3, "x") + cals2.add_parameter_value(param_val2, "amp", 3, "x") + + self.assertTrue(cals1 == cals2) + class TestOverrideDefaults(QiskitExperimentsTestCase): """ @@ -384,6 +456,18 @@ def setUp(self): self.cals.add_schedule(xp, num_qubits=1) self.cals.add_schedule(xp_drag, (3,)) + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_parameter_value_adding_and_filtering(self): """Test that adding parameter values behaves in the expected way.""" @@ -643,6 +727,18 @@ def setUp(self): self.cals.add_parameter_value(40, self.sigma_xp, schedule="xp") self.cals.add_parameter_value(160, self.duration_xp, schedule="xp") + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_meas_schedule(self): """Test that we get a properly assigned measure schedule without drive channels.""" sched = self.cals.get_schedule("meas", (0,)) @@ -771,6 +867,18 @@ def setUp(self): 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_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_call_registration(self): """Check that by registering the call we registered three schedules.""" @@ -923,6 +1031,18 @@ class TestControlChannels(CrossResonanceTest): support parameters with the same names. """ + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_get_schedule(self): """Check that we can get a CR schedule with a built in Call.""" @@ -1054,6 +1174,18 @@ def setUp(self): self.cals.add_parameter_value(0.3, "amp", (3,), "xp") self.cals.add_parameter_value(40, "σ", (), "xp") + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_short_key(self): """Test simple value assignment""" sched = self.cals.get_schedule("xp", (2,), assign_params={"amp": 0.1}) @@ -1199,6 +1331,18 @@ def setUp(self): self.cals.add_parameter_value(160, "duration", (4,), "xp") self.cals.add_parameter_value(40, "σ", (), "xp") + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_reference_replaced(self): """Test that we get an error when there is an inconsistency in subroutines.""" @@ -1282,6 +1426,18 @@ def setUp(self): self.cals.add_parameter_value(640, "w", (3, 2), "cr_p") self.cals.add_parameter_value(800, "duration", (3, 2), "cr_p") + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_assign_coupled_explicitly(self): """Test that we get the proper schedules when they are coupled.""" @@ -1407,6 +1563,18 @@ def setUp(self): ParameterValue(0.4, self.date_time2, group="super_cal"), "amp", (0,), "xp" ) + def test_calibration_save_json(self): + """Test that the calibration under test can be serialized through JSON.""" + filename = self.__class__.__name__ + + try: + self.cals.save(file_type="json", file_prefix=filename) + loaded = self.cals.load(file_path=f"{filename}.json") + self.assertEqual(self.cals, loaded) + finally: + if os.path.exists(f"{filename}.json"): + os.remove(f"{filename}.json") + def test_parameter_table_most_recent(self): """Test the most_recent argument to the parameter_table method.""" @@ -1449,14 +1617,15 @@ def tearDown(self): """Clean-up after the test.""" super().tearDown() - for file in ["parameter_values.csv", "parameter_config.csv", "schedules.csv"]: + for file in ["parameter_values.csv", "parameter_config.csv", "schedules.csv", ".json"]: if os.path.exists(self._prefix + file): os.remove(self._prefix + file) def test_save_load_parameter_values(self): """Test that we can save and load parameter values.""" - self.cals.save("csv", overwrite=True, file_prefix=self._prefix) + with self.assertWarns(DeprecationWarning): + self.cals.save("csv", overwrite=True, file_prefix=self._prefix) self.assertEqual(self.cals.get_parameter_value("amp", (3,), "xp"), 0.1 + 0.01j) self.cals._params = defaultdict(list) @@ -1465,7 +1634,8 @@ def test_save_load_parameter_values(self): self.cals.get_parameter_value("amp", (3,), "xp") # Load the parameters, check value and type. - self.cals.load_parameter_values(self._prefix + "parameter_values.csv") + with self.assertWarns(DeprecationWarning): + self.cals.load_parameter_values(self._prefix + "parameter_values.csv") val = self.cals.get_parameter_value("amp", (3,), "xp") self.assertEqual(val, 0.1 + 0.01j) @@ -1480,10 +1650,12 @@ def test_save_load_parameter_values(self): self.assertTrue(isinstance(val, float)) # Check that we cannot rewrite files as they already exist. - with self.assertRaises(CalibrationError): - self.cals.save("csv", file_prefix=self._prefix) + with self.assertWarns(DeprecationWarning): + with self.assertRaises(CalibrationError): + self.cals.save("csv", file_prefix=self._prefix) - self.cals.save("csv", overwrite=True, file_prefix=self._prefix) + with self.assertWarns(DeprecationWarning): + self.cals.save("csv", overwrite=True, file_prefix=self._prefix) def test_alternate_date_formats(self): """Test that we can reload dates with or without time-zone.""" @@ -1492,9 +1664,11 @@ def test_alternate_date_formats(self): value = ParameterValue(0.222, date_time=new_date) self.cals.add_parameter_value(value, "amp", (3,), "xp") - self.cals.save("csv", overwrite=True, file_prefix=self._prefix) + with self.assertWarns(DeprecationWarning): + self.cals.save("csv", overwrite=True, file_prefix=self._prefix) self.cals._params = defaultdict(list) - self.cals.load_parameter_values(self._prefix + "parameter_values.csv") + with self.assertWarns(DeprecationWarning): + self.cals.load_parameter_values(self._prefix + "parameter_values.csv") def test_save_load_library(self): """Test that we can load and save a library. @@ -1509,9 +1683,11 @@ def test_save_load_library(self): cals.parameters_table() - cals.save(file_type="csv", overwrite=True, file_prefix=self._prefix) + with self.assertWarns(DeprecationWarning): + cals.save(file_type="csv", overwrite=True, file_prefix=self._prefix) - cals.load_parameter_values(self._prefix + "parameter_values.csv") + with self.assertWarns(DeprecationWarning): + cals.load_parameter_values(self._prefix + "parameter_values.csv") # Test the value of a few loaded params. self.assertEqual(cals.get_parameter_value("amp", (0,), "x"), 0.5) @@ -1520,6 +1696,21 @@ def test_save_load_library(self): BackendData(backend).drive_freqs[0], ) + def test_json_round_trip(self): + """Test round trip test for JSON file format. + + This method guarantees full equality including parameterized template schedules + and we can still generate schedules with loaded calibration instance, + even though calibrations is instantiated outside built-in library. + """ + self.cals.save(file_type="json", overwrite="True", file_prefix=self._prefix) + loaded = self.cals.load(file_path=self._prefix + ".json") + self.assertEqual(self.cals, loaded) + + original_sched = self.cals.get_schedule("cr", (3, 2)) + roundtrip_sched = loaded.get_schedule("cr", (3, 2)) + self.assertEqual(original_sched, roundtrip_sched) + class TestInstructionScheduleMap(QiskitExperimentsTestCase): """Class to test the functionality of a Calibrations""" @@ -1755,62 +1946,3 @@ def test_serialization(self): cals.add_parameter_value(0.12345, "amp", 3, "x") self.assertRoundTripSerializable(cals) - - def test_equality(self): - """Test the equal method on calibrations.""" - backend = FakeBelemV2() - library = FixedFrequencyTransmon(basis_gates=["sx", "x"]) - - cals1 = Calibrations.from_backend( - backend, libraries=[library], add_parameter_defaults=False - ) - cals2 = Calibrations.from_backend( - backend, libraries=[library], add_parameter_defaults=False - ) - self.assertTrue(cals1 == cals2) - - date_time = datetime.now(timezone.utc).astimezone() - param_val = ParameterValue(0.12345, date_time=date_time) - cals1.add_parameter_value(param_val, "amp", 3, "x") - - # The two objects are different due to missing parameter value - self.assertFalse(cals1 == cals2) - - # The two objects are different due to time stamps - param_val2 = ParameterValue(0.12345, date_time=date_time - timedelta(seconds=1)) - cals2.add_parameter_value(param_val2, "amp", 3, "x") - self.assertFalse(cals1 == cals2) - - # The two objects are different due to missing parameter value - cals3 = Calibrations.from_backend( - backend, libraries=[library], add_parameter_defaults=False - ) - self.assertFalse(cals1 == cals3) - - # The two objects are identical due to time stamps - cals2.add_parameter_value(param_val, "amp", 3, "x") - self.assertFalse(cals1 == cals3) - - # The schedules contained in the cals are different. - library2 = FixedFrequencyTransmon(basis_gates=["sx", "x", "y"]) - cals1 = Calibrations.from_backend(backend, libraries=[library]) - cals2 = Calibrations.from_backend(backend, libraries=[library2]) - self.assertFalse(cals1 == cals2) - - # Ensure that the equality is not sensitive to parameter adding order. - cals1 = Calibrations.from_backend( - backend, libraries=[library], add_parameter_defaults=False - ) - cals2 = Calibrations.from_backend( - backend, libraries=[library], add_parameter_defaults=False - ) - param_val1 = ParameterValue(0.54321, date_time=date_time) - param_val2 = ParameterValue(0.12345, date_time=date_time - timedelta(seconds=1)) - - cals1.add_parameter_value(param_val2, "amp", 3, "x") - cals1.add_parameter_value(param_val1, "amp", 3, "x") - - cals2.add_parameter_value(param_val1, "amp", 3, "x") - cals2.add_parameter_value(param_val2, "amp", 3, "x") - - self.assertTrue(cals1 == cals2)